61 lines
4.2 KiB
Python
61 lines
4.2 KiB
Python
"""Read-only recovery of orphaned history links using unique byte-identical source videos."""
|
|
import base64,collections,hashlib,importlib.util,json,pathlib,sqlite3,subprocess
|
|
ROOT=pathlib.Path('/home/ps/huobao-video-cleanup');OUT=ROOT/'historical-20261112'
|
|
assert (OUT/'probe-done.json').exists()
|
|
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)
|
|
cfg=json.loads((ROOT/'config.json').read_text());bridges={b['name']:b for b in cfg['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())
|
|
pending=[r for r in probes if not r['verified'] and r.get('diagnostic',{}).get('historyPresent') is False]
|
|
SCAN=r'''
|
|
import hashlib,json,os,pathlib,stat,sys
|
|
q=json.load(sys.stdin);sizes=set(q['sizes']);matches=[];examined=0
|
|
root=pathlib.Path('/app/output')
|
|
for directory,dirs,files in os.walk(root,followlinks=False):
|
|
dirs[:]=[n for n in dirs if not (pathlib.Path(directory)/n).is_symlink()]
|
|
for name in files:
|
|
p=pathlib.Path(directory)/name
|
|
if p.suffix.lower() not in ('.mp4','.webm','.mov','.mkv') or p.is_symlink():continue
|
|
s=p.stat()
|
|
if not stat.S_ISREG(s.st_mode) or s.st_nlink!=1 or s.st_size not in sizes:continue
|
|
h=hashlib.sha256();examined+=1
|
|
with p.open('rb') as f:
|
|
for b in iter(lambda:f.read(1024*1024),b''):h.update(b)
|
|
after=p.stat()
|
|
if (s.st_size,s.st_mtime_ns,s.st_ino)!=(after.st_size,after.st_mtime_ns,after.st_ino):continue
|
|
if h.hexdigest() in q['hashes']:
|
|
sub=p.parent.relative_to(root).as_posix()
|
|
matches.append({'locator':{'filename':p.name,'subfolder':'' if sub=='.' else sub,'type':'output'},'sha256':h.hexdigest(),'bytes':s.st_size})
|
|
print('HB_CLEANUP_RESULT:'+json.dumps({'ok':True,'result':{'matches':matches,'examined':examined}}))
|
|
'''
|
|
byworker=collections.defaultdict(list)
|
|
for p in pending:
|
|
b=bridges[p['bridge']];w=b['workers'][p['diagnostic']['provider']];byworker[w].append(p)
|
|
scans={}
|
|
for worker,items in byworker.items():
|
|
receipts=[inventory[p['task']]['receipt'] for p in items]
|
|
scans[worker]=m.docker_python(worker,SCAN,{'sizes':list({r['bytes'] for r in receipts}),'hashes':list({r['sha256'] for r in receipts})})
|
|
(OUT/'orphan-source-matches.json').write_text(json.dumps(scans,indent=2))
|
|
results=[]
|
|
for p in pending:
|
|
r=inventory[p['task']];b=bridges[p['bridge']];provider=p['diagnostic']['provider'];worker=b['workers'][provider]
|
|
out={'task':p['task'],'id':p['id'],'bridge':p['bridge'],'verified':False,'method':'unique-byte-identical-orphan'}
|
|
try:
|
|
matches=[x for x in scans[worker]['matches'] if x['sha256']==r['receipt']['sha256'] and x['bytes']==r['receipt']['bytes']]
|
|
if len(matches)!=1:raise RuntimeError('No byte-identical source video' if not matches else 'Multiple byte-identical source videos; association ambiguous')
|
|
same=[x for x in pending if x['task']!=p['task'] and inventory[x['task']]['receipt']==r['receipt'] and x['id']!=p['id']]
|
|
if same:raise RuntimeError('Multiple local tasks share identical content; association ambiguous')
|
|
raw=base64.urlsafe_b64decode(p['id']+'='*(-len(p['id'])%4)).decode();decoded,uuid=raw.split(':',1)
|
|
if decoded!=provider:raise RuntimeError('Provider identity mismatch')
|
|
manifest={'provider':provider,'prompt':uuid,'locator':matches[0]['locator'],'historyHash':hashlib.sha256(b'{}').hexdigest()}
|
|
guard=m.docker_python(b['normalizer'],m.NORMALIZER_CODE,{'mode':'guard','id':p['id'],'manifest':manifest})
|
|
if not guard['historyMissing']:raise RuntimeError('History reappeared; require normal verification')
|
|
file=m.docker_python(worker,m.WORKER_CODE,{'mode':'inspect','locator':manifest['locator']})
|
|
if {k:file[k] for k in ('sha256','bytes')}!=r['receipt']:raise RuntimeError('Source changed after scan')
|
|
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__
|
|
results.append(out)
|
|
(OUT/'orphan-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')}),flush=True)
|
|
(OUT/'orphan-probe-done.json').write_text(json.dumps({'total':len(results),'verified':sum(r['verified'] for r in results)}))
|