104 lines
4.3 KiB
Python
104 lines
4.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Return the latest 1C saved-state watch run under an output root."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import importlib.util
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from list_1c_saved_state_watch_runs import list_runs, write_json
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def build_latest(root: Path, *, require_delta: bool, require_changed: bool) -> dict[str, Any]:
|
|
root = root.resolve()
|
|
listing = list_runs(root, limit=None, only_with_delta=require_delta, only_changed=require_changed)
|
|
latest = listing.get("latest")
|
|
return {
|
|
"schema": "onec_saved_state_latest_watch_run.v1",
|
|
"output_root": str(root),
|
|
"require_delta": require_delta,
|
|
"require_changed": require_changed,
|
|
"found": latest is not None,
|
|
"latest": latest,
|
|
"counts": listing.get("counts") or {},
|
|
"safety": {
|
|
"read_only": True,
|
|
"sql_write_performed": False,
|
|
},
|
|
}
|
|
|
|
|
|
def check_latest(latest_path: Path, check_output: Path) -> None:
|
|
module_path = REPO_ROOT / "scripts" / "check_1c_saved_state_latest_watch_run.py"
|
|
spec = importlib.util.spec_from_file_location("saved_state_latest_watch_check", module_path)
|
|
if spec is None or spec.loader is None:
|
|
raise RuntimeError(f"Cannot load latest watch checker: {module_path}")
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
result = module.check_latest(latest_path)
|
|
module.write_json(check_output, result)
|
|
if not result.get("passed"):
|
|
raise RuntimeError(f"Saved-state latest watch check failed: {check_output}")
|
|
|
|
|
|
def render_markdown(data: dict[str, Any]) -> str:
|
|
module_path = REPO_ROOT / "scripts" / "render_1c_saved_state_latest_watch_run_markdown.py"
|
|
spec = importlib.util.spec_from_file_location("saved_state_latest_watch_markdown", module_path)
|
|
if spec is None or spec.loader is None:
|
|
raise RuntimeError(f"Cannot load latest watch renderer: {module_path}")
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module.render(data)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Return latest 1C saved-state watch run.")
|
|
parser.add_argument("--root", type=Path, required=True)
|
|
parser.add_argument("--require-delta", action="store_true")
|
|
parser.add_argument("--require-changed", action="store_true")
|
|
parser.add_argument("--output", type=Path)
|
|
parser.add_argument("--markdown-output", type=Path)
|
|
parser.add_argument("--skip-markdown", action="store_true")
|
|
parser.add_argument("--check-output", type=Path)
|
|
parser.add_argument("--skip-check", action="store_true")
|
|
args = parser.parse_args()
|
|
|
|
result = build_latest(args.root, require_delta=args.require_delta, require_changed=args.require_changed)
|
|
markdown_output = args.markdown_output
|
|
if markdown_output is None and args.output and not args.skip_markdown:
|
|
markdown_output = args.output.with_suffix(".md")
|
|
if markdown_output and not args.skip_markdown:
|
|
markdown_output = markdown_output.resolve()
|
|
result["markdown"] = str(markdown_output)
|
|
check_output = args.check_output
|
|
if check_output is None and args.output and not args.skip_check:
|
|
check_output = args.output.with_name(f"{args.output.stem}-check.json")
|
|
if check_output and not args.skip_check:
|
|
check_output = check_output.resolve()
|
|
result["check"] = str(check_output)
|
|
if args.output:
|
|
write_json(args.output, result)
|
|
if markdown_output and not args.skip_markdown:
|
|
markdown_output.parent.mkdir(parents=True, exist_ok=True)
|
|
markdown_output.write_text(render_markdown(result), encoding="utf-8")
|
|
if args.output:
|
|
write_json(args.output, result)
|
|
if check_output and not args.skip_check:
|
|
if not args.output:
|
|
raise SystemExit("Use --output when latest watch check output is enabled.")
|
|
check_latest(args.output, check_output)
|
|
print(json.dumps({"output": str(args.output) if args.output else None, "schema": result["schema"], "found": result["found"]}, ensure_ascii=False))
|
|
return 0 if result["found"] else 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|