#!/usr/bin/env python3
from __future__ import annotations
import datetime as dt, hashlib, json, os, subprocess, time, urllib.parse, urllib.request
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'
ENDPOINT='http://192.168.40.32:7861/voice/ngoc-huyen-clone'
NARR=ROOT/'story/spoken-narration.txt'; CHUNKS=ROOT/'audio/chunks'; ATTEMPTS=ROOT/'log/tts-attempts'
HOT=ROOT/'work/tts/hot-manifest.json'; FINAL=ROOT/'script/tts-manifest.json'; OUTPUT=ROOT/'audio/story-full.wav'
CFG={'style':'doc_truyen','speed':'1.0','denoise':'true'}
def now(): return dt.datetime.now(dt.timezone.utc).isoformat().replace('+00:00','Z')
def sha(p:Path):
 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:Path,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',encoding='utf-8');os.replace(q,p)
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(NARR)==CANON
def probe(p:Path):
 x=json.loads(subprocess.check_output(['ffprobe','-v','error','-show_entries','format=duration:stream=codec_type,codec_name,sample_rate,channels','-of','json',str(p)],text=True));d=float(x['format']['duration']);a=[s for s in x['streams'] if s['codec_type']=='audio'];assert d>0 and len(a)==1;return {'duration':d,'stream':a[0]}
def plan(text:str,max_chars=2600):
 paras=[p.strip() for p in text.split('\n\n') if p.strip()];out=[];cur=''
 for p in paras:
  if len(p)>max_chars:
   sentences=[];buf=''
   for token in p.split(' '):
    if len(buf)+len(token)+1>max_chars: sentences.append(buf);buf=token
    else: buf=(buf+' '+token).strip()
   if buf: sentences.append(buf)
  else: sentences=[p]
  for unit in sentences:
   test=(cur+'\n\n'+unit).strip()
   if cur and len(test)>max_chars: out.append(cur);cur=unit
   else: cur=test
 if cur: out.append(cur)
 return out
def main():
 guard();CHUNKS.mkdir(parents=True,exist_ok=True);ATTEMPTS.mkdir(parents=True,exist_ok=True)
 text=NARR.read_text(encoding='utf-8');chunks=plan(text);assert ''.join(''.join(x.split()) for x in chunks)==''.join(text.split())
 hot=json.loads(HOT.read_text()) if HOT.exists() else {'schema_version':1,'project_id':PROJECT,'run_id':RUN,'status':'running','source_canon_sha256':CANON,'spoken_narration_sha256':CANON,'request_config':CFG,'segment_count':len(chunks),'segments':[],'created_at':now()}
 assert hot['source_canon_sha256']==CANON and hot['segment_count']==len(chunks) and hot['request_config']==CFG
 done={x['index']:x for x in hot.get('segments',[])}
 verified=[]
 for i,text_chunk in enumerate(chunks,1):
  guard();th=hashlib.sha256(text_chunk.encode()).hexdigest();p=CHUNKS/f'segment-{i:04d}.wav'
  old=done.get(i)
  if old and old.get('text_sha256')==th and p.exists():
   pr=probe(p);ah=sha(p)
   if old.get('audio_sha256')==ah and abs(old.get('duration',0)-pr['duration'])<0.02: verified.append(old);continue
  attempt={'schema_version':1,'project_id':PROJECT,'run_id':RUN,'attempt_id':f'tts-{i:04d}-{int(time.time())}','status':'submitting','index':i,'source_canon_sha256':CANON,'spoken_narration_sha256':CANON,'text_sha256':th,'request_config':CFG,'output_path':str(p),'created_at':now()};atomic(ATTEMPTS/f'segment-{i:04d}.json',attempt)
  data=urllib.parse.urlencode({'text':text_chunk,**CFG}).encode();req=urllib.request.Request(ENDPOINT,data=data,headers={'Content-Type':'application/x-www-form-urlencoded'},method='POST');tmp=p.with_suffix('.wav.part')
  try:
   with urllib.request.urlopen(req,timeout=1200) as r, tmp.open('wb') as f:
    assert r.status==200
    while True:
     b=r.read(1024*1024)
     if not b:break
     f.write(b)
   os.replace(tmp,p);pr=probe(p);seg={'index':i,'text_sha256':th,'output_path':str(p),'duration':pr['duration'],'bytes':p.stat().st_size,'audio_sha256':sha(p),'probe':pr,'status':'completed','verified':True,'completed_at':now()};verified.append(seg);attempt.update({'status':'completed','audio_sha256':seg['audio_sha256'],'duration':seg['duration'],'bytes':seg['bytes'],'completed_at':now()});atomic(ATTEMPTS/f'segment-{i:04d}.json',attempt)
   hot.update({'status':'running','segments':verified,'updated_at':now()});atomic(HOT,hot)
  except Exception as e:
   if tmp.exists():tmp.unlink()
   attempt.update({'status':'failed','error_type':type(e).__name__,'error':str(e)[:500],'failed_at':now()});atomic(ATTEMPTS/f'segment-{i:04d}.json',attempt);hot.update({'status':'failed','verified':False,'failed_segment_index':i,'error_type':type(e).__name__,'updated_at':now()});atomic(HOT,hot);raise
 assert len(verified)==len(chunks)
 concat=ROOT/'work/tts/concat.txt';concat.write_text(''.join("file '"+str(Path(x['output_path'])).replace("'","'\\''")+"'\n" for x in verified),encoding='utf-8')
 tmp=OUTPUT.with_name('story-full.tmp.wav');subprocess.run(['ffmpeg','-y','-v','error','-f','concat','-safe','0','-i',str(concat),'-c','copy',str(tmp)],check=True);os.replace(tmp,OUTPUT);pr=probe(OUTPUT)
 final={'schema_version':1,'project_id':PROJECT,'run_id':RUN,'status':'completed','verified':True,'source_canon_sha256':CANON,'spoken_narration_sha256':CANON,'tts_words_per_minute':231,'request_config':CFG,'segment_count':len(verified),'segments':verified,'output':str(OUTPUT),'output_duration':pr['duration'],'output_bytes':OUTPUT.stat().st_size,'output_sha256':sha(OUTPUT),'output_probe':pr,'hash_method':'sha256_stream_8MiB','voice_test':'not_run_forbidden','completed_at':now()};atomic(FINAL,final);hot.update({'status':'completed','verified':True,'segments':verified,'output':str(OUTPUT),'output_sha256':final['output_sha256'],'completed_at':now()});atomic(HOT,hot);print(json.dumps({'status':'completed','segments':len(verified),'duration':pr['duration'],'bytes':OUTPUT.stat().st_size,'sha256':final['output_sha256']}))
if __name__=='__main__':main()