#!/usr/bin/env python3
import fcntl,json,sqlite3,subprocess,sys,time,re
from datetime import datetime,timezone
from pathlib import Path
DB=Path('/opt/struktur/youtube-research/knowledge.db')
WRITER='/opt/struktur/obsidian-graphiti-import/aa043/aa043_single_writer.py'
LOCK=Path('/run/graphiti-control/aa049-queue-worker.lock')
LOG=Path('/opt/struktur/reports/aa049-worker-events.jsonl')
def now(): return datetime.now(timezone.utc)
def due(v):
    if not v: return True
    try:
        x=datetime.fromisoformat(v.replace('Z','+00:00'))
        if x.tzinfo is None: x=x.replace(tzinfo=timezone.utc)
        return x<=now()
    except Exception: return False

def safe_historical_youtube(row):
    """Allow only identifiable, existing historical YouTube source documents."""
    iid=row['import_identity'] or ''
    path=row['obsidian_path'] or ''
    return (
        row['source_type']=='youtube' and row['video_id'] is None and
        row['graphiti_status']=='retry_wait' and
        (row['graphiti_release'] is None or int(row['graphiti_release'])==0) and
        bool(re.fullmatch(r'[0-9a-fA-F]{64}', iid)) and
        path.startswith('/opt/obsidian-vault/YouTube-Research/') and
        Path(path).is_file()
    )
def pick():
    c=sqlite3.connect(f'file:{DB}?mode=ro',uri=True); c.row_factory=sqlite3.Row
    rows=c.execute("select id,video_id,source_type,obsidian_path,import_identity,graphiti_status,next_attempt_at,graphiti_release from graphiti_import_queue where ((source_type='youtube' and (video_id is not null or (video_id is null and obsidian_path is not null and import_identity is not null))) or source_type='k-daten') and graphiti_status in ('queued','retry_wait') order by id").fetchall(); c.close()
    for r in rows:
        eligible = (r['source_type']=='k-daten' or r['video_id'] is not None or safe_historical_youtube(r))
        if eligible and (r['graphiti_release'] is None or int(r['graphiti_release'])==0) and due(r['next_attempt_at']): return dict(r)
    return None
def log(x):
    x={'ts_utc':now().isoformat(),**x}; LOG.parent.mkdir(parents=True,exist_ok=True); LOG.open('a').write(json.dumps(x,ensure_ascii=False)+'\n'); print(json.dumps(x,ensure_ascii=False),flush=True)
LOCK.parent.mkdir(parents=True,exist_ok=True)
with LOCK.open('w') as f:
    try: fcntl.flock(f,fcntl.LOCK_EX|fcntl.LOCK_NB)
    except BlockingIOError: log({'event':'SKIP_LOCK_HELD'}); sys.exit(0)
    row=pick()
    if not row: log({'event':'NO_DUE_YOUTUBE'}); sys.exit(0)
    log({'event':'START','queue_id':row['id'],'video_id':row['video_id'],'status':row['graphiti_status'],'next_attempt_at':row['next_attempt_at']})
    p=subprocess.run(['/opt/struktur/youtube-research/venv/bin/python',WRITER,'--queue-id',str(row['id'])],capture_output=True,text=True,timeout=1100)
    log({'event':'END','queue_id':row['id'],'returncode':p.returncode,'stdout':p.stdout[-2000:],'stderr':p.stderr[-2000:]})
    # A failure is source-local; the next timer invocation must remain available.
    sys.exit(0)