#!/usr/bin/env python3 """Sync src/shared/ aus dem Monitor-Repo + Drift-Check. Hintergrund: src/shared/ enthält lokale Kopien aus dem Monitor-Repo (source_rules, services/source_health, services/source_suggester, agents/claude_client). Wenn der Monitor diese Module ändert, drifted die Verwaltung auseinander - dieses Skript hält beide synchron. Modi: python scripts/sync_shared.py --check # nur Drift-Diagnose, kein Schreiben python scripts/sync_shared.py --apply # Drift sichtbar machen + anwenden python scripts/sync_shared.py --apply --quiet # für CI-/Hook-Aufrufe Mojibake-Schutz: Falls beim Kopieren Doppel-Encoded UTF-8 erkannt wird (was im Monitor noch in einigen Texten steckt), wird ftfy zur Reparatur aufgerufen. Exit-Codes: 0 in sync (oder Sync erfolgreich) 1 Drift gefunden (--check Modus) oder Fehler beim Apply 2 Quelle nicht gefunden / Konfigurationsproblem """ from __future__ import annotations import argparse import difflib import sys from pathlib import Path # Fest verdrahtete Quell-Mappings: (Monitor-relative Quelle, Verwaltung-relative Destination) SHARED_FILES = [ ("src/source_rules.py", "src/shared/source_rules.py"), ("src/services/source_health.py", "src/shared/services/source_health.py"), ("src/services/source_suggester.py", "src/shared/services/source_suggester.py"), ("src/agents/claude_client.py", "src/shared/agents/claude_client.py"), ] # Files mit verwaltungs-spezifischen Anpassungen, die NICHT auto-synced werden # duerfen. Drift wird gemeldet (zur Information), aber --apply schreibt nicht. # Wenn der Monitor-Source auch sinnvoll ins Update soll: Aenderungen manuell # in den Verwaltungs-Fork einarbeiten, dann Eintrag pruefen oder entfernen. LOCKED_FILES = { "src/shared/services/source_health.py": "Phase 2 (Verwaltung): tenant_id-Filter weg + Historie-Archivierung " "+ User-Agent/Timeout aus Config - waere im Monitor unsinnig.", } DEFAULT_MONITOR = Path("/home/claude-dev/AegisSight-Monitor") DEFAULT_VERWALTUNG = Path(__file__).resolve().parent.parent def fix_mojibake(text: str) -> tuple[str, bool]: """Repariert Doppel-Encoded UTF-8 falls vorhanden. Gibt (text, fixed?) zurück.""" try: import ftfy # type: ignore except ImportError: return text, False fixed = ftfy.fix_text(text) return fixed, fixed != text def patch_imports_for_shared(text: str) -> str: """Patcht 'from agents.' -> 'from shared.agents.' damit Module innerhalb von src/shared/ ihre Geschwister-Module korrekt finden. 'from config import ...' bleibt unverändert (config.py liegt in beiden Apps in src/ Root). """ lines = text.splitlines(keepends=True) out = [] for line in lines: stripped = line.lstrip() indent = line[: len(line) - len(stripped)] if stripped.startswith("from agents."): line = indent + "from shared.agents." + stripped[len("from agents."):] elif stripped.startswith("from services."): line = indent + "from shared.services." + stripped[len("from services."):] out.append(line) return "".join(out) def diff_summary(old: str, new: str, label_old: str, label_new: str, max_lines: int = 30) -> str: diff = list(difflib.unified_diff( old.splitlines(keepends=True), new.splitlines(keepends=True), fromfile=label_old, tofile=label_new, n=2, )) if not diff: return "" if len(diff) <= max_lines: return "".join(diff) return "".join(diff[:max_lines]) + f"\n ... ({len(diff) - max_lines} weitere Zeilen abgeschnitten)\n" def main() -> int: parser = argparse.ArgumentParser(description="Sync src/shared/ aus Monitor-Repo") parser.add_argument("--check", action="store_true", help="nur prüfen, nichts schreiben") parser.add_argument("--apply", action="store_true", help="Sync ausführen") parser.add_argument("--quiet", action="store_true", help="weniger Output (für Hooks/CI)") parser.add_argument("--monitor", type=Path, default=DEFAULT_MONITOR, help="Pfad zum Monitor-Repo") parser.add_argument("--verwaltung", type=Path, default=DEFAULT_VERWALTUNG, help="Pfad zum Verwaltungs-Repo") args = parser.parse_args() if not args.check and not args.apply: parser.error("entweder --check oder --apply angeben") if not args.monitor.exists(): print(f"FEHLER: Monitor-Pfad nicht gefunden: {args.monitor}", file=sys.stderr) return 2 drift_count = 0 applied_count = 0 locked_drift_count = 0 for monitor_rel, verwaltung_rel in SHARED_FILES: src_path = args.monitor / monitor_rel dst_path = args.verwaltung / verwaltung_rel is_locked = verwaltung_rel in LOCKED_FILES if not src_path.exists(): print(f"FEHLER: Monitor-Quelle fehlt: {src_path}", file=sys.stderr) return 2 src_text = src_path.read_text(encoding="utf-8") src_text, fixed_mojibake = fix_mojibake(src_text) src_text = patch_imports_for_shared(src_text) existing = dst_path.read_text(encoding="utf-8") if dst_path.exists() else "" if existing == src_text: if not args.quiet: print(f" = {verwaltung_rel}: in sync") continue if is_locked: locked_drift_count += 1 if not args.quiet: print(f" L LOCKED-DRIFT: {verwaltung_rel}") print(f" Grund: {LOCKED_FILES[verwaltung_rel]}") print(f" -> NICHT ueberschrieben. Manuell pruefen ob Monitor-Aenderung") print(f" in den Verwaltungs-Fork eingearbeitet werden muss.") continue drift_count += 1 diff = diff_summary(existing, src_text, dst_path.name + " (Verwaltung)", src_path.name + " (Monitor)") if not args.quiet: print(f" ! DRIFT: {verwaltung_rel}") if fixed_mojibake: print(f" (Mojibake im Monitor-Original gefixt)") if diff: for line in diff.splitlines()[:20]: print(f" {line}") if args.apply: dst_path.parent.mkdir(parents=True, exist_ok=True) dst_path.write_text(src_text, encoding="utf-8") applied_count += 1 if args.check: if drift_count == 0 and locked_drift_count == 0: if not args.quiet: print(f"OK: src/shared/ ist in sync mit {args.monitor}") return 0 msg_parts = [] if drift_count: msg_parts.append(f"{drift_count} Datei(en) drift'en, mit --apply synchronisieren") if locked_drift_count: msg_parts.append(f"{locked_drift_count} LOCKED-Datei(en) drift'en (manuell pruefen)") if not args.quiet: print("\n" + ", ".join(msg_parts) + ".", file=sys.stderr) # Exit-Code: 1 wenn auto-sync ausstehend, 0 wenn nur LOCKED-Drift (informativ) return 1 if drift_count else 0 # apply if applied_count == 0 and locked_drift_count == 0: if not args.quiet: print(f"OK: src/shared/ war schon in sync, nichts zu tun.") return 0 if not args.quiet: if applied_count: print(f"\n{applied_count} Datei(en) synchronisiert.") print("Vergiss nicht: git diff src/shared/ → git add → commit") if locked_drift_count: print(f"\nHINWEIS: {locked_drift_count} LOCKED-Datei(en) NICHT geschrieben.") print("Manuell pruefen ob Monitor-Aenderung in den Verwaltungs-Fork uebernommen werden muss.") return 0 if __name__ == "__main__": sys.exit(main())