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