cloudinit-bot
a8ec7b80de
---ci--- project: inboxos phase: 1 milestone: v1.4 status: execute wave: 2 tasks: [T3, T4, T5, T6, T7, T8] ---/ci---
101 lines
3.5 KiB
Python
Executable File
101 lines
3.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Explain a run: structured-log → human summary.
|
|
|
|
v1.4 REQ-INB-42.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from datetime import datetime
|
|
|
|
|
|
ARTIFACTS_DIR = os.environ.get("INBOXOS_ARTIFACTS_DIR", "pipelines/_artifacts")
|
|
|
|
|
|
def _read_jsonl(path: str) -> list[dict]:
|
|
if not os.path.exists(path):
|
|
return []
|
|
with open(path) as f:
|
|
return [json.loads(line) for line in f if line.strip()]
|
|
|
|
|
|
def _read_last_run() -> dict:
|
|
path = os.path.join(ARTIFACTS_DIR, "last_run.json")
|
|
if not os.path.exists(path):
|
|
return {}
|
|
with open(path) as f:
|
|
return json.load(f)
|
|
|
|
|
|
def _duration(start: str, end: str) -> str:
|
|
try:
|
|
s = datetime.fromisoformat(start.replace("Z", "+00:00"))
|
|
e = datetime.fromisoformat(end.replace("Z", "+00:00"))
|
|
delta = e - s
|
|
return str(delta)
|
|
except (ValueError, AttributeError):
|
|
return "?"
|
|
|
|
|
|
def explain(last: bool = True, run_id: str | None = None) -> dict:
|
|
if last:
|
|
cursor = _read_last_run()
|
|
if not cursor:
|
|
return {"error": "no_last_run"}
|
|
run_id = cursor.get("run_id", "last")
|
|
events = _read_jsonl(os.path.join(ARTIFACTS_DIR, f"{run_id}.jsonl"))
|
|
error_events = [e for e in events if e.get("event", "").endswith("_error")][:3]
|
|
return {
|
|
"run_id": run_id,
|
|
"run_type": cursor.get("run_type", "?"),
|
|
"started_at": cursor.get("started_at", "?"),
|
|
"finished_at": cursor.get("finished_at", "?"),
|
|
"duration": _duration(cursor.get("started_at", ""), cursor.get("finished_at", "")),
|
|
"mids_total": cursor.get("mids_total", 0),
|
|
"mids_upserted": cursor.get("mids_upserted", 0),
|
|
"mids_skipped": cursor.get("mids_skipped", 0),
|
|
"mids_errored": cursor.get("mids_errored", 0),
|
|
"modules": cursor.get("modules", {}),
|
|
"top_errors": [
|
|
{"event": e.get("event"), "message": e.get("message", "")}
|
|
for e in error_events
|
|
],
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
p = argparse.ArgumentParser(description="Explain an InboxOS run")
|
|
g = p.add_mutually_exclusive_group()
|
|
g.add_argument("--last", action="store_true", default=True, help="Read last_run.json")
|
|
g.add_argument("--run-id", help="Specific run ID to explain")
|
|
p.add_argument("--json", action="store_true", help="Emit JSON instead of human summary")
|
|
args = p.parse_args()
|
|
|
|
result = explain(last=not args.run_id, run_id=args.run_id)
|
|
if "error" in result and result["error"] == "no_last_run":
|
|
sys.stderr.write("No last_run.json found. Run a pipeline first.\n")
|
|
return 1
|
|
|
|
if args.json:
|
|
print(json.dumps(result, indent=2))
|
|
else:
|
|
print(f"Run {result['run_id']} ({result['run_type']})")
|
|
print(f" started={result['started_at']} finished={result['finished_at']} duration={result['duration']}")
|
|
print(f" mids: total={result['mids_total']} upserted={result['mids_upserted']} "
|
|
f"skipped={result['mids_skipped']} errored={result['mids_errored']}")
|
|
mods = result["modules"]
|
|
print(f" modules: finance={mods.get('finance', 0)} crm={mods.get('crm', 0)} "
|
|
f"intel={mods.get('intel', 0)} promotions={mods.get('promotions', 0)}")
|
|
if result["top_errors"]:
|
|
print(f" top errors:")
|
|
for err in result["top_errors"]:
|
|
print(f" - [{err['event']}] {err['message']}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|