Codex
d3f3151218
Expose a business schema catalog and explicit template runner for Kingdee sales-order facts. Update wiki-query guidance so the LLM routes operational questions through templates first and guarded SQL only when needed. Refresh user docs and ops aliases, and add coverage for schema metadata, template execution, and CLI JSON output. Validation: python3 -m unittest discover -v; python3 -m compileall obsidian_wiki tests scripts; git diff --check
150 lines
4.4 KiB
Python
Executable File
150 lines
4.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Single user-facing operations entrypoint for this repository."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Sequence
|
|
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
REPO_ROOT = SCRIPT_DIR.parent
|
|
|
|
for path in (REPO_ROOT, SCRIPT_DIR):
|
|
if str(path) not in sys.path:
|
|
sys.path.insert(0, str(path))
|
|
|
|
from obsidian_wiki.cli import main as obsidian_cli_main
|
|
|
|
import manage_business_sync
|
|
import manage_mcp_service
|
|
|
|
|
|
HELP_TEXT = """obsidian-wiki user operations
|
|
|
|
Usage:
|
|
scripts/obsidian-wiki-ops.py help
|
|
|
|
scripts/obsidian-wiki-ops.py setup [args...]
|
|
scripts/obsidian-wiki-ops.py info
|
|
scripts/obsidian-wiki-ops.py file-index rebuild [args...]
|
|
|
|
scripts/obsidian-wiki-ops.py kingdee config [args...]
|
|
scripts/obsidian-wiki-ops.py kingdee auth-test [args...]
|
|
scripts/obsidian-wiki-ops.py kingdee sync [args...]
|
|
scripts/obsidian-wiki-ops.py kingdee schema --json
|
|
scripts/obsidian-wiki-ops.py kingdee template order_count --vault /path/to/vault --year 2026 --month 6 --json
|
|
scripts/obsidian-wiki-ops.py kingdee sql "SELECT COUNT(*) FROM orders" [args...]
|
|
|
|
scripts/obsidian-wiki-ops.py timer install [args...]
|
|
scripts/obsidian-wiki-ops.py timer run-now [args...]
|
|
scripts/obsidian-wiki-ops.py timer status
|
|
scripts/obsidian-wiki-ops.py timer stop
|
|
scripts/obsidian-wiki-ops.py timer uninstall [args...]
|
|
|
|
scripts/obsidian-wiki-ops.py mcp start [args...]
|
|
scripts/obsidian-wiki-ops.py mcp tools list
|
|
scripts/obsidian-wiki-ops.py mcp status
|
|
scripts/obsidian-wiki-ops.py mcp stop
|
|
scripts/obsidian-wiki-ops.py mcp uninstall
|
|
|
|
scripts/obsidian-wiki-ops.py daily-update [args...]
|
|
|
|
Notes:
|
|
timer install is an alias for installing and starting the Kingdee business sync timer.
|
|
Use --help after any forwarded command to see that command's detailed options.
|
|
"""
|
|
|
|
|
|
CLI_COMMANDS = {
|
|
"setup",
|
|
"list",
|
|
"info",
|
|
"file-index",
|
|
"business",
|
|
"mcp-server",
|
|
"bot-view",
|
|
}
|
|
|
|
|
|
def print_help() -> int:
|
|
print(HELP_TEXT, end="")
|
|
return 0
|
|
|
|
|
|
def run_script(script_name: str, args: Sequence[str]) -> int:
|
|
script = SCRIPT_DIR / script_name
|
|
completed = subprocess.run([str(script), *args], check=False)
|
|
return int(completed.returncode)
|
|
|
|
|
|
def route_kingdee(args: Sequence[str]) -> int:
|
|
if not args:
|
|
return print_help()
|
|
action, *rest = args
|
|
if action == "config":
|
|
return obsidian_cli_main(["business", "config", "kingdee", *rest])
|
|
if action == "auth-test":
|
|
return obsidian_cli_main(["business", "sync", "kingdee", "--dry-run-auth", *rest])
|
|
if action == "sync":
|
|
return obsidian_cli_main(["business", "sync", "kingdee", *rest])
|
|
if action == "schema":
|
|
return obsidian_cli_main(["business", "schema", *rest])
|
|
if action == "template":
|
|
return obsidian_cli_main(["business", "template", *rest])
|
|
if action == "query":
|
|
return obsidian_cli_main(["business", "query", *rest])
|
|
if action == "sql":
|
|
return obsidian_cli_main(["business", "sql", *rest])
|
|
print(f"Unknown kingdee action: {action}", file=sys.stderr)
|
|
return 2
|
|
|
|
|
|
def route_timer(args: Sequence[str]) -> int:
|
|
if not args:
|
|
return print_help()
|
|
action, *rest = args
|
|
action_aliases = {
|
|
"install": "start",
|
|
"start": "start",
|
|
"run": "run-now",
|
|
"now": "run-now",
|
|
"run-now": "run-now",
|
|
"status": "status",
|
|
"stop": "stop",
|
|
"uninstall": "uninstall",
|
|
}
|
|
mapped = action_aliases.get(action)
|
|
if mapped is None:
|
|
print(f"Unknown timer action: {action}", file=sys.stderr)
|
|
return 2
|
|
return manage_business_sync.main([mapped, *rest])
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
args = list(sys.argv[1:] if argv is None else argv)
|
|
if not args or args[0] in {"help", "-h", "--help"}:
|
|
return print_help()
|
|
|
|
command, *rest = args
|
|
if command in CLI_COMMANDS:
|
|
return obsidian_cli_main([command, *rest])
|
|
if command == "kingdee":
|
|
return route_kingdee(rest)
|
|
if command == "timer":
|
|
return route_timer(rest)
|
|
if command == "mcp":
|
|
return manage_mcp_service.main(rest)
|
|
if command == "daily-update":
|
|
return run_script("daily-update.sh", rest)
|
|
|
|
print(f"Unknown command: {command}", file=sys.stderr)
|
|
print("Run `scripts/obsidian-wiki-ops.py help` for available operations.", file=sys.stderr)
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|