#!/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/018-Hon-Uoc-Khong-Duoc-Phep-Yeu');PROJECT=ROOT.name;RUN='run-20260721T040505Z-cd9f944a';OWNER='zoro'
CANON='d12e1733f81bf173b5db789b0028a0be478d0d2e565c05a91e236c95cd33bd78';AUDIO='1f78d70445770dea08b5690880a53bdff83cdeb7ca0e735f93510ad6edd0de79';LAYOUT='2405027103a759b350a6c3a735ea754b3c5d5831d333a0c19e1f4fb48dfc6103'
BASE='http://192.168.1.104:8022';INTENT=ROOT/'log/footage-intent.json';RECEIPT=ROOT/'log/footage-render.json';OUT=ROOT/'output/footage/footage.mp4'
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:
  for b in iter(lambda:f.read(8*1024*1024),b''):h.update(b)
 return h.hexdigest()
def atomic(p,v):p.parent.mkdir(parents=True,exist_ok=True);q=p.with_suffix(p.suffix+'.tmp');q.write_text(json.dumps(v,ensure_ascii=False,indent=2)+'\n');os.replace(q,p)
def get(path):
 with urllib.request.urlopen(BASE+path,timeout=20) as r:return json.load(r)
def guard():
 x=json.loads((ROOT/'.ownership-lock.json').read_text());assert (x['project_id'],x['run_id'],x['owner'],x['status'])==(PROJECT,RUN,OWNER,'main_session_pipeline');assert sha(ROOT/'story/story-canon.txt')==CANON and sha(ROOT/'audio/story-full.wav')==AUDIO and sha(ROOT/'image/normalized/layout-1920x1080.png')==LAYOUT
def probe(p):return json.loads(subprocess.check_output(['ffprobe','-v','error','-show_entries','format=duration:stream=codec_type,codec_name,width,height,sample_rate,channels,r_frame_rate','-of','json',str(p)],text=True))
def main():
 guard();assert not INTENT.exists() and not RECEIPT.exists() and not OUT.exists();start=time.time()
 while True:
  h=get('/health');assert h['status']=='ok' and h['h264_nvenc'] is True and h['default_footage_dir']=='/data/video-pipeline/GacMaiAudio/footage' and h['output_policy']=='request_path_under_video_root'
  if h.get('running',[])==[]:break
  if time.time()-start>14400:raise RuntimeError('worker remained busy for four hours; no job submitted')
  time.sleep(20)
 guard();payload={'layout':str(ROOT/'image/normalized/layout-1920x1080.png'),'footage_dir':'/data/video-pipeline/GacMaiAudio/footage','pattern':'*.mp4','recursive':True,'audio':str(ROOT/'audio/story-full.wav'),'output':str(OUT),'seed':20260721,'overwrite':False,'width':1920,'height':1080,'fps':30,'center_x':656,'center_y':0,'center_width':608,'center_height':1080,'video_encoder':'h264_nvenc','preset':'p4','cq':21}
 intent={'schema_version':1,'project_id':PROJECT,'run_id':RUN,'status':'submitting','source_canon_sha256':CANON,'source_audio_sha256':AUDIO,'layout_sha256':LAYOUT,'horizontal_mirror':'hflip','source_audio_discarded_required':True,'request':payload,'created_at':now()};atomic(INTENT,intent)
 req=urllib.request.Request(BASE+'/v1/footage-render-huyenan/jobs',data=json.dumps(payload).encode(),headers={'Content-Type':'application/json'},method='POST')
 with urllib.request.urlopen(req,timeout=60) as r:created=json.load(r)
 jid=created.get('job_id') or created.get('id');assert jid;intent.update({'status':'submitted','job_id':jid,'submitted_at':now()});atomic(INTENT,intent)
 while True:
  job=get('/v1/footage-render-huyenan/jobs/'+jid);st=job.get('status')
  if st in ('completed','failed','cancelled'):break
  time.sleep(10)
 if st!='completed':atomic(RECEIPT,{'schema_version':1,'project_id':PROJECT,'run_id':RUN,'status':st,'verified':False,'job_id':jid,'job':job,'failed_at':now()});raise RuntimeError('footage job terminal '+str(st))
 assert job.get('verification',{}).get('verified') is True and Path(job['output']).resolve()==OUT.resolve() and OUT.exists();pr=probe(OUT);v=[x for x in pr['streams'] if x['codec_type']=='video'];a=[x for x in pr['streams'] if x['codec_type']=='audio'];d=float(pr['format']['duration']);assert len(v)==1 and len(a)==1 and v[0]['width']==1920 and v[0]['height']==1080 and v[0]['r_frame_rate']=='30/1' and a[0]['sample_rate']=='48000' and abs(d-2446.47)<0.05
 verification=job.get('verification',{});mirror=verification.get('mirror_applied',job.get('mirror_applied'));discarded=verification.get('source_audio_discarded',job.get('source_audio_discarded'));assert mirror is True and discarded is True
 receipt={'schema_version':1,'project_id':PROJECT,'run_id':RUN,'status':'completed','verified':True,'job_id':jid,'source_canon_sha256':CANON,'source_audio_sha256':AUDIO,'layout_sha256':LAYOUT,'horizontal_mirror':'hflip','mirror_applied':True,'source_audio_discarded':True,'output_path':str(OUT),'output_sha256':sha(OUT),'output_bytes':OUT.stat().st_size,'output_duration':d,'probe':pr,'job':job,'hash_method':'sha256_stream_8MiB','completed_at':now()};atomic(RECEIPT,receipt);print(json.dumps({'status':'completed','job_id':jid,'output_sha256':receipt['output_sha256'],'bytes':receipt['output_bytes']}))
if __name__=='__main__':main()