#!/usr/bin/env python3
"""
Wave-2 Graphiti-Import fuer YouTube-Research — BEREINIGTE VERSION
Qualitaetsfix gegenueber Wave-1:
- Quellenanker wird NICHT gesendet (ersetzt durch neutrale 1-Zeile)
- Content-Cleaner filtert Video-Meta-Saetze, Channel-Provenienz, Tautologien
- source_description enthaelt KEIN context-JSON (nur youtube_id)
- Keine privaten Labels (LueftungsProfi etc.) erreichbar aus diesem Script
"""
import sqlite3, os, re, json, time, urllib.request, urllib.error
from datetime import datetime, timezone
DB_PATH = "/opt/struktur/youtube-research/knowledge.db"
VAULT_PATH = "/opt/obsidian-vault/YouTube-Research/Welle-2" # Wave-2 Pfad
LOG_PATH = "/opt/struktur/graphiti/youtube_research_wave2_log.json"
GRAPHITI_URL = "http://127.0.0.1:8644/episodes"
PAUSE_SECS = 2.5
# === CONTENT CLEANER ===
# Saetze die video-provenienz beschreiben statt Wissen enthalten
VIDEO_META_PATTERNS = [
r"(?i)the video[^.]*(?:discusses|is about|covers|explores|explains|shows|presents|titled|hosted|posted|published)[^.]*\.",
r"(?i)das video[^.]*(?:handelt|erklaert|zeigt|bespricht|behandelt|wurde)[^.]*\.",
r"(?i)(?:this|the) (?:youtube )?video[^.]*youtube[^.]*\.",
r"(?i)(?:is|are) (?:hosted|published|available|featured) on youtube[^.]*\.",
r"(?i)(?:published|posted|created|produced|released) (?:a |the |this )?(?:youtube )?video[^.]*\.",
r"(?i)youtube channel[^.]*provides[^.]*\.",
r"(?i)(?:channel|kanal)[^.]*(?:provides|creates|produces|publishes) (?:content|tutorials|insights|videos)[^.]*\.",
r"(?i)the episode[^.]*(?:is about|discusses|covers|titled)[^.]*\.",
r"(?i)(?:Webronaq|Jack Roberts|Der KI-Doktor|Mike Mildenberger|Julian Ivanov|Marc De Fanti|Philip Thomas|AI mit Arnie|Alex Sprogis|Prompt Engineer)[^.]*(?:published|posted|created|produced|made|released)[^.]*\.",
r"(?i)(?:Webronaq|Jack Roberts|Der KI-Doktor|Mike Mildenberger|Julian Ivanov|Marc De Fanti|Philip Thomas|AI mit Arnie|Alex Sprogis|Prompt Engineer) (?:produced|created|made|released) a youtube[^.]*\.",
]
# Tautologie-Muster
TAUTOLOGY_PATTERNS = [
r"(?i)\w+(?:['s])?\s+(?:creator|entwickler|author)\s+created\s+\w+", # "X's Creator created X"
r"(?i)\w+ is used in \w+ environment", # "X is used in X environment"
r"(?i)\w+ is a (?:tool|product|service) by \w+\s*\.", # "X is a tool by Y." (too generic)
]
# Private/projektspezifische Labels die NICHT in globalen Graph sollen
PRIVATE_LABEL_PATTERNS = [
r"LüftungsProfi", r"Lueftungsprofi", r"LP-spezifisch", r"LP-Skills",
r"Agent Solutions", r"inkasso2025", r"K:\\\\",
r"projekte-LP", r"projekte-AG",
]
COMPILE = lambda lst: [re.compile(p) for p in lst]
_VIDEO_META_RE = COMPILE(VIDEO_META_PATTERNS)
_TAUTOLOGY_RE = COMPILE(TAUTOLOGY_PATTERNS)
_PRIVATE_RE = COMPILE(PRIVATE_LABEL_PATTERNS)
def clean_sentence(sentence: str) -> bool:
"""Gibt True zurueck wenn Satz behalten werden soll, False wenn herausfiltern."""
s = sentence.strip()
if len(s) < 20:
return False
for pat in _VIDEO_META_RE:
if pat.search(s):
return False
for pat in _TAUTOLOGY_RE:
if pat.search(s):
return False
for pat in _PRIVATE_RE:
if pat.search(s):
return False
return True
def graphiti_content_cleaner(raw_content: str) -> str:
"""
Bereinigt den Content vor dem Graphiti-POST:
- Splittet nach Satzgrenzen
- Filtert Video-Meta, Tautologien, private Labels
- Gibt bereinigten Text zurueck
"""
# Bullet-Points und Zeilen einzeln behandeln
lines = raw_content.split("\n")
cleaned_lines = []
for line in lines:
stripped = line.strip()
# Leerzeilen und Sektionsheader behalten
if not stripped or stripped.startswith("#"):
cleaned_lines.append(line)
continue
# Bullet-Points aufsplitten und pruefen
if stripped.startswith("- ") or stripped.startswith("* "):
text = stripped[2:]
if clean_sentence(text):
cleaned_lines.append(line)
else:
# Normaler Satz: nur wenn nicht herausgefiltert
if clean_sentence(stripped):
cleaned_lines.append(line)
return "\n".join(cleaned_lines)
# === ERLAUBTE SEKTIONEN ===
# Quellenanker wird NICHT mehr als vollstaendige Sektion gesendet
# Stattdessen: neutrale 1-Zeile am Ende des Contents
CONTENT_SECTIONS = [
"Graphiti-Ready Summary",
"Kernaussagen",
"Begriffe / Entitaeten",
]
BLOCKED_SECTIONS = ["Volltext-Kontext", "Quellenanker"]
def build_neutral_source_line(meta: dict) -> str:
"""Erzeugt neutrale Quellenzeile ohne private Labels."""
title = meta.get("title", "").replace("LüftungsProfi", "").strip()
channel = meta.get("channel", "").strip()
yt_id = meta.get("youtube_id", "")
lang = meta.get("language", "")
return f"Quelle: {title} | Kanal: {channel} | ID: {yt_id} | Sprache: {lang}"
def extract_graphiti_content(filepath: str, meta: dict) -> tuple:
"""Liest Artefakt, extrahiert nur erlaubte Sektionen, bereinigt Content."""
with open(filepath, "r", encoding="utf-8") as f:
raw = f.read()
# YAML-Frontmatter entfernen
raw = re.sub(r"^---\n.*?\n---\n?", "", raw, flags=re.DOTALL)
# Titel extrahieren
title_match = re.search(r"^# (.+)$", raw, re.MULTILINE)
title = title_match.group(1).strip() if title_match else meta.get("title", "?")
# Sektionen extrahieren
parts = []
section_pattern = re.compile(r"^## (.+)$", re.MULTILINE)
section_positions = [(m.group(1).strip(), m.start(), m.end())
for m in section_pattern.finditer(raw)]
for i, (sec_name, start, end) in enumerate(section_positions):
# Blockierte Sektionen ueberspringen (inkl. Quellenanker)
if any(b in sec_name for b in BLOCKED_SECTIONS):
continue
# Nur erlaubte Sektionen
if not any(a in sec_name for a in CONTENT_SECTIONS):
continue
next_start = (section_positions[i + 1][1]
if i + 1 < len(section_positions) else len(raw))
content = raw[end:next_start].strip()
# Code-Bloecke entfernen
content = re.sub(r"```.*?```", "", content, flags=re.DOTALL).strip()
if content:
parts.append(f"## {sec_name}\n\n{content}")
# Rohen Content zusammensetzen
raw_content = "\n\n---\n\n".join(parts)
# === CONTENT CLEANER ANWENDEN ===
cleaned_content = graphiti_content_cleaner(raw_content)
# Neutrale Quellenzeile anfuegen (kein Quellenanker-Block)
neutral_source = build_neutral_source_line(meta)
final_content = cleaned_content.rstrip() + "\n\n" + neutral_source
return title, final_content
def registry_check(vid: str) -> tuple:
conn = sqlite3.connect(DB_PATH)
cur = conn.cursor()
cur.execute("""SELECT graphiti_status, quality_status, artifact_status
FROM source_processing_registry
WHERE source_system='youtube-research' AND source_id=?""", (str(vid),))
row = cur.fetchone()
conn.close()
return row
def registry_update(vid: str, yt_id: str, status: str, error_msg: str = None):
conn = sqlite3.connect(DB_PATH)
cur = conn.cursor()
now = datetime.now().isoformat()
ik = f"yt-{yt_id}"
if status == "imported":
cur.execute("""UPDATE source_processing_registry SET
quality_status='passed', graphiti_status='imported',
graphiti_import_key=?, graphiti_actor=?,
graphiti_group_id='youtube-research-wave-2',
updated_at=?, notes='Wave-2 Import OK (bereinigt)'
WHERE source_system='youtube-research' AND source_id=?
""", (ik, ik, now, str(vid)))
else:
note = ("Import FEHLER: " + str(error_msg))[:200]
cur.execute("""UPDATE source_processing_registry SET
graphiti_status='failed', updated_at=?, notes=?
WHERE source_system='youtube-research' AND source_id=?
""", (now, note, str(vid)))
conn.commit()
conn.close()
def post_episode(content: str, source: str, actor: str,
context_dict: dict, timestamp_str: str) -> tuple:
payload = {
"content": content,
"source": source,
"actor": actor,
# Minimaler context: NUR youtube_id — kein JSON-Dump mit privaten Labels
"context": context_dict.get("youtube_id", ""),
"timestamp": timestamp_str,
}
data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
req = urllib.request.Request(
GRAPHITI_URL, data=data,
headers={"Content-Type": "application/json"}, method="POST"
)
try:
with urllib.request.urlopen(req, timeout=60) as resp:
return resp.status, resp.read().decode("utf-8")
except urllib.error.HTTPError as e:
body = e.read().decode("utf-8") if e.fp else str(e)
return e.code, body
except Exception as e:
return 0, str(e)
def load_log() -> dict:
if os.path.exists(LOG_PATH):
try:
with open(LOG_PATH) as f:
return json.load(f)
except Exception:
return {}
return {}
def save_log(log_data: dict):
with open(LOG_PATH, "w", encoding="utf-8") as f:
json.dump(log_data, f, ensure_ascii=False, indent=2)
def main(wave_meta: dict):
"""
wave_meta: dict mit vid_str -> (yt_id, lang, channel, filename)
Wird vom jeweiligen Wave-Script befuellt.
"""
log = load_log()
ok_count = fail_count = skip_count = 0
results = []
print(f"\n=== Wave Graphiti-Import (bereinigt) === {datetime.now().isoformat()}")
print(f"Ziel: {len(wave_meta)} Artefakte | Vault: {VAULT_PATH}\n")
for vid_str in sorted(wave_meta.keys(), key=lambda x: int(x)):
yt_id, lang, channel, fn = wave_meta[vid_str]
print(f"[{vid_str}] {fn}")
# Duplikatschutz
if yt_id in log and log[yt_id].get("http_status") == 200:
print(f" [SKIP] Log-Duplikat")
skip_count += 1
results.append({"id": vid_str, "yt_id": yt_id, "result": "skip_log"})
continue
reg = registry_check(vid_str)
if reg and reg[0] == "imported":
print(f" [SKIP] Registry-Duplikat")
skip_count += 1
results.append({"id": vid_str, "yt_id": yt_id, "result": "skip_registry"})
continue
fp = os.path.join(VAULT_PATH, fn)
if not os.path.exists(fp):
print(f" [FEHLER] Datei fehlt: {fp}")
fail_count += 1
results.append({"id": vid_str, "yt_id": yt_id, "result": "file_not_found"})
continue
meta = {"internal_id": vid_str, "youtube_id": yt_id,
"language": lang, "channel": channel,
"graphiti_group_id": "youtube-research-wave-2",
"import_key": f"wave2-{yt_id}"}
title, graphiti_content = extract_graphiti_content(fp, meta)
meta["title"] = title
if len(graphiti_content.strip()) < 50:
print(f" [FEHLER] Content zu kurz nach Bereinigung: {len(graphiti_content)}")
fail_count += 1
results.append({"id": vid_str, "yt_id": yt_id, "result": "too_short"})
continue
print(f" Content nach Cleaner: {len(graphiti_content)} Zeichen")
ts = datetime.now(timezone.utc).isoformat()
http_status, body = post_episode(
content=graphiti_content,
source="youtube-research-wave-2",
actor=f"yt-{yt_id}",
context_dict=meta,
timestamp_str=ts,
)
print(f" HTTP: {http_status}")
if http_status == 200:
ok_count += 1
imported_at = datetime.now().isoformat()
log[yt_id] = {
"internal_id": vid_str, "youtube_id": yt_id, "title": title,
"channel": channel, "language": lang, "filename": fn,
"http_status": 200, "imported_at": imported_at,
"content_len": len(graphiti_content),
"graphiti_group_id": "youtube-research-wave-2",
"cleaner_applied": True,
}
save_log(log)
registry_update(vid_str, yt_id, "imported")
print(f" [OK]")
results.append({"id": vid_str, "yt_id": yt_id, "result": "ok", "title": title})
else:
fail_count += 1
err_msg = body[:200] if body else "?"
log[yt_id] = {"internal_id": vid_str, "youtube_id": yt_id, "filename": fn,
"http_status": http_status, "error": err_msg,
"imported_at": datetime.now().isoformat()}
save_log(log)
registry_update(vid_str, yt_id, "failed", err_msg)
print(f" [FEHLER] {http_status}: {err_msg[:60]}")
results.append({"id": vid_str, "yt_id": yt_id, "result": "failed"})
if http_status == 200:
time.sleep(PAUSE_SECS)
print(f"\n=== ERGEBNIS === OK:{ok_count} Skip:{skip_count} Fehler:{fail_count}")
return results, ok_count, fail_count, skip_count
if __name__ == "__main__":
print("Dieses Script ist ein Template — Wave-2-Meta-Daten in WAVE2_META eintragen.")
print("Dann: from wave2_import_template import main; main(WAVE2_META)")