#!/usr/bin/env python3
import argparse, hashlib, json, os, re, shutil, sqlite3, sys
from datetime import datetime, timezone
from pathlib import Path

KDB = "/opt/struktur/youtube-research/knowledge.db"
VAULT = Path("/opt/obsidian-vault")
VALID_EVIDENCE = {"UNSICHER", "PLAUSIBEL", "BESTÄTIGT", "GESICHERT", "WIDERSPRÜCHLICH"}
PROMPT_VERSION = "p33-ku-extraction-prompt-v1"
EXTRACTION_PROMPT = """Du extrahierst Knowledge Units aus genau einem übergebenen kanonischen Content Record.\n\nRegeln:\n- Verwende ausschließlich Informationen aus dem übergebenen Quelltext.\n- Erzeuge nur kleinste, atomare, eigenständig prüfbare fachliche Aussagen.\n- Eine unabhängige Aussage pro Knowledge Unit.\n- Keine Vermutungen, externen Ergänzungen oder Weltwissen.\n- Keine Zusammenfassung des gesamten Dokuments.\n- Führe für jede Aussage eine genaue Fundstelle (Abschnitt, Chunk oder Zeilenbereich) mit.\n- Erhalte Unsicherheit, Einschränkungen und Zeitbezüge aus der Quelle.\n- Glätte oder verschweige Widersprüche nicht; klassifiziere sie als WIDERSPRÜCHLICH.\n- Überschriften, Stichworte und reine Entitäten ohne Aussage sind keine Knowledge Units.\n- Gib strukturierte Kandidaten mit statement, source_id, source_type, content_version, source_locator, evidence_class und confidence_reason aus.\n"""

def sha(b): return hashlib.sha256(b).hexdigest()
def norm(s): return re.sub(r"\\s+", " ", s).strip()
def words(s): return re.findall(r"[A-Za-zÄÖÜäöüß0-9][A-Za-zÄÖÜäöüß0-9'_-]*", s)
def identity(source_id, version, statement):
    return sha(("v1\\n" + source_id + "\\n" + version + "\\n" + norm(statement)).encode())
def import_identity(source_id, version):
    return sha((source_id + ":" + version).encode())

def db():
    return sqlite3.connect("file:" + KDB + "?mode=ro", uri=True)

def youtube_rows():
    c=db(); c.row_factory=sqlite3.Row
    rows=[]
    q="""SELECT v.id,v.youtube_id,v.title,v.transcript,v.transcript_hash,v.transcript_status,
                 v.transcript_source,v.transcript_updated_at,e.obsidian_path,e.content_hash,e.processed_at
          FROM videos v LEFT JOIN e2e_extractions e ON e.video_id=v.id
          WHERE v.youtube_id IS NOT NULL AND length(trim(COALESCE(v.transcript,'')))>0
          ORDER BY v.id"""
    for r in c.execute(q): rows.append(dict(r))
    c.close(); return rows

def source_versions(source_id):
    c=db(); c.row_factory=sqlite3.Row
    out=[]
    # source_versions is keyed by version_id; identity/path lives in the registry.
    q="""SELECT sv.version_id,sv.content_hash,sv.content_version_hash,sv.import_identity,
                 sv.artifact_path,sv.created_at,r.source_identity,r.canonical_obsidian_path
          FROM source_versions sv JOIN source_processing_registry r
            ON (r.latest_observed_version_id=sv.version_id OR r.current_graph_version_id=sv.version_id)
          WHERE r.source_identity=? OR r.source_id=? OR r.canonical_obsidian_path=?
          ORDER BY sv.version_id DESC"""
    for r in c.execute(q, (source_id, source_id, source_id.removeprefix("obsidian:"))):
        out.append(dict(r))
    c.close(); return out

def canonical_youtube(r, kind):
    text=r["transcript"] or ""
    content_hash=sha(text.encode("utf-8"))
    artifact=r.get("obsidian_path")
    artifact_exists=bool(artifact and (VAULT / artifact).is_file())
    return {"source_id":"youtube:"+r["youtube_id"],"source_type":"youtube_transcript", "youtube_id":r["youtube_id"],
            "title":r.get("title"),"text":text,"content_version":content_hash,"raw_source_hash":r.get("transcript_hash"),
            "obsidian_path":artifact if artifact_exists else None,"artifact_exists":artifact_exists,
            "transcript_status":r.get("transcript_status"),"transcript_source":r.get("transcript_source"),
            "selection_kind":kind,"source_db_video_id":r.get("id"),"source_versions":source_versions("youtube:"+r["youtube_id"])}

def canonical_obs(path):
    b=path.read_bytes(); rel=path.relative_to(VAULT).as_posix(); text=b.decode("utf-8", errors="replace")
    return {"source_id":"obsidian:"+rel,"source_type":"obsidian","youtube_id":None,"title":rel,
            "text":text,"content_version":sha(b),"raw_source_hash":sha(b),"obsidian_path":rel,
            "artifact_exists":True,"transcript_status":None,"transcript_source":None,"selection_kind":"native_obsidian",
            "source_db_video_id":None,"source_versions":source_versions("obsidian:"+rel)}

def select_sources():
    ys=youtube_rows()
    pure=[r for r in ys if not (r.get("obsidian_path") and (VAULT/r["obsidian_path"]).is_file())]
    dual=[r for r in ys if r.get("obsidian_path") and (VAULT/r["obsidian_path"]).is_file()]
    if len(pure)<4 or len(dual)<2: raise RuntimeError(f"selection impossible pure={len(pure)} dual={len(dual)}")
    # Deterministic, diverse by shortest stable IDs; no historical KU rows are read as input.
    chosen=[]
    chosen += [canonical_youtube(r,"pure_youtube_transcript") for r in pure[:4]]
    chosen += [canonical_obs(VAULT/x) for x in [
        "Systemstruktur/SYSTEM-REFERENZ.md",
        "Systemstruktur/Review-Routing-Decision-Units.md",
        "Systemstruktur/Knowledge-Pipeline-Review-Unter-100-2026-07-22.md",
        "05-Ressourcen/Hermes-Sessions-2026-07/TranscriptBackfillWorker-untersuchen.md"]
        if (VAULT/x).is_file()]
    if len(chosen)!=8: raise RuntimeError(f"native obsidian selection impossible got={len(chosen)-4}")
    chosen += [canonical_youtube(r,"youtube_plus_obsidian_artifact") for r in dual[:2]]
    if len(chosen)!=10: raise RuntimeError("exactly 10 sources not selected")
    ids=[x["source_id"] for x in chosen]
    if len(set(ids))!=10: raise RuntimeError("source identity collision")
    return chosen

def sentence_candidates(src):
    text=src["text"]
    out=[]; offset=0
    # Preserve line provenance; sentence fragments are never silently repaired.
    for li,line in enumerate(text.splitlines(),1):
        s=line.strip()
        if not s or s.startswith("#") or s.startswith("-") or s.startswith("*") or s.startswith(">"):
            continue
        for frag in re.split(r"(?<=[.!?])\s+", s):
            st=norm(frag).strip("` ")
            if len(words(st))<8: continue
            if re.match(r"^(quelle|quellen|inhalt|zusammenfassung|status|todo|hinweis)\s*[:#]", st, re.I): continue
            # Candidate-level refusal for obvious multi-claim constructions.
            if st.count(";") or re.search(r"\b(sowohl|einerseits|andererseits|währenddessen)\b", st, re.I):
                continue
            out.append((st, f"line:{li}"))
    # Bound the pilot output while remaining source-derived and deterministic.
    return out[:12]

def classify(stmt):
    if re.search(r"\b(widerspruch|widersprüchlich|unklar|nicht nachgewiesen|unsicher|möglicherweise|vermutlich)\b", stmt, re.I):
        return "WIDERSPRÜCHLICH" if re.search(r"\bwidersprüch",stmt,re.I) else "UNSICHER", "Die Quelle enthält eine explizite Unsicherheits-/Widerspruchsmarkierung."
    return "PLAUSIBEL", "Einzelquelle direkt belegt; keine unabhängige Zweitquelle im P33-Pilot nachgewiesen; daher nicht GESICHERT."

def valid(c):
    req=["statement","source_id","content_version","source_locator","evidence_class","confidence_reason","knowledge_unit_identity"]
    if any(not c.get(k) for k in req): return False,"required_field_missing"
    if c["statement"].lower().startswith("knowledge unit from") or len(words(c["statement"]))<8: return False,"placeholder_or_too_short"
    if len(words(c["statement"]))>45: return False,"too_broad"
    if c["statement"].endswith(":"): return False,"heading_like"
    if c["evidence_class"] not in VALID_EVIDENCE: return False,"invalid_evidence"
    return True,None

def jaccard(a,b):
    A=set(x.lower() for x in words(a)); B=set(x.lower() for x in words(b))
    return len(A&B)/max(1,len(A|B))

def run():
    ap=argparse.ArgumentParser(); ap.add_argument("--output",required=True); args=ap.parse_args()
    out=Path(args.output); out.mkdir(parents=True,exist_ok=False); staging=out/"p33_staging.sqlite"
    sources=select_sources(); conn=sqlite3.connect(staging); conn.executescript("""
    PRAGMA foreign_keys=ON;
    CREATE TABLE sources(source_id TEXT PRIMARY KEY, source_type TEXT, youtube_id TEXT, title TEXT, content_version TEXT NOT NULL, raw_source_hash TEXT, text_bytes INTEGER, word_count INTEGER, transcript_status TEXT, obsidian_path TEXT, artifact_exists INTEGER, selection_kind TEXT, source_db_video_id INTEGER, source_versions_json TEXT, frozen_at TEXT);
    CREATE TABLE candidates(candidate_id INTEGER PRIMARY KEY, source_id TEXT NOT NULL, statement TEXT, source_locator TEXT, evidence_class TEXT, confidence_reason TEXT, knowledge_unit_identity TEXT, valid_status TEXT, reject_reason TEXT, dedupe_status TEXT DEFAULT 'UNIQUE', FOREIGN KEY(source_id) REFERENCES sources(source_id));
    CREATE TABLE provenance(candidate_id INTEGER, source_id TEXT, content_version TEXT, source_locator TEXT, source_path TEXT, youtube_id TEXT, representation TEXT, FOREIGN KEY(candidate_id) REFERENCES candidates(candidate_id));
    """)
    frozen=datetime.now(timezone.utc).isoformat(); allc=[]
    for s in sources:
        conn.execute("INSERT INTO sources VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", (s["source_id"],s["source_type"],s["youtube_id"],s["title"],s["content_version"],s["raw_source_hash"],len(s["text"].encode()),len(words(s["text"])),s["transcript_status"],s["obsidian_path"],int(s["artifact_exists"]),s["selection_kind"],s["source_db_video_id"],json.dumps(s["source_versions"],sort_keys=True),frozen))
        for st,loc in sentence_candidates(s):
            ev,reason=classify(st); c={"statement":st,"source_id":s["source_id"],"content_version":s["content_version"],"source_locator":loc,"evidence_class":ev,"confidence_reason":reason,"knowledge_unit_identity":identity(s["source_id"],s["content_version"],st)}
            ok,rej=valid(c); allc.append((s,c,ok,rej))
    # Exact and semantic dedup across the ten canonical records.
    accepted=[]
    for s,c,ok,rej in allc:
        status="VALID" if ok else "REJECT"
        d="UNIQUE"
        if ok:
            for prior in accepted:
                if prior[1]["knowledge_unit_identity"]==c["knowledge_unit_identity"] or jaccard(prior[1]["statement"],c["statement"])>=0.88:
                    d="DUPLICATE"; break
            if d=="UNIQUE": accepted.append((s,c))
        cur=conn.execute("INSERT INTO candidates(source_id,statement,source_locator,evidence_class,confidence_reason,knowledge_unit_identity,valid_status,reject_reason,dedupe_status) VALUES(?,?,?,?,?,?,?,?,?)",(c["source_id"],c["statement"],c["source_locator"],c["evidence_class"],c["confidence_reason"],c["knowledge_unit_identity"],status,rej,d))
        cid=cur.lastrowid
        conn.execute("INSERT INTO provenance VALUES(?,?,?,?,?,?,?)",(cid,s["source_id"],s["content_version"],c["source_locator"],s["obsidian_path"],s["youtube_id"],s["selection_kind"]))
    conn.commit(); conn.close()
    # JSON export contains all source metadata and all candidate rows without source text.
    conn=sqlite3.connect(staging); conn.row_factory=sqlite3.Row
    sources_out=[dict(r) for r in conn.execute("SELECT * FROM sources ORDER BY rowid")]
    cand_out=[dict(r) for r in conn.execute("SELECT * FROM candidates ORDER BY candidate_id")]
    conn.close()
    dual=[s for s in sources if s["selection_kind"]=="youtube_plus_obsidian_artifact"]
    dedup_pass=(len({s["source_id"] for s in sources})==10 and all(s["artifact_exists"] for s in dual) and len(dual)==2)
    summary={"p33_version":"1.0.0","checked_at":frozen,"source_count":len(sources),"sources":sources_out,"candidates":cand_out,"prompt_version":PROMPT_VERSION,"prompt":EXTRACTION_PROMPT,"model":{"mode":"deterministic_no_external_model_call","configured_productive_model":"not queried or changed","reason":"P33 staging extractor is provider-free; no model request made"},"metrics":{"candidates":len(allc),"valid":sum(1 for _,_,ok,_ in allc if ok),"rejects":sum(1 for _,_,ok,_ in allc if not ok),"duplicates":sum(1 for r in cand_out if r["dedupe_status"]=="DUPLICATE"),"evidence_distribution":{e:sum(1 for r in cand_out if r["valid_status"]=="VALID" and r["evidence_class"]==e) for e in sorted(VALID_EVIDENCE)},"hallucinations":"not detected by source-only deterministic extractor; independent fact verification not performed"},"canonical_source_dedup":{"result":"CANONICAL_SOURCE_DEDUP_PASS" if dedup_pass else "CANONICAL_SOURCE_DEDUP_FAIL","dual_sources":[{"source_id":s["source_id"],"youtube_id":s["youtube_id"],"obsidian_path":s["obsidian_path"],"canonical_content_version":s["content_version"]} for s in dual],"proof":"Each dual representation is one canonical source row and was processed once; artifact is provenance metadata, not a second input row."},"production_non_change":{"knowledge_units_writes":0,"historical_kus_changed":0,"backfill_started":0,"graphiti_release":0,"graphiti_posts":0,"neo4j_writes":0,"gold_promotions":0}}
    (out/"p33-result.json").write_text(json.dumps(summary,ensure_ascii=False,indent=2,sort_keys=True),encoding="utf-8")
    print(json.dumps({"output":str(out),"sources":len(sources),"candidates":len(allc),"valid":summary["metrics"]["valid"],"rejects":summary["metrics"]["rejects"],"duplicates":summary["metrics"]["duplicates"],"dedup":summary["canonical_source_dedup"]["result"]},ensure_ascii=False))

if __name__=="__main__": run()