"""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)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()