#!/usr/bin/env python3 from __future__ import annotations import datetime as dt,hashlib,json,os,time,urllib.request,subprocess from pathlib import Path ROOT=Path('/data/video-pipeline/HaTramAudio/project/016-Nguoi-Giu-Am-Thanh-Cuoi-Cung');RUN='run-20260720T233725Z-24504f17';CANON='fae6fce617a14dc4f8b3bd52968cc688a456c45a60bc2ed7701ccd8164875e02';BASE='http://192.168.1.104:8024' INPUT=ROOT/'output/final.mp4';OUTPUT=ROOT/'output/final-upload.mp4';INTENT=ROOT/'log/transcode-intent.json';RECEIPT=ROOT/'log/transcode-upload.json' def now():return dt.datetime.now(dt.timezone.utc).isoformat().replace('+00:00','Z') def sha(p): h=hashlib.sha256() with p.open('rb') as f: while b:=f.read(8*1024*1024):h.update(b) return h.hexdigest() def atomic(p,o):p.parent.mkdir(parents=True,exist_ok=True);t=p.with_suffix(p.suffix+'.tmp');t.write_text(json.dumps(o,ensure_ascii=False,indent=2)+'\n');os.replace(t,p) def get(path): with urllib.request.urlopen(BASE+path,timeout=30) as r:return json.load(r) def post(path,obj): q=urllib.request.Request(BASE+path,data=json.dumps(obj).encode(),headers={'Content-Type':'application/json'},method='POST') with urllib.request.urlopen(q,timeout=60) as r:return json.load(r) def probe(p): q=subprocess.run(['ffprobe','-v','error','-show_entries','format=duration,size:stream=codec_type,codec_name,width,height,r_frame_rate,sample_rate,channels','-of','json',str(p)],capture_output=True,text=True,check=True);return json.loads(q.stdout) def guard(): l=json.loads((ROOT/'.ownership-lock.json').read_text());assert (l['project_id'],l['run_id'],l['owner'],l['status'])==(ROOT.name,RUN,'zoro','main_session_pipeline');assert sha(ROOT/'story/story-canon.txt')==CANON f=json.loads((ROOT/'log/final-render.json').read_text());assert f['status']=='completed' and f['verified'] and f['visual_qa']['verified'] and f['output_sha256']==sha(INPUT) def main(): guard();payload={'input_video_path':str(INPUT),'output_video_path':str(OUTPUT),'target_output_bytes':800000000,'max_output_bytes':1000000000,'width':1920,'height':1080,'fps':30,'audio_bitrate_kbps':96,'overwrite':False};request_sha=hashlib.sha256(json.dumps(payload,sort_keys=True,separators=(',',':')).encode()).hexdigest();job=None if RECEIPT.exists(): r=json.loads(RECEIPT.read_text());assert r['request_sha256']==request_sha if r.get('status')=='completed' and r.get('verified') and OUTPUT.exists() and r.get('output_sha256')==sha(OUTPUT):print(json.dumps({'status':'completed_reused','job_id':r['job_id']}));return job=r.get('job_id') if not job: if OUTPUT.exists():raise RuntimeError('unreceipted upload copy exists') if INTENT.exists() and json.loads(INTENT.read_text()).get('status')=='submitting':raise RuntimeError('ambiguous transcode intent') h=get('/health');assert h.get('status')=='ok' and h.get('h264_nvenc') is True and not h.get('running') i={'schema_version':1,'project_id':ROOT.name,'run_id':RUN,'status':'submitting','verified':False,'request_sha256':request_sha,'payload':payload,'source_canon_sha256':CANON,'source_final_sha256':sha(INPUT),'created_at':now()};atomic(INTENT,i) response=post('/v1/huyenan-transcode-upload/jobs',payload);job=response.get('job_id') or response.get('id');assert job;i.update(status='submitted',job_id=job,updated_at=now());atomic(INTENT,i);atomic(RECEIPT,{**i,'submitted_response':response}) while True: guard();j=get('/v1/huyenan-transcode-upload/jobs/'+job);r=json.loads(RECEIPT.read_text());r.update(status=j.get('status',r['status']),last_response=j,updated_at=now());atomic(RECEIPT,r) if j.get('status') in ('failed','cancelled'):raise RuntimeError('transcode terminal failure') if j.get('status')=='completed': v=j.get('verification') or {};assert v.get('verified') is True and v.get('under_one_gb') is True and OUTPUT.exists() and 0