"""Autonomous YouTube Research E2E worker and independent completion validator."""
from __future__ import annotations
import hashlib, json, logging, os, re, shutil, socket, sqlite3, tempfile, time
from datetime import datetime, timezone, timedelta
from pathlib import Path
from supadata_native import NativeTranscriptUnavailable, QuotaWait, TransientSupadataError, account_status, fetch_native
DB = Path('/opt/struktur/youtube-research/knowledge.db')
CE_DB = Path('/opt/struktur/content-extraction/data/content_extraction.db')
VAULT = Path('/opt/obsidian-vault/YouTube-Research/Automatisch')
TMP = Path('/opt/struktur/youtube-research/.e2e-tmp')
VERSION = 'e2e-extraction-v1'
VALIDATOR_VERSION = 'content-validator-v2'
WORKER = f'{socket.gethostname()}:{os.getpid()}'
LOG = logging.getLogger('youtube-e2e')
def now(): return datetime.now(timezone.utc).isoformat()
def sha(s): return hashlib.sha256(s.encode('utf-8')).hexdigest()
def conn():
c=sqlite3.connect(DB, timeout=15, isolation_level=None); c.row_factory=sqlite3.Row
c.execute('PRAGMA busy_timeout=10000'); c.execute('PRAGMA foreign_keys=ON'); return c
def ensure_schema():
import db
db.init_db()
c=conn()
for col in ('classification TEXT','rejection_reason TEXT','metrics_json TEXT','validator_version TEXT'):
try: c.execute('ALTER TABLE e2e_rejections ADD COLUMN '+col)
except sqlite3.OperationalError: pass
c.close()
class ContentReject(ValueError):
def __init__(self, classification, metrics):
self.classification=classification; self.metrics=metrics
super().__init__(classification)
class TranscriptFetchError(RuntimeError):
pass
def content_classify(segs, text):
"""General content-value check; no single keyword decides the result."""
import difflib
clean=[re.sub(r'\s+',' ',s[3].lower()).strip() for s in segs if s[3].strip()]
words=re.findall(r"[a-zA-ZÀ-ÿÄÖÜäöüß]{3,}",text.lower())
unique_words=len(set(words)); lexical=unique_words/max(min(len(words),5000),1)
unique_segments=len(set(clean))/max(len(clean),1)
duplicate_count=len(clean)-len(set(clean))
adjacent=sum(difflib.SequenceMatcher(None,a,b).ratio()>=0.82 for a,b in zip(clean,clean[1:]))/max(len(clean)-1,1)
sentences=[x.strip() for x in re.split(r'(?<=[.!?])\s+',text) if len(x.strip())>=45]
informational=sum(bool(re.search(r'\b(weil|daher|deshalb|schritt|zuerst|dann|how|because|therefore|first|then|install|config|api|system|methode|prozess|funktion|problem|lösung|is|are|will|can)\b',x.lower())) for x in sentences)/max(len(sentences),1)
symbol_ratio=sum(1 for x in text if x in '♪♫🎵🎶')/max(len(text),1)
music_shape=(symbol_ratio>0.002 and lexical<0.22 and (adjacent>0.12 or unique_segments<0.72) and len(sentences)<max(4,len(segs)//12))
metrics={'repeat_ratio':round(1-unique_segments,4),'unique_segment_ratio':round(unique_segments,4),'lexical_diversity':round(lexical,4),'duplicate_segment_count':duplicate_count,'adjacent_similarity_ratio':round(adjacent,4),'informational_sentence_ratio':round(informational,4),'segment_count':len(segs),'sentence_count':len(sentences),'validator_version':VALIDATOR_VERSION}
if music_shape: return 'non_informational_music',metrics
if unique_segments<0.55 or adjacent>0.35 or lexical<0.18: return 'repetitive_low_information',metrics
if informational<0.15 or len(sentences)<3: return 'uncertain_content',metrics
return 'informational',metrics
def reserve(c):
"""Reserve one queued Supadata job, with a one-credit safety reserve."""
row=c.execute("SELECT * FROM e2e_jobs WHERE status IN ('queued','supadata_retry','supadata_quota_wait') AND (local_retry_after IS NULL OR local_retry_after<=?) ORDER BY priority,job_id LIMIT 1",(now(),)).fetchone()
if not row: return None
acct=account_status()
if acct.get('http') != 200 or acct.get('available') is None:
LOG.error('Supadata account status unavailable; job remains waiting')
if row['status'] != 'supadata_quota_wait':
c.execute("UPDATE e2e_jobs SET status='supadata_quota_wait',step='quota_wait',updated_at=?,error_code='account_status_unavailable',error_message='Supadata account status unavailable' WHERE job_id=? AND status IN ('queued','supadata_retry')",(now(),row['job_id'])); c.commit()
return None
if acct['available'] <= 1:
if row['status'] != 'supadata_quota_wait':
c.execute("UPDATE e2e_jobs SET status='supadata_quota_wait',step='quota_wait',updated_at=?,error_code='quota_reserve',error_message='Supadata quota reserve active' WHERE job_id=? AND status IN ('queued','supadata_retry')",(now(),row['job_id'])); c.commit()
return None
try: c.execute('BEGIN IMMEDIATE')
except sqlite3.OperationalError as ex:
if 'locked' not in str(ex).lower() and 'busy' not in str(ex).lower(): raise
c.rollback(); LOG.warning('database busy during reserve; retrying on next poll'); return None
changed=c.execute("UPDATE e2e_jobs SET status='supadata_processing',step='supadata_fetch',attempts=attempts+1,started_at=COALESCE(started_at,?),reserved_at=?,updated_at=?,lease_until=NULL,error_code=NULL,error_message=NULL WHERE job_id=? AND status IN ('queued','supadata_retry','supadata_quota_wait')",(now(),now(),now(),row['job_id']))
if changed.rowcount != 1: c.rollback(); return None
c.execute('COMMIT')
row=c.execute('SELECT * FROM e2e_jobs WHERE job_id=?',(row['job_id'],)).fetchone()
LOG.info('job %s reserved for Supadata Native (available=%s)', row['job_id'], acct['available'])
return row
def fetch_segments(yid):
result=fetch_native(yid)
return result.segments, result
def extract(row, segs):
text=' '.join(x[3] for x in segs); sentences=[x.strip() for x in re.split(r'(?<=[.!?])\s+',text) if len(x.strip())>30]
payload={'summary':' '.join(sentences[:5]),'key_points':sentences[:20],'facts':[], 'technical_details':[], 'procedures':[], 'reasons':[], 'warnings':[], 'examples':[], 'open_questions':[], 'topics':[], 'keywords':[], 'source_spans':[{'segment_id':s[0],'start':s[1],'end':s[2],'text':s[3]} for s in segs]}
return payload
def write_obsidian(row,payload,th, ch):
VAULT.mkdir(parents=True,exist_ok=True); TMP.mkdir(parents=True,exist_ok=True)
matches=[]
for p in VAULT.rglob('*.md'):
try:
if re.search(rf'(?im)^youtube_id:\s*{re.escape(row["youtube_id"])}\s*$',p.read_text(errors='ignore')): matches.append(p)
except OSError: pass
if len(matches)>1: raise RuntimeError('duplicate_obsidian')
if matches: return str(matches[0])
slug=re.sub(r'[^a-zA-Z0-9_-]+','-',(row['title'] or row['youtube_id']))[:80].strip('-')
target=VAULT/f'{row["youtube_id"]}-{slug}.md'; fd,tmp=tempfile.mkstemp(prefix='.e2e-',suffix='.md',dir=TMP); os.close(fd)
content='---\n'+f'youtube_id: {row["youtube_id"]}\ntitle: {json.dumps(row["title"] or "",ensure_ascii=False)}\nchannel: {json.dumps(row["channel"] or "",ensure_ascii=False)}\nsource_url: https://www.youtube.com/watch?v={row["youtube_id"]}\npublished_at: {row["published_at"] or ""}\nprocessed_at: {now()}\ntranscript_hash: {th}\ncontent_hash: {ch}\nextraction_version: {VERSION}\nprocessing_status: complete\n---\n\n# {row["title"] or row["youtube_id"]}\n\n## Kurzfassung\n{payload["summary"]}\n\n## Zentrale Aussagen\n'+'\n'.join(f'- {x}' for x in payload['key_points'])+'\n\n## Quellenstellen\n'+'\n'.join(f'- Segment {x["segment_id"]} ({x["start"]:.2f}s–{x["end"]:.2f}s): {x["text"]}' for x in payload['source_spans'])+'\n'
Path(tmp).write_text(content,encoding='utf-8'); os.replace(tmp,target); return str(target)
def register_graphiti(path, video_id):
"""Best-effort local enqueue; Graphiti is deliberately never on the hot path."""
try:
import sys
sys.path.insert(0, '/opt/struktur/graphiti')
from youtube_graphiti_queue import enqueue
enqueue(path)
except Exception:
LOG.exception('graphiti enqueue failed; reconciliation will recover it')
def sync_ce(row,path,th,ch):
c=sqlite3.connect(CE_DB,timeout=15); c.execute('PRAGMA busy_timeout=10000')
for col in ('description TEXT','transcript TEXT','transcript_hash TEXT','content_hash TEXT','extraction_version TEXT','processed_at TEXT'):
try: c.execute('ALTER TABLE ce_sources ADD COLUMN '+col)
except sqlite3.OperationalError: pass
c.execute("INSERT INTO ce_sources(source_system,source_video_id,youtube_id,title,channel,published_at,youtube_url,transcript_status,obsidian_artifact_path,imported_at,source_snapshot_hash,summary,description,transcript,transcript_hash,content_hash,extraction_version,processed_at) VALUES('youtube-research',?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT(source_system,source_video_id) DO UPDATE SET youtube_id=excluded.youtube_id,title=excluded.title,description=excluded.description,transcript=excluded.transcript,obsidian_artifact_path=excluded.obsidian_artifact_path,transcript_status=excluded.transcript_status,source_snapshot_hash=excluded.source_snapshot_hash,summary=excluded.summary,transcript_hash=excluded.transcript_hash,content_hash=excluded.content_hash,extraction_version=excluded.extraction_version,processed_at=excluded.processed_at",(row['id'],row['youtube_id'],row['title'],row['channel'],row['published_at'],f'https://www.youtube.com/watch?v={row["youtube_id"]}','complete',path,now(),ch,row['summary'] or '',row['description'] or '',row['transcript'] or '',th,ch,VERSION,now()))
c.commit(); c.close()
def validate(c,job,payload,path,th,ch):
p=Path(path); ok=p.exists() and row_id(p,job['youtube_id'])
ce=sqlite3.connect(CE_DB); x=ce.execute('SELECT obsidian_artifact_path,transcript_status FROM ce_sources WHERE source_system=? AND source_video_id=?',('youtube-research',job['video_id'])).fetchone(); ce.close()
return ok and x and x[0]==path and x[1]=='complete'
def row_id(p,y): return bool(re.search(rf'(?im)^youtube_id:\s*{re.escape(y)}\s*$',p.read_text(errors='ignore')))
def process(job):
c=conn(); row=c.execute('SELECT * FROM videos WHERE id=?',(job['video_id'],)).fetchone()
try:
segs,native_meta=fetch_segments(row['youtube_id']); text=' '.join(s[3] for s in segs)
if len(text)<500 or len(segs)<2: raise ValueError('technically_unusable')
classification,metrics=content_classify(segs,text)
if classification != 'informational': raise ContentReject(classification,metrics)
th=sha(text); payload=extract(row,segs); raw=json.dumps(payload,ensure_ascii=False,sort_keys=True); ch=sha(raw)
c.execute('BEGIN'); c.execute('DELETE FROM transcript_segments WHERE video_id=?',(row['id'],))
c.executemany('INSERT INTO transcript_segments(video_id,start_seconds,text) VALUES(?,?,?)',[(row['id'],round(s[1]),s[3]) for s in segs]); c.execute("UPDATE videos SET transcript=?,transcript_hash=?,transcript_status='done',transcript_source='supadata_native',transcript_updated_at=? WHERE id=?",(text,th,now(),row['id']))
row=dict(row); row['transcript']=text
from summary_generator import generate_summary
from summary_lifecycle import is_stale_fallback
if is_stale_fallback(row['summary']):
fresh=generate_summary(row['title'] or row['youtube_id'], row['description'] or '', text)
if fresh: c.execute("UPDATE videos SET summary=?,updated_at=CURRENT_TIMESTAMP WHERE id=?",(fresh,row['id'])); row['summary']=fresh
c.execute('COMMIT')
path=write_obsidian(row,payload,th,ch); sync_ce(row,path,th,ch); register_graphiti(path,row['youtube_id'])
c.execute("INSERT OR REPLACE INTO e2e_extractions(video_id,youtube_id,extraction_version,transcript_hash,content_hash,payload_json,obsidian_path,processed_at) VALUES(?,?,?,?,?,?,?,?)",(row['id'],row['youtube_id'],VERSION,th,ch,raw,path,now()))
c.execute("UPDATE e2e_jobs SET status='validation_pending',step='validation',updated_at=? WHERE job_id=?",(now(),job['job_id']))
import subprocess
if subprocess.run(['/opt/struktur/youtube-research/venv/bin/python','/opt/struktur/youtube-research/e2e_validator.py',str(job['job_id'])],timeout=30).returncode != 0: raise RuntimeError('validation_failed')
c.execute("UPDATE e2e_jobs SET status='complete',step='complete',completed_at=?,updated_at=?,lease_until=NULL WHERE job_id=?",(now(),now(),job['job_id']))
except ContentReject as ex:
c.rollback()
reason=ex.classification
c.execute(
"INSERT OR REPLACE INTO e2e_rejections"
"(youtube_id,url,reason,classification,rejection_reason,metrics_json,validator_version) "
"VALUES(?,?,?,?,?,?,?)",
(
row['youtube_id'],
f'https://www.youtube.com/watch?v={row["youtube_id"]}',
reason,reason,reason,
json.dumps(ex.metrics,sort_keys=True),
VALIDATOR_VERSION
)
)
# WICHTIG: Der videos-Masterdatensatz wird NIEMALS automatisch gelöscht.
c.execute(
"UPDATE e2e_jobs SET "
"status='rejected_no_usable_transcript',step='rejected',"
"completed_at=?,updated_at=?,lease_until=NULL,"
"error_code=?,error_message=? WHERE job_id=?",
(now(),now(),reason,json.dumps(ex.metrics,sort_keys=True),job['job_id'])
)
except NativeTranscriptUnavailable as ex:
c.rollback()
c.execute("UPDATE videos SET transcript_status='unavailable',transcript_source=NULL,transcript_updated_at=? WHERE id=? AND (transcript IS NULL OR trim(transcript)='')",(now(),row['id']))
c.execute("UPDATE e2e_jobs SET status='supadata_native_unavailable',step='supadata_terminal',completed_at=?,updated_at=?,lease_until=NULL,error_code='native_unavailable',error_message=? WHERE job_id=?",(now(),now(),str(ex)[:500],job['job_id']))
c.commit()
except QuotaWait as ex:
c.rollback()
retry=(datetime.now(timezone.utc)+timedelta(minutes=15)).isoformat()
c.execute("UPDATE e2e_jobs SET status='supadata_quota_wait',step='quota_wait',updated_at=?,local_retry_after=?,lease_until=NULL,error_code='quota_wait',error_message=? WHERE job_id=?",(now(),retry,str(ex)[:200],job['job_id']))
c.commit()
except TransientSupadataError as ex:
c.rollback()
retry=(datetime.now(timezone.utc)+timedelta(minutes=5)).isoformat()
c.execute("UPDATE e2e_jobs SET status='supadata_retry',step='supadata_retry',updated_at=?,local_retry_after=?,lease_until=NULL,error_code=?,error_message=? WHERE job_id=?",(now(),retry,f'supadata_http_{ex.status}' if ex.status else 'supadata_transient_error',str(ex)[:500],job['job_id']))
c.commit()
except TranscriptFetchError as ex:
c.rollback()
# Technische/API/IP-/HTTP-Fehler sind KEIN Beweis für fehlendes Transkript.
# Nach drei Versuchen bleibt das Video erhalten und der Job wird blockiert
# statt verworfen oder als inhaltlich abgelehnt markiert.
c.execute(
"UPDATE e2e_jobs SET "
"status=CASE WHEN attempts>=3 THEN 'blocked_transcript_fetch' ELSE 'queued' END,"
"step='transcript_error',updated_at=?,lease_until=NULL,"
"error_code='transcript_fetch_error',error_message=? "
"WHERE job_id=?",
(now(),str(ex)[:1000],job['job_id'])
)
except ValueError as ex:
c.rollback()
reason=str(ex)
# Z.B. technisch zu kurzer, aber tatsächlich abgerufener Inhalt.
# Auch hier darf niemals der Masterdatensatz gelöscht werden.
c.execute(
"INSERT OR REPLACE INTO e2e_rejections"
"(youtube_id,url,reason,rejection_reason,validator_version) "
"VALUES(?,?,?,?,?)",
(
row['youtube_id'],
f'https://www.youtube.com/watch?v={row["youtube_id"]}',
reason,reason,VALIDATOR_VERSION
)
)
c.execute(
"UPDATE e2e_jobs SET "
"status='rejected_no_usable_transcript',step='rejected',"
"completed_at=?,updated_at=?,lease_until=NULL,"
"error_code=?,error_message=? WHERE job_id=?",
(now(),now(),reason,reason,job['job_id'])
)
except RuntimeError as ex:
c.rollback()
c.execute(
"UPDATE e2e_jobs SET "
"status=CASE WHEN attempts>=3 THEN 'discarded_after_processing_error' ELSE 'queued' END,"
"step='error',updated_at=?,lease_until=NULL,"
"error_code='processing_error',error_message=? WHERE job_id=?",
(now(),str(ex)[:1000],job['job_id'])
)
return
except Exception as ex:
c.rollback(); c.execute("UPDATE e2e_jobs SET status=CASE WHEN attempts>=3 THEN 'discarded_after_processing_error' ELSE 'queued' END,step='error',updated_at=?,lease_until=NULL,error_code='processing_error',error_message=? WHERE job_id=?",(now(),str(ex)[:1000],job['job_id'])); LOG.exception('job failed')
finally: c.close()
def main():
logging.basicConfig(level=logging.INFO); ensure_schema(); LOG.info('worker started')
while True:
c = conn()
try:
j = reserve(c)
finally:
c.close()
if j: process(j)
else: time.sleep(2)
if __name__=='__main__': main()