"""Enqueue only the user-authorized, locally fully decoded, preflighted historical videos.

Missing-history tasks require a unique byte-identical file on the original worker,
no live history, no references from any queued/running/other completed task.
The existing exact-file cleanup executor performs the actual unlink and history removal.
"""
import contextlib,importlib.util,json,pathlib,sqlite3,time,traceback
ROOT=pathlib.Path('/home/ps/huobao-video-cleanup');OUT=ROOT/'historical-20261112'
spec=importlib.util.spec_from_file_location('cleaner',ROOT/'comfy_video_cleanup.py');m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m)
bridges={b['name']:b for b in json.loads((ROOT/'config.json').read_text())['bridges']}
inventory={r['id']:r for r in json.loads((ROOT/'historical-inventory.json').read_text())}
probes=json.loads((OUT/'probe-results.json').read_text())+json.loads((OUT/'orphan-probe-results.json').read_text())
probes={r['task']:r for r in probes if r['verified']}
approved=json.loads((ROOT/'historical-approved.json').read_text())
assert len({r['task'] for r in approved})==len(approved)
assert (OUT/'probe-done.json').exists() and (OUT/'orphan-probe-done.json').exists()
# Consistent receipts/journal backup before adding any historical acknowledgements.
for b in bridges.values():
 backup=OUT/(b['name']+'-before-enqueue.sqlite3')
 if not backup.exists():
  with contextlib.closing(sqlite3.connect('file:'+b['database']+'?mode=ro',uri=True)) as db,contextlib.closing(sqlite3.connect(backup)) as target:db.backup(target)
results=[]
for a in approved:
 out={'task':a['task'],'id':a['id'],'enqueued':False}
 try:
  r=inventory[a['task']];p=probes[a['task']];b=bridges[a['bridge']]
  assert r['eligible'] and r['localVerified'] and r['status']=='completed'
  assert (a['id'],a['receipt'],a['index'])==(r['task_id'],r['receipt'],r['index'])
  assert all(a[k]==p[k] for k in ('id','bridge','manifest','worker','fileReceipt')),'Preflight manifest mismatch'
  assert b['workers'][a['manifest']['provider']]==a['worker'],'Worker changed'
  with contextlib.closing(sqlite3.connect(b['database'],timeout=15)) as db:
   db.row_factory=sqlite3.Row
   assert db.execute('select 1 from jobs where id=?',(a['id'],)).fetchone(),'Unknown bridge job'
   existing=db.execute('select * from video_cleanup_acks where id=?',(a['id'],)).fetchone()
   if existing:
    assert (existing['output_index'],existing['sha256'],existing['bytes'])==(a['index'],a['receipt']['sha256'],a['receipt']['bytes']),'Conflicting receipt'
    assert existing['manifest'] and json.loads(existing['manifest'])['fileReceipt']==a['fileReceipt'],'Conflicting prepared file'
    out.update(enqueued=True,status=existing['status']);results.append(out);continue
   if a['method']=='live-history-and-download':
    fresh=m.docker_python(b['normalizer'],m.NORMALIZER_CODE,{'mode':'verify','id':a['id'],'index':a['index'],**a['receipt']})
    assert fresh==a['manifest'],'History changed after preflight'
   else:
    assert a['method']=='unique-byte-identical-orphan'
    assert {k:a['fileReceipt'][k] for k in ('sha256','bytes')}==a['receipt'],'Source is not identical to retained local video'
   guard=m.docker_python(b['normalizer'],m.NORMALIZER_CODE,{'mode':'guard','id':a['id'],'manifest':a['manifest']})
   if a['method']=='unique-byte-identical-orphan':assert guard['historyMissing'],'History reappeared'
   fresh=m.docker_python(a['worker'],m.WORKER_CODE,{'mode':'inspect','locator':a['manifest']['locator']})
   assert fresh==a['fileReceipt'],'Video changed since preflight'
   manifest={**a['manifest'],'worker':a['worker'],'fileReceipt':a['fileReceipt'],'historicalEvidence':{'method':a['method'],'localTask':a['task'],'localReceipt':a['receipt'],'fullDecodeVerified':True}}
   now=time.time()
   with db:db.execute("insert into video_cleanup_acks(id,output_index,sha256,bytes,status,manifest,created,updated) values(?,?,?,?,'prepared',?,?,?)",(a['id'],a['index'],a['receipt']['sha256'],a['receipt']['bytes'],json.dumps(manifest),now,now))
   out.update(enqueued=True,status='prepared')
 except Exception as e:out['reason']=str(e) if type(e) in (RuntimeError,AssertionError) else type(e).__name__
 results.append(out)
 (OUT/'enqueue-results.json').write_text(json.dumps(results,indent=2))
 print(json.dumps(out),flush=True)
(OUT/'enqueue-done.json').write_text(json.dumps({'total':len(results),'enqueued':sum(r['enqueued'] for r in results)}))