#!/usr/bin/env python3 import json import re import time import urllib.request from pathlib import Path PROJECT = Path('/data/video-pipeline/Channel-Loi-Co-Nhan/project/13062026-048-Dung-De-Long-Ganh-Ti-Dot-Chay-Phuc-Phan') ENDPOINT = 'http://192.168.1.103:8001/transcribe-and-save' HEALTH_ENDPOINT = 'http://192.168.1.103:8001/health' MASTER_AUDIO = PROJECT / 'audio/master_narration.wav' TIMELINE = PROJECT / 'audio/master_narration_timeline.json' ASR_OUTPUT = PROJECT / 'subtitles/asr_word_timing.json' SCENE_ASR_DIR = PROJECT / 'subtitles/asr_scene_words' SRT_OUTPUT = PROJECT / 'subtitles/master_subtitles_asr_timing_script_text.srt' DEBUG_OUTPUT = PROJECT / 'subtitles/subtitle_asr_script_text_debug.json' MANIFEST = PROJECT / 'logs/asr_subtitle_rebuild_manifest.json' WORDS_PER_CUE = 5 MIN_WORD_RATIO = 0.90 def log(msg): print(time.strftime('%F %T ') + msg, flush=True) def post_json(url, payload=None, timeout=7200): data = None if payload is None else json.dumps(payload, ensure_ascii=False).encode('utf-8') req = urllib.request.Request( url, data=data, headers={'Content-Type': 'application/json'} if payload is not None else {}, method='POST' if payload is not None else 'GET', ) with urllib.request.urlopen(req, timeout=timeout) as resp: return json.loads(resp.read().decode('utf-8')) def collect_words(doc): words = [] for seg in doc.get('segments', []) or []: for w in seg.get('words', []) or []: if w.get('start') is not None and w.get('end') is not None: words.append(w) if not words and isinstance(doc.get('words'), list): words = [w for w in doc['words'] if w.get('start') is not None and w.get('end') is not None] return words def srt_time(sec): ms = int(round(max(0.0, sec) * 1000)) h = ms // 3600000 ms %= 3600000 m = ms // 60000 ms %= 60000 s = ms // 1000 ms %= 1000 return f'{h:02}:{m:02}:{s:02},{ms:03}' def script_by_scene(): data = json.loads((PROJECT / 'script/scenes.json').read_text(encoding='utf-8')) return { sc['scene_id']: re.findall(r'\S+', (sc.get('narration') or '').replace('\n', ' ')) for sc in data['scenes'] } def build_scene_cues(scene_id, script_words, asr_words, scene_start, speech_end): if not script_words: return [], {'scene_id': scene_id, 'script_words': 0, 'asr_words': len(asr_words), 'ratio': 1.0} ratio = len(asr_words) / max(1, len(script_words)) if ratio < MIN_WORD_RATIO: raise SystemExit(f'ASR scene word count too low: {scene_id} asr={len(asr_words)} script={len(script_words)} ratio={ratio:.3f}') cues = [] idx = 0 while idx < len(script_words): take = min(WORDS_PER_CUE, len(script_words) - idx) if len(script_words) - idx == WORDS_PER_CUE + 1: take = 3 text = ' '.join(script_words[idx:idx + take]) # Map script-word indexes onto ASR word timestamps within this scene only. # This prevents any ASR omissions in one scene from shifting later scenes. a0 = min(len(asr_words) - 1, round(idx * len(asr_words) / len(script_words))) a1 = min(len(asr_words) - 1, max(a0, round((idx + take) * len(asr_words) / len(script_words)) - 1)) start = scene_start + float(asr_words[a0]['start']) end = scene_start + float(asr_words[a1]['end']) if end <= start: end = min(speech_end, start + 0.35) end = min(end, speech_end) cues.append({'scene_id': scene_id, 'start': start, 'end': end, 'text': text}) idx += take return cues, {'scene_id': scene_id, 'script_words': len(script_words), 'asr_words': len(asr_words), 'ratio': ratio} def main(): if not MASTER_AUDIO.exists(): raise SystemExit(f'missing master audio: {MASTER_AUDIO}') if not TIMELINE.exists(): raise SystemExit(f'missing timeline: {TIMELINE}') SRT_OUTPUT.parent.mkdir(parents=True, exist_ok=True) SCENE_ASR_DIR.mkdir(parents=True, exist_ok=True) MANIFEST.parent.mkdir(parents=True, exist_ok=True) health = post_json(HEALTH_ENDPOINT, None, timeout=30) timeline = json.loads(TIMELINE.read_text(encoding='utf-8'))['items'] scene_scripts = script_by_scene() all_cues = [] scene_debug = [] total_asr = 0 total_script = 0 for item in timeline: scene_id = item['scene_id'] audio_path = Path(item['audio_path']) out = SCENE_ASR_DIR / f'{scene_id}.json' payload = { 'input_path': str(audio_path), 'output_path': str(out), 'language': 'vi', 'task': 'transcribe', 'word_timestamps': True, } log(f'POST ASR scene {scene_id}') response = post_json(ENDPOINT, payload, timeout=7200) if not out.exists(): out.write_text(json.dumps(response, ensure_ascii=False, indent=2), encoding='utf-8') asr_words = collect_words(json.loads(out.read_text(encoding='utf-8'))) script_words = scene_scripts.get(scene_id, []) cues, dbg = build_scene_cues( scene_id, script_words, asr_words, float(item['start']), float(item.get('speech_end', item['end'])), ) all_cues.extend(cues) scene_debug.append(dbg) total_asr += len(asr_words) total_script += len(script_words) lines = [] for idx, c in enumerate(all_cues, 1): lines.append(str(idx)) lines.append(f"{srt_time(c['start'])} --> {srt_time(c['end'])}") lines.append(c['text']) lines.append('') SRT_OUTPUT.write_text('\n'.join(lines), encoding='utf-8') DEBUG_OUTPUT.write_text(json.dumps({ 'mode': 'scene_level_asr_timing', 'asr_words': total_asr, 'script_words': total_script, 'ratio': total_asr / max(1, total_script), 'scene_debug': scene_debug, 'cues': all_cues, }, ensure_ascii=False, indent=2), encoding='utf-8') MANIFEST.write_text(json.dumps({ 'status': 'completed', 'mode': 'scene_level_asr_timing', 'endpoint': ENDPOINT, 'health': health, 'input_audio_path': str(MASTER_AUDIO), 'scene_asr_dir': str(SCENE_ASR_DIR), 'srt_output': str(SRT_OUTPUT), 'asr_words': total_asr, 'script_words': total_script, 'cues': len(all_cues), 'note': 'Subtitle timing rebuilt from per-scene ASR word timestamps; display text comes from the matching scene script so ASR omissions cannot shift later scenes.', }, ensure_ascii=False, indent=2), encoding='utf-8') log(f'DONE ASR scene-level subtitle cues={len(all_cues)} asr_words={total_asr} script_words={total_script}') if __name__ == '__main__': main()