#!/usr/bin/env python3 """ Processing-Registry Setup fuer YouTube-Research. Legt source_processing_registry in knowledge.db an und traegt Pilotstatus ein. Idempotent: mehrfach ausfuehrbar ohne Datenverlust. """ import hashlib, json, re, sqlite3, sys from datetime import datetime, timezone from pathlib import Path DB_PATH = Path("/opt/struktur/youtube-research/knowledge.db") PILOT_DIR = Path("/opt/obsidian-vault/YouTube-Research/Pilot") PILOT_LOG = Path("/opt/struktur/graphiti/youtube_research_pilot_log.json") NOW = datetime.now(timezone.utc).isoformat() # Pilot-Video-IDs (intern) PILOT_IDS = [46, 428, 857, 892, 944, 969, 972, 982, 993, 994] # Mapping: interne ID -> MD-Dateiname ARTIFACT_FILES = { 46: "ID-046-claude-cowork.md", 428: "ID-428-claude-code-stitch.md", 857: "ID-857-openclaw-ollama-gratis.md", 892: "ID-892-openclaw-mission-control.md", 944: "ID-944-hermes-desktop.md", 969: "ID-969-langdock-dokumente.md", 972: "ID-972-claude-notebooklm.md", 982: "ID-982-n8n-vps-deploy.md", 993: "ID-993-claude-full-course.md", 994: "ID-994-lokale-ki-odysseus.md", } def log(msg): print(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}", flush=True) def sha256(text: str) -> str: return hashlib.sha256(text.encode("utf-8")).hexdigest() def file_sha256(path: Path) -> str: if not path.exists(): return "" return hashlib.sha256(path.read_bytes()).hexdigest() def load_pilot_log(): if not PILOT_LOG.exists(): return {} entries = json.loads(PILOT_LOG.read_text(encoding="utf-8")) # Key: interne ID (als String) -> Eintrag return {str(e.get("internal_id")): e for e in entries} CREATE_TABLE_SQL = """ CREATE TABLE IF NOT EXISTS source_processing_registry ( id INTEGER PRIMARY KEY AUTOINCREMENT, source_system TEXT NOT NULL, source_id TEXT NOT NULL, source_external_id TEXT, source_title TEXT, source_hash TEXT, artifact_status TEXT, artifact_path TEXT, artifact_hash TEXT, quality_status TEXT, graphiti_status TEXT, graphiti_import_key TEXT, graphiti_actor TEXT, graphiti_group_id TEXT, processed_at TEXT, updated_at TEXT, notes TEXT, UNIQUE(source_system, source_id) ); """ log("=== Processing-Registry Setup ===") log(f"DB: {DB_PATH}") # 1. Tabelle anlegen log("Lege source_processing_registry an (idempotent)...") con = sqlite3.connect(DB_PATH) cur = con.cursor() cur.executescript(CREATE_TABLE_SQL) con.commit() log("Tabelle angelegt oder bereits vorhanden.") # 2. Pilot-Log laden log("Lade Pilot-Log...") pilot_log = load_pilot_log() log(f"Pilot-Log Eintraege: {len(pilot_log)}") imported_internal_ids = set(pilot_log.keys()) log(f"Importierte IDs: {imported_internal_ids}") # 3. Video-Daten aus DB laden log("Lade Video-Metadaten aus knowledge.db...") rows = cur.execute(""" SELECT v.id, v.youtube_id, v.title, v.channel, v.language, v.transcript_hash, v.transcript_status, v.transcript_quality, COUNT(ts.id) as seg_count FROM videos v LEFT JOIN transcript_segments ts ON ts.video_id = v.id WHERE v.id IN (46,428,857,892,944,969,972,982,993,994) GROUP BY v.id """).fetchall() video_data = {} for row in rows: vid_id, yt_id, title, channel, lang, tr_hash, tr_status, tr_quality, seg_count = row video_data[vid_id] = { "youtube_id": yt_id, "title": title or "", "channel": channel or "", "language": lang or "", "transcript_hash": tr_hash or "", "transcript_status": tr_status or "", "transcript_quality": tr_quality or "", "seg_count": seg_count or 0, } log(f" Video {vid_id}: yt={yt_id} segs={seg_count} tr_quality={tr_quality} tr_hash={'JA' if tr_hash else 'LEER'}") # 4. Jeden Pilot-Eintrag aufbereiten und upsert log("\nSchreibe Registry-Eintraege...") results_table = [] for vid_id in PILOT_IDS: if vid_id not in video_data: log(f" WARNUNG: Video {vid_id} nicht in DB gefunden, uebersprungen.") continue vd = video_data[vid_id] internal_id_str = str(vid_id) yt_id = vd["youtube_id"] seg_count = vd["seg_count"] # source_hash: youtube_id + seg_count + transcript_hash (stabil, reprodzierbar) existing_tr_hash = vd["transcript_hash"] source_hash_input = f"{yt_id}|{seg_count}|{existing_tr_hash or 'no_hash'}" source_hash = sha256(source_hash_input) # artifact_hash: Hash der MD-Datei artifact_filename = ARTIFACT_FILES.get(vid_id, "") artifact_path_obj = PILOT_DIR / artifact_filename if artifact_filename else None artifact_path_str = str(artifact_path_obj) if artifact_path_obj else "" artifact_hash = file_sha256(artifact_path_obj) if artifact_path_obj else "" # Status bestimmen if vid_id == 46: artifact_status = "created" quality_status = "needs_verification" graphiti_status = "blocked" graphiti_key = "" graphiti_actor = "" notes = "graphiti_ready:false -- Transkript-Qualitaet zu niedrig. Manuell pruefen." else: artifact_status = "created" quality_status = "passed" graphiti_status = "imported" # Aus Pilot-Log pl_entry = pilot_log.get(internal_id_str, {}) graphiti_key = pl_entry.get("actor", f"yt-{yt_id}") graphiti_actor = pl_entry.get("source", "youtube-research-pilot") imported_at = pl_entry.get("imported_at", "") notes = f"Pilot-Import {imported_at}" # UPSERT (INSERT OR REPLACE wuerde id verlieren; nutze INSERT OR IGNORE + UPDATE) cur.execute(""" INSERT OR IGNORE INTO source_processing_registry (source_system, source_id, source_external_id, source_title, source_hash, artifact_status, artifact_path, artifact_hash, quality_status, graphiti_status, graphiti_import_key, graphiti_actor, processed_at, updated_at, notes) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) """, ( "youtube-research", internal_id_str, yt_id, vd["title"], source_hash, artifact_status, artifact_path_str, artifact_hash, quality_status, graphiti_status, graphiti_key, graphiti_actor, NOW, NOW, notes, )) if cur.rowcount == 0: # Bereits vorhanden -> UPDATE (idempotent: nur Hashes + Status updaten) cur.execute(""" UPDATE source_processing_registry SET source_hash=?, artifact_hash=?, artifact_status=?, quality_status=?, graphiti_status=?, graphiti_import_key=?, graphiti_actor=?, updated_at=?, notes=? WHERE source_system='youtube-research' AND source_id=? """, ( source_hash, artifact_hash, artifact_status, quality_status, graphiti_status, graphiti_key, graphiti_actor, NOW, notes, internal_id_str, )) action = "AKTUALISIERT" else: action = "NEU EINGETRAGEN" log(f" [{vid_id}] {action}: artifact={artifact_status} quality={quality_status} graphiti={graphiti_status} source_hash={source_hash[:12]}... artifact_hash={artifact_hash[:12] if artifact_hash else 'FEHLT'}...") results_table.append({ "id": vid_id, "yt_id": yt_id, "artifact_file": artifact_filename, "artifact_status": artifact_status, "quality_status": quality_status, "graphiti_status": graphiti_status, "source_hash": source_hash[:16] + "...", "artifact_hash": (artifact_hash[:16] + "...") if artifact_hash else "FEHLT", }) con.commit() log("Alle Eintraege committed.") # 5. Ergebnis-Dump zur Kontrolle log("\n=== Alle Registry-Eintraege ===") check = cur.execute(""" SELECT source_id, source_external_id, artifact_status, quality_status, graphiti_status, graphiti_import_key, substr(source_hash,1,16), substr(artifact_hash,1,16) FROM source_processing_registry WHERE source_system='youtube-research' ORDER BY CAST(source_id AS INTEGER) """).fetchall() for row in check: log(f" ID={row[0]} yt={row[1]} art={row[2]} qual={row[3]} gfti={row[4]} key={row[5]} src_h={row[6]}.. art_h={row[7]}..") log(f"\nGesamt registriert: {len(check)} Pilotquellen") # 6. Tabellen-Statistik total_graphiti = sum(1 for r in check if r[4] == "imported") total_blocked = sum(1 for r in check if r[4] == "blocked") log(f"graphiti_status=imported: {total_graphiti}") log(f"graphiti_status=blocked: {total_blocked}") # 7. Maschinenlesbares JSON fuer Bericht-Skript import json as _json summary = { "total": len(check), "imported": total_graphiti, "blocked": total_blocked, "results_table": results_table, } print("SUMMARY_JSON:" + _json.dumps(summary, ensure_ascii=False)) con.close() log("=== Registry-Setup abgeschlossen ===")