122 lines
4.6 KiB
Python
122 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Proaktiver Health-Check für Second Brain.
|
||
Erstellt alle 6h ein Engramm mit System-Status.
|
||
Nur bei Problemen wird eine Warnung generiert.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import sqlite3
|
||
import subprocess
|
||
import sys
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
|
||
BRAIN_DIR = Path("/root/.openclaw/workspace/second-brain")
|
||
DB_PATH = BRAIN_DIR / "data" / "brain.sqlite"
|
||
|
||
def get_db_stats():
|
||
conn = sqlite3.connect(str(DB_PATH))
|
||
conn.row_factory = sqlite3.Row
|
||
c = conn.cursor()
|
||
total = c.execute("SELECT COUNT(*) FROM engrams").fetchone()[0]
|
||
confirmed_true = c.execute("SELECT COUNT(*) FROM engrams WHERE json_extract(correctness_json, '$.verdict') = 'confirmed_true' OR (json_extract(correctness_json, '$.verdict') IS NULL AND json_extract(correctness_json, '$.confirmed') = 1)").fetchone()[0]
|
||
confirmed_false = c.execute("SELECT COUNT(*) FROM engrams WHERE json_extract(correctness_json, '$.verdict') = 'confirmed_false' OR (json_extract(correctness_json, '$.verdict') IS NULL AND json_extract(correctness_json, '$.confirmed') = 0 AND COALESCE(json_extract(correctness_json, '$.rejections'), 0) > 0)").fetchone()[0]
|
||
pending = total - confirmed_true - confirmed_false
|
||
latest = c.execute("SELECT created_at FROM engrams ORDER BY created_at DESC LIMIT 1").fetchone()
|
||
latest_created = latest[0] if latest else None
|
||
conn.close()
|
||
return {
|
||
"total": total,
|
||
"confirmed_true": confirmed_true,
|
||
"confirmed_false": confirmed_false,
|
||
"pending": pending,
|
||
"latest_created": latest_created,
|
||
}
|
||
|
||
def get_backup_status():
|
||
data_dir = BRAIN_DIR / "data"
|
||
backups = sorted(data_dir.glob("backup_*.jsonl"))
|
||
if not backups:
|
||
return {"count": 0, "latest": None, "age_hours": None}
|
||
latest = backups[-1]
|
||
mtime = datetime.fromtimestamp(latest.stat().st_mtime, tz=timezone.utc)
|
||
age_hours = (datetime.now(timezone.utc) - mtime).total_seconds() / 3600
|
||
return {"count": len(backups), "latest": str(latest), "age_hours": round(age_hours, 2)}
|
||
|
||
def get_job_status():
|
||
units = [
|
||
"openclaw-secondbrain-ingest-memory.service",
|
||
"openclaw-secondbrain-index-vectors.service",
|
||
"openclaw-secondbrain-review.service",
|
||
"openclaw-secondbrain-heartbeat.service",
|
||
"openclaw-secondbrain-verify-pending.service",
|
||
]
|
||
status = {}
|
||
for u in units:
|
||
try:
|
||
out = subprocess.check_output(["systemctl", "is-active", u], text=True, stderr=subprocess.DEVNULL).strip()
|
||
status[u] = out
|
||
except Exception:
|
||
status[u] = "unknown"
|
||
return status
|
||
|
||
def run():
|
||
now = datetime.now(timezone.utc).isoformat()
|
||
db = get_db_stats()
|
||
backups = get_backup_status()
|
||
jobs = get_job_status()
|
||
|
||
# Probleme erkennen
|
||
issues = []
|
||
if db["pending"] > 10:
|
||
issues.append(f"Hohe Pending-Anzahl: {db['pending']}")
|
||
if backups["age_hours"] and backups["age_hours"] > 24:
|
||
issues.append(f"Backup zu alt: {backups['age_hours']}h")
|
||
for unit, state in jobs.items():
|
||
if state not in ("active", "running"):
|
||
issues.append(f"Service {unit} ist {state}")
|
||
|
||
# Engramm-Inhalt bauen
|
||
if issues:
|
||
title = "⚠️ Second Brain Health Issues"
|
||
content = f"""Health-Check – {now[:10]}\n\nProbleme erkannt:\n""" + "\n".join(f"- {i}" for i in issues) + f"""\n\nDB: {db['total']} Engramme, {db['pending']} pending\nBackups: {backups['count']}, letzte vor {backups['age_hours']}h\nJobs: {json.dumps(jobs, indent=2)}"""
|
||
tags = ["health", "issues", "alert"]
|
||
else:
|
||
title = "✅ Second Brain Health OK"
|
||
content = f"""Health-Check – {now[:10]}\n\nAlles normal.\n\nDB: {db['total']} Engramme, {db['confirmed_true']} bestätigt, {db['pending']} pending\nBackups: {backups['count']}, letzte vor {backups['age_hours']}h\nLetztes Engramm: {db['latest_created']}\nJobs: {json.dumps(jobs, indent=2)}"""
|
||
tags = ["health", "ok"]
|
||
|
||
# Engramm speichern
|
||
sys.path.insert(0, str(BRAIN_DIR))
|
||
from src.store import EngramStore
|
||
from src.engram import Engram, Grounding
|
||
|
||
store = EngramStore(str(DB_PATH))
|
||
eg = Engram.create(
|
||
content=content,
|
||
source="system",
|
||
tags=tags,
|
||
grounding=Grounding.ASSUMPTION,
|
||
)
|
||
eg.metadata.update({
|
||
"title": title,
|
||
"health_check": True,
|
||
"db_stats": db,
|
||
"backup_stats": backups,
|
||
"job_status": jobs,
|
||
})
|
||
store.save(eg)
|
||
|
||
print(json.dumps({
|
||
"success": True,
|
||
"time": now,
|
||
"engram_id": str(eg.id),
|
||
"issues_found": len(issues),
|
||
}, indent=2, ensure_ascii=False))
|
||
|
||
if __name__ == "__main__":
|
||
run()
|