130 lines
4.6 KiB
Python
130 lines
4.6 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import report_failed_turns
|
|
import report_quality_metrics
|
|
import report_repeated_failures
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
DEFAULT_REPORT_DIR = ROOT / "reports" / "observability"
|
|
DEFAULT_OUTPUT_DIR = DEFAULT_REPORT_DIR / "quality"
|
|
|
|
|
|
def timestamp_slug(now: datetime | None = None) -> str:
|
|
current = now or datetime.now(timezone.utc)
|
|
return current.strftime("%Y%m%dT%H%M%SZ")
|
|
|
|
|
|
def snapshot_paths(output_dir: Path, *, slug: str) -> dict[str, Path]:
|
|
return {
|
|
"quality": output_dir / f"weekly-quality-{slug}.json",
|
|
"repeated_failures": output_dir / f"repeated-failures-{slug}.json",
|
|
"failed_turns": output_dir / f"failed-turns-{slug}.json",
|
|
"index": output_dir / "index.json",
|
|
}
|
|
|
|
|
|
def read_index(path: Path) -> dict[str, Any]:
|
|
if not path.exists():
|
|
return {"snapshots": []}
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
if not isinstance(data, dict):
|
|
return {"snapshots": []}
|
|
snapshots = data.get("snapshots")
|
|
if not isinstance(snapshots, list):
|
|
data["snapshots"] = []
|
|
return data
|
|
|
|
|
|
def write_index(path: Path, entry: dict[str, Any], *, keep: int = 50) -> dict[str, Any]:
|
|
data = read_index(path)
|
|
snapshots = [item for item in data.get("snapshots", []) if isinstance(item, dict)]
|
|
snapshots.append(entry)
|
|
snapshots.sort(key=lambda item: str(item.get("created_at") or ""), reverse=True)
|
|
data["snapshots"] = snapshots[:keep]
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
return data
|
|
|
|
|
|
def build_snapshot(
|
|
*,
|
|
report_dir: Path,
|
|
output_dir: Path,
|
|
days: int,
|
|
limit: int,
|
|
min_repeat_count: int,
|
|
failed_limit: int,
|
|
) -> dict[str, Any]:
|
|
records = report_quality_metrics.iter_records(report_dir)
|
|
filtered = report_quality_metrics.filter_recent(records, days=days)
|
|
quality_payload = report_quality_metrics.summarize(filtered, limit=limit)
|
|
|
|
turn_records = report_failed_turns.iter_turn_audit(report_dir)
|
|
failed_payload = {"records": len(turn_records), "failed_turns": report_failed_turns.filter_failed(turn_records, limit=failed_limit)}
|
|
repeated_payload = {
|
|
"records": len(turn_records),
|
|
"repeated_failures": report_repeated_failures.group_repeated_failures(turn_records, min_count=min_repeat_count, limit=limit),
|
|
}
|
|
|
|
slug = timestamp_slug()
|
|
paths = snapshot_paths(output_dir, slug=slug)
|
|
report_quality_metrics.write_report(paths["quality"], quality_payload)
|
|
report_repeated_failures.write_report(paths["repeated_failures"], repeated_payload)
|
|
report_failed_turns.write_report(paths["failed_turns"], failed_payload)
|
|
|
|
created_at = datetime.now(timezone.utc).isoformat()
|
|
entry = {
|
|
"created_at": created_at,
|
|
"slug": slug,
|
|
"days": days,
|
|
"limit": limit,
|
|
"min_repeat_count": min_repeat_count,
|
|
"failed_limit": failed_limit,
|
|
"artifacts": {
|
|
"quality": str(paths["quality"]),
|
|
"repeated_failures": str(paths["repeated_failures"]),
|
|
"failed_turns": str(paths["failed_turns"]),
|
|
},
|
|
"summary": {
|
|
"records": quality_payload.get("records"),
|
|
"top_failure_types": quality_payload.get("top_failure_types") or [],
|
|
"repeated_failures": len(repeated_payload.get("repeated_failures") or []),
|
|
"failed_turns": len(failed_payload.get("failed_turns") or []),
|
|
},
|
|
}
|
|
write_index(paths["index"], entry)
|
|
return {"snapshot": entry, "paths": {key: str(value) for key, value in paths.items()}}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Export dated observability quality snapshot bundle and update index.")
|
|
parser.add_argument("--report-dir", type=Path, default=DEFAULT_REPORT_DIR)
|
|
parser.add_argument("--output-dir", type=Path, default=DEFAULT_OUTPUT_DIR)
|
|
parser.add_argument("--days", type=int, default=7)
|
|
parser.add_argument("--limit", type=int, default=10)
|
|
parser.add_argument("--min-repeat-count", type=int, default=2)
|
|
parser.add_argument("--failed-limit", type=int, default=100)
|
|
args = parser.parse_args()
|
|
|
|
payload = build_snapshot(
|
|
report_dir=args.report_dir,
|
|
output_dir=args.output_dir,
|
|
days=args.days,
|
|
limit=args.limit,
|
|
min_repeat_count=args.min_repeat_count,
|
|
failed_limit=args.failed_limit,
|
|
)
|
|
print(json.dumps(payload, ensure_ascii=False, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|