58 lines
4.3 KiB
Python
58 lines
4.3 KiB
Python
"""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)}))
|