#!/usr/bin/env python3 import importlib.util, json, re, shutil, time, urllib.request from pathlib import Path PROJECT = Path('/data/video-pipeline/Youtube-Video-Maker/Project/28052026-002-Tran-Quang-Dieu-Bui-Thi-Xuan') ASR_ENDPOINT = 'http://192.168.1.103:8001/transcribe-and-save' SCENES = PROJECT / 'script/scenes.json' AUDIO_DIR = PROJECT / 'audio/scenes' ASR_DIR = PROJECT / 'subtitles/scene_pronunciation_asr' BACKUP_DIR = PROJECT / 'audio/scene_backups' REPORT = PROJECT / 'logs/step07_pronunciation_qa_report.json' STEP7 = PROJECT / 'scripts_run_step_07_tts_v2.py' STEP8 = PROJECT / 'scripts_run_step_08_compose_tts.py' WATCH = { 'trần': ['tần'], 'nhưng': ['nưng'], 'đây': ['dây'], } def log(msg): print(time.strftime('%Y-%m-%d %H:%M:%S ') + msg, flush=True) def post_json(url, payload, timeout=1800): req = urllib.request.Request(url, data=json.dumps(payload, ensure_ascii=False).encode(), headers={'Content-Type':'application/json'}, method='POST') with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read().decode()) def norm(s): s = s.lower() s = re.sub(r'[^a-zà-ỹđ\s]', ' ', s) return re.sub(r'\s+', ' ', s).strip() def transcript_for(scene_id, audio_path): out = ASR_DIR / f'{scene_id}.json' payload = {'input_path': str(audio_path), 'output_path': str(out), 'language': 'vi', 'task': 'transcribe', 'word_timestamps': True} resp = post_json(ASR_ENDPOINT, payload, timeout=1800) if out.exists(): doc = json.loads(out.read_text(encoding='utf-8')) else: doc = resp out.write_text(json.dumps(doc, ensure_ascii=False, indent=2), encoding='utf-8') text = doc.get('text') or doc.get('transcript') or ' '.join(seg.get('text','') for seg in doc.get('segments',[]) or []) return text, str(out) def expected_hits(text): n = norm(text) return [w for w in WATCH if re.search(rf'\b{re.escape(w)}\b', n)] def flag_scene(narration, transcript): exp = expected_hits(narration) nt = norm(transcript) flags = [] for w in exp: bads = WATCH[w] good_count = len(re.findall(rf'\b{re.escape(w)}\b', nt)) src_count = len(re.findall(rf'\b{re.escape(w)}\b', norm(narration))) bad_found = [b for b in bads if re.search(rf'\b{re.escape(b)}\b', nt)] if bad_found or good_count < src_count: flags.append({'word': w, 'src_count': src_count, 'asr_good_count': good_count, 'bad_found': bad_found}) return flags def cased_payload(tts_text): # Keep display/subtitle text untouched; only nudge OmniVoice with case on risky words. repl = {'trần':'Trần', 'nhưng':'Nhưng', 'đây':'Đây'} out = tts_text for k,v in repl.items(): out = re.sub(rf'\b{re.escape(k)}\b', v, out, flags=re.I) return out def main(): ASR_DIR.mkdir(parents=True, exist_ok=True); BACKUP_DIR.mkdir(parents=True, exist_ok=True); REPORT.parent.mkdir(parents=True, exist_ok=True) spec = importlib.util.spec_from_file_location('step7', STEP7) mod = importlib.util.module_from_spec(spec); spec.loader.exec_module(mod) scenes = json.loads(SCENES.read_text(encoding='utf-8'))['scenes'] candidates = [s for s in scenes if expected_hits(s['narration'])] log(f'QA candidates={len(candidates)}') report = {'project_id': PROJECT.name, 'watchlist': WATCH, 'candidates': len(candidates), 'items': [], 'regenerated': []} flagged = [] for i,s in enumerate(candidates,1): sid = s['scene_id']; audio = AUDIO_DIR/f'{sid}.wav' if not audio.exists(): report['items'].append({'scene_id': sid, 'status': 'missing_audio'}) continue log(f'ASR {i}/{len(candidates)} {sid}') try: tr, asr_path = transcript_for(sid, audio) flags = flag_scene(s['narration'], tr) item = {'scene_id': sid, 'status': 'flagged' if flags else 'ok', 'expected': expected_hits(s['narration']), 'flags': flags, 'asr_text': tr, 'asr_path': asr_path} report['items'].append(item) if flags: flagged.append((s,item)) except Exception as e: report['items'].append({'scene_id': sid, 'status': 'asr_error', 'error': str(e)}) log(f'FLAGGED {len(flagged)}') for s,item in flagged: sid = s['scene_id']; old = AUDIO_DIR/f'{sid}.wav' backup = BACKUP_DIR/f'{sid}_pronunciation_{time.strftime("%Y%m%d_%H%M%S")}.wav' if old.exists(): shutil.copy2(old, backup) base_tts, changes = mod.normalize(s['narration']) item_payload = {'scene_id':sid, 'narration':s['narration'], 'tts_text':cased_payload(base_tts), 'normalization':changes+['watchlist_casing_exception'], 'output':str(old)} log(f'REGEN {sid}') r = mod.generate(item_payload) item['regenerate_result'] = r item['backup_path'] = str(backup) if r.get('status') == 'ok': report['regenerated'].append(sid) # Rebuild Step 7 manifest from current files. all_results=[] for s in scenes: p=AUDIO_DIR/f"{s['scene_id']}.wav" if p.exists(): all_results.append({'scene_id':s['scene_id'],'status':'ok','output':str(p),'bytes':p.stat().st_size,'duration':mod.duration(p)}) else: all_results.append({'scene_id':s['scene_id'],'status':'failed','output':str(p),'error':'missing'}) failed=[r for r in all_results if r['status']=='failed'] (PROJECT/'logs/step_07_tts_v2_manifest.json').write_text(json.dumps({'project_id':PROJECT.name,'step':7,'version':'v2_pronunciation_qa','status':'completed' if not failed else 'partial_failed','total_scenes':len(scenes),'done':len(scenes)-len(failed),'failed':failed,'total_duration_seconds':sum((r.get('duration') or 0) for r in all_results),'results':all_results},ensure_ascii=False,indent=2),encoding='utf-8') report['status'] = 'completed' REPORT.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding='utf-8') if report['regenerated']: log('RUN_STEP8 after regenerated scenes') import subprocess, sys subprocess.check_call([sys.executable, str(STEP8)], cwd=str(PROJECT), timeout=3600) log('DONE '+json.dumps({'candidates':len(candidates),'flagged':len(flagged),'regenerated':len(report['regenerated'])}, ensure_ascii=False)) if __name__ == '__main__': main()