import sqlite3,json,glob,os,datetime,hashlib,re
from pathlib import Path
DB='/opt/struktur/youtube-research/knowledge.db'; BASE='/opt/struktur/youtube-research/reports/p29'; SCRIPT='/opt/struktur/youtube-research/supadata_native_batch.py'
c=sqlite3.connect('file:'+DB+'?mode=ro',uri=True); c.row_factory=sqlite3.Row
src=Path(SCRIPT).read_text(errors='ignore')
# Current exact candidate query/function source lines
source_lines=[l for l in src.splitlines() if 'def candidates' in l or 'SELECT e.job_id' in l or "e.status IN" in l or "coalesce(v.transcript_status" in l or "transcript_source" in l]
def row(d):
 d=dict(d); d['transcript_present']=bool((d.get('transcript') or '').strip()); d['transcript_len']=len(d.get('transcript') or ''); return d
blocked=[row(r) for r in c.execute("SELECT e.job_id,e.video_id,e.youtube_id,e.status e2e_status,e.attempts,e.error_code,e.error_message,e.updated_at,e.reserved_at,e.started_at,e.completed_at,v.title,v.transcript_status,v.transcript,v.transcript_source,v.youtube_id v_youtube_id FROM e2e_jobs e JOIN videos v ON v.id=e.video_id WHERE e.status='blocked_transcript_fetch' ORDER BY e.job_id")]
retry=[row(r) for r in c.execute("SELECT v.id video_id,v.youtube_id,v.title,v.transcript_status,v.transcript,v.transcript_source,v.updated_at video_updated_at,e.job_id,e.status e2e_status,e.attempts,e.error_code,e.error_message,e.updated_at e_updated_at FROM videos v LEFT JOIN e2e_jobs e ON e.video_id=v.id AND e.job_id=(SELECT e2.job_id FROM e2e_jobs e2 WHERE e2.video_id=v.id ORDER BY COALESCE(e2.updated_at,'') DESC,e2.job_id DESC LIMIT 1) WHERE v.transcript_status='retry' ORDER BY v.id")]
cands=[dict(r) for r in c.execute("SELECT e.job_id,e.video_id,e.youtube_id,e.status e2e_status,v.transcript_status,v.transcript_source,length(trim(coalesce(v.transcript,''))) transcript_len FROM e2e_jobs e JOIN videos v ON v.id=e.video_id WHERE v.youtube_id IS NOT NULL AND length(trim(coalesce(v.transcript,'')))=0 AND coalesce(v.transcript_status,'')<>'done' AND coalesce(v.transcript_source,'') NOT IN ('supadata_native','windows_local') AND e.status IN ('blocked_transcript_fetch','local_transcript_retry','local_transcript_pending','supadata_retry')")]
def cb(d):
 if d['transcript_status']=='done' and d['transcript_present']: return 'A_stale_done_transcript_present'
 if d['transcript_status']=='partial' or (d['transcript_present'] and d['transcript_status']!='done'): return 'C_partial_or_present_non_done'
 if not d['v_youtube_id']: return 'D_missing_video_id'
 if d['transcript_source'] in ('supadata_native','windows_local'): return 'F_excluded_source'
 if not d['transcript_present']: return 'B_transcript_empty'
 return 'G_unclear'
def cr(d):
 if d['transcript_present']: return '6_retry_despite_transcript'
 s=d.get('e2e_status') or ''
 if s=='blocked_transcript_fetch': return '1_retry_blocked'
 if s=='local_transcript_pending': return '2_retry_local_pending'
 if s=='local_transcript_retry': return '3_retry_local_retry'
 if s=='supadata_retry': return '4_retry_supadata_retry'
 if s in ('rejected_no_usable_transcript','supadata_native_unavailable'): return '7_terminal'
 if d.get('job_id') is None: return '5_retry_without_e2e'
 return '8_other_unknown'
for d in blocked:d['classification']=cb(d)
for d in retry:d['classification']=cr(d)
errs=[]
for f in sorted(glob.glob(BASE+'/account2-batch-*-result.json')):
 for r in json.load(open(f)).get('results',[]):
  if r.get('result')=='error':
   x={'report':os.path.basename(f),**r}; x['later_records']=[]
   for g in sorted(glob.glob(BASE+'/account2-batch-*-result.json')):
    for z in json.load(open(g)).get('results',[]):
     if z.get('job_id')==r.get('job_id') and os.path.basename(g)!=os.path.basename(f): x['later_records'].append({'report':os.path.basename(g),**z})
   rr=c.execute("SELECT e.job_id,e.youtube_id,e.status,e.error_code,e.error_message,e.updated_at,v.transcript_status,v.transcript_source,length(v.transcript) transcript_len FROM e2e_jobs e JOIN videos v ON v.id=e.video_id WHERE e.job_id=?",(r['job_id'],)).fetchone(); x['current']=dict(rr) if rr else None; errs.append(x)
def counts(a,k):
 keys=sorted(set(x.get(k) for x in a)); return {str(v):sum(x.get(k)==v for x in a) for v in keys}
blocked_ids={x['video_id'] for x in blocked}; retry_ids={x['video_id'] for x in retry}; union=blocked_ids|retry_ids
empty={x['video_id'] for x in blocked+retry if not x['transcript_present']}; partial={x['video_id'] for x in blocked+retry if x['transcript_status']=='partial'}; stale={x['video_id'] for x in blocked+retry if x['transcript_present']}
queue={s:c.execute('SELECT COUNT(*) FROM graphiti_import_queue WHERE graphiti_status=?',(s,)).fetchone()[0] for s in ('done','processing','retry_wait','failed_permanent','queued')}; queue['total']=c.execute('SELECT COUNT(*) FROM graphiti_import_queue').fetchone()[0]
out={'checked_at_utc':datetime.datetime.now(datetime.timezone.utc).isoformat(),'source_rule_lines':source_lines,'live':{'videos_total':c.execute('SELECT COUNT(*) FROM videos').fetchone()[0],'video_status':{s:c.execute('SELECT COUNT(*) FROM videos WHERE transcript_status=?',(s,)).fetchone()[0] for s in ('done','retry','partial','none')},'e2e_status':{r['status']:r['n'] for r in c.execute('SELECT status,COUNT(*) n FROM e2e_jobs GROUP BY status')},'transcript_source':{str(r['transcript_source']):r['n'] for r in c.execute('SELECT transcript_source,COUNT(*) n FROM videos GROUP BY transcript_source')},'blocked_count':len(blocked),'retry_count':len(retry),'candidate_count':len(cands),'candidate_ids':cands,'queue':queue,'queue_dupe_import':c.execute('SELECT COUNT(*) FROM (SELECT import_identity FROM graphiti_import_queue GROUP BY import_identity HAVING COUNT(*)>1)').fetchone()[0],'queue_dupe_record':c.execute('SELECT COUNT(*) FROM (SELECT reconciliation_record_id FROM graphiti_import_queue GROUP BY reconciliation_record_id HAVING COUNT(*)>1)').fetchone()[0],'release_nonzero':c.execute('SELECT COUNT(*) FROM graphiti_import_queue WHERE COALESCE(graphiti_release,0)<>0').fetchone()[0]},'blocked':blocked,'blocked_class_counts':counts(blocked,'classification'),'retry':retry,'retry_class_counts':counts(retry,'classification'),'unique_sets':{'blocked':len(blocked_ids),'retry':len(retry_ids),'union':len(union),'empty':len(empty),'partial':len(partial),'stale_present':len(stale)},'timeouts':errs}
print(json.dumps(out,ensure_ascii=False,default=str))
c.close()