yaojingang
65 lines
1.8 KiB
Python
65 lines
1.8 KiB
Python
"""Shared runtime helpers for the Yao CLI."""
|
|
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
SCRIPT_INTERFACE = "internal-module"
|
|
SCRIPT_INTERFACE_REASON = "Imported by yao.py and command modules for shared subprocess execution and JSON payload parsing."
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
SCRIPTS = ROOT / "scripts"
|
|
|
|
|
|
def script_path(name: str) -> str:
|
|
return str(SCRIPTS / name)
|
|
|
|
|
|
def load_json_maybe(text: str) -> dict | None:
|
|
text = text.strip()
|
|
if not text:
|
|
return None
|
|
try:
|
|
return json.loads(text)
|
|
except json.JSONDecodeError:
|
|
return None
|
|
|
|
|
|
def run_script(name: str, args: list[str], cwd: Path | None = None) -> dict:
|
|
proc = subprocess.run(
|
|
[sys.executable, script_path(name), *args],
|
|
cwd=cwd or ROOT,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
payload = load_json_maybe(proc.stdout)
|
|
return {
|
|
"command": f"{name} {' '.join(args)}".strip(),
|
|
"returncode": proc.returncode,
|
|
"ok": proc.returncode == 0,
|
|
"stdout": proc.stdout,
|
|
"stderr": proc.stderr,
|
|
"payload": payload,
|
|
}
|
|
|
|
|
|
def run_adoption_drift_if_source_exists(skill_dir: Path | None = None) -> dict:
|
|
target = (skill_dir or ROOT).resolve()
|
|
events_path = target / "reports" / "telemetry_events.jsonl"
|
|
if events_path.exists():
|
|
return run_script("render_adoption_drift_report.py", [str(target)])
|
|
return {
|
|
"command": "render_adoption_drift_report.py skipped: missing reports/telemetry_events.jsonl",
|
|
"returncode": 0,
|
|
"ok": True,
|
|
"stdout": "",
|
|
"stderr": "",
|
|
"payload": {
|
|
"ok": True,
|
|
"skipped": True,
|
|
"reason": "raw telemetry event logs are local-only; keeping committed adoption_drift_report artifacts",
|
|
},
|
|
}
|