90 lines
2.6 KiB
Python
90 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Tägliche Zusammenfassung der Second Brain Aktivitäten.
|
||
Erstellt ein Engramm mit Highlights des Vortags.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import sqlite3
|
||
import sys
|
||
from datetime import datetime, timezone, timedelta
|
||
from pathlib import Path
|
||
|
||
BRAIN_DIR = Path("/root/.openclaw/workspace/second-brain")
|
||
DB_PATH = BRAIN_DIR / "data" / "brain.sqlite"
|
||
|
||
def run():
|
||
now = datetime.now(timezone.utc)
|
||
yesterday = now - timedelta(days=1)
|
||
date_str = yesterday.strftime("%Y-%m-%d")
|
||
|
||
conn = sqlite3.connect(str(DB_PATH))
|
||
conn.row_factory = sqlite3.Row
|
||
c = conn.cursor()
|
||
|
||
# Engramme von gestern (created_at innerhalb des Tages)
|
||
c.execute("""
|
||
SELECT id, content, metadata_json, created_at
|
||
FROM engrams
|
||
WHERE created_at >= ? AND created_at < ?
|
||
""", (yesterday.isoformat(), now.isoformat()))
|
||
rows = c.fetchall()
|
||
|
||
total_yesterday = len(rows)
|
||
sources = {}
|
||
tags = {}
|
||
for r in rows:
|
||
meta = json.loads(r["metadata_json"] or "{}")
|
||
src = meta.get("source", "unknown")
|
||
sources[src] = sources.get(src, 0) + 1
|
||
for t in meta.get("tags", []):
|
||
tags[t] = tags.get(t, 0) + 1
|
||
|
||
conn.close()
|
||
|
||
# Zusammenfassung bauen
|
||
top_sources = sorted(sources.items(), key=lambda x: x[1], reverse=True)[:5]
|
||
top_tags = sorted(tags.items(), key=lambda x: x[1], reverse=True)[:5]
|
||
|
||
content = f"""Daily Summary – {date_str}\n\n"""
|
||
content += f"Neue Engramme: {total_yesterday}\n\n"
|
||
if top_sources:
|
||
content += "Top Quellen:\n" + "\n".join(f"- {src}: {cnt}" for src, cnt in top_sources) + "\n\n"
|
||
if top_tags:
|
||
content += "Top Tags:\n" + "\n".join(f"- {tag}: {cnt}" for tag, cnt in top_tags) + "\n\n"
|
||
content += f"Generiert am {now.isoformat()}"
|
||
|
||
# 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=["daily-summary", "auto"],
|
||
grounding=Grounding.ASSUMPTION,
|
||
)
|
||
eg.metadata.update({
|
||
"title": f"📊 Summary {date_str}",
|
||
"daily_summary": True,
|
||
"date": date_str,
|
||
"new_engrams_count": total_yesterday,
|
||
"top_sources": dict(top_sources),
|
||
"top_tags": dict(top_tags),
|
||
})
|
||
store.save(eg)
|
||
|
||
print(json.dumps({
|
||
"success": True,
|
||
"date": date_str,
|
||
"engram_id": str(eg.id),
|
||
"new_engrams": total_yesterday,
|
||
}, indent=2, ensure_ascii=False))
|
||
|
||
if __name__ == "__main__":
|
||
run()
|