Files
dj/integration-work/historical_cleanup_snapshot_check.py
2026-11-13 11:46:25 +08:00

13 lines
848 B
Python

import pathlib,json,subprocess
root=pathlib.Path('/home/ps/huobao-video-cleanup');out=root/'historical-20261112'
approved=json.loads((root/'historical-approved.json').read_text());target={a['manifest']['prompt'] for a in approved}
names=json.loads((out/'services-before.json').read_text());result=[]
for i in json.loads(subprocess.check_output(['docker','inspect',*[n.lstrip('/') for n in names]],text=True)):
env=dict(x.split('=',1) for x in i['Config'].get('Env',[]) if '=' in x);dest=env.get('H3_HISTORY_SNAPSHOT')
if not dest:continue
mount=next((m for m in i['Mounts'] if m['Destination']==dest),None)
if not mount:continue
snapshot=json.loads(pathlib.Path(mount['Source']).read_text());overlap=sorted(set(snapshot.get('history',{}))&target)
result.append({'worker':i['Name'],'matchingCleanedPrompts':overlap})
print(json.dumps(result))