Codex
b85c26e0ba
Add scripts/obsidian-wiki-ops.py as the repo-local command surface for common wiki, Kingdee, timer, daily update, and MCP operations. Route the existing business and MCP shell wrappers through the unified entrypoint for compatibility, and update docs to recommend the single command path. Validation: python3 -m unittest discover -v; python3 -m compileall obsidian_wiki tests scripts; git diff --check.
293 lines
10 KiB
Python
293 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Manage the user-level Kingdee business fact sync timer."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import os
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Sequence
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
if str(REPO_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(REPO_ROOT))
|
|
|
|
from obsidian_wiki.config import resolve_vault_path
|
|
|
|
|
|
SERVICE_NAME = "obsidian-wiki-business-sync.service"
|
|
TIMER_NAME = "obsidian-wiki-business-sync.timer"
|
|
DEFAULT_CONFIG_RELATIVE_PATH = Path(".runtime") / "kingdee.env"
|
|
DEFAULT_ON_CALENDAR = "*-*-* 07:10:00"
|
|
DEFAULT_RANDOMIZED_DELAY_SEC = "15m"
|
|
DEFAULT_LOOKBACK_HOURS = 48
|
|
DEFAULT_INITIAL_DAYS = 7
|
|
PROXY_ENV_NAMES = ("http_proxy", "https_proxy", "all_proxy", "no_proxy", "HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "NO_PROXY")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BusinessSyncServicePaths:
|
|
project_root: Path
|
|
service_unit_path: Path
|
|
timer_unit_path: Path
|
|
config_path: Path
|
|
|
|
|
|
def systemd_quote(value: str | Path) -> str:
|
|
return shlex.quote(str(value))
|
|
|
|
|
|
def resolve_service_paths(project_root: Path, config_home: Path | None = None) -> BusinessSyncServicePaths:
|
|
root = project_root.expanduser().resolve()
|
|
base_config = config_home or Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config"))
|
|
unit_dir = base_config.expanduser() / "systemd" / "user"
|
|
return BusinessSyncServicePaths(
|
|
project_root=root,
|
|
service_unit_path=unit_dir / SERVICE_NAME,
|
|
timer_unit_path=unit_dir / TIMER_NAME,
|
|
config_path=root / DEFAULT_CONFIG_RELATIVE_PATH,
|
|
)
|
|
|
|
|
|
def resolve_vault_for_service(project_root: Path, vault_path: Path | str | None = None) -> Path:
|
|
if vault_path is not None:
|
|
return Path(vault_path).expanduser().resolve()
|
|
configured = resolve_vault_path(None, start=project_root)
|
|
if configured:
|
|
return Path(configured).expanduser().resolve()
|
|
return project_root.expanduser().resolve()
|
|
|
|
|
|
def render_service_unit(
|
|
*,
|
|
project_root: Path,
|
|
vault_path: Path,
|
|
config_path: Path,
|
|
python_path: Path | None = None,
|
|
lookback_hours: int = DEFAULT_LOOKBACK_HOURS,
|
|
initial_days: int = DEFAULT_INITIAL_DAYS,
|
|
) -> str:
|
|
resolved_python = python_path or Path(sys.executable)
|
|
command = " ".join(
|
|
[
|
|
systemd_quote(resolved_python),
|
|
"-m",
|
|
"obsidian_wiki",
|
|
"business",
|
|
"sync",
|
|
"kingdee",
|
|
"--vault",
|
|
systemd_quote(vault_path),
|
|
"--config",
|
|
systemd_quote(config_path),
|
|
"--incremental",
|
|
"--lookback-hours",
|
|
str(lookback_hours),
|
|
"--initial-days",
|
|
str(initial_days),
|
|
]
|
|
)
|
|
return (
|
|
"[Unit]\n"
|
|
"Description=obsidian-wiki Kingdee business fact sync\n"
|
|
"After=network-online.target\n"
|
|
"Wants=network-online.target\n"
|
|
"\n"
|
|
"[Service]\n"
|
|
"Type=oneshot\n"
|
|
f"WorkingDirectory={systemd_quote(project_root)}\n"
|
|
f"EnvironmentFile={systemd_quote(config_path)}\n"
|
|
f"PassEnvironment={' '.join(PROXY_ENV_NAMES)}\n"
|
|
f"ExecStart={command}\n"
|
|
)
|
|
|
|
|
|
def render_timer_unit(
|
|
*,
|
|
on_calendar: str = DEFAULT_ON_CALENDAR,
|
|
randomized_delay_sec: str = DEFAULT_RANDOMIZED_DELAY_SEC,
|
|
) -> str:
|
|
return (
|
|
"[Unit]\n"
|
|
"Description=Run obsidian-wiki Kingdee business sync on a schedule\n"
|
|
"\n"
|
|
"[Timer]\n"
|
|
f"OnCalendar={on_calendar}\n"
|
|
f"RandomizedDelaySec={randomized_delay_sec}\n"
|
|
"Persistent=true\n"
|
|
f"Unit={SERVICE_NAME}\n"
|
|
"\n"
|
|
"[Install]\n"
|
|
"WantedBy=timers.target\n"
|
|
)
|
|
|
|
|
|
def run_systemctl(args: Sequence[str], *, check: bool = True) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(["systemctl", "--user", *args], text=True, check=check)
|
|
|
|
|
|
def write_units(
|
|
paths: BusinessSyncServicePaths,
|
|
*,
|
|
vault_path: Path,
|
|
python_path: Path | None = None,
|
|
on_calendar: str = DEFAULT_ON_CALENDAR,
|
|
randomized_delay_sec: str = DEFAULT_RANDOMIZED_DELAY_SEC,
|
|
lookback_hours: int = DEFAULT_LOOKBACK_HOURS,
|
|
initial_days: int = DEFAULT_INITIAL_DAYS,
|
|
) -> None:
|
|
paths.service_unit_path.parent.mkdir(parents=True, exist_ok=True)
|
|
paths.service_unit_path.write_text(
|
|
render_service_unit(
|
|
project_root=paths.project_root,
|
|
vault_path=vault_path,
|
|
config_path=paths.config_path,
|
|
python_path=python_path,
|
|
lookback_hours=lookback_hours,
|
|
initial_days=initial_days,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
paths.timer_unit_path.write_text(
|
|
render_timer_unit(on_calendar=on_calendar, randomized_delay_sec=randomized_delay_sec),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def start_timer(
|
|
*,
|
|
project_root: Path,
|
|
vault_path: Path | str | None = None,
|
|
python_path: Path | None = None,
|
|
on_calendar: str = DEFAULT_ON_CALENDAR,
|
|
randomized_delay_sec: str = DEFAULT_RANDOMIZED_DELAY_SEC,
|
|
lookback_hours: int = DEFAULT_LOOKBACK_HOURS,
|
|
initial_days: int = DEFAULT_INITIAL_DAYS,
|
|
) -> str:
|
|
paths = resolve_service_paths(project_root)
|
|
if not paths.config_path.exists():
|
|
raise RuntimeError(
|
|
f"Missing Kingdee config file: {paths.config_path}. "
|
|
"Run `scripts/obsidian-wiki-ops.py kingdee config` first."
|
|
)
|
|
resolved_vault = resolve_vault_for_service(paths.project_root, vault_path=vault_path)
|
|
write_units(
|
|
paths,
|
|
vault_path=resolved_vault,
|
|
python_path=python_path,
|
|
on_calendar=on_calendar,
|
|
randomized_delay_sec=randomized_delay_sec,
|
|
lookback_hours=lookback_hours,
|
|
initial_days=initial_days,
|
|
)
|
|
run_systemctl(["daemon-reload"])
|
|
run_systemctl(["enable", "--now", TIMER_NAME])
|
|
return (
|
|
"Business sync timer is enabled.\n"
|
|
f"Service: {paths.service_unit_path}\n"
|
|
f"Timer: {paths.timer_unit_path}\n"
|
|
f"Vault: {resolved_vault}\n"
|
|
f"Config: {paths.config_path}\n"
|
|
f"Schedule: {on_calendar} randomized by {randomized_delay_sec}\n"
|
|
)
|
|
|
|
|
|
def run_now() -> None:
|
|
run_systemctl(["start", SERVICE_NAME])
|
|
|
|
|
|
def stop_timer() -> None:
|
|
run_systemctl(["disable", "--now", TIMER_NAME], check=False)
|
|
|
|
|
|
def uninstall_timer(project_root: Path) -> None:
|
|
paths = resolve_service_paths(project_root)
|
|
stop_timer()
|
|
if paths.service_unit_path.exists():
|
|
paths.service_unit_path.unlink()
|
|
if paths.timer_unit_path.exists():
|
|
paths.timer_unit_path.unlink()
|
|
run_systemctl(["daemon-reload"], check=False)
|
|
run_systemctl(["reset-failed", SERVICE_NAME], check=False)
|
|
run_systemctl(["reset-failed", TIMER_NAME], check=False)
|
|
|
|
|
|
def timer_status() -> None:
|
|
run_systemctl(["status", TIMER_NAME], check=False)
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description="Manage the Kingdee business sync user timer.")
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
|
|
start_parser = subparsers.add_parser("start", help="Install or update the user timer.")
|
|
start_parser.add_argument("--project-root", default=Path(__file__).resolve().parent.parent, type=Path)
|
|
start_parser.add_argument("--vault", dest="vault_path", type=Path, help="Vault path to sync into.")
|
|
start_parser.add_argument("--python", dest="python_path", type=Path, help="Python executable for the systemd service.")
|
|
start_parser.add_argument("--on-calendar", default=DEFAULT_ON_CALENDAR)
|
|
start_parser.add_argument("--randomized-delay-sec", default=DEFAULT_RANDOMIZED_DELAY_SEC)
|
|
start_parser.add_argument("--lookback-hours", type=int, default=DEFAULT_LOOKBACK_HOURS)
|
|
start_parser.add_argument("--initial-days", type=int, default=DEFAULT_INITIAL_DAYS)
|
|
|
|
run_now_parser = subparsers.add_parser("run-now", help="Run the sync service immediately.")
|
|
run_now_parser.add_argument("--project-root", default=Path(__file__).resolve().parent.parent, type=Path)
|
|
run_now_parser.add_argument("--vault", dest="vault_path", type=Path)
|
|
run_now_parser.add_argument("--python", dest="python_path", type=Path)
|
|
run_now_parser.add_argument("--lookback-hours", type=int, default=DEFAULT_LOOKBACK_HOURS)
|
|
run_now_parser.add_argument("--initial-days", type=int, default=DEFAULT_INITIAL_DAYS)
|
|
|
|
subparsers.add_parser("status", help="Show timer status.")
|
|
subparsers.add_parser("stop", help="Disable and stop the timer.")
|
|
|
|
uninstall_parser = subparsers.add_parser("uninstall", help="Remove managed timer/service unit files.")
|
|
uninstall_parser.add_argument("--project-root", default=Path(__file__).resolve().parent.parent, type=Path)
|
|
|
|
args = parser.parse_args(argv)
|
|
if args.command == "start":
|
|
print(
|
|
start_timer(
|
|
project_root=args.project_root,
|
|
vault_path=args.vault_path,
|
|
python_path=args.python_path,
|
|
on_calendar=args.on_calendar,
|
|
randomized_delay_sec=args.randomized_delay_sec,
|
|
lookback_hours=args.lookback_hours,
|
|
initial_days=args.initial_days,
|
|
),
|
|
end="",
|
|
)
|
|
return 0
|
|
if args.command == "run-now":
|
|
paths = resolve_service_paths(args.project_root)
|
|
if not paths.service_unit_path.exists():
|
|
start_timer(
|
|
project_root=args.project_root,
|
|
vault_path=args.vault_path,
|
|
python_path=args.python_path,
|
|
lookback_hours=args.lookback_hours,
|
|
initial_days=args.initial_days,
|
|
)
|
|
run_now()
|
|
print(f"Started {SERVICE_NAME}.")
|
|
return 0
|
|
if args.command == "status":
|
|
timer_status()
|
|
return 0
|
|
if args.command == "stop":
|
|
stop_timer()
|
|
print(f"Stopped {TIMER_NAME}.")
|
|
return 0
|
|
if args.command == "uninstall":
|
|
uninstall_timer(args.project_root)
|
|
print(f"Uninstalled {TIMER_NAME}.")
|
|
return 0
|
|
raise AssertionError(f"unhandled command: {args.command}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|