"""Read-only preflight for explicitly requested historical-video cleanup."""
import importlib.util,json,pathlib,sqlite3,subprocess,time,traceback
ROOT=pathlib.Path('/home/ps/huobao-video-cleanup')
OUT=ROOT/'historical-20261112';OUT.mkdir(exist_ok=True)
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)
config=json.loads((ROOT/'config.json').read_text())
normal=next(b for b in config['bridges'] if 'vdn' not in b['name'])
vdn=next(b for b in config['bridges'] if 'vdn' in b['name'])
SNAPSHOT=r'''
import os,json,pathlib,hashlib
files={}
for folder in ('/app/input','/app/output'):
 for root,dirs,names in os.walk(folder,followlinks=False):
  for name in names:
   p=pathlib.Path(root)/name
   if p.suffix.lower() not in ('.png','.jpg','.jpeg','.webp','.bmp','.gif') or p.is_symlink():continue
   s=p.stat();files[str(p)]={'bytes':s.st_size,'mtime':s.st_mtime_ns}
print('HB_CLEANUP_RESULT:'+json.dumps({'ok':True,'result':files}))
'''
DIAG=r'''
import asyncio,sys,json,httpx,os,app
q=json.load(sys.stdin)
async def go():
 p,pid,e,outputs=await app.job_state(q['id'])
 async with httpx.AsyncClient(timeout=30) as c:
  r=await c.get('http://127.0.0.1:8105/v1/jobs/'+q['id'],headers={'Authorization':'Bearer '+os.getenv('COMFY_ADAPTER_TOKEN','')});r.raise_for_status();state=r.json()
 return {'provider':p.id,'historyPresent':bool(e),'historyStatus':e.get('status',{}).get('status_str'),'historyCompleted':e.get('status',{}).get('completed'),'outputCount':len(outputs),'adapterStatus':state.get('status')}
try:print('HB_CLEANUP_RESULT:'+json.dumps({'ok':True,'result':asyncio.run(go())}))
except Exception as e:print('HB_CLEANUP_RESULT:'+json.dumps({'ok':False,'error':type(e).__name__}))
'''
try:
 if not (OUT/'image-snapshot-before.json').exists():
  snapshots={w:m.docker_python(w,SNAPSHOT,{}) for w in sorted(set(normal['workers'].values())|set(vdn['workers'].values()))}
  (OUT/'image-snapshot-before.json').write_text(json.dumps(snapshots))
  containers=list(snapshots)+[normal['normalizer'],vdn['normalizer']]
  inspection=json.loads(subprocess.check_output(['docker','inspect',*containers],text=True))
  (OUT/'services-before.json').write_text(json.dumps({i['Name']:{'id':i['Id'],'started':i['State']['StartedAt']} for i in inspection}))
 results=[]
 for r in json.loads((ROOT/'historical-inventory.json').read_text()):
  if not r['eligible']:continue
  entry=vdn if r['config_base_url'].endswith(':18794') else normal
  out={'task':r['id'],'id':r['task_id'],'bridge':entry['name'],'verified':False}
  try:
   with sqlite3.connect('file:'+entry['database']+'?mode=ro',uri=True) as db:
    if not db.execute('select 1 from jobs where id=?',(r['task_id'],)).fetchone():raise RuntimeError('Unknown bridge video')
   manifest=m.docker_python(entry['normalizer'],m.NORMALIZER_CODE,{'mode':'verify','id':r['task_id'],'index':r['index'],**r['receipt']})
   m.docker_python(entry['normalizer'],m.NORMALIZER_CODE,{'mode':'guard','id':r['task_id'],'manifest':manifest})
   worker=entry['workers'][manifest['provider']]
   file=m.docker_python(worker,m.WORKER_CODE,{'mode':'inspect','locator':manifest['locator']})
   out.update(verified=True,manifest=manifest,worker=worker,fileReceipt=file)
  except Exception as e:
   out['reason']=str(e) if type(e) is RuntimeError else type(e).__name__
   try:out['diagnostic']=m.docker_python(entry['normalizer'],DIAG,{'id':r['task_id']})
   except Exception:pass
  results.append(out)
  (OUT/'probe-results.json').write_text(json.dumps(results,indent=2))
  print(json.dumps({k:v for k,v in out.items() if k in ('task','verified','reason','diagnostic')}),flush=True)
 (OUT/'probe-done.json').write_text(json.dumps({'total':len(results),'verified':sum(r['verified'] for r in results)}))
except Exception:
 (OUT/'probe-failed.txt').write_text(traceback.format_exc());raise