#!/usr/bin/env python3 import difflib import json import re import time import unicodedata from pathlib import Path PROJECT = Path('/data/video-pipeline/Youtube-Video-Maker/Project/08062026-003-Le-Hoan') SCENES_JSON = PROJECT / 'script/scenes.json' ASR_JSON = PROJECT / 'subtitles/master_narration_asr_words.json' OUTPUT_SRT = PROJECT / 'subtitles/master_subtitles.srt' OUTPUT_JSON = PROJECT / 'subtitles/master_subtitles_mapping.json' MANIFEST = PROJECT / 'logs/step11_subtitle_mapping_manifest.json' MAX_CUE_CHARS = 78 MAX_CUE_SECONDS = 6.5 MIN_CUE_SECONDS = 0.7 TOKEN_RE = re.compile(r"\w+", re.UNICODE) def log(msg): print(time.strftime('%Y-%m-%d %H:%M:%S ') + msg, flush=True) def norm_token(text): text = text.lower().replace('đ', 'd') text = ''.join(ch for ch in unicodedata.normalize('NFD', text) if unicodedata.category(ch) != 'Mn') return re.sub(r'[^a-z0-9]+', '', text) def tokenize_with_spans(text): return [{'text': m.group(0), 'norm': norm_token(m.group(0)), 'start_char': m.start(), 'end_char': m.end()} for m in TOKEN_RE.finditer(text)] def collect_asr_words(doc): words = [] for seg in doc.get('segments', []) or []: for w in seg.get('words', []) or []: if w.get('start') is None or w.get('end') is None: continue words.append({'text': w.get('word', '').strip(), 'norm': norm_token(w.get('word', '')), 'start': float(w['start']), 'end': float(w['end'])}) return [w for w in words if w['norm']] def build_original_text(scenes): chunks = [] scene_ranges = [] cursor = 0 for scene in scenes: if chunks: chunks.append('\n') cursor += 1 start = cursor narration = scene.get('narration', '').strip() chunks.append(narration) cursor += len(narration) scene_ranges.append({'scene_id': scene['scene_id'], 'scene_type': scene.get('scene_type', ''), 'start_char': start, 'end_char': cursor}) return ''.join(chunks), scene_ranges def align_tokens(orig_tokens, asr_words): orig_norm = [t['norm'] for t in orig_tokens] asr_norm = [w['norm'] for w in asr_words] sm = difflib.SequenceMatcher(None, orig_norm, asr_norm, autojunk=False) token_times = [None] * len(orig_tokens) matched = 0 for tag, i1, i2, j1, j2 in sm.get_opcodes(): if tag == 'equal': for oi, aj in zip(range(i1, i2), range(j1, j2)): token_times[oi] = {'start': asr_words[aj]['start'], 'end': asr_words[aj]['end'], 'source': 'matched'} matched += 1 elif tag == 'replace' and (i2 - i1) == (j2 - j1): # Same-length fuzzy region: borrow timing positionally for near matches. for oi, aj in zip(range(i1, i2), range(j1, j2)): ratio = difflib.SequenceMatcher(None, orig_norm[oi], asr_norm[aj]).ratio() if ratio >= 0.72: token_times[oi] = {'start': asr_words[aj]['start'], 'end': asr_words[aj]['end'], 'source': 'fuzzy'} matched += 1 # Fill gaps by linear interpolation between nearest timed tokens. timed_indices = [i for i, v in enumerate(token_times) if v] if not timed_indices: raise RuntimeError('No aligned tokens between original narration and ASR words') for i, tok in enumerate(orig_tokens): if token_times[i]: continue prevs = [x for x in timed_indices if x < i] nexts = [x for x in timed_indices if x > i] prev_i = prevs[-1] if prevs else None next_i = nexts[0] if nexts else None if prev_i is not None and next_i is not None: span = next_i - prev_i frac = (i - prev_i) / span start = token_times[prev_i]['end'] + frac * (token_times[next_i]['start'] - token_times[prev_i]['end']) elif prev_i is not None: start = token_times[prev_i]['end'] + 0.25 * (i - prev_i) else: start = max(0.0, token_times[next_i]['start'] - 0.25 * (next_i - i)) token_times[i] = {'start': start, 'end': start + 0.2, 'source': 'interpolated'} return token_times, matched def split_cues(text, tokens, token_times): cues = [] cur_start_token = 0 last_break_token = 0 for i, tok in enumerate(tokens): snippet = text[tokens[cur_start_token]['start_char']:tok['end_char']].strip() cue_duration = token_times[i]['end'] - token_times[cur_start_token]['start'] next_char = text[tok['end_char']:tok['end_char'] + 1] punctuation_break = next_char in '.?!;:' scene_boundary_break = next_char == '\n' comma_break = next_char == ',' and len(snippet) >= 38 too_long = len(snippet) >= MAX_CUE_CHARS too_slow = cue_duration >= MAX_CUE_SECONDS and len(snippet) >= 28 if i == len(tokens) - 1 or scene_boundary_break or punctuation_break or comma_break or too_long or too_slow: start_char = tokens[cur_start_token]['start_char'] end_char = tok['end_char'] # Include immediately trailing punctuation in display text. while end_char < len(text) and text[end_char] in '.,?!;:': end_char += 1 cue_text = re.sub(r'\s+', ' ', text[start_char:end_char]).strip() start = token_times[cur_start_token]['start'] end = token_times[i]['end'] if end - start < MIN_CUE_SECONDS: end = start + MIN_CUE_SECONDS cues.append({'index': len(cues) + 1, 'start': round(start, 3), 'end': round(end, 3), 'text': cue_text, 'start_token': cur_start_token, 'end_token': i}) cur_start_token = i + 1 last_break_token = i return [c for c in cues if c['text']] def fix_overlaps(cues): for i, cue in enumerate(cues): if cue['start'] < 0: cue['start'] = 0.0 if i > 0 and cue['start'] < cues[i - 1]['end']: prev = cues[i - 1] midpoint = round((prev['start'] + cue['end']) / 2, 3) prev['end'] = min(prev['end'], max(prev['start'] + 0.2, cue['start'] - 0.02)) if cue['start'] > prev['start'] else midpoint cue['start'] = round(prev['end'] + 0.02, 3) if cue['end'] <= cue['start']: cue['end'] = round(cue['start'] + MIN_CUE_SECONDS, 3) cue['index'] = i + 1 return cues def fmt_time(sec): ms = int(round(sec * 1000)) h, rem = divmod(ms, 3600000) m, rem = divmod(rem, 60000) s, ms = divmod(rem, 1000) return f'{h:02d}:{m:02d}:{s:02d},{ms:03d}' def write_srt(cues, path): lines = [] for cue in cues: lines.append(str(cue['index'])) lines.append(f"{fmt_time(cue['start'])} --> {fmt_time(cue['end'])}") lines.append(cue['text']) lines.append('') path.write_text('\n'.join(lines), encoding='utf-8') def main(): scenes = json.loads(SCENES_JSON.read_text(encoding='utf-8'))['scenes'] asr_doc = json.loads(ASR_JSON.read_text(encoding='utf-8')) text, scene_ranges = build_original_text(scenes) orig_tokens = tokenize_with_spans(text) asr_words = collect_asr_words(asr_doc) log(f'ALIGN original_tokens={len(orig_tokens)} asr_words={len(asr_words)}') token_times, matched = align_tokens(orig_tokens, asr_words) cues = fix_overlaps(split_cues(text, orig_tokens, token_times)) validation_errors = [] if not cues: validation_errors.append('no subtitle cues generated') for i, cue in enumerate(cues): if not cue['text']: validation_errors.append(f'empty text at cue {i+1}') break if cue['start'] < 0 or cue['end'] <= cue['start']: validation_errors.append(f'invalid time at cue {i+1}') break if i and cue['start'] < cues[i - 1]['end']: validation_errors.append(f'overlap at cue {i+1}') break OUTPUT_SRT.parent.mkdir(parents=True, exist_ok=True) MANIFEST.parent.mkdir(parents=True, exist_ok=True) write_srt(cues, OUTPUT_SRT) mapping = { 'project_id': PROJECT.name, 'step': 11, 'source_text': 'script/scenes.json:scene.narration', 'timing_source': 'subtitles/master_narration_asr_words.json:segments[].words', 'asr_transcript_is_display_text': False, 'original_tokens': len(orig_tokens), 'asr_words': len(asr_words), 'aligned_tokens': matched, 'cues_count': len(cues), 'scene_ranges': scene_ranges, 'cues': cues, } OUTPUT_JSON.write_text(json.dumps(mapping, ensure_ascii=False, indent=2), encoding='utf-8') status = 'completed' if not validation_errors else 'failed_validation' manifest = { 'project_id': PROJECT.name, 'step': 11, 'status': status, 'output_srt': str(OUTPUT_SRT), 'output_json': str(OUTPUT_JSON), 'cues_count': len(cues), 'original_tokens': len(orig_tokens), 'asr_words': len(asr_words), 'aligned_tokens': matched, 'validation_errors': validation_errors, 'note': 'Subtitle display text is original scene.narration; ASR words are used only for timing.', } MANIFEST.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding='utf-8') if validation_errors: raise SystemExit('Validation failed: ' + '; '.join(validation_errors)) log('DONE ' + json.dumps({'status': status, 'cues': len(cues), 'aligned_tokens': matched}, ensure_ascii=False)) if __name__ == '__main__': main()