| 1 | #!/usr/bin/env python3 |
| 2 | """ |
| 3 | whisper_transcribe.py — 对 douyin-downloader 下载的视频进行 Whisper 语音识别 |
| 4 | |
| 5 | 安装: |
| 6 | pip install openai-whisper rich |
| 7 | # ffmpeg: conda install -c conda-forge ffmpeg 或放 ffmpeg.exe 到同目录 |
| 8 | |
| 9 | 用法: |
| 10 | python whisper_transcribe.py # 扫描 ./Downloaded/ 下所有mp4 |
| 11 | python whisper_transcribe.py -d ./Downloaded/ # 指定目录 |
| 12 | python whisper_transcribe.py -f video.mp4 # 单个文件 |
| 13 | python whisper_transcribe.py -d ./Downloaded/ -m medium # 用medium模型 |
| 14 | python whisper_transcribe.py -d ./Downloaded/ --srt # 同时输出SRT |
| 15 | python whisper_transcribe.py --skip-existing --sc # 跳过已有 + 繁转简 |
| 16 | """ |
| 17 | |
| 18 | import argparse |
| 19 | import os |
| 20 | import shutil |
| 21 | import subprocess |
| 22 | import sys |
| 23 | import tempfile |
| 24 | from pathlib import Path |
| 25 | from typing import Optional |
| 26 | |
| 27 | from rich.console import Console |
| 28 | from rich.panel import Panel |
| 29 | from rich.progress import ( |
| 30 | BarColumn, |
| 31 | Progress, |
| 32 | SpinnerColumn, |
| 33 | TaskProgressColumn, |
| 34 | TextColumn, |
| 35 | TimeElapsedColumn, |
| 36 | ) |
| 37 | from rich.table import Table |
| 38 | from rich.text import Text |
| 39 | |
| 40 | console = Console() |
| 41 | |
| 42 | # ── 颜色主题 (区别于 douyin-downloader 的 cyan/magenta) ── |
| 43 | THEME = { |
| 44 | "accent": "bright_green", |
| 45 | "banner": "bold bright_green", |
| 46 | "info": "dodger_blue1", |
| 47 | "success": "green", |
| 48 | "warning": "yellow", |
| 49 | "error": "red", |
| 50 | "dim": "dim white", |
| 51 | "file": "bright_cyan", |
| 52 | "model": "orchid", |
| 53 | } |
| 54 | |
| 55 | |
| 56 | # ============================================================ |
| 57 | # TranscribeDisplay — rich 进度显示 |
| 58 | # ============================================================ |
| 59 | class TranscribeDisplay: |
| 60 | def __init__(self): |
| 61 | self.console = console |
| 62 | self._progress_ctx: Optional[Progress] = None |
| 63 | self._progress: Optional[Progress] = None |
| 64 | self._overall_id: Optional[int] = None |
| 65 | self._file_id: Optional[int] = None |
| 66 | self._file_index = 0 |
| 67 | self._file_total = 0 |
| 68 | self._stats = {"success": 0, "failed": 0, "skipped": 0} |
| 69 | |
| 70 | # ── banner ── |
| 71 | def show_banner(self): |
| 72 | banner = Text() |
| 73 | banner.append(" 🎙 Whisper 视频转录工具\n", style="bold bright_green") |
| 74 | banner.append(" ── Video → Text via OpenAI Whisper ──", style="dim bright_green") |
| 75 | panel = Panel(banner, border_style="bright_green", expand=False, padding=(0, 2)) |
| 76 | self.console.print(panel) |
| 77 | self.console.print() |
| 78 | |
| 79 | # ── progress lifecycle ── |
| 80 | def start_session(self, total: int): |
| 81 | self._file_total = total |
| 82 | self._file_index = 0 |
| 83 | self._stats = {"success": 0, "failed": 0, "skipped": 0} |
| 84 | |
| 85 | self._progress_ctx = Progress( |
| 86 | SpinnerColumn(style="bright_green"), |
| 87 | TextColumn("[progress.description]{task.description}"), |
| 88 | BarColumn(bar_width=30, complete_style="bright_green", finished_style="green"), |
| 89 | TaskProgressColumn(), |
| 90 | TimeElapsedColumn(), |
| 91 | TextColumn("[dim]{task.fields[detail]}"), |
| 92 | console=self.console, |
| 93 | transient=True, |
| 94 | refresh_per_second=6, |
| 95 | ) |
| 96 | self._progress = self._progress_ctx.__enter__() |
| 97 | self._overall_id = self._progress.add_task( |
| 98 | "[bright_green]总体进度[/]", |
| 99 | total=max(total, 1), |
| 100 | completed=0, |
| 101 | detail=f"共 {total} 个视频", |
| 102 | ) |
| 103 | |
| 104 | def stop_session(self): |
| 105 | if self._file_id is not None and self._progress: |
| 106 | self._progress.remove_task(self._file_id) |
| 107 | self._file_id = None |
| 108 | if self._progress_ctx is not None: |
| 109 | self._progress_ctx.__exit__(None, None, None) |
| 110 | self._progress_ctx = None |
| 111 | self._progress = None |
| 112 | self._overall_id = None |
| 113 | |
| 114 | # ── per-file ── |
| 115 | def start_file(self, index: int, name: str): |
| 116 | self._file_index = index |
| 117 | if self._file_id is not None and self._progress: |
| 118 | self._progress.remove_task(self._file_id) |
| 119 | if not self._progress: |
| 120 | return |
| 121 | self._file_id = self._progress.add_task( |
| 122 | self._file_desc("提取音频"), |
| 123 | total=4, # 提取音频 → 识别 → 转换 → 保存 |
| 124 | completed=0, |
| 125 | detail=self._shorten(name, 50), |
| 126 | ) |
| 127 | |
| 128 | def advance_file(self, step: str, detail: str = ""): |
| 129 | if not self._progress or self._file_id is None: |
| 130 | return |
| 131 | self._progress.advance(self._file_id, 1) |
| 132 | self._progress.update( |
| 133 | self._file_id, |
| 134 | description=self._file_desc(step), |
| 135 | detail=detail, |
| 136 | ) |
| 137 | |
| 138 | def complete_file(self, status: str, detail: str = ""): |
| 139 | if status in self._stats: |
| 140 | self._stats[status] += 1 |
| 141 | if self._progress: |
| 142 | if self._file_id is not None: |
| 143 | self._progress.update( |
| 144 | self._file_id, |
| 145 | completed=4, |
| 146 | description=self._file_desc( |
| 147 | "完成" if status == "success" else "跳过" if status == "skipped" else "失败" |
| 148 | ), |
| 149 | detail=detail, |
| 150 | ) |
| 151 | self._progress.remove_task(self._file_id) |
| 152 | self._file_id = None |
| 153 | if self._overall_id is not None: |
| 154 | self._progress.advance(self._overall_id, 1) |
| 155 | self._progress.update( |
| 156 | self._overall_id, |
| 157 | detail=f"✓{self._stats['success']} ✗{self._stats['failed']} ⊘{self._stats['skipped']}", |
| 158 | ) |
| 159 | |
| 160 | # ── summary table ── |
| 161 | def show_summary(self): |
| 162 | table = Table( |
| 163 | title="Transcription Summary", |
| 164 | show_header=True, |
| 165 | header_style=f"bold {THEME['accent']}", |
| 166 | border_style=THEME["accent"], |
| 167 | ) |
| 168 | table.add_column("Metric", style=THEME["info"]) |
| 169 | table.add_column("Count", justify="right", style=THEME["success"]) |
| 170 | |
| 171 | total = self._stats["success"] + self._stats["failed"] + self._stats["skipped"] |
| 172 | table.add_row("Total", str(total)) |
| 173 | table.add_row("Success", str(self._stats["success"])) |
| 174 | table.add_row("Failed", str(self._stats["failed"])) |
| 175 | table.add_row("Skipped", str(self._stats["skipped"])) |
| 176 | if total > 0: |
| 177 | rate = self._stats["success"] / total * 100 |
| 178 | table.add_row("Success Rate", f"{rate:.1f}%") |
| 179 | |
| 180 | self.console.print() |
| 181 | self.console.print(table) |
| 182 | |
| 183 | # ── logging ── |
| 184 | def info(self, msg: str): |
| 185 | self._out().print(f"[{THEME['info']}]ℹ[/] {msg}") |
| 186 | |
| 187 | def success(self, msg: str): |
| 188 | self._out().print(f"[{THEME['success']}]✓[/] {msg}") |
| 189 | |
| 190 | def warning(self, msg: str): |
| 191 | self._out().print(f"[{THEME['warning']}]⚠[/] {msg}") |
| 192 | |
| 193 | def error(self, msg: str): |
| 194 | self._out().print(f"[{THEME['error']}]✗[/] {msg}") |
| 195 | |
| 196 | def dep_ok(self, name: str, detail: str = ""): |
| 197 | self._out().print(f" [{THEME['success']}]✓[/] {name} [{THEME['dim']}]{detail}[/]") |
| 198 | |
| 199 | def dep_fail(self, name: str, hint: str): |
| 200 | self._out().print(f" [{THEME['error']}]✗[/] {name} [{THEME['dim']}]{hint}[/]") |
| 201 | |
| 202 | # ── internal ── |
| 203 | def _file_desc(self, step: str) -> str: |
| 204 | return f"[{THEME['accent']}]{self._file_index}/{self._file_total}[/] · {step}" |
| 205 | |
| 206 | def _out(self) -> Console: |
| 207 | return self._progress.console if self._progress else self.console |
| 208 | |
| 209 | @staticmethod |
| 210 | def _shorten(text: str, max_len: int = 50) -> str: |
| 211 | t = (text or "").strip() |
| 212 | return t if len(t) <= max_len else f"{t[: max_len - 3]}..." |
| 213 | |
| 214 | |
| 215 | display = TranscribeDisplay() |
| 216 | |
| 217 | |
| 218 | # ============================================================ |
| 219 | # 核心功能 |
| 220 | # ============================================================ |
| 221 | def find_ffmpeg(): |
| 222 | p = shutil.which("ffmpeg") |
| 223 | if p: |
| 224 | return p |
| 225 | local = Path(__file__).parent / "ffmpeg.exe" |
| 226 | if local.exists(): |
| 227 | return str(local) |
| 228 | try: |
| 229 | import imageio_ffmpeg |
| 230 | |
| 231 | return imageio_ffmpeg.get_ffmpeg_exe() |
| 232 | except ImportError: |
| 233 | pass |
| 234 | return None |
| 235 | |
| 236 | |
| 237 | def extract_audio(video_path, audio_path, ffmpeg_path="ffmpeg"): |
| 238 | cmd = [ |
| 239 | ffmpeg_path, |
| 240 | "-i", |
| 241 | str(video_path), |
| 242 | "-vn", |
| 243 | "-acodec", |
| 244 | "pcm_s16le", |
| 245 | "-ar", |
| 246 | "16000", |
| 247 | "-ac", |
| 248 | "1", |
| 249 | str(audio_path), |
| 250 | "-y", |
| 251 | "-loglevel", |
| 252 | "error", |
| 253 | ] |
| 254 | result = subprocess.run(cmd, capture_output=True, text=True) |
| 255 | if result.returncode != 0: |
| 256 | console.print(f" [{THEME['error']}]ffmpeg错误: {result.stderr.strip()}[/]") |
| 257 | return result.returncode == 0 and Path(audio_path).exists() |
| 258 | |
| 259 | |
| 260 | def _format_srt_time(seconds): |
| 261 | h, r = divmod(seconds, 3600) |
| 262 | m, r = divmod(r, 60) |
| 263 | s = int(r) |
| 264 | ms = int((r - s) * 1000) |
| 265 | return f"{int(h):02d}:{int(m):02d}:{s:02d},{ms:03d}" |
| 266 | |
| 267 | |
| 268 | def _safe_stem(stem): |
| 269 | """清洗文件名: 去掉换行、#、特殊符号，避免 Windows 路径报错""" |
| 270 | import re |
| 271 | |
| 272 | # 换行符 → 空格 |
| 273 | stem = stem.replace("\n", " ").replace("\r", " ") |
| 274 | # Windows 不允许的字符 + # → 下划线 |
| 275 | stem = re.sub(r'[<>:"/\\|?*#]', "_", stem) |
| 276 | # 连续空格/下划线 → 单个下划线 |
| 277 | stem = re.sub(r"[\s_]+", "_", stem) |
| 278 | # 去首尾下划线 |
| 279 | stem = stem.strip("_ ") |
| 280 | # 限制长度 (Windows MAX_PATH) |
| 281 | if len(stem) > 150: |
| 282 | stem = stem[:150] |
| 283 | return stem |
| 284 | |
| 285 | |
| 286 | def transcribe_file( |
| 287 | video_path, model, ffmpeg_path, output_formats, language, converter, output_dir=None |
| 288 | ): |
| 289 | video_path = Path(video_path) |
| 290 | stem = _safe_stem(video_path.stem) |
| 291 | |
| 292 | # 确定输出目录 |
| 293 | out_dir = None |
| 294 | if output_dir: |
| 295 | out_dir = Path(output_dir) |
| 296 | else: |
| 297 | # 尝试用原目录，但很多抖音文件夹名含换行/#等字符，写入会失败 |
| 298 | # 所以先试 mkdir + 写入测试，失败就 fallback |
| 299 | try: |
| 300 | candidate = video_path.parent |
| 301 | candidate.mkdir(parents=True, exist_ok=True) |
| 302 | # 测试是否真的能写文件 |
| 303 | test_file = candidate / ".whisper_test" |
| 304 | test_file.write_text("ok", encoding="utf-8") |
| 305 | test_file.unlink() |
| 306 | out_dir = candidate |
| 307 | except Exception: |
| 308 | out_dir = None |
| 309 | |
| 310 | if out_dir is None: |
| 311 | out_dir = Path("./transcripts") |
| 312 | |
| 313 | out_dir.mkdir(parents=True, exist_ok=True) |
| 314 | |
| 315 | txt_path = out_dir / f"{stem}.transcript.txt" |
| 316 | srt_path = out_dir / f"{stem}.transcript.srt" |
| 317 | |
| 318 | tmpdir = tempfile.mkdtemp(prefix="whisper_") |
| 319 | try: |
| 320 | # 先把视频复制到临时目录，避免原路径含特殊字符导致 ffmpeg/写入失败 |
| 321 | tmp_video = os.path.join(tmpdir, "input.mp4") |
| 322 | try: |
| 323 | shutil.copy2(str(video_path), tmp_video) |
| 324 | except Exception as e: |
| 325 | # 长路径/特殊字符 fallback: 用 Windows 短路径 |
| 326 | try: |
| 327 | import ctypes |
| 328 | |
| 329 | buf = ctypes.create_unicode_buffer(512) |
| 330 | ctypes.windll.kernel32.GetShortPathNameW(str(video_path), buf, 512) |
| 331 | short_path = buf.value |
| 332 | if short_path: |
| 333 | shutil.copy2(short_path, tmp_video) |
| 334 | else: |
| 335 | raise |
| 336 | except Exception: |
| 337 | console.print(f" [{THEME['error']}]无法访问视频文件: {e}[/]") |
| 338 | display.advance_file("失败", "路径不可达") |
| 339 | return False |
| 340 | |
| 341 | # Step 1: 提取音频 |
| 342 | audio_path = os.path.join(tmpdir, "audio.wav") |
| 343 | if not extract_audio(tmp_video, audio_path, ffmpeg_path): |
| 344 | display.advance_file("失败", "音频提取失败") |
| 345 | return False |
| 346 | audio_mb = os.path.getsize(audio_path) / 1024 / 1024 |
| 347 | display.advance_file("识别中", f"音频 {audio_mb:.1f}MB") |
| 348 | |
| 349 | # Step 2: Whisper 识别 |
| 350 | result = model.transcribe(audio_path, language=language, verbose=False) |
| 351 | segments = result.get("segments", []) |
| 352 | detected_lang = result.get("language", language) |
| 353 | |
| 354 | if not segments: |
| 355 | display.advance_file("无内容", "未检测到语音") |
| 356 | return False |
| 357 | |
| 358 | # Step 3: 繁转简 |
| 359 | def _cv(text): |
| 360 | return converter.convert(text) if converter and text else text |
| 361 | |
| 362 | text_lines = [_cv(seg["text"].strip()) for seg in segments if seg.get("text", "").strip()] |
| 363 | tag = "→简" if converter else "" |
| 364 | display.advance_file("保存", f"{len(segments)}段 lang={detected_lang} {tag}") |
| 365 | |
| 366 | # Step 4: 写文件 |
| 367 | saved = [] |
| 368 | if "txt" in output_formats: |
| 369 | txt_path.write_text("\n".join(text_lines), encoding="utf-8") |
| 370 | saved.append(txt_path.name) |
| 371 | if "srt" in output_formats: |
| 372 | srt_lines = [] |
| 373 | for i, seg in enumerate(segments, 1): |
| 374 | text = _cv(seg["text"].strip()) |
| 375 | if text: |
| 376 | srt_lines.append( |
| 377 | f"{i}\n{_format_srt_time(seg['start'])} --> {_format_srt_time(seg['end'])}\n{text}\n" |
| 378 | ) |
| 379 | srt_path.write_text("\n".join(srt_lines), encoding="utf-8") |
| 380 | saved.append(srt_path.name) |
| 381 | |
| 382 | display.advance_file("完成", " + ".join(saved)) |
| 383 | return True |
| 384 | |
| 385 | finally: |
| 386 | shutil.rmtree(tmpdir, ignore_errors=True) |
| 387 | |
| 388 | |
| 389 | def find_videos(directory, skip_existing=False, output_dir=None): |
| 390 | directory = Path(directory) |
| 391 | if not directory.exists(): |
| 392 | display.error(f"目录不存在: {directory}") |
| 393 | return [] |
| 394 | |
| 395 | videos = sorted(directory.rglob("*.mp4")) |
| 396 | |
| 397 | if skip_existing: |
| 398 | filtered = [] |
| 399 | for v in videos: |
| 400 | safe = _safe_stem(v.stem) |
| 401 | dirs_to_check = [v.parent] |
| 402 | if output_dir: |
| 403 | dirs_to_check.append(Path(output_dir)) |
| 404 | dirs_to_check.append(Path("./transcripts")) |
| 405 | found = any((d / f"{safe}.transcript.txt").exists() for d in dirs_to_check) |
| 406 | if found: |
| 407 | display.info(f"跳过 {safe[:50]}... (已有transcript)") |
| 408 | else: |
| 409 | filtered.append(v) |
| 410 | videos = filtered |
| 411 | |
| 412 | return videos |
| 413 | |
| 414 | |
| 415 | # ============================================================ |
| 416 | # Main |
| 417 | # ============================================================ |
| 418 | def main(): |
| 419 | parser = argparse.ArgumentParser( |
| 420 | description="Whisper 视频转录工具 — 批量语音识别", |
| 421 | formatter_class=argparse.RawDescriptionHelpFormatter, |
| 422 | epilog=( |
| 423 | "示例:\n" |
| 424 | " python whisper_transcribe.py -d ./Downloaded/\n" |
| 425 | " python whisper_transcribe.py -f video.mp4 -m medium\n" |
| 426 | " python whisper_transcribe.py -d ./Downloaded/ --srt --sc --skip-existing" |
| 427 | ), |
| 428 | ) |
| 429 | parser.add_argument("-d", "--dir", default="./Downloaded", help="视频目录 (默认 ./Downloaded/)") |
| 430 | parser.add_argument("-f", "--file", help="单个视频文件") |
| 431 | parser.add_argument( |
| 432 | "-m", |
| 433 | "--model", |
| 434 | default="base", |
| 435 | choices=["tiny", "base", "small", "medium", "large"], |
| 436 | help="Whisper模型 (默认 base)", |
| 437 | ) |
| 438 | parser.add_argument("-l", "--language", default="zh", help="语言 (默认 zh)") |
| 439 | parser.add_argument("--srt", action="store_true", help="同时输出SRT字幕") |
| 440 | parser.add_argument("--skip-existing", action="store_true", help="跳过已有transcript的视频") |
| 441 | parser.add_argument("--sc", action="store_true", help="繁体转简体 (需 pip install OpenCC)") |
| 442 | parser.add_argument( |
| 443 | "-o", |
| 444 | "--output", |
| 445 | default=None, |
| 446 | help="转录文件输出目录 (默认与视频同目录, 路径异常时自动fallback到 ./transcripts)", |
| 447 | ) |
| 448 | |
| 449 | args = parser.parse_args() |
| 450 | |
| 451 | # ── Banner ── |
| 452 | display.show_banner() |
| 453 | |
| 454 | # ── 依赖检查 ── |
| 455 | console.print(f" [{THEME['dim']}]检查依赖...[/]") |
| 456 | |
| 457 | ffmpeg_path = find_ffmpeg() |
| 458 | if not ffmpeg_path: |
| 459 | display.dep_fail("ffmpeg", "conda install -c conda-forge ffmpeg 或放 ffmpeg.exe 到同目录") |
| 460 | sys.exit(1) |
| 461 | display.dep_ok("ffmpeg", ffmpeg_path) |
| 462 | |
| 463 | try: |
| 464 | import whisper |
| 465 | except ImportError: |
| 466 | display.dep_fail("openai-whisper", "pip install openai-whisper") |
| 467 | sys.exit(1) |
| 468 | display.dep_ok("whisper", "已安装") |
| 469 | |
| 470 | converter = None |
| 471 | if args.sc: |
| 472 | try: |
| 473 | from opencc import OpenCC |
| 474 | |
| 475 | converter = OpenCC("t2s") |
| 476 | display.dep_ok("OpenCC", "繁体→简体") |
| 477 | except ImportError: |
| 478 | display.dep_fail("OpenCC", "pip install OpenCC") |
| 479 | sys.exit(1) |
| 480 | |
| 481 | console.print() |
| 482 | |
| 483 | # ── 收集视频 ── |
| 484 | if args.file: |
| 485 | videos = [Path(args.file)] |
| 486 | if not videos[0].exists(): |
| 487 | display.error(f"文件不存在: {args.file}") |
| 488 | sys.exit(1) |
| 489 | else: |
| 490 | videos = find_videos(args.dir, skip_existing=args.skip_existing, output_dir=args.output) |
| 491 | |
| 492 | if not videos: |
| 493 | display.warning("没有找到需要处理的视频文件") |
| 494 | return |
| 495 | |
| 496 | display.info(f"找到 {len(videos)} 个视频") |
| 497 | |
| 498 | # ── 加载模型 ── |
| 499 | display.info(f"加载 Whisper 模型: [{THEME['model']}]{args.model}[/] (首次需下载)") |
| 500 | model = whisper.load_model(args.model) |
| 501 | display.success(f"模型 [{THEME['model']}]{args.model}[/] 加载完成") |
| 502 | console.print() |
| 503 | |
| 504 | # ── 输出格式 ── |
| 505 | output_formats = {"txt"} |
| 506 | if args.srt: |
| 507 | output_formats.add("srt") |
| 508 | |
| 509 | # ── 处理 ── |
| 510 | display.start_session(len(videos)) |
| 511 | try: |
| 512 | for i, video in enumerate(videos, 1): |
| 513 | display.start_file(i, video.name) |
| 514 | try: |
| 515 | ok = transcribe_file( |
| 516 | video, model, ffmpeg_path, output_formats, args.language, converter, args.output |
| 517 | ) |
| 518 | display.complete_file( |
| 519 | "success" if ok else "failed", video.name if ok else "识别失败" |
| 520 | ) |
| 521 | except KeyboardInterrupt: |
| 522 | display.complete_file("failed", "用户中断") |
| 523 | raise |
| 524 | except Exception as e: |
| 525 | display.complete_file("failed", str(e)[:60]) |
| 526 | console.print(f" [{THEME['error']}]错误详情: {e}[/]") |
| 527 | import traceback |
| 528 | |
| 529 | console.print(f"[{THEME['dim']}]{traceback.format_exc()}[/]") |
| 530 | except KeyboardInterrupt: |
| 531 | display.warning("用户中断") |
| 532 | finally: |
| 533 | display.stop_session() |
| 534 | |
| 535 | # ── 汇总 ── |
| 536 | display.show_summary() |
| 537 | |
| 538 | |
| 539 | if __name__ == "__main__": |
| 540 | main() |
| 541 |