57 lines
4.0 KiB
Python
57 lines
4.0 KiB
Python
"""Read-only media verification and consistent DB backup before explicit historical cleanup."""
|
|
import concurrent.futures, contextlib, hashlib, json, os, pathlib, re, sqlite3, subprocess, time
|
|
from urllib.parse import urlsplit, unquote
|
|
ROOT=pathlib.Path(__file__).resolve().parent.parent
|
|
DATA=ROOT/'huobao-data'
|
|
OUT=ROOT/'integration-work/historical-video-cleanup-20261112'
|
|
OUT.mkdir(exist_ok=True)
|
|
FF=ROOT/'HuobaoDrama/resources/bin'
|
|
def digest(file):
|
|
h=hashlib.sha256()
|
|
with file.open('rb') as f:
|
|
for b in iter(lambda:f.read(1024*1024),b''):h.update(b)
|
|
return {'sha256':h.hexdigest(),'bytes':file.stat().st_size}
|
|
def verify(row):
|
|
result={k:row[k] for k in ('id','status','local_path','task_id','config_id','config_base_url','provider','model')}
|
|
try:
|
|
relative=row['local_path']
|
|
if not re.fullmatch(r'static/videos/[^/\\]+\.(mp4|webm)',relative,re.I):raise ValueError('Unsupported local path')
|
|
file=(DATA/relative).resolve(strict=True)
|
|
if not file.is_relative_to((DATA/'static/videos').resolve()):raise ValueError('Local path outside video directory')
|
|
before=digest(file)
|
|
flags={'creationflags':subprocess.CREATE_NO_WINDOW,'capture_output':True,'timeout':180}
|
|
p=subprocess.run([str(FF/'ffprobe.exe'),'-v','error','-show_format','-show_streams','-of','json',str(file)],**flags)
|
|
if p.returncode:raise ValueError('ffprobe failed')
|
|
info=json.loads(p.stdout);v=next((s for s in info['streams'] if s['codec_type']=='video'),{})
|
|
duration=float(info['format'].get('duration',0))
|
|
if not v.get('width') or not v.get('height') or duration<=0:raise ValueError('Invalid dimensions/duration')
|
|
p=subprocess.run([str(FF/'ffmpeg.exe'),'-nostdin','-v','error','-xerror','-threads','2','-i',str(file),'-map','0:v:0','-map','0:a?','-f','null','-'],**flags)
|
|
if p.returncode:raise ValueError('Full decode failed')
|
|
with file.open('r+b') as f:os.fsync(f.fileno())
|
|
if before!=digest(file):raise ValueError('Local video changed during verification')
|
|
result.update(localVerified=True,receipt=before,duration=duration,width=v['width'],height=v['height'])
|
|
if row['status']!='completed':raise ValueError('Not completed')
|
|
if row['provider'] not in ('minimax','localvideo'):raise ValueError('Unsupported provider')
|
|
base=urlsplit(row['config_base_url']);url=urlsplit(row['result_url'] or '')
|
|
if base.scheme!='http' or base.hostname!='192.168.110.111' or base.port not in (18792,18794) or base.username or base.password:raise ValueError('Unsupported bridge')
|
|
parts=url.path.split('/')
|
|
if (url.scheme,url.netloc)!=(base.scheme,base.netloc) or len(parts)<5 or parts[1]!='media' or unquote(parts[2])!=row['task_id'] or not parts[3].isdigit():raise ValueError('Source URL does not match task')
|
|
result['index']=int(parts[3]);result['eligible']=True
|
|
except Exception as e:result.update(eligible=False,reason=str(e) if isinstance(e,ValueError) else type(e).__name__)
|
|
return result
|
|
with contextlib.closing(sqlite3.connect((DATA/'huobao.sqlite3').as_uri()+'?mode=ro',uri=True)) as db:
|
|
db.row_factory=sqlite3.Row
|
|
backup=OUT/'before.sqlite3'
|
|
if backup.exists():raise RuntimeError('Refusing to replace initial backup')
|
|
with contextlib.closing(sqlite3.connect(backup)) as target:db.backup(target)
|
|
rows=[dict(r) for r in db.execute("select * from sys_task where type='video' and local_path is not null and local_path!='' order by id")]
|
|
media={p.relative_to(DATA).as_posix():digest(p) for p in (DATA/'static').rglob('*') if p.is_file()}
|
|
(OUT/'media-before.json').write_text(json.dumps(media,indent=2),'utf-8')
|
|
results=[]
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
|
|
for item in pool.map(verify,rows):
|
|
results.append(item)
|
|
(OUT/'inventory.json').write_text(json.dumps(results,indent=2),'utf-8')
|
|
print(json.dumps({'task':item['id'],'verified':item.get('localVerified',False),'eligible':item['eligible'],'reason':item.get('reason')}),flush=True)
|
|
print(json.dumps({'total':len(results),'verified':sum(r.get('localVerified',False) for r in results),'eligible':sum(r['eligible'] for r in results),'mediaSnapshot':len(media)}))
|