補齊 AI Smoke 趨勢管理

services/ai_automation_smoke_service.py

@@ -60,13 +60,17 @@ def _compact_history_record(result: Dict[str, Any]) -> Dict[str, Any]:
     }
 
 
+def _read_history_lines() -> List[str]:
+    with _HISTORY_LOCK:
+        with open(_HISTORY_PATH, "r", encoding="utf-8") as fh:
+            return fh.readlines()
+
+
 def _load_history(limit: int = 20) -> List[Dict[str, Any]]:
     if limit <= 0:
         return []
     try:
-        with _HISTORY_LOCK:
-            with open(_HISTORY_PATH, "r", encoding="utf-8") as fh:
-                lines = fh.readlines()
+        lines = _read_history_lines()
     except FileNotFoundError:
         return []
 
@@ -108,9 +112,50 @@ def _history_summary(records: List[Dict[str, Any]]) -> Dict[str, Any]:
         "counts": counts,
         "recent": records,
         "latest": records[-1] if records else None,
+        "daily": _daily_summary(records),
     }
 
 
+def _daily_summary(records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    by_day: Dict[str, Dict[str, int]] = {}
+    for record in records:
+        day = str(record.get("generated_at") or "unknown")[:10]
+        bucket = by_day.setdefault(day, {"ok": 0, "warning": 0, "critical": 0, "total": 0})
+        status = record.get("status", "critical")
+        if status in bucket:
+            bucket[status] += 1
+        bucket["total"] += 1
+    return [
+        {"date": day, **counts}
+        for day, counts in sorted(by_day.items())[-14:]
+    ]
+
+
+def export_smoke_history_jsonl() -> Dict[str, Any]:
+    try:
+        lines = _read_history_lines()
+    except FileNotFoundError:
+        lines = []
+    return {
+        "content": "".join(lines),
+        "count": sum(1 for line in lines if line.strip()),
+        "path": _HISTORY_PATH,
+    }
+
+
+def clear_smoke_history() -> Dict[str, Any]:
+    try:
+        cleared = _count_jsonl_lines(_HISTORY_PATH)
+        with _HISTORY_LOCK:
+            try:
+                os.remove(_HISTORY_PATH)
+            except FileNotFoundError:
+                cleared = 0
+        return {"cleared": cleared, "path": _HISTORY_PATH}
+    except Exception as exc:
+        return {"cleared": 0, "path": _HISTORY_PATH, "error": str(exc)[:300]}
+
+
 def _event_router_check() -> Dict[str, Any]:
     try:
         from services import event_router