#!/usr/bin/env python3 import json import re import time from pathlib import Path PROJECT = Path('/data/video-pipeline/Youtube-Video-Maker/Project/31052026-001-Nguyen-Hue-Quang-Trung') INPUT_SRT = PROJECT / 'subtitles/master_subtitles.srt' PROBE_JSON = PROJECT / 'audio/master_audio_probe.json' OUTPUT_SRT = PROJECT / 'subtitles/master_subtitles_clamped.srt' VALIDATION_JSON = PROJECT / 'subtitles/master_subtitles_validation.json' MANIFEST = PROJECT / 'logs/step12_clamp_subtitles_manifest.json' SAFE_GAP = 0.02 MIN_DURATION = 0.50 MAX_DURATION = 7.50 def log(msg): print(time.strftime('%Y-%m-%d %H:%M:%S ') + msg, flush=True) def parse_time(value): m = re.match(r'^(\d+):(\d+):(\d+),(\d+)$', value.strip()) if not m: raise ValueError(f'Invalid SRT time: {value}') h, mi, s, ms = map(int, m.groups()) return h * 3600 + mi * 60 + s + ms / 1000.0 def fmt_time(sec): sec = max(0.0, sec) ms_total = int(round(sec * 1000)) h, rem = divmod(ms_total, 3600000) mi, rem = divmod(rem, 60000) s, ms = divmod(rem, 1000) return f'{h:02d}:{mi:02d}:{s:02d},{ms:03d}' def read_srt(path): content = path.read_text(encoding='utf-8-sig').strip() if not content: return [] blocks = re.split(r'\n\s*\n', content) cues = [] for block in blocks: lines = [line.rstrip('\r') for line in block.splitlines() if line.strip()] if len(lines) < 2: continue idx_line = lines[0] time_line = lines[1] text_lines = lines[2:] if '-->' not in time_line: time_line = lines[0] idx_line = str(len(cues) + 1) text_lines = lines[1:] start_s, end_s = [x.strip() for x in time_line.split('-->', 1)] cues.append({ 'index': len(cues) + 1, 'original_index': idx_line, 'start': parse_time(start_s), 'end': parse_time(end_s), 'text': '\n'.join(text_lines).strip(), 'changes': [], }) return cues def write_srt(cues, path): lines = [] for i, cue in enumerate(cues, start=1): lines.append(str(i)) lines.append(f"{fmt_time(cue['start'])} --> {fmt_time(cue['end'])}") lines.extend(cue['text'].splitlines()) lines.append('') path.write_text('\n'.join(lines), encoding='utf-8') def main(): if not INPUT_SRT.exists(): raise SystemExit(f'Missing input subtitle: {INPUT_SRT}') if not PROBE_JSON.exists(): raise SystemExit(f'Missing Step 9 probe json: {PROBE_JSON}') master_duration = float(json.loads(PROBE_JSON.read_text(encoding='utf-8'))['master_duration_sec']) cues = read_srt(INPUT_SRT) adjusted = 0 for i, cue in enumerate(cues): orig = (cue['start'], cue['end']) if cue['start'] < 0: cue['start'] = 0.0 cue['changes'].append('clamped_start_to_zero') if cue['end'] > master_duration: cue['end'] = master_duration cue['changes'].append('clamped_end_to_master_duration') if cue['end'] <= cue['start']: cue['end'] = min(master_duration, cue['start'] + MIN_DURATION) cue['changes'].append('expanded_invalid_duration') if cue['end'] - cue['start'] < MIN_DURATION: cue['end'] = min(master_duration, cue['start'] + MIN_DURATION) cue['changes'].append('expanded_short_duration') if cue['end'] - cue['start'] > MAX_DURATION: cue['end'] = cue['start'] + MAX_DURATION cue['changes'].append('trimmed_long_duration') if i > 0: prev = cues[i - 1] min_start = prev['end'] + SAFE_GAP if cue['start'] < min_start: # Prefer shortening previous cue to preserve current speech onset. target_prev_end = max(prev['start'] + MIN_DURATION, cue['start'] - SAFE_GAP) if target_prev_end < prev['end']: prev['end'] = target_prev_end prev['changes'].append('shortened_for_safe_gap') if cue['start'] < prev['end'] + SAFE_GAP: cue['start'] = min(master_duration, prev['end'] + SAFE_GAP) cue['changes'].append('shifted_start_for_safe_gap') if cue['end'] <= cue['start']: cue['end'] = min(master_duration, cue['start'] + MIN_DURATION) cue['changes'].append('expanded_after_shift') if (round(cue['start'], 3), round(cue['end'], 3)) != (round(orig[0], 3), round(orig[1], 3)): adjusted += 1 overlaps = 0 invalid = 0 outside = 0 too_short = 0 for i, cue in enumerate(cues): if not cue['text'].strip() or cue['start'] < 0 or cue['end'] <= cue['start']: invalid += 1 if cue['end'] > master_duration + 0.001: outside += 1 if cue['end'] - cue['start'] < MIN_DURATION - 0.001: too_short += 1 if i > 0 and cue['start'] < cues[i - 1]['end']: overlaps += 1 write_srt(cues, OUTPUT_SRT) validation = { 'project_id': PROJECT.name, 'step': 12, 'input_srt': str(INPUT_SRT), 'output_srt': str(OUTPUT_SRT), 'master_duration_sec': master_duration, 'safe_gap_sec': SAFE_GAP, 'min_duration_sec': MIN_DURATION, 'max_duration_sec': MAX_DURATION, 'total_cues': len(cues), 'adjusted_cues': sum(1 for c in cues if c['changes']), 'overlap_remaining': overlaps, 'invalid_cues': invalid, 'outside_master_duration': outside, 'too_short_cues': too_short, 'within_master_duration': outside == 0, 'adjustments': [c for c in cues if c['changes']], } VALIDATION_JSON.write_text(json.dumps(validation, ensure_ascii=False, indent=2), encoding='utf-8') status = 'completed' if overlaps == 0 and invalid == 0 and outside == 0 else 'failed_validation' manifest = { 'project_id': PROJECT.name, 'step': 12, 'status': status, 'output_srt': str(OUTPUT_SRT), 'validation_json': str(VALIDATION_JSON), 'total_cues': len(cues), 'adjusted_cues': validation['adjusted_cues'], 'overlap_remaining': overlaps, 'invalid_cues': invalid, 'outside_master_duration': outside, } MANIFEST.parent.mkdir(parents=True, exist_ok=True) MANIFEST.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding='utf-8') if status != 'completed': raise SystemExit(json.dumps(manifest, ensure_ascii=False)) log('DONE ' + json.dumps({'status': status, 'cues': len(cues), 'adjusted': validation['adjusted_cues'], 'overlap_remaining': overlaps, 'invalid': invalid}, ensure_ascii=False)) if __name__ == '__main__': main()