feat: Curator Auto-Archivierung (14-Tage-Inaktivität)

2026-07-24 13:46:49 +02:00
parent 8996c0d012
commit ce030d6817
3 changed files with 497 additions and 0 deletions
@@ -0,0 +1,199 @@
+#!/usr/bin/env python3
+"""
+Curator Auto-Archivierung: Archiviert Skills nach 14 Tagen Inaktivität.
+
+Liest `.curator_state` und `.usage.json` aus ~/.hermes/skills/, prüft
+auf Inaktivität (>14 Tage last_used_at), schließt aus (pinned, system)
+und verschiebt in .archive/<Kategorie>/.
+"""
+
+import json
+import os
+import shutil
+from datetime import datetime, timezone
+from pathlib import Path
+
+HERMES_SKILLS_DIR = Path.home() / ".hermes" / "skills"
+CURATOR_STATE_FILE = HERMES_SKILLS_DIR / ".curator_state"
+USAGE_JSON_FILE = HERMES_SKILLS_DIR / ".usage.json"
+ARCHIVE_DIR = HERMES_SKILLS_DIR / ".archive"
+LOG_DIR = Path.home() / ".hermes" / "logs" / "curator"
+
+
+def load_json(path: Path) -> dict:
+    """JSON-Datei laden."""
+    with open(path, "r", encoding="utf-8") as f:
+        return json.load(f)
+
+
+def save_json(path: Path, data: dict) -> None:
+    """JSON-Datei speichern (mit Pretty-Print)."""
+    with open(path, "w", encoding="utf-8") as f:
+        json.dump(data, f, indent=2, ensure_ascii=False)
+        f.write("\n")
+
+
+def is_stale(skill_usage: dict, threshold_days: int = 14) -> bool:
+    """Prüft, ob ein Skill älter als threshold_days ist."""
+    last_used = skill_usage.get("last_used_at")
+    if not last_used:
+        return False
+
+    # ISO-Format mit optionaler Zeitzone
+    last_used_str = last_used.replace("Z", "+00:00")
+    try:
+        last_used_dt = datetime.fromisoformat(last_used_str)
+    except ValueError:
+        return False
+
+    now = datetime.now(timezone.utc)
+    delta = now - last_used_dt
+    return delta.days >= threshold_days
+
+
+def is_excluded(skill_name: str, excluded: list) -> bool:
+    """Prüft, ob ein Skill ausgeschlossen ist."""
+    return skill_name in excluded
+
+
+def is_pinned(skill_usage: dict) -> bool:
+    """Prüft, ob ein Skill angepinnt ist."""
+    return skill_usage.get("pinned", False) is True
+
+
+def get_skill_category(skill_name: str) -> str:
+    """
+    Ermittelt die Kategorie eines Skills basierend auf dem Verzeichnisnamen.
+    Gibt 'unknown' zurück, wenn keine Kategorie gefunden wird.
+    """
+    # Versuche, die Kategorie aus dem Verzeichnis zu ermitteln
+    # z.B. 'betrieb-playbook' → 'betrieb-playbook' (direkt unter skills/)
+    # 'apple' (in .archive/apple/) → 'apple'
+    # Falls der Skill direkt in skills/ liegt ohne Unterverzeichnis:
+    return "unknown"
+
+
+def ensure_archive_category(category: str) -> Path:
+    """Erstellt das Archiv-Verzeichnis für eine Kategorie, falls nötig."""
+    category_path = ARCHIVE_DIR / category
+    category_path.mkdir(parents=True, exist_ok=True)
+    return category_path
+
+
+def move_skill_to_archive(skill_name: str, category: str) -> Path:
+    """
+    Verschiebt ein Skill-Verzeichnis in das Archiv.
+    Gibt den neuen Pfad zurück.
+    """
+    source = HERMES_SKILLS_DIR / skill_name
+    target_category = ensure_archive_category(category)
+    target = target_category / skill_name
+
+    if source.exists():
+        shutil.move(str(source), str(target))
+
+    return target
+
+
+def archive_skill(skill_name: str, usage_data: dict) -> dict:
+    """
+    Archiviert einen Skill:
+    - Setzt state auf 'archived'
+    - Setzt archived_at auf aktuelle Zeit
+    - Verschiebt das Verzeichnis nach .archive/
+    Gibt das aktualisierte Usage-Dict zurück.
+    """
+    usage_data[skill_name]["state"] = "archived"
+    usage_data[skill_name]["archived_at"] = datetime.now(timezone.utc).isoformat()
+    return usage_data
+
+
+def log_archive_action(skill_name: str, category: str, success: bool, message: str) -> None:
+    """Protokolliert eine Archivierungsaktion."""
+    LOG_DIR.mkdir(parents=True, exist_ok=True)
+    log_file = LOG_DIR / f"archive-{datetime.now(timezone.utc).date().isoformat()}.log"
+
+    timestamp = datetime.now(timezone.utc).isoformat()
+    status = "SUCCESS" if success else "FAILED"
+    log_line = f"[{timestamp}] {status} | {skill_name} ({category}) | {message}\n"
+
+    with open(log_file, "a", encoding="utf-8") as f:
+        f.write(log_line)
+
+
+def main() -> None:
+    """Hauptfunktion: Archiviert stale Skills."""
+    # 1. Config laden
+    if not CURATOR_STATE_FILE.exists():
+        print(f"Fehler: {CURATOR_STATE_FILE} nicht gefunden.")
+        return
+
+    curator_state = load_json(CURATOR_STATE_FILE)
+    archivierung = curator_state.get("archivierung", {})
+    if not archivierung.get("enabled", False):
+        print("Archivierung ist nicht aktiviert.")
+        return
+
+    threshold_days = archivierung.get("threshold_days", 14)
+    excluded_skills = archivierung.get("excluded_skills", [])
+
+    # 2. Usage-Daten laden
+    if not USAGE_JSON_FILE.exists():
+        print(f"Fehler: {USAGE_JSON_FILE} nicht gefunden.")
+        return
+
+    usage_data = load_json(USAGE_JSON_FILE)
+
+    # 3. Skills prüfen
+    archived_count = 0
+    skipped_count = 0
+
+    for skill_name, skill_usage in usage_data.items():
+        # Ausschlusskriterien prüfen
+        if is_excluded(skill_name, excluded_skills):
+            print(f"Übersprungen (Exkludiert): {skill_name}")
+            skipped_count += 1
+            continue
+
+        if is_pinned(skill_usage):
+            print(f"Übersprungen (Gepinnt): {skill_name}")
+            skipped_count += 1
+            continue
+
+        # Inaktivität prüfen
+        if not is_stale(skill_usage, threshold_days):
+            print(f"Übersprungen (Aktiv): {skill_name}")
+            skipped_count += 1
+            continue
+
+        # Archivierung durchführen
+        category = get_skill_category(skill_name) or "unknown"
+        print(f"Archiviere: {skill_name} → {category}")
+
+        try:
+            # Usage aktualisieren
+            usage_data = archive_skill(skill_name, usage_data)
+
+            # Verzeichnis verschieben
+            move_skill_to_archive(skill_name, category)
+
+            # Erfolg protokollieren
+            log_archive_action(skill_name, category, True, "Skill archiviert")
+
+            archived_count += 1
+        except Exception as e:
+            log_archive_action(skill_name, category, False, str(e))
+            print(f"Fehler beim Archivieren von {skill_name}: {e}")
+
+    # 4. Usage-Daten speichern
+    save_json(USAGE_JSON_FILE, usage_data)
+
+    # 5. Archivierungs-Zeitstempel aktualisieren
+    curator_state["archivierung"]["last_archive_run_at"] = datetime.now(timezone.utc).isoformat()
+    save_json(CURATOR_STATE_FILE, curator_state)
+
+    print(f"\nArchivierung abgeschlossen: {archived_count} archiviert, {skipped_count} übersprungen.")
+
+
+if __name__ == "__main__":
+    main()