mission-control-v2/deploy/curator_archive_logic.py

#!/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 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()