#!/usr/bin/env python3 """Local Chinese subtitles from speech; faster-whisper 1.2.1 / Python 3.12. Example: python h3_subtitle_local.py final.mp4 final.srt --model ./models/whisper-turbo Download the complete CTranslate2 model once before running. No network/API is used by this script. Output includes editable UTF-8 SRT and a timing audit JSON. """ from __future__ import annotations import argparse from dataclasses import asdict, dataclass import json import math import os from pathlib import Path import re import sys from typing import Iterable @dataclass class Cue: start: float end: float text: str @dataclass class WordSpan: start: float end: float word: str def clean_text(value: str) -> str: return re.sub(r'\s+', ' ', str(value)).strip().replace('-->', '→') def timestamp(seconds: float) -> str: """Round once to integer milliseconds, including minute/hour carries.""" if not math.isfinite(seconds) or seconds < 0: raise ValueError('Invalid subtitle timestamp') milliseconds = int(math.floor(seconds * 1000 + 0.5)) hours, remainder = divmod(milliseconds, 3_600_000) minutes, remainder = divmod(remainder, 60_000) secs, millis = divmod(remainder, 1000) return f'{hours:02d}:{minutes:02d}:{secs:02d},{millis:03d}' def wrapped_text(text: str, line_chars: int = 16) -> str: """Chinese-oriented wrapping; it changes layout, never word timing.""" text = clean_text(text) if len(text) <= line_chars: return text # A single ASR token can exceed a line. Preserve it rather than fabricating # character timestamps. Normal cues use at most two short display lines. return '\n'.join(text[i:i + line_chars] for i in range(0, len(text), line_chars)) def segment_cues(segment, max_chars: int = 24, max_seconds: float = 4.0) -> list[Cue]: """Group the model's real word spans; never divide time by text length.""" words = getattr(segment, 'words', None) if not words: text = clean_text(segment.text) start, end = float(segment.start), float(segment.end) if text and math.isfinite(start) and math.isfinite(end) and end > start >= 0: return [Cue(start, end, text)] return [] normalized: list[WordSpan] = [] prefix = '' for word in words: text, start, end = str(word.word), float(word.start), float(word.end) if not (math.isfinite(start) and math.isfinite(end) and end >= start >= 0): raise ValueError('ASR produced an invalid word time; inspect the input audio.') if end == start: # Alignment can assign zero duration to punctuation. Attach its text # to the adjoining timed token, retaining the model's real bounds. if normalized: normalized[-1].word += text else: prefix += text else: normalized.append(WordSpan(start, end, prefix + text)) prefix = '' if not normalized: text = clean_text(segment.text) start, end = float(segment.start), float(segment.end) return [Cue(start, end, text)] if text and end > start >= 0 else [] cues: list[Cue] = [] pending: list = [] def flush(): if pending: text = clean_text(''.join(str(w.word) for w in pending)) if text: cues.append(Cue(float(pending[0].start), float(pending[-1].end), text)) pending.clear() for word in normalized: text = str(word.word) start, end = float(word.start), float(word.end) if not clean_text(text): continue if not (math.isfinite(start) and math.isfinite(end) and end > start >= 0): raise ValueError('ASR produced an invalid word time; inspect the input audio.') candidate = clean_text(''.join(str(w.word) for w in pending) + text) if pending and (len(candidate) > max_chars or end - float(pending[0].start) > max_seconds or start - float(pending[-1].end) > 0.5): flush() pending.append(word) if text.rstrip().endswith(('。', '！', '？', '!', '?', ';', '；')): flush() flush() return cues def make_cues(segments: Iterable, max_chars: int = 24, max_seconds: float = 4.0) -> list[Cue]: cues: list[Cue] = [] for segment in segments: cues.extend(segment_cues(segment, max_chars, max_seconds)) # Whisper alignment can put two adjoining word spans a few ms over each # other. Trim only the earlier cue's display end at the next actual start. for current, following in zip(cues, cues[1:]): if following.start < current.start: raise ValueError('ASR produced non-monotonic subtitle timing.') current.end = min(current.end, following.start) return [cue for cue in cues if round(cue.end * 1000) > round(cue.start * 1000)] def srt_text(cues: Iterable[Cue], line_chars: int = 16) -> str: return ''.join(f'{index}\n{timestamp(cue.start)} --> {timestamp(cue.end)}\n' f'{wrapped_text(cue.text, line_chars)}\n\n' for index, cue in enumerate(cues, 1)) def parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser(description='全本地中文语音识别与SRT字幕，默认CPU INT8。') p.add_argument('input', type=Path, help='本机视频或音频文件') p.add_argument('output', type=Path, help='输出字幕，例如 final.srt') p.add_argument('--model', type=Path, required=True, help='完整的本地CTranslate2模型目录') p.add_argument('--device', choices=['cpu', 'cuda'], default='cpu') p.add_argument('--compute-type', choices=['int8', 'float32', 'float16', 'int8_float16'], default=None) p.add_argument('--language', default='zh', help='识别语言代码，默认zh；不翻译') p.add_argument('--prompt', default='简体中文。阿木，橙色雨衣，迟到的包裹。', help='专有名词/简体中文提示') p.add_argument('--max-chars', type=int, default=24, help='每条字幕目标字符上限') p.add_argument('--line-chars', type=int, default=16, help='中文显示折行字符数') p.add_argument('--max-seconds', type=float, default=4.0) p.add_argument('--threads', type=int, default=min(8, os.cpu_count() or 4)) p.add_argument('--overwrite', action='store_true', help='覆盖已有的本次字幕与审计文件') return p def main(argv=None) -> int: args = parser().parse_args(argv) if not args.input.is_file(): raise ValueError('输入文件不存在：' + str(args.input)) if args.output.suffix.lower() != '.srt': raise ValueError('输出文件名必须以.srt结尾') if min(args.max_chars, args.line_chars, args.threads) < 1 or args.max_seconds <= 0: raise ValueError('字幕长度、线程与时长参数必须大于0') required = ('model.bin', 'config.json', 'tokenizer.json', 'preprocessor_config.json', 'vocabulary.json') missing = [name for name in required if not (args.model / name).is_file()] if missing: raise ValueError('本地模型目录缺少：' + ', '.join(missing)) output = args.output.resolve() audit = output.with_suffix('.asr.json') for path in (output, audit): if path == args.input.resolve(): raise ValueError('输出路径与输入文件相同') if path.exists() and not args.overwrite: raise ValueError('输出已存在；使用新文件名或添加--overwrite：' + str(path)) os.environ['HF_HUB_OFFLINE'] = '1' os.environ['TRANSFORMERS_OFFLINE'] = '1' try: from faster_whisper import WhisperModel except ImportError as exc: raise RuntimeError('请在字幕独立环境安装faster-whisper==1.2.1与ctranslate2==4.8.4') from exc compute_type = args.compute_type or ('int8' if args.device == 'cpu' else 'float16') model = WhisperModel(str(args.model.resolve()), device=args.device, compute_type=compute_type, cpu_threads=args.threads, num_workers=1, local_files_only=True) segments, info = model.transcribe( str(args.input.resolve()), language=args.language, task='transcribe', beam_size=5, temperature=0.0, word_timestamps=True, vad_filter=True, vad_parameters={'min_silence_duration_ms': 500}, condition_on_previous_text=False, initial_prompt=args.prompt, hallucination_silence_threshold=1.0, ) # Iteration is required: faster-whisper returns a lazy generator. segments = list(segments) cues = make_cues(segments, args.max_chars, args.max_seconds) output.parent.mkdir(parents=True, exist_ok=True) output.write_text(srt_text(cues, args.line_chars), encoding='utf-8', newline='\n') report = { 'input': str(args.input.resolve()), 'model': str(args.model.resolve()), 'device': args.device, 'compute_type': compute_type, 'language': info.language, 'duration_seconds': info.duration, 'cue_count': len(cues), 'word_timestamps': True, 'network_disabled': True, 'cues': [asdict(cue) for cue in cues], 'segments': [asdict(segment) for segment in segments], } audit.write_text(json.dumps(report, ensure_ascii=False, indent=2) + '\n', encoding='utf-8') print(f'完成：{output}\n字幕条数：{len(cues)}\n时间戳记录：{audit}', flush=True) if not cues: print('未检测到有效对白。纯环境声片段保持空字幕；请核对输入音轨。', flush=True) return 0 if __name__ == '__main__': try: sys.exit(main()) except (ValueError, RuntimeError) as error: print('错误：' + str(error), file=sys.stderr) sys.exit(2)