From 7de5ab3902270c1e35546cacf6ddaa0f681657f1 Mon Sep 17 00:00:00 2001 From: hmo Date: Sun, 16 Aug 2026 20:17:14 +0800 Subject: [PATCH] =?UTF-8?q?subtitle-studio:=20=E5=AD=97=E5=B9=95=E7=94=9F?= =?UTF-8?q?=E6=88=90=E7=B3=BB=E7=BB=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 选择文件夹/单文件 → SenseVoice 转录 → LLM 修正 → 多路线翻译 → 双语 SRT - 路线：直译中文 / 英转中 / 仅转录 - 并发流水线：转录(可调) + LLM 并发(可调)，降噪拆锁并发 - 断点续跑：.subtitle-work/ 中间产物，三阶段独立续跑 + 翻译段级续跑 - 去重：history 跟随视频文件夹，防重复任务保护 - 实时进度：SSE 推送 + 耗时显示 + 子任务状态 + 重新生成按钮 - 时间戳调优：VAD silence_schedule + noisereduce 降噪 + 完整性校验 --- .gitignore | 17 + ARCHITECTURE.md | 124 +++++++ README.md | 63 ++++ scripts/start.bat | 46 +++ scripts/start_hidden.bat | 56 +++ scripts/start_server.py | 55 +++ scripts/stop.bat | 23 ++ src/substudio/__init__.py | 1 + src/substudio/config.py | 36 ++ src/substudio/llm.py | 69 ++++ src/substudio/main.py | 300 ++++++++++++++++ src/substudio/pipeline/__init__.py | 0 src/substudio/pipeline/_fix_worker.py | 50 +++ src/substudio/pipeline/fix.py | 93 +++++ src/substudio/pipeline/srt.py | 70 ++++ src/substudio/pipeline/transcribe.py | 258 ++++++++++++++ src/substudio/pipeline/translate.py | 96 ++++++ src/substudio/taskmanager.py | 470 ++++++++++++++++++++++++++ static/app.js | 438 ++++++++++++++++++++++++ templates/index.html | 145 ++++++++ 20 files changed, 2410 insertions(+) create mode 100644 .gitignore create mode 100644 ARCHITECTURE.md create mode 100644 README.md create mode 100644 scripts/start.bat create mode 100644 scripts/start_hidden.bat create mode 100644 scripts/start_server.py create mode 100644 scripts/stop.bat create mode 100644 src/substudio/__init__.py create mode 100644 src/substudio/config.py create mode 100644 src/substudio/llm.py create mode 100644 src/substudio/main.py create mode 100644 src/substudio/pipeline/__init__.py create mode 100644 src/substudio/pipeline/_fix_worker.py create mode 100644 src/substudio/pipeline/fix.py create mode 100644 src/substudio/pipeline/srt.py create mode 100644 src/substudio/pipeline/transcribe.py create mode 100644 src/substudio/pipeline/translate.py create mode 100644 src/substudio/taskmanager.py create mode 100644 static/app.js create mode 100644 templates/index.html diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..6cc4c08 --- /dev/null +++ b/.gitignore @@ -0,0 +1,17 @@ +# 运行时产物 +logs/ +output/ +__pycache__/ +*.pyc + +# 环境与密钥 +.env +env.local +*.env + +# 临时 +temp/ +*.log + +# 中间产物（在视频目录下，不提交） +.subtitle-work/ diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md new file mode 100644 index 0000000..1420b7f --- /dev/null +++ b/ARCHITECTURE.md @@ -0,0 +1,124 @@ +# 字幕生成系统 subtitle-studio — 架构设计 + +> 设计日期：2026-08-15 ｜ 架构：纯 Windows 本地，自包含 ｜ 目标：文件夹/单文件 → 多路线字幕 + +## 1. 总体架构 + +``` +┌─────────────────────────── Windows 本机 ───────────────────────────┐ +│ │ +│ [Web 前端] 浏览器页面 │ +│ - 选择文件夹 / 单个文件 │ +│ - 选择翻译路线（直译 / 英转中 / 仅转录） │ +│ - 实时查看任务进度（SSE 推送） │ +│ - 已完成视频列表 + 字幕文件下载/打开 │ +│ │ HTTP/SSE │ +│ [FastAPI 后端] :8788 │ +│ - 任务管理器（队列 + 并发控制，默认 1 任务串行防 GPU 争抢） │ +│ - 进度状态机：pending → transcribing → fixing → translating → srt → done │ +│ │ │ +│ [管道核心] subsudio.pipeline │ +│ - transcribe.py SenseVoice + fsmn-vad（GPU，16kHz wav） │ +│ - fix.py LLM 分块并发修正日文/韩文转录 │ +│ - translate.py LLM 全文翻译（路线1 直译 / 路线2 英转中） │ +│ - srt.py 生成 .ja.zh.srt / .zh.srt / .zh.en.srt │ +│ │ │ +│ [运行时] py312_cuda（conda） + OCG Router (192.168.1.246:19878) │ +└─────────────────────────────────────────────────────────────────────┘ +``` + +## 2. 设计决策 + +### D1. 复用 py312_cuda 环境，不新建 venv +- 转录需要 faster-whisper/funasr/torch-CUDA，已全部就绪于 `D:\ProgramData\anaconda3\envs\py312_cuda` +- 后端 FastAPI/uvicorn 也装这个环境（检查缺包再补） +- 避免重复装 CUDA 依赖（虚拟环境铁律：不污染系统，但可复用专用环境） + +### D2. LLM 走 OCG Router（本地中转，自动选 key） +- base_url `http://192.168.1.246:19878/v1`，apiKey `ocg-router-local` +- 模型 `deepseek-v4-flash`（1M 上下文，全文一次请求） +- 关键经验：**不指定 max_tokens**（推理模型会烧光截断）；全文翻译 1746 段 105 秒 + +### D3. 翻译路线（用户可选） +| 路线 | 流程 | 产物 | +|------|------|------| +| direct (默认) | 转录 → LLM修正 → 直译中文 | .ja.zh.srt + .zh.srt | +| via_en | 转录 → LLM修正 → 英转中 | .r2.zh.en.srt + .r2.zh.srt | +| transcribe_only | 只转录，不翻译 | _transcript.json | + +### D4. 任务队列串行执行 +- GPU 转录是重资源操作，一次只跑 1 个任务 +- 翻译阶段可并发分块（fix 用 4 worker） +- 进度通过 SSE 实时推送到前端 + +### D5. 输出目录 +- 默认：字幕文件生成到**视频同目录**（PotPlayer 自动加载） +- 同时复制一份到 `output//` 便于管理/下载 + +## 3. 目录布局 + +``` +subtitle-studio/ +├── src/substudio/ +│ ├── __init__.py +│ ├── main.py # FastAPI 入口（uvicorn） +│ ├── taskmanager.py # 任务队列 + 状态机 + SSE 广播 +│ ├── pipeline/ +│ │ ├── __init__.py +│ │ ├── transcribe.py # SenseVoice 转录（复用技能逻辑） +│ │ ├── fix.py # LLM 分块并发修正 +│ │ ├── translate.py # LLM 全文翻译（路线选择） +│ │ └── srt.py # SRT 生成 +│ ├── config.py # 路径/模型/API 配置 +│ └── llm.py # OCG Router 客户端（流式兼容） +├── templates/index.html # 前端页面 +├── static/app.js # 前端逻辑 +├── output/ # 任务输出 +├── scripts/start.bat # 启动脚本 +└── README.md +``` + +## 4. 进度状态机 + +``` +pending → transcribing → fixing → translating → srt → done + └────── error（失败即停，可重试） +``` + +每阶段有 `progress`（0-100）和 `message`（如 "转录中 320/1746 段"） +SSE 事件：`task_update` 推送 `{task_id, status, progress, message}` + +## 5. 关键技术点 + +### 5.1 转录（复用已验证逻辑） +- ffmpeg 提取 16kHz 单声道 wav +- **降噪（默认开启）**：noisereduce 频谱门控，prop_decrease=0.7（0.9 过度会字间隙），实测 RMS 降 64%，句子边界更清晰 +- **fsmn-vad 切段（时间戳精准关键）**： + - `max_single_segment_time` 参数**无效**（实测） + - 必须传自定义 `silence_schedule`：`[(8000,500),(12000,300),(20000,200),(inf,100)]` + - 效果：段长从默认 15.67s → 4.64s 上限，平均 1.59s，时间戳严重错位修复 + - 根因：默认 schedule 产生长段 → 只能按字符比例估算句内时间 → 字幕与说话时间对不上 +- SenseVoiceSmall 逐段转录，rich_transcription_postprocess 清理 +- 输出句子级时间戳 JSON + +### 5.2 修正（分块并发） +- 按字符量分块（块边界=段边界，不切断句子） +- 每块一个 worker 进程并发（4 worker） +- prompt：结合上下文修正 ASR 错误，保留 [idx] + +### 5.3 翻译（全文一次请求） +- **不指定 max_tokens**（1M 上下文） +- 全文 30K 字符一次请求，1746 段 105 秒 +- 输出 [idx] 行式解析 + +### 5.4 OCG Router 兼容 +- 兼容 SSE 流式返回（data: {...} 解析） +- 有限重试（3 次退避，不疯狂重试） +- 失败即停 + 进度保存（断点续传） + +## 6. 安全与资源 + +- 无 Docker 操作，纯本地文件处理 +- 只读视频，写入同目录 SRT + output/ +- 一次一个任务（GPU 转录），翻译并发仅限 LLM API +- 不碰系统环境（复用 py312_cuda 专用环境） diff --git a/README.md b/README.md new file mode 100644 index 0000000..7144759 --- /dev/null +++ b/README.md @@ -0,0 +1,63 @@ +# subtitle-studio 字幕生成系统 + +选择文件夹或单个视频 → 自动生成多路线双语字幕，实时查看进度。 + +## 快速开始 + +```powershell +# 方式一（推荐）：后台启动，无窗口，日志写 logs/server.log +scripts\start_hidden.bat + +# 方式二：前台启动（关窗口即停止） +scripts\start.bat + +# 停止后台服务 +scripts\stop.bat +``` + +浏览器打开 **http://127.0.0.1:8788**（start_hidden.bat 会自动打开） + +## 功能 + +1. **选择视频**：浏览文件系统（支持盘符切换/上级目录），或直接输入路径；选文件夹批量处理 +2. **选择路线**： + - 路线 A 直译中文：转录 → LLM 修正 → 日文直译中文 → `.ja.zh.srt` + `.zh.srt` + - 路线 B 英转中：转录 → LLM 修正 → 日→英→中 → `.r2.zh.en.srt` + `.r2.zh.srt` + - 仅转录：只做语音识别 → `_transcript.json` +3. **实时进度**：SSE 推送任务状态（等待/转录/修正/翻译/生成字幕/完成/失败） +4. **完成列表**：已生成字幕文件可直接下载，失败任务可一键重试 + +## 技术栈 + +| 组件 | 说明 | +|------|------| +| 前端 | 原生 HTML/JS，SSE 实时进度 | +| 后端 | FastAPI + uvicorn（:8788） | +| 转录 | SenseVoiceSmall + fsmn-vad（GPU，py312_cuda 环境） | +| 修正 | LLM 分块并发（4 worker） | +| 翻译 | DeepSeek V4 Flash 全文一次请求（1M 上下文，不指定 max_tokens） | +| LLM | OCG Router（192.168.1.246:19878，自动选 key） | + +## 目录 + +``` +subtitle-studio/ +├── src/substudio/ +│ ├── main.py # FastAPI 入口 +│ ├── taskmanager.py # 任务队列 + 状态机 + SSE +│ ├── llm.py # OCG Router 客户端（流式兼容） +│ ├── config.py # 配置 +│ └── pipeline/ +│ ├── transcribe.py # SenseVoice 转录 +│ ├── fix.py # 分块并发修正 +│ ├── translate.py # 路线翻译 +│ └── srt.py # SRT 生成 +├── templates/index.html +├── static/app.js +├── output/ # 任务输出（含副本） +└── scripts/start.bat +``` + +## 架构文档 + +见 [ARCHITECTURE.md](ARCHITECTURE.md) diff --git a/scripts/start.bat b/scripts/start.bat new file mode 100644 index 0000000..9d8fd48 --- /dev/null +++ b/scripts/start.bat @@ -0,0 +1,46 @@ +@echo off +chcp 65001 >nul +title subtitle-studio 字幕生成系统 +setlocal + +REM ===== subtitle-studio 启动脚本 ===== +REM 自动定位项目根目录（bat 所在目录的上上级） +set SCRIPT_DIR=%~dp0 +set PROJECT_ROOT=%SCRIPT_DIR%.. +set APP_DIR=%PROJECT_ROOT%\src +set PYTHON=D:\ProgramData\anaconda3\envs\py312_cuda\python.exe +set PORT=8788 + +REM 检查 Python 环境 +if not exist "%PYTHON%" ( + echo [错误] 找不到 Python 环境: %PYTHON% + echo 请确认 py312_cuda 环境存在 + pause + exit /b 1 +) + +REM 检查端口是否已被占用（已有实例在跑） +netstat -ano | findstr ":%PORT%" | findstr "LISTENING" >nul 2>&1 +if %errorlevel%==0 ( + echo subtitle-studio 已在运行: http://127.0.0.1:%PORT% + start http://127.0.0.1:%PORT% + exit /b 0 +) + +echo ============================================ +echo subtitle-studio 字幕生成系统 +echo 服务地址: http://127.0.0.1:%PORT% +echo 关闭本窗口即停止服务 +echo ============================================ +echo. + +REM 后台清理旧日志（可选） +if exist "%PROJECT_ROOT%\logs\server.log" del "%PROJECT_ROOT%\logs\server.log" >nul 2>&1 + +REM 启动服务（前台，窗口关闭即停止） +cd /d "%PROJECT_ROOT%" +"%PYTHON%" -m uvicorn substudio.main:app --host 127.0.0.1 --port %PORT% --app-dir "%APP_DIR%" + +echo. +echo 服务已停止。 +pause diff --git a/scripts/start_hidden.bat b/scripts/start_hidden.bat new file mode 100644 index 0000000..cdf31a6 --- /dev/null +++ b/scripts/start_hidden.bat @@ -0,0 +1,56 @@ +@echo off +chcp 65001 >nul +title subtitle-studio 启动器 +setlocal + +REM ===== subtitle-studio 后台启动（无窗口）===== +set SCRIPT_DIR=%~dp0 +set PROJECT_ROOT=%SCRIPT_DIR%.. +set APP_DIR=%PROJECT_ROOT%\src +set PYTHON=D:\ProgramData\anaconda3\envs\py312_cuda\python.exe +set PORT=8788 +set LOG_DIR=%PROJECT_ROOT%\logs + +if not exist "%PYTHON%" ( + echo [错误] 找不到 Python 环境: %PYTHON% + pause + exit /b 1 +) + +netstat -ano | findstr ":%PORT%" | findstr "LISTENING" >nul 2>&1 +if %errorlevel%==0 ( + echo subtitle-studio 已在运行: http://127.0.0.1:%PORT% + start http://127.0.0.1:%PORT% + exit /b 0 +) + +if not exist "%LOG_DIR%" mkdir "%LOG_DIR%" + +REM 后台启动：用 cmd /c 包装重定向，start 本身不阻塞 +cd /d "%PROJECT_ROOT%" +start "subtitle-studio-server" /min cmd /c ""%PYTHON%" -m uvicorn substudio.main:app --host 127.0.0.1 --port %PORT% --app-dir "%APP_DIR%" >"%LOG_DIR%\server.log" 2>&1" + +REM 等待端口就绪 +set /a tries=0 +:waitloop +set /a tries+=1 +if %tries% GTR 20 goto timeout +netstat -ano | findstr ":%PORT%" | findstr "LISTENING" >nul 2>&1 +if %errorlevel%==0 goto ready +timeout /t 1 /nobreak >nul +goto waitloop + +:ready +echo. +echo ✅ subtitle-studio 已启动 +echo http://127.0.0.1:%PORT% +echo 日志: %LOG_DIR%\server.log +echo. +start http://127.0.0.1:%PORT% +exit /b 0 + +:timeout +echo. +echo ⚠️ 启动超时，请查看日志: %LOG_DIR%\server.log +pause +exit /b 1 diff --git a/scripts/start_server.py b/scripts/start_server.py new file mode 100644 index 0000000..f2baac1 --- /dev/null +++ b/scripts/start_server.py @@ -0,0 +1,55 @@ +# -*- coding: utf-8 -*- +""" +subtitle-studio 后台启动器（无阻塞） +用 subprocess.Popen 启动 uvicorn，shell 调用立即返回 +日志写到 logs/server.log +用法: python start_server.py [port] +""" +import sys, os, subprocess, time, threading + +PYTHON = sys.executable +PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +APP_DIR = os.path.join(PROJECT_ROOT, "src") +LOG_DIR = os.path.join(PROJECT_ROOT, "logs") +PORT = sys.argv[1] if len(sys.argv) > 1 else "8788" + +os.makedirs(LOG_DIR, exist_ok=True) +log_path = os.path.join(LOG_DIR, "server.log") + + +def is_running(port): + """端口是否已被监听""" + try: + import socket + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.settimeout(1) + return s.connect_ex(("127.0.0.1", int(port))) == 0 + except Exception: + return False + + +def main(): + if is_running(PORT): + print(f"subtitle-studio 已在运行: http://127.0.0.1:{PORT}") + return + + # 打开日志文件（追加模式） + logf = open(log_path, 'a', encoding='utf-8', buffering=1) + + # 真正的异步启动：不等待，不继承句柄 + proc = subprocess.Popen( + [PYTHON, "-m", "uvicorn", "substudio.main:app", + "--host", "127.0.0.1", "--port", PORT, "--app-dir", APP_DIR], + cwd=PROJECT_ROOT, + stdout=logf, + stderr=subprocess.STDOUT, + creationflags=subprocess.CREATE_NO_WINDOW, + ) + print(f"已启动 subtitle-studio (PID {proc.pid})") + print(f"服务地址: http://127.0.0.1:{PORT}") + print(f"日志: {log_path}") + # 不等待进程，直接返回 + + +if __name__ == "__main__": + main() diff --git a/scripts/stop.bat b/scripts/stop.bat new file mode 100644 index 0000000..77f25be --- /dev/null +++ b/scripts/stop.bat @@ -0,0 +1,23 @@ +@echo off +chcp 65001 >nul +title subtitle-studio 停止 +setlocal + +REM ===== 停止 subtitle-studio 服务 ===== +set PORT=8788 + +echo 查找端口 %PORT% 上的进程... +for /f "tokens=5" %%p in ('netstat -ano ^| findstr ":%PORT%" ^| findstr "LISTENING"') do ( + echo 停止 PID %%p + taskkill /f /pid %%p >nul 2>&1 +) + +timeout /t 1 /nobreak >nul + +netstat -ano | findstr ":%PORT%" | findstr "LISTENING" >nul 2>&1 +if %errorlevel%==0 ( + echo ⚠️ 端口仍被占用，可能需要手动结束 +) else ( + echo ✅ subtitle-studio 已停止 +) +pause diff --git a/src/substudio/__init__.py b/src/substudio/__init__.py new file mode 100644 index 0000000..40a96af --- /dev/null +++ b/src/substudio/__init__.py @@ -0,0 +1 @@ +# -*- coding: utf-8 -*- diff --git a/src/substudio/config.py b/src/substudio/config.py new file mode 100644 index 0000000..43f9e7f --- /dev/null +++ b/src/substudio/config.py @@ -0,0 +1,36 @@ +# -*- coding: utf-8 -*- +"""subtitle-studio 配置""" +import os + +# 环境路径 +PYTHON = r"D:\ProgramData\anaconda3\envs\py312_cuda\python.exe" +FFMPEG = r"D:\ProgramData\anaconda3\envs\py312_cuda\ffmpeg.exe" + +# SenseVoice 模型缓存 +MODEL_DIR = r"C:\Users\hmo\.cache\modelscope\models\iic--SenseVoiceSmall\snapshots\master" +VAD_DIR = r"C:\Users\hmo\.cache\modelscope\models\iic--speech_fsmn_vad_zh-cn-16k-common-pytorch\snapshots\master" + +# OCG Router（可用环境变量覆盖：LLM_BASE_URL / LLM_API_KEY / LLM_MODEL） +LLM_BASE_URL = os.environ.get("LLM_BASE_URL", "http://192.168.1.246:19878/v1") +LLM_API_KEY = os.environ.get("LLM_API_KEY", "ocg-router-local") +LLM_MODEL = os.environ.get("LLM_MODEL", "deepseek-v4-flash") + +# 翻译路线 +ROUTES = { + "direct": "直译中文（转录→修正→日文直译中文）", + "via_en": "英转中（转录→修正→日文→英文→中文）", + "transcribe_only": "仅转录（不翻译，输出 transcript.json）", +} + +# 并发 +FIX_WORKERS = 4 # 修正分块并发数 +FIX_BLOCK_CHARS = 5000 # 每块字符数 +MAX_CONCURRENT_TASKS = 2 # 转录并发（GPU 模型单例共享，16G 显存可并发 2；CPU 为实际瓶颈） +LLM_CONCURRENCY = 6 # LLM 并发数（修正/翻译，3 key × 2，可调） + +# 输出（项目根 output/，即 src/substudio/config.py 上三级） +_OUTPUT_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), "output") +OUTPUT_DIR = _OUTPUT_DIR + +# 中间产物统一子文件夹（放在视频所在文件夹下，固定名，一个文件夹存所有中间产物） +WORK_DIRNAME = ".subtitle-work" diff --git a/src/substudio/llm.py b/src/substudio/llm.py new file mode 100644 index 0000000..ce39ab3 --- /dev/null +++ b/src/substudio/llm.py @@ -0,0 +1,69 @@ +# -*- coding: utf-8 -*- +"""OCG Router LLM 客户端（兼容 SSE 流式返回 + 有限重试）""" +import json, time, re +import urllib.request +import urllib.error + +from .config import LLM_BASE_URL, LLM_API_KEY, LLM_MODEL + + +def call_llm(messages, timeout=180, temperature=0.3): + """调用 LLM，兼容 SSE 流式返回和普通 JSON 返回。 + 关键：不指定 max_tokens（推理模型会烧光截断） + timeout=180：OCG Router 内部已有 30s 首字节超时+换key，客户端 180s 足够， + 避免卡住请求等 20 分钟才重试 + """ + body = json.dumps({ + "model": LLM_MODEL, + "messages": messages, + "temperature": temperature, + }).encode('utf-8') + req = urllib.request.Request(LLM_BASE_URL + "/chat/completions", data=body, headers={ + "Content-Type": "application/json", + "Authorization": f"Bearer {LLM_API_KEY}", + "Accept": "text/event-stream", + }) + last_ex = None + for attempt in range(3): # 有限重试，失败即停，不疯狂重试 + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + data = resp.read() + text = data.decode('utf-8', errors='replace') + if text.strip().startswith('data:'): + # SSE 流式：拼 content + full = [] + for ev in text.split('\n\n'): + for line in ev.split('\n'): + if line.startswith('data: '): + ds = line[6:].strip() + if ds == '[DONE]': + continue + try: + obj = json.loads(ds) + delta = obj["choices"][0]["delta"].get("content", "") + if delta: + full.append(delta) + except Exception: + pass + return "".join(full) + return json.loads(text)["choices"][0]["message"].get("content", "") + except urllib.error.HTTPError as ex: + last_ex = ex + if ex.code in (400, 429) or 500 <= ex.code < 600: + time.sleep(2 ** attempt) + continue + raise + except Exception as ex: + last_ex = ex + time.sleep(2 ** attempt) + raise last_ex + + +def parse_idx_lines(resp): + """解析 [idx] 行式输出 -> {idx: text}""" + result = {} + for line in resp.splitlines(): + m = re.match(r'^\[(\d+)\]\s*(.+)$', line.strip()) + if m and m.group(2).strip(): + result[int(m.group(1))] = m.group(2).strip() + return result diff --git a/src/substudio/main.py b/src/substudio/main.py new file mode 100644 index 0000000..17843f5 --- /dev/null +++ b/src/substudio/main.py @@ -0,0 +1,300 @@ +# -*- coding: utf-8 -*- +"""subtitle-studio 主入口：FastAPI Web 服务""" +import os, sys, json, queue, glob +import threading + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from fastapi import FastAPI, Request +from fastapi.responses import HTMLResponse, StreamingResponse, FileResponse, JSONResponse +from fastapi.staticfiles import StaticFiles +from fastapi.templating import Jinja2Templates +from pydantic import BaseModel + +from substudio.taskmanager import manager +from substudio.config import ROUTES + +PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +app = FastAPI(title="subtitle-studio") +templates = Jinja2Templates(directory=os.path.join(PROJECT_ROOT, "templates")) + +# 静态文件 +static_dir = os.path.join(PROJECT_ROOT, "static") +if os.path.isdir(static_dir): + app.mount("/static", StaticFiles(directory=static_dir), name="static") + + +@app.middleware("http") +async def no_cache(request, call_next): + """禁用缓存：开发期确保前端改动立即生效""" + response = await call_next(request) + if request.url.path.startswith("/static") or request.url.path == "/": + response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate" + response.headers["Pragma"] = "no-cache" + return response + + +class CreateTaskReq(BaseModel): + source: str + route: str = "direct" + language: str = "ja" + + +@app.get("/", response_class=HTMLResponse) +def index(request: Request): + return templates.TemplateResponse("index.html", {"request": request, "routes": ROUTES}) + + +@app.post("/api/tasks") +def create_task(req: CreateTaskReq): + if not os.path.exists(req.source): + return JSONResponse({"error": f"路径不存在: {req.source}"}, status_code=400) + if req.route not in ROUTES: + return JSONResponse({"error": f"未知路线: {req.route}"}, status_code=400) + task, skipped, reason = manager.create_task(req.source, req.route, req.language) + if task is None: + # 正在处理中 或 全部已处理过 + notice = reason or f"所选内容已用该路线处理过（{len(skipped)} 个视频），未重复创建任务" + return JSONResponse({ + "error": None, + "skipped_all": True, + "skipped": [os.path.basename(s) for s in skipped], + "notice": notice, + }, status_code=200) + resp = task.to_dict() + if skipped: + resp["skipped"] = [os.path.basename(s) for s in skipped] + resp["notice"] = f"跳过已处理的 {len(skipped)} 个视频（同路线去重）" + return resp + + +@app.get("/api/tasks") +def list_tasks(): + return manager.list_tasks() + + +@app.get("/api/tasks/{tid}") +def get_task(tid: str): + t = manager.get_task(tid) + if not t: + return JSONResponse({"error": "not found"}, status_code=404) + return t + + +@app.post("/api/tasks/{tid}/retry") +def retry_task(tid: str): + t = manager.retry_task(tid) + if not t: + return JSONResponse({"error": "无法重试"}, status_code=400) + return t + + +@app.post("/api/tasks/{tid}/cancel") +def cancel_task(tid: str): + t = manager.cancel_task(tid) + if not t: + return JSONResponse({"error": "not found"}, status_code=404) + return t + + +@app.get("/api/events") +def sse(request: Request): + """SSE 实时进度推送""" + q = queue.Queue() + manager.subscribe(q) + + def gen(): + try: + # 先发当前全部任务快照 + yield f"data: {json.dumps({'type': 'snapshot', 'tasks': manager.list_tasks()}, ensure_ascii=False)}\n\n" + while True: + if request.is_disconnected: + break + try: + ev = q.get(timeout=10) + yield f"data: {ev}\n\n" + except queue.Empty: + yield ": keepalive\n\n" + finally: + manager.unsubscribe(q) + + return StreamingResponse(gen(), media_type="text/event-stream") + + +@app.get("/api/browse") +def browse(path: str = ""): + """浏览文件系统，用于前端选择文件/文件夹""" + base = path if path else os.path.expanduser("~") + if not os.path.isdir(base): + base = os.path.dirname(base) + entries = [] + try: + for name in sorted(os.listdir(base)): + full = os.path.join(base, name) + entries.append({ + "name": name, + "path": full, + "is_dir": os.path.isdir(full), + }) + except Exception: + pass + return {"cwd": base, "entries": entries} + + +@app.get("/api/drives") +def drives(): + import string + out = [] + for letter in string.ascii_uppercase: + if os.path.exists(f"{letter}:\\"): + out.append(f"{letter}:\\") + return {"drives": out} + + +@app.get("/api/download") +def download(path: str = ""): + if not path or not os.path.isfile(path): + return JSONResponse({"error": "not found"}, status_code=404) + return FileResponse(path, filename=os.path.basename(path)) + + +@app.post("/api/video/regenerate") +def regenerate_video(req: CreateTaskReq): + """重新生成单个视频：清空 .subtitle-work/ 中间产物 + SRT + history 记录，然后重跑""" + from substudio.config import WORK_DIRNAME + video = req.source + route = req.route + if not os.path.isfile(video): + return JSONResponse({"error": f"不是文件: {video}"}, status_code=400) + folder = os.path.dirname(video) + base = os.path.splitext(os.path.basename(video))[0] + work_dir = os.path.join(folder, WORK_DIRNAME) + + # 1. 删除中间产物 + removed = [] + for suffix in ("_transcript.json", "_fixed.json", "_translated.json", "_audio.wav"): + fp = os.path.join(work_dir, f"{base}{suffix}") + if os.path.exists(fp): + os.remove(fp) + removed.append(os.path.basename(fp)) + + # 2. 删除 SRT（视频同目录） + for suffix in (".ja.zh.srt", ".zh.srt", ".r2.zh.en.srt", ".r2.zh.srt"): + fp = os.path.splitext(video)[0] + suffix + if os.path.exists(fp): + os.remove(fp) + removed.append(os.path.basename(fp)) + + # 3. 删除 history 记录 + from substudio.taskmanager import _load_history, _save_history + history = _load_history(video) + hkey = f"{video}|{route}" + if hkey in history: + del history[hkey] + _save_history(video, history) + + # 4. 创建新任务重跑 + task, skipped, reason = manager.create_task(video, route, req.language) + return {"removed": removed, "task_id": task.id if task else None, "status": "ok", "notice": reason} + + +@app.get("/api/folder_status") +def folder_status(request: Request): + """选定文件夹后，返回每个视频的处理状态（供任务列表自动加载） + 状态推断：.subtitle-work/ 中间产物 + history + """ + from substudio.taskmanager import _load_history + from substudio.config import WORK_DIRNAME + from .pipeline.transcribe import VIDEO_EXTS + + path = request.query_params.get("path", "") + route = request.query_params.get("route", "direct") + if not os.path.isdir(path): + return {"error": f"不是文件夹: {path}"} + + work_dir = os.path.join(path, WORK_DIRNAME) + history = _load_history(path) + + videos = [] + for f in sorted(os.listdir(path)): + if not f.lower().endswith(VIDEO_EXTS): + continue + full = os.path.join(path, f) + base = os.path.splitext(f)[0] + # 中间产物存在性 + has_transcript = os.path.exists(os.path.join(work_dir, f"{base}_transcript.json")) + has_fixed = os.path.exists(os.path.join(work_dir, f"{base}_fixed.json")) + has_translated = os.path.exists(os.path.join(work_dir, f"{base}_translated.json")) + # history 完成记录 + hkey = f"{full}|{route}" + done = hkey in history + warnings = history[hkey].get("warnings", []) if done else [] + + # 推断状态 + if done: + status = "done" + progress = 100 + message = "已完成" + ("（含警告）" if warnings else "") + elif has_translated: + status = "translating" # 翻译中（部分完成，可续跑） + progress = 70 + message = "翻译部分完成（可续跑）" + elif has_fixed: + status = "fixing" + progress = 45 + message = "修正完成，待翻译（可续跑）" + elif has_transcript: + status = "transcribing" + progress = 20 + message = "转录完成，待修正（可续跑）" + else: + status = "pending" + progress = 0 + message = "未开始" + + videos.append({ + "video": full, + "name": f, + "status": status, + "progress": progress, + "message": message, + "warnings": warnings, + }) + + return {"path": path, "work_dir": work_dir, "videos": videos} + + +class ConcurrencyReq(BaseModel): + llm: int = 6 + transcribe: int = 2 + + +@app.get("/api/concurrency") +def get_concurrency(): + """获取当前并发配置""" + from substudio.config import MAX_CONCURRENT_TASKS, LLM_CONCURRENCY + return { + "transcribe": manager.transcribe_concurrency, # 转录并发（GPU 模型共享） + "llm": manager.llm_concurrency, # LLM 并发（可调） + } + + +@app.post("/api/concurrency") +def set_concurrency(req: ConcurrencyReq): + """调整并发数（transcribe 1-4, llm 1-20，可调）""" + t = manager.set_transcribe_concurrency(req.transcribe) if req.transcribe else None + l = manager.set_llm_concurrency(req.llm) if req.llm else None + notice = [] + if t is not None: + notice.append(f"转录并发 {t}") + if l is not None: + notice.append(f"LLM 并发 {l}") + return {"transcribe": manager.transcribe_concurrency, "llm": manager.llm_concurrency, + "notice": "，".join(notice)} + + +if __name__ == "__main__": + import uvicorn + port = int(sys.argv[1]) if len(sys.argv) > 1 else 8788 + print(f"subtitle-studio http://127.0.0.1:{port}") + uvicorn.run(app, host="127.0.0.1", port=port) diff --git a/src/substudio/pipeline/__init__.py b/src/substudio/pipeline/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/substudio/pipeline/_fix_worker.py b/src/substudio/pipeline/_fix_worker.py new file mode 100644 index 0000000..d22fd4d --- /dev/null +++ b/src/substudio/pipeline/_fix_worker.py @@ -0,0 +1,50 @@ +# -*- coding: utf-8 -*- +"""修正 worker：单块日文转录修正（流式接收）""" +import sys, io, os, json, time, re +sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace') + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) +from substudio.llm import call_llm, parse_idx_lines + +FIX_PROMPT = ( + "You are an expert in Japanese speech recognition (ASR) post-processing. " + "The following lines are automatic speech recognition output from a Japanese video. " + "Many lines contain recognition errors: wrong kanji, missing or extra particles, " + "misheard words, broken segments. " + "\n\n" + "For EACH line:\n" + "1. Fix obvious ASR errors using the surrounding context to infer the correct Japanese.\n" + "2. If a line is already correct, return it unchanged.\n" + "3. Keep the [index] prefix exactly like [0], [1].\n" + "4. Return ONLY the CORRECTED Japanese text, one line per input line, in the same order.\n" + "5. Do NOT translate, do NOT merge lines, do NOT omit any line." +) + + +def fix_block(items): + lines = "\n".join(f"[{i}] {it['text']}" for i, it in enumerate(items)) + resp = call_llm([ + {"role": "system", "content": FIX_PROMPT}, + {"role": "user", "content": lines}, + ], temperature=0.1) + result = parse_idx_lines(resp) + out = [] + for i, it in enumerate(items): + fixed = result.get(i, it['text']) + out.append({**it, 'text': fixed}) + return out + + +def main(): + in_path = sys.argv[1] + out_path = sys.argv[2] + with open(in_path, encoding='utf-8') as f: + items = json.load(f) + out = fix_block(items) + with open(out_path, 'w', encoding='utf-8') as f: + json.dump(out, f, ensure_ascii=False, indent=1) + print(f"worker DONE: {len(out)} -> {out_path}") + + +if __name__ == "__main__": + main() diff --git a/src/substudio/pipeline/fix.py b/src/substudio/pipeline/fix.py new file mode 100644 index 0000000..5a93dc8 --- /dev/null +++ b/src/substudio/pipeline/fix.py @@ -0,0 +1,93 @@ +# -*- coding: utf-8 -*- +"""LLM 修正管道：按字符量分块，并发 worker 修正，保留 [idx]""" +import os, json, subprocess, sys, time + +from ..config import PYTHON, FIX_WORKERS, FIX_BLOCK_CHARS + +WORKER = os.path.join(os.path.dirname(os.path.abspath(__file__)), "_fix_worker.py") + + +def split_blocks(items, num_blocks): + """按字符量尽量均分 N 块（块边界=段边界，不切断句子）""" + total_chars = sum(len(it['text']) for it in items) + target = total_chars / num_blocks + blocks = [] + cur_block, cur_chars = [], 0 + for i, it in enumerate(items): + cur_block.append(i) + cur_chars += len(it['text']) + if cur_chars >= target and len(blocks) < num_blocks - 1: + blocks.append(cur_block) + cur_block, cur_chars = [], 0 + if cur_block: + blocks.append(cur_block) + return blocks + + +def fix(items, progress_cb=None): + """修正转录文本（分块并发）。items: [{start_ms, end_ms, text}] -> 修正后的同结构 list""" + def report(msg, pct): + if progress_cb: + progress_cb(msg, pct) + + n = len(items) + # 按字符量切块（动态块数，每块 ≤ FIX_BLOCK_CHARS） + blocks = [] + cur_block, cur_chars = [], 0 + for i, it in enumerate(items): + cur_block.append(i) + cur_chars += len(it['text']) + if cur_chars >= FIX_BLOCK_CHARS: + blocks.append(cur_block) + cur_block, cur_chars = [], 0 + if cur_block: + blocks.append(cur_block) + + report(f"修正分块 {len(blocks)} 块，启动并发 worker...", 0) + tmp_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "_fix_tmp") + os.makedirs(tmp_dir, exist_ok=True) + + procs = [] + tasks = [] + for b_idx, blk in enumerate(blocks): + blk_items = [dict(items[i]) for i in blk] + bfile = os.path.join(tmp_dir, f"block{b_idx}.json") + bfile_out = os.path.join(tmp_dir, f"block{b_idx}_out.json") + with open(bfile, 'w', encoding='utf-8') as f: + json.dump(blk_items, f, ensure_ascii=False, indent=1) + tasks.append((b_idx, blk, bfile, bfile_out)) + p = subprocess.Popen( + [PYTHON, WORKER, bfile, bfile_out], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + ) + procs.append(p) + + t0 = time.time() + done_blocks = 0 + while any(p.poll() is None for p in procs): + time.sleep(3) + cur_done = sum(1 for p in procs if p.poll() is not None) + if cur_done > done_blocks: + done_blocks = cur_done + pct = int(done_blocks / len(procs) * 95) + report(f"修正中 {done_blocks}/{len(procs)} 块完成", pct) + for p in procs: + p.wait() + + # 合并 + out_list = [dict(it) for it in items] + total_fixed = 0 + for b_idx, blk, bfile, bfile_out in tasks: + if os.path.exists(bfile_out): + with open(bfile_out, encoding='utf-8') as f: + fixed_items = json.load(f) + if len(fixed_items) == len(blk): + for j, orig_idx in enumerate(blk): + if fixed_items[j].get('text'): + out_list[orig_idx]['text'] = fixed_items[j]['text'] + total_fixed += 1 + for fp in (bfile, bfile_out): + if os.path.exists(fp): + os.remove(fp) + report(f"修正完成 {total_fixed}/{n} 段", 100) + return out_list diff --git a/src/substudio/pipeline/srt.py b/src/substudio/pipeline/srt.py new file mode 100644 index 0000000..fd99501 --- /dev/null +++ b/src/substudio/pipeline/srt.py @@ -0,0 +1,70 @@ +# -*- coding: utf-8 -*- +"""SRT 生成模块""" +import os + + +def fmt_ts(ms): + ms = max(0, int(ms)) + return "%02d:%02d:%02d,%03d" % (ms // 3600000, (ms % 3600000) // 60000, (ms % 60000) // 1000, ms % 1000) + + +def build_srt_ja_zh(items): + """日文+中文双语（日上中下）""" + lines = [] + for i, it in enumerate(items, 1): + ja = it.get('text', '').strip() + zh = it.get('zh', '').strip() + if not ja and not zh: + continue + lines.append(f"{i}\n{fmt_ts(it['start_ms'])} --> {fmt_ts(it['end_ms'])}\n{ja}\n{zh}\n") + return "\n".join(lines) + + +def build_srt_zh(items): + lines = [] + for i, it in enumerate(items, 1): + zh = it.get('zh', '').strip() + if not zh: + continue + lines.append(f"{i}\n{fmt_ts(it['start_ms'])} --> {fmt_ts(it['end_ms'])}\n{zh}\n") + return "\n".join(lines) + + +def build_srt_zh_en(items): + """中英双语（中上英下）""" + lines = [] + for i, it in enumerate(items, 1): + zh = it.get('zh', '').strip() + en = it.get('en', '').strip() + if not zh and not en: + continue + lines.append(f"{i}\n{fmt_ts(it['start_ms'])} --> {fmt_ts(it['end_ms'])}\n{zh}\n{en}\n") + return "\n".join(lines) + + +def write_srts(items, video_path, route, out_dir=None): + """按路线生成 SRT 到视频同目录 + 可选 output 目录。返回生成的文件列表""" + base = os.path.splitext(video_path)[0] + generated = [] + + def _write(path, content): + with open(path, 'w', encoding='utf-8-sig') as f: + f.write(content) + generated.append(path) + # 副本到 output 目录 + if out_dir: + os.makedirs(out_dir, exist_ok=True) + dup = os.path.join(out_dir, os.path.basename(path)) + with open(dup, 'w', encoding='utf-8-sig') as f: + f.write(content) + generated.append(dup) + + if route == "direct": + _write(base + ".ja.zh.srt", build_srt_ja_zh(items)) + _write(base + ".zh.srt", build_srt_zh(items)) + elif route == "via_en": + _write(base + ".r2.zh.en.srt", build_srt_zh_en(items)) + _write(base + ".r2.zh.srt", build_srt_zh(items)) + elif route == "transcribe_only": + pass # 只输出 transcript.json + return generated diff --git a/src/substudio/pipeline/transcribe.py b/src/substudio/pipeline/transcribe.py new file mode 100644 index 0000000..427fd43 --- /dev/null +++ b/src/substudio/pipeline/transcribe.py @@ -0,0 +1,258 @@ +# -*- coding: utf-8 -*- +"""转录管道：ffmpeg 提取音频 → fsmn-vad 切段 → SenseVoice 逐段转录 → 句子级时间戳""" +import os, json, re, subprocess, tempfile + +import numpy as np +import wave + +from funasr import AutoModel +from funasr.utils.postprocess_utils import rich_transcription_postprocess + +from ..config import FFMPEG, MODEL_DIR, VAD_DIR + +VIDEO_EXTS = ('.mp4', '.mkv', '.avi', '.mov', '.flv', '.webm', '.ts', '.m4v') + +# ---- 模型单例（全局复用，避免每次转录重新加载 ~3.2G 显存）---- +_model_lock = __import__('threading').Lock() +_models = {"vad": None, "asr": None} + + +def get_models(): + """懒加载 + 线程安全获取 VAD 和 SenseVoice 模型（全局单例）""" + global _models + with _model_lock: + if _models["vad"] is None: + _models["vad"] = AutoModel(model=VAD_DIR, device="cuda:0", disable_update=True) + if _models["asr"] is None: + _models["asr"] = AutoModel(model=MODEL_DIR, device="cuda:0", disable_update=True) + return _models["vad"], _models["asr"] + + +def extract_audio(video_path, wav_path): + subprocess.run([FFMPEG, '-y', '-hide_banner', '-loglevel', 'error', '-i', video_path, + '-vn', '-acodec', 'pcm_s16le', '-ar', '16000', '-ac', '1', wav_path], + check=True) + + +def load_wav_int16(path): + with wave.open(path, 'rb') as w: + sr = w.getframerate() + n = w.getnframes() + data = np.frombuffer(w.readframes(n), dtype=np.int16) + return data, sr + + +def write_segment_wav(path, samples): + with wave.open(path, 'wb') as w: + w.setnchannels(1) + w.setsampwidth(2) + w.setframerate(16000) + w.writeframes(samples.astype(np.int16).tobytes()) + + +def split_sentences(text): + parts = re.split(r'(?<=[。！？!?])', text) + return [p.strip() for p in parts if p.strip()] + + +def refine_segment(seg): + sentences = split_sentences(seg['text']) + if len(sentences) <= 1: + return [seg] + total_chars = sum(len(s) for s in sentences) + dur = seg['end_ms'] - seg['start_ms'] + out, cursor = [], seg['start_ms'] + for s in sentences: + frac = len(s) / total_chars + seg_dur = dur * frac + out.append({'start_ms': int(cursor), 'end_ms': int(cursor + seg_dur), 'text': s}) + cursor += seg_dur + return out + + +def merge_fragments(segs, min_ms=2000): + if not segs: + return [] + merged = [dict(segs[0])] + for seg in segs[1:]: + last = merged[-1] + gap = seg['start_ms'] - last['end_ms'] + new_dur = seg['end_ms'] - last['start_ms'] + if (last['end_ms'] - last['start_ms']) < min_ms and gap < 500 and new_dur < 8000: + last['end_ms'] = seg['end_ms'] + last['text'] += seg['text'] + else: + merged.append(dict(seg)) + return merged + + +def split_long_segment(s_ms, e_ms, window_ms=8000): + if e_ms - s_ms <= window_ms: + return [(s_ms, e_ms)] + out, cur = [], s_ms + while cur < e_ms: + nxt = min(cur + window_ms, e_ms) + out.append((cur, nxt)) + cur = nxt + return out + + +def extract_and_denoise(video_path, wav_path, denoise=True, progress_cb=None): + """提取音频 + 降噪（纯 CPU，可并发执行，不占 GPU） + 返回 wav_path；denoise=True 时原地降噪 + """ + def report(msg, pct): + if progress_cb: + progress_cb(msg, pct) + + report("提取音频...", 30) + os.makedirs(os.path.dirname(wav_path), exist_ok=True) + extract_audio(video_path, wav_path) + + if denoise: + try: + report("降噪中...", 60) + import noisereduce as nr + data, sr = load_wav_int16(wav_path) + float_data = data.astype(np.float32) / 32768.0 + # prop_decrease=0.7：平衡降噪效果与语音保留（0.9 会过度降噪产生字间隙） + denoised = nr.reduce_noise(y=float_data, sr=sr, prop_decrease=0.7) + out16 = (denoised * 32768).astype(np.int16) + write_segment_wav(wav_path, out16) + except Exception as ex: + print(f"降噪跳过: {ex}", flush=True) + report("音频就绪", 100) + return wav_path + + +def transcribe(video_path, out_json, language="ja", max_seg_ms=8000, progress_cb=None, denoise=True, wav_path=None): + """完整转录（提取音频+降噪+VAD+ASR）。progress_cb(msg, percent) 可选回调 + denoise=True: 用 noisereduce 降噪（压制背景噪音，提升 ASR 准确率） + wav_path: 指定音频路径（若已存在则跳过提取/降噪，只做 GPU 转录） + """ + t0 = __import__('time').time() + + def report(msg, pct): + if progress_cb: + progress_cb(msg, pct) + + # 1. 提取音频 + 降噪（CPU，可并发）——若 wav 已存在（调用方已做）则跳过 + tmp_dir = os.path.dirname(out_json) + os.makedirs(tmp_dir, exist_ok=True) + if wav_path is None: + # 唯一命名，避免并发冲突 + base_name = os.path.splitext(os.path.basename(video_path))[0] if os.path.isfile(video_path) else "audio" + safe_base = re.sub(r'[^\w\-.]', '_', base_name)[:30] + wav_path = os.path.join(tmp_dir, f"_audio_{safe_base}_{os.getpid()}.wav") + if not os.path.exists(wav_path) or os.path.getsize(wav_path) == 0: + extract_and_denoise(video_path, wav_path, denoise, progress_cb) + else: + report("音频已就绪（跳过提取/降噪）", 10) + + # 2. VAD（全局单例） + report("加载 VAD 模型...", 5) + vad, asr = get_models() + report("VAD 语音检测...", 8) + # 积极切分：自定义 silence_schedule 让 VAD 在短静音就切段， + # 得到 ≤4.6s 的自然句段（默认会产生 15s 长段，只能靠字符比例估算时间 → 字幕错位） + # max_single_segment_time 参数实测无效，必须用 silence_schedule + AGGRESSIVE_SILENCE = [(8000, 500), (12000, 300), (20000, 200), (float('inf'), 100)] + vad_res = vad.generate(input=wav_path, silence_schedule=AGGRESSIVE_SILENCE) + segments = vad_res[0]["value"] + total = len(segments) + report(f"VAD 检测到 {total} 段", 10) + + # 3. SenseVoice（asr 已从 get_models 获得） + + results = [] + samples_all, sr = load_wav_int16(wav_path) + # 临时段文件按视频唯一命名（并发转录时避免文件冲突 WinError 32） + base_name = os.path.splitext(os.path.basename(video_path))[0] if os.path.isfile(video_path) else "seg" + safe_base = re.sub(r'[^\w\-.]', '_', base_name)[:30] + tmp_wav = os.path.join(tmp_dir, f"_seg_tmp_{safe_base}_{os.getpid()}.wav") + + for i, (s_ms, e_ms) in enumerate(segments): + for seg_s, seg_e in split_long_segment(s_ms, e_ms, max_seg_ms): + s, e = int(seg_s / 1000 * sr), int(seg_e / 1000 * sr) + seg_samples = samples_all[max(0, s):e] + if len(seg_samples) < sr * 0.3: + continue + write_segment_wav(tmp_wav, seg_samples) + try: + res = asr.generate(input=tmp_wav, language=language, use_itn=True, batch_size_s=60) + raw_text = res[0]["text"] + text = rich_transcription_postprocess(raw_text) + text = re.sub(r'<\|[^|]*\|>', '', text).strip() + except Exception: + continue + if not text: + continue + for refined in refine_segment({'start_ms': seg_s, 'end_ms': seg_e, 'text': text}): + results.append(refined) + if (i + 1) % 50 == 0 or (i + 1) == total: + pct = 12 + int((i + 1) / total * 78) + report(f"转录中 {i+1}/{total} 段", pct) + + # 转录完成前读音频时长（用于完整性校验） + audio_duration_ms = 0 + try: + with wave.open(wav_path, 'rb') as w: + audio_duration_ms = int(w.getnframes() / w.getframerate() * 1000) + except Exception: + pass + + if os.path.exists(tmp_wav): + os.remove(tmp_wav) + if os.path.exists(wav_path): + os.remove(wav_path) + + results = merge_fragments(results) + + # 完整性校验：残缺转录直接抛错；空洞仅警告（视频可能本身无对白） + ok, warnings = validate_transcript(results, audio_duration_ms) + if not ok: + raise RuntimeError(f"转录校验失败: {warnings[0] if warnings else '未知'}") + + with open(out_json, 'w', encoding='utf-8') as f: + json.dump(results, f, ensure_ascii=False, indent=1) + report(f"转录完成 {len(results)} 段（覆盖 {audio_duration_ms/1000:.0f}s）", 95) + # 返回 (out_json, warnings) 供上层记录警告 + return out_json, warnings + +def validate_transcript(results, audio_duration_ms, min_coverage=0.2, min_segments=5, + max_gap_ms=30000): + """转录完整性校验：防止转录残缺但被标记完成 + 返回 (ok, warnings) + - 硬失败：空结果 / 段数过少 / 覆盖 <20%（AV 视频语音占比低，实测 ~25%） + - 警告（不失败）：段间空洞 >30s（可能是视频本身无对白，仅记录） + """ + warnings = [] + if not results: + return False, ["转录结果为空（0 段）"] + if len(results) < min_segments: + return False, [f"转录段数过少: {len(results)} < {min_segments}"] + if audio_duration_ms <= 0: + return False, ["音频时长为 0"] + + # 1. 总覆盖比例（硬失败） + covered = results[-1]['end_ms'] - results[0]['start_ms'] + ratio = covered / audio_duration_ms + if ratio < min_coverage: + return False, [f"转录覆盖不足: {covered/1000:.0f}s/{audio_duration_ms/1000:.0f}s ({ratio:.0%} < {min_coverage:.0%})"] + + # 2. 空洞检测（警告，不失败——视频可能本身无对白） + prev_end = results[0]['end_ms'] + for i in range(1, len(results)): + gap = results[i]['start_ms'] - prev_end + if gap > max_gap_ms: + warnings.append(f"段{i}前有 {gap/1000:.0f}s 无对白间隔（可能正常）") + prev_end = max(prev_end, results[i]['end_ms']) + + # 3. 尾部未覆盖（警告） + tail = audio_duration_ms - results[-1]['end_ms'] + if tail > max_gap_ms * 2: + warnings.append(f"音频尾部有 {tail/1000:.0f}s 未覆盖（可能无对白）") + + return True, warnings + + diff --git a/src/substudio/pipeline/translate.py b/src/substudio/pipeline/translate.py new file mode 100644 index 0000000..cda6449 --- /dev/null +++ b/src/substudio/pipeline/translate.py @@ -0,0 +1,96 @@ +# -*- coding: utf-8 -*- +"""LLM 翻译管道：全文一次请求（1M 上下文）+ 段级增量续跑 +支持 existing: 已有翻译结果的段跳过，只翻缺失段（断点续跑） +""" +from ..llm import call_llm, parse_idx_lines + +TRANSLATE_PROMPT_TMPL = ( + "You are a professional subtitle translator. Translate ALL the following " + "subtitle lines into {lang_name}. " + "Preserve the [index] prefix exactly like [0], [1]. " + "Translate naturally for subtitle display. One translation per line, " + "in the same order. Output ONLY the translated lines, nothing else. " + "Do not omit, merge, or skip any line." +) + +LANG_NAMES = {"en": "English", "zh": "Simplified Chinese (简体中文)"} + + +def translate_all(items, target_lang, progress_cb=None, existing=None): + """翻译（支持增量续跑）。 + items: [{text}] 全部段 + existing: {idx: 已翻译文本} —— 这些段跳过，只翻缺失段 + 返回 {idx: text} + """ + existing = existing or {} + # 找出缺失段 + missing_idx = [i for i in range(len(items)) if i not in existing or not existing[i]] + if not missing_idx: + if progress_cb: + progress_cb(f"全部已翻译，跳过", 100) + return dict(existing) + + # 缺失段按字符量分组（每块 ~5000 字符，避免超长请求截断） + BLOCK_CHARS = 5000 + blocks = [] + cur_block, cur_chars = [], 0 + for idx in missing_idx: + cur_block.append(idx) + cur_chars += len(items[idx]['text']) + if cur_chars >= BLOCK_CHARS: + blocks.append(cur_block) + cur_block, cur_chars = [], 0 + if cur_block: + blocks.append(cur_block) + + result = dict(existing) + lang_name = LANG_NAMES[target_lang] + total_blocks = len(blocks) + for b_idx, blk in enumerate(blocks): + # 组内重新编号 [0..n] 发给模型，再映射回原 idx + blk_items = [items[i] for i in blk] + rel_map = {j: orig_idx for j, orig_idx in enumerate(blk)} # 组内位置 -> 原idx + lines = "\n".join(f"[{j}] {it['text']}" for j, it in enumerate(blk_items)) + if progress_cb: + progress_cb(f"翻译为{'英文' if target_lang=='en' else '中文'}：块 {b_idx+1}/{total_blocks} ({len(blk)} 段)", 5 + int(b_idx / total_blocks * 90)) + resp = call_llm([ + {"role": "system", "content": TRANSLATE_PROMPT_TMPL.format(lang_name=lang_name)}, + {"role": "user", "content": lines}, + ], timeout=600) # 翻译大请求生成慢，超时放宽到 600s + parsed = parse_idx_lines(resp) + for j, orig_idx in rel_map.items(): + if j in parsed and parsed[j]: + result[orig_idx] = parsed[j] + if progress_cb: + progress_cb(f"翻译完成 {len(result)}/{len(items)} 段", 100) + return result + + +def translate_direct(items, progress_cb=None, existing_zh=None): + """路线1：直译中文（支持增量续跑）。 + items: [{start_ms,end_ms,text}] -> 同结构 + zh 字段 + existing_zh: {idx: 已翻译中文} + """ + out = [dict(it) for it in items] + zh_map = translate_all(items, "zh", progress_cb, existing=existing_zh) + for i, it in enumerate(out): + if i in zh_map and zh_map[i]: + it['zh'] = zh_map[i] + return out + + +def translate_via_en(items, progress_cb=None, existing_en=None, existing_zh=None): + """路线2：日文→英文→中文（支持增量续跑）。返回 items + en + zh 字段""" + out = [dict(it) for it in items] + # 步骤1：日→英 + en_map = translate_all(items, "en", progress_cb, existing=existing_en) + for i, it in enumerate(out): + if i in en_map and en_map[i]: + it['en'] = en_map[i] + # 步骤2：英→中 + en_items = [{'text': it.get('en', it['text'])} for it in out] + zh_map = translate_all(en_items, "zh", progress_cb, existing=existing_zh) + for i, it in enumerate(out): + if i in zh_map and zh_map[i]: + it['zh'] = zh_map[i] + return out diff --git a/src/substudio/taskmanager.py b/src/substudio/taskmanager.py new file mode 100644 index 0000000..6553188 --- /dev/null +++ b/src/substudio/taskmanager.py @@ -0,0 +1,470 @@ +# -*- coding: utf-8 -*- +"""任务管理器：队列 + 状态机 + SSE 广播""" +import os, json, threading, time, uuid, shutil +from dataclasses import dataclass, field, asdict +from concurrent.futures import ThreadPoolExecutor + +from .config import OUTPUT_DIR, MAX_CONCURRENT_TASKS, LLM_CONCURRENCY, WORK_DIRNAME +from .pipeline.transcribe import transcribe, extract_and_denoise, VIDEO_EXTS +from .pipeline.fix import fix +from .pipeline.translate import translate_direct, translate_via_en +from .pipeline.srt import write_srts + +# 去重记录文件：放在被处理视频的所在文件夹下 +HISTORY_FILENAME = ".subtitle-history.json" + + +def _history_path(video_path): + """视频/文件夹的 history 文件路径（跟视频在同一文件夹）""" + if os.path.isdir(video_path): + folder = video_path + else: + folder = os.path.dirname(video_path) + return os.path.join(folder, HISTORY_FILENAME) + + +def _load_history(video_path): + """从视频所在文件夹加载 history（按 视频|路线 记录）""" + hp = _history_path(video_path) + if os.path.exists(hp): + try: + with open(hp, encoding='utf-8') as f: + return json.load(f) + except Exception: + return {} + return {} + + +def _save_history(video_path, history): + """保存 history 到视频所在文件夹""" + hp = _history_path(video_path) + try: + with open(hp, 'w', encoding='utf-8') as f: + json.dump(history, f, ensure_ascii=False, indent=1) + except Exception: + pass + + +@dataclass +class Task: + id: str + source: str # 文件或文件夹路径 + route: str # direct / via_en / transcribe_only + language: str = "ja" + status: str = "pending" # pending/transcribing/fixing/translating/srt/done/error + progress: int = 0 + message: str = "" + created: float = field(default_factory=time.time) + finished: float = None + error: str = "" + files: list = field(default_factory=list) # 生成的 SRT 文件 + transcript_path: str = "" + video_errors: list = field(default_factory=list) # 单视频失败记录 [{video, error}] + processed_count: int = 0 # 成功处理视频数 + failed_count: int = 0 # 失败视频数 + total_videos: int = 0 # 总视频数（文件夹） + started_at: float = None # 任务开始时间（总体耗时） + subtasks: dict = field(default_factory=dict) # {video: {status, progress, message, started_at, elapsed}} 子任务状态 + + def to_dict(self): + d = asdict(self) + # 动态耗时：运行中 = now - started_at, 完成 = finished - created + if self.finished: + d["elapsed"] = round(self.finished - self.created, 1) + elif self.started_at: + d["elapsed"] = round(time.time() - self.started_at, 1) + else: + d["elapsed"] = 0 + return d + + +class TaskManager: + def __init__(self): + self.tasks = {} # id -> Task + self.queue = [] # 等待队列（串行执行） + self.lock = threading.Lock() + self.worker = None # 当前执行线程 + self.subscribers = [] # SSE 客户端队列 + # 并发控制 + self.transcribe_lock = threading.BoundedSemaphore(MAX_CONCURRENT_TASKS) # 转录并发（GPU 模型共享，默认 2） + self.llm_sem = threading.BoundedSemaphore(LLM_CONCURRENCY) # LLM 并发信号量 + self.llm_concurrency = LLM_CONCURRENCY + self.transcribe_concurrency = MAX_CONCURRENT_TASKS + + def set_llm_concurrency(self, n): + """动态调整 LLM 并发数（重建信号量）""" + n = max(1, min(int(n), 20)) + self.llm_concurrency = n + self.llm_sem = threading.BoundedSemaphore(n) + return n + + def set_transcribe_concurrency(self, n): + """动态调整转录并发数（1-4，重建信号量）""" + n = max(1, min(int(n), 4)) + self.transcribe_concurrency = n + self.transcribe_lock = threading.BoundedSemaphore(n) + return n + + # ---- SSE ---- + def subscribe(self, q): + self.subscribers.append(q) + + def unsubscribe(self, q): + if q in self.subscribers: + self.subscribers.remove(q) + + def _broadcast(self, task): + ev = json.dumps({"type": "task_update", "task": task.to_dict()}, ensure_ascii=False) + for q in list(self.subscribers): + try: + q.put(ev) + except Exception: + pass + + # ---- 任务管理 ---- + def create_task(self, source, route, language="ja"): + # 防重复：同 source+route 已有 pending/running 任务则拒绝（避免重复处理） + with self.lock: + for t in self.tasks.values(): + if t.source == source and t.route == route and t.status in ("pending", "transcribing", "fixing", "translating", "srt"): + return None, [], f"该内容正在处理中（任务 {t.id[:8]}），请等待完成后再提交" + + # 去重检查：文件夹 → 找出未处理/处理过的视频清单 + skipped = [] + if os.path.isdir(source): + history = _load_history(source) + for f in sorted(os.listdir(source)): + if f.lower().endswith(VIDEO_EXTS): + key = os.path.join(source, f) + hkey = f"{key}|{route}" + if hkey in history: + skipped.append(key) + if skipped and len(skipped) == sum(1 for f in os.listdir(source) if f.lower().endswith(VIDEO_EXTS)): + # 全部已处理过，拒绝创建 + return None, skipped, "" + elif os.path.isfile(source): + # 单文件也检查：同路线已处理过则拒绝 + history = _load_history(source) + hkey = f"{source}|{route}" + if hkey in history: + return None, [source], "" + + task = Task(id=uuid.uuid4().hex[:12], source=source, route=route, language=language) + with self.lock: + self.tasks[task.id] = task + self.queue.append(task.id) + self._broadcast(task) + self._maybe_start() + return task, skipped, "" + + def _maybe_start(self): + with self.lock: + if self.worker and self.worker.is_alive(): + return + running = [t for t in self.tasks.values() if t.status in ("transcribing", "fixing", "translating", "srt")] + if len(running) >= MAX_CONCURRENT_TASKS: + return + if not self.queue: + return + task_id = self.queue.pop(0) + task = self.tasks[task_id] + task.status = "transcribing" + task.message = "开始处理" + self.worker = threading.Thread(target=self._run_task, args=(task,), daemon=True) + self.worker.start() + self._broadcast(task) + + def _run_task(self, task): + try: + self._process(task) + task.status = "done" + task.progress = 100 + if task.failed_count > 0: + task.message = f"部分完成: 成功 {task.processed_count}, 失败 {task.failed_count}" + else: + task.message = f"完成 ({task.processed_count} 个视频)" + task.finished = time.time() + except Exception as ex: + task.status = "error" + task.error = str(ex) + task.message = f"失败: {ex}" + task.finished = time.time() + self._broadcast(task) + # 处理下一个 + self._maybe_start() + + def _update(self, task, msg, pct=None): + task.message = msg + if pct is not None: + task.progress = int(pct) + self._broadcast(task) + + def _process(self, task): + # 收集视频文件 + videos = [] + skipped_in_run = [] + if os.path.isfile(task.source): + if task.source.lower().endswith(VIDEO_EXTS): + videos = [task.source] + elif os.path.isdir(task.source): + # 过滤已处理（同路线）的视频 + history = _load_history(task.source) + for f in sorted(os.listdir(task.source)): + if f.lower().endswith(VIDEO_EXTS): + key = os.path.join(task.source, f) + if f"{key}|{task.route}" in history: + skipped_in_run.append(key) + else: + videos.append(key) + if skipped_in_run: + task.message = f"跳过 {len(skipped_in_run)} 个已处理视频（同路线去重）" + self._update(task, task.message, 1) + if not videos: + raise ValueError(f"所选内容均已处理过，无需重复处理: {task.source}") + + out_dir = os.path.join(OUTPUT_DIR, task.id) + os.makedirs(out_dir, exist_ok=True) + + total = len(videos) + task.total_videos = total + task.started_at = time.time() + # 初始化子任务状态 + for v in videos: + task.subtasks[v] = {"status": "pending", "progress": 0, "message": "等待中", "started_at": None, "elapsed": 0} + task.message = f"开始处理 {total} 个视频（转录 {MAX_CONCURRENT_TASKS} + LLM {self.llm_concurrency} 并发）" + self._update(task, task.message, 2) + + # 流水线并发：每个视频一个 worker 线程 + # 转录内部拿 GPU 锁（串行），修正/翻译拿 LLM 信号量（限并发） + done_count = 0 + done_lock = threading.Lock() + + def _update_overall(): + """更新总体进度：已完成/失败/总数""" + done_total = task.processed_count + task.failed_count + overall_pct = int(done_total / total * 100) if total else 100 + # 聚合消息 + running = [st for st in task.subtasks.values() if st["status"] in ("transcribing", "fixing", "translating", "srt")] + running_names = [v for v, st in task.subtasks.items() if st["status"] in ("transcribing", "fixing", "translating", "srt")] + msg = f"总体 {done_total}/{total} 完成" + if running_names: + msg += f" | 处理中: {', '.join(os.path.basename(n) for n in running_names[:3])}" + if task.failed_count: + msg += f" | 失败 {task.failed_count}" + task.message = msg + self._update(task, msg, overall_pct) + + def _sub_update(video, status, progress, message): + """更新单个视频子任务状态（含耗时）""" + cur = task.subtasks.get(video, {}) + started = cur.get("started_at") + if started is None: + started = time.time() + elapsed = (time.time() - started) if started else 0 + task.subtasks[video] = { + "status": status, "progress": progress, "message": message, + "started_at": started, "elapsed": round(elapsed, 1), + } + self._broadcast(task) + + def _video_worker(v_idx, video): + nonlocal done_count + try: + _sub_update(video, "transcribing", 5, "转录中") + self._process_one(task, video, out_dir, v_idx, total, _sub_update) + with done_lock: + task.processed_count += 1 + done_count += 1 + _sub_update(video, "done", 100, "完成") + except Exception as ex: + with done_lock: + task.failed_count += 1 + done_count += 1 + task.video_errors.append({"video": video, "error": str(ex)[:300]}) + # 记录完整 traceback 到日志（排查用） + try: + import traceback as _tb + tb = _tb.format_exc() + with open(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_error.log"), 'a', encoding='utf-8') as _ef: + _ef.write(f"\n=== {os.path.basename(video)} ===\n{tb}\n") + except Exception: + pass + _sub_update(video, "error", 0, f"失败: {str(ex)[:80]}") + finally: + _update_overall() + + # 并发执行（worker 数 = 转录锁 + LLM 并发，转录阶段大部分在等锁） + pool_size = max(MAX_CONCURRENT_TASKS, self.llm_concurrency) + with ThreadPoolExecutor(max_workers=pool_size) as pool: + futures = [pool.submit(_video_worker, i, v) for i, v in enumerate(videos)] + # 等待全部完成（含失败） + for f in futures: + f.result() + + if task.failed_count > 0: + task.message = f"部分完成: 成功 {task.processed_count}, 失败 {task.failed_count}" + else: + task.message = f"完成 ({task.processed_count} 个视频)" + self._update(task, task.message, 100) + + def _process_one(self, task, video, out_dir, v_idx=0, total=1, sub_update=None): + """处理单个视频（转录→修正→翻译→SRT）——支持断点续跑 + 中间产物在 视频所在文件夹/.subtitle-work/（固定目录，不按视频分） + 续跑：有 transcript 无 fixed → 从修正开始；有 fixed → 从翻译开始 + 转录拿 GPU 锁（串行），修正/翻译拿 LLM 信号量（限并发） + """ + def sub(status, progress, message): + if sub_update: + sub_update(video, status, progress, message) + + base = os.path.splitext(os.path.basename(video))[0] + fname = os.path.basename(video) + # 中间产物目录：视频所在文件夹 .subtitle-work/ + work_dir = os.path.join(os.path.dirname(video), WORK_DIRNAME) + os.makedirs(work_dir, exist_ok=True) + transcript_json = os.path.join(work_dir, f"{base}_transcript.json") + fixed_json = os.path.join(work_dir, f"{base}_fixed.json") + translated_json = os.path.join(work_dir, f"{base}_translated.json") + wav_path = os.path.join(work_dir, f"{base}_audio.wav") + + # ===== 阶段 1: 转录（可续跑）===== + if os.path.exists(transcript_json): + sub("transcribing", 100, f"转录已存在（续跑）{fname}") + else: + # 提取音频 + 降噪（CPU，可并发——在 GPU 锁外做） + sub("transcribing", 2, f"提取音频+降噪 {fname}") + extract_and_denoise(video, wav_path, denoise=True, + progress_cb=lambda msg, pct: sub("transcribing", 2 + int(pct * 0.06), f"{msg} {fname}")) + # 转录（GPU 锁） + with self.transcribe_lock: + sub("transcribing", 10, f"转录中 {fname}") + def _tcb(msg, pct, _fname=fname): + if msg.startswith("转录中"): + sub("transcribing", 10 + int(pct * 0.3), f"{_fname}: {msg.split('转录中')[1].strip()}") + else: + sub("transcribing", 10 + int(pct * 0.3), f"{msg} {_fname}") + _result, _warnings = transcribe(video, transcript_json, language=task.language, + progress_cb=_tcb, wav_path=wav_path) + if _warnings: + task.video_errors.append({"video": video, "error": "；".join(_warnings), "warn": True}) + task.transcript_path = transcript_json + + if task.route == "transcribe_only": + task.files.append(transcript_json) + self._save_history_record(video, task, [transcript_json]) + return + + with open(transcript_json, encoding='utf-8') as f: + items = json.load(f) + + # ===== 阶段 2+3: 修正 + 翻译（LLM 信号量，可续跑）===== + with self.llm_sem: + # 修正（可续跑） + if os.path.exists(fixed_json): + sub("fixing", 100, f"修正已存在（续跑）{fname}") + with open(fixed_json, encoding='utf-8') as f: + items = json.load(f) + else: + sub("fixing", 45, f"LLM 修正 {fname}") + items = fix(items, progress_cb=lambda msg, pct: sub("fixing", 45 + int(pct * 0.15), f"{msg} [{fname}]")) + with open(fixed_json, 'w', encoding='utf-8') as f: + json.dump(items, f, ensure_ascii=False, indent=1) + + # 翻译（可续跑：从 translated_json 加载已有结果，只翻缺失段） + existing_zh = {} + existing_en = {} + if os.path.exists(translated_json): + try: + with open(translated_json, encoding='utf-8') as f: + prev = json.load(f) + # 提取已有翻译 + for i, it in enumerate(prev): + if it.get('zh'): + existing_zh[i] = it['zh'] + if it.get('en'): + existing_en[i] = it['en'] + sub("translating", 30, f"翻译已有 {len(existing_zh)}/{len(items)} 段（续跑）{fname}") + except Exception: + existing_zh, existing_en = {}, {} + + sub("translating", 40, f"翻译中 {fname}") + if task.route == "direct": + items = translate_direct(items, + progress_cb=lambda msg, pct: sub("translating", 40 + int(pct * 0.5), f"{msg} [{fname}]"), + existing_zh=existing_zh) + elif task.route == "via_en": + items = translate_via_en(items, + progress_cb=lambda msg, pct: sub("translating", 40 + int(pct * 0.5), f"{msg} [{fname}]"), + existing_en=existing_en, existing_zh=existing_zh) + with open(translated_json, 'w', encoding='utf-8') as f: + json.dump(items, f, ensure_ascii=False, indent=1) + + # 4. SRT（写视频同目录，PotPlayer 需要；副本到 output） + sub("srt", 92, f"生成 SRT {fname}") + srt_files = write_srts(items, video, task.route, out_dir) + task.files.extend(srt_files) + sub("srt", 98, f"SRT 完成 {fname}") + + # 5. 记录 history + self._save_history_record(video, task, srt_files) + + def _save_history_record(self, video, task, srt_files): + """记录 history（精确到视频，存在视频所在文件夹）""" + try: + history = _load_history(video) + hkey = f"{video}|{task.route}" + # 收集警告（transcribe 的 warnings 记录在 video_errors 里 warn=true） + warnings = [] + for ve in task.video_errors: + if ve.get("video") == video and ve.get("warn"): + warnings.append(ve["error"]) + history[hkey] = { + "video": video, + "route": task.route, + "files": list(srt_files), + "warnings": warnings, + "time": time.time(), + } + _save_history(video, history) + except Exception: + pass + + # ---- API ---- + def list_tasks(self): + return [t.to_dict() for t in sorted(self.tasks.values(), key=lambda t: -t.created)] + + def get_task(self, tid): + t = self.tasks.get(tid) + return t.to_dict() if t else None + + def retry_task(self, tid): + t = self.tasks.get(tid) + if not t or t.status not in ("error", "done"): + return None + # 复用 ID 重新入队 + t.status = "pending" + t.error = "" + t.message = "重试" + t.progress = 0 + with self.lock: + self.queue.append(t.id) + self._broadcast(t) + self._maybe_start() + return t.to_dict() + + def cancel_task(self, tid): + """从队列移除（无法中断运行中的，只移出等待队列）""" + t = self.tasks.get(tid) + if not t: + return None + if t.status == "pending": + with self.lock: + if tid in self.queue: + self.queue.remove(tid) + t.status = "cancelled" + t.message = "已取消" + self._broadcast(t) + return t.to_dict() + + +manager = TaskManager() diff --git a/static/app.js b/static/app.js new file mode 100644 index 0000000..ea39bf1 --- /dev/null +++ b/static/app.js @@ -0,0 +1,438 @@ +// subtitle-studio 前端逻辑 +let selectedPath = ""; +let tasksMap = {}; + +// ---- 文件浏览 ---- +async function loadDrives() { + const r = await fetch("/api/drives"); + const d = await r.json(); + const sel = document.getElementById("drive-select"); + sel.innerHTML = d.drives.map(x => ``).join(""); + sel.onchange = () => { document.getElementById("path-input").value = sel.value; browse(sel.value); }; + if (d.drives.length) { document.getElementById("path-input").value = d.drives[0]; browse(d.drives[0]); } +} + +async function browse(path) { + path = cleanPath(path); + const r = await fetch("/api/browse?path=" + encodeURIComponent(path)); + const d = await r.json(); + document.getElementById("path-input").value = d.cwd; + const list = document.getElementById("browser-list"); + list.innerHTML = ""; + const entries = d.entries.filter(e => !e.name.startsWith("$")); + for (const e of entries) { + const div = document.createElement("div"); + div.className = "browser-item" + (e.path === selectedPath ? " selected" : ""); + div.innerHTML = `${e.is_dir ? "📁" : "🎬"}${e.name}` + + (e.is_dir ? "" : '文件'); + div.onclick = () => { + selectedPath = e.path; + updateSelection(e); + document.querySelectorAll(".browser-item").forEach(x => x.classList.remove("selected")); + div.classList.add("selected"); + }; + if (e.is_dir) { + div.ondblclick = () => browse(e.path); + } + list.appendChild(div); + } +} + +function updateSelection(e) { + const preview = document.getElementById("sel-preview"); + if (e.is_dir) { + preview.innerHTML = `已选择文件夹：${e.path}（将批量处理其中所有视频）`; + loadFolderStatus(e.path); // 自动加载该文件夹视频状态 + } else { + preview.innerHTML = `已选择文件：${e.path}`; + } +} + +// 加载文件夹内所有视频的处理状态（含已完成/警告/续跑点） +async function loadFolderStatus(path) { + const route = document.querySelector('input[name="route"]:checked')?.value || "direct"; + try { + const r = await fetch(`/api/folder_status?path=${encodeURIComponent(path)}&route=${route}`); + const d = await r.json(); + if (d.error) { return; } + renderFolderVideos(d.videos); + } catch (e) {} +} + +// 渲染文件夹视频状态为任务卡片列表 +function renderFolderVideos(videos) { + const list = document.getElementById("task-list"); + if (!videos || !videos.length) { return; } + folderVideos = videos; // 保存状态，供 mergeTaskProgress 更新 + folderRendered = true; + // 转成任务格式渲染（模拟一个总任务 + 子任务） + const total = videos.length; + const doneCount = videos.filter(v => v.status === "done").length; + const subTasks = {}; + videos.forEach(v => { + let msg = v.message; + if (v.warnings && v.warnings.length) { + msg += ' ⚠️ ' + v.warnings[0].slice(0, 60); + } + subTasks[v.video] = { + status: v.status, progress: v.progress, message: msg, + started_at: null, elapsed: 0, warnings: v.warnings || [] + }; + }); + const fakeTask = { + id: "folder-" + Math.random().toString(36).slice(2, 8), + source: videos[0].video.split(/[\\/]/).slice(0, -1).join("\\"), + route: "direct", status: doneCount === total ? "done" : "pending", + progress: Math.round(doneCount / total * 100), + message: `文件夹状态：${doneCount}/${total} 已完成`, + created: Date.now() / 1000, files: [], subtasks: subTasks, + total_videos: total, processed_count: doneCount, failed_count: 0, + }; + // 临时切到 tasks 视图渲染 folder 卡片（renderTasks 有 guard 会挡 folder） + currentView = "tasks"; + renderTasks([fakeTask]); + currentView = "folder"; // 渲染完成后切回 folder 视图，阻止任务轮询覆盖 +} + +function goUp() { + const cur = document.getElementById("path-input").value; + const parent = cur.replace(/[\\/][^\\/]*[\\/]?$/, "") || cur; + if (parent !== cur) browse(parent); +} + +function refreshBrowse() { browse(document.getElementById("path-input").value); } + +// 清理路径：去除首尾双引号（用户从资源管理器复制）、多余空白、尾部斜杠 +function cleanPath(p) { + if (!p) return ""; + let s = p.trim(); + // 去掉首尾双引号（成对或单个） + if ((s.startsWith('"') && s.endsWith('"')) || (s.startsWith("'") && s.endsWith("'"))) { + s = s.slice(1, -1); + } else if (s.startsWith('"')) { + s = s.slice(1); + } else if (s.endsWith('"')) { + s = s.slice(0, -1); + } + // 去除尾部多余斜杠（保留盘符根如 C:\） + s = s.replace(/[\\/]+$/, ""); + if (/^[A-Za-z]:$/.test(s)) s += "\\"; + return s.trim(); +} + +// 手动输入路径（回车触发 change）——若是文件夹自动加载视频状态 +document.getElementById("path-input").addEventListener("change", () => { + const p = cleanPath(document.getElementById("path-input").value); + document.getElementById("path-input").value = p; + selectedPath = p; + document.getElementById("sel-preview").innerHTML = `路径：${p}`; + // 文件夹 → 自动加载任务列表（loadFolderStatus 内部会判断，文件路径不渲染） + loadFolderStatus(p); +}); +// 回车键也触发（有些浏览器回车不触发 change） +document.getElementById("path-input").addEventListener("keydown", (e) => { + if (e.key === "Enter") { + const p = cleanPath(document.getElementById("path-input").value); + document.getElementById("path-input").value = p; + selectedPath = p; + document.getElementById("sel-preview").innerHTML = `路径：${p}`; + loadFolderStatus(p); + } +}); + +// ---- 非阻塞提示 toast ---- +function showToast(msg, type) { + let toast = document.getElementById("toast"); + if (!toast) { + toast = document.createElement("div"); + toast.id = "toast"; + toast.style.cssText = "position:fixed;top:16px;right:16px;z-index:9999;max-width:380px;background:#1c212b;border:1px solid #2a2f3a;border-radius:8px;padding:10px 14px;font-size:13px;box-shadow:0 4px 16px rgba(0,0,0,.4);"; + document.body.appendChild(toast); + } + toast.innerHTML = `${type === "warn" ? "⚠️" : type === "err" ? "❌" : "✅"}${msg}`; + toast.style.borderColor = type === "warn" ? "#d29922" : type === "err" ? "#f85149" : "#3fb950"; + clearTimeout(toast._t); + toast._t = setTimeout(() => toast.remove(), 5000); +} + +// ---- 区域折叠 ---- +function collapseSections() { + const s1 = document.getElementById("section-1"); + const s2 = document.getElementById("section-2"); + if (s1) s1.removeAttribute("open"); + if (s2) s2.removeAttribute("open"); + const s3 = document.getElementById("section-3"); + if (s3 && !s3.hasAttribute("open")) s3.setAttribute("open", ""); +} + +// ---- 任务创建 ---- +async function createTask() { + selectedPath = cleanPath(selectedPath); + document.getElementById("path-input").value = selectedPath; + if (!selectedPath) { showToast("请先选择文件或文件夹", "warn"); return; } + const route = document.querySelector('input[name="route"]:checked').value; + const language = document.getElementById("lang-select").value; + const btn = document.getElementById("submit-btn"); + btn.disabled = true; + try { + const r = await fetch("/api/tasks", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ source: selectedPath, route, language }), + }); + let data = null; + try { data = await r.json(); } catch (e) { + showToast("服务器错误: " + (r.status || "") , "err"); + return; + } + if (data.error) { showToast(data.error, "err"); return; } + if (data.skipped_all) { showToast(data.notice || "所选内容已处理过", "warn"); return; } + if (data.notice) { showToast(data.notice, "warn"); } + // 提交成功：折叠 1、2 区。若在 folder 视图则保持（实时进度会合并进来） + collapseSections(); + if (currentView === "folder") { + loadFolderStatus(selectedPath); // 刷新 folder 状态（含新任务进度） + } else { + fetch("/api/tasks").then(r => r.json()).then(renderTasks); + } + } finally { + btn.disabled = false; + } +} + +// ---- 耗时格式化 ---- +function fmtDur(sec) { + sec = Math.max(0, Math.round(sec || 0)); + const h = Math.floor(sec / 3600), m = Math.floor((sec % 3600) / 60), s = sec % 60; + if (h > 0) return `${h}h${m}m`; + if (m > 0) return `${m}m${s}s`; + return `${s}s`; +} + +// ---- 任务渲染 ---- +const STATUS_LABEL = { + pending: "等待中", transcribing: "转录中", fixing: "修正中", + translating: "翻译中", srt: "生成字幕", done: "完成", error: "失败", cancelled: "已取消", +}; + +// 当前视图：'tasks'=真实任务列表, 'folder'=文件夹状态视图 +let currentView = "tasks"; +let folderVideos = null; // 最近一次 folder_status 返回的视频状态 +let folderRendered = false; // folder 视图是否已渲染 + +// 合并真实任务进度到 folder 视图（已完成的视频保留，运行中的更新进度） +function mergeTaskProgress(tasks) { + if (!folderVideos) return; + let changed = false; + for (const t of tasks) { + if (!t.subtasks) continue; + for (const [video, st] of Object.entries(t.subtasks)) { + const fv = folderVideos.find(v => v.video === video); + if (fv) { + if (fv.status !== st.status || fv.progress !== st.progress) { + fv.status = st.status; + fv.progress = st.progress; + fv.message = st.message || fv.message; + changed = true; + } + } + } + } + if (changed) renderFolderVideos(folderVideos); +} + +function renderTasks(tasks) { + // folder 视图下：合并真实任务进度到 folder 视图，不覆盖已完成的视频 + if (currentView === "folder") { + if (folderVideos && folderRendered) { + mergeTaskProgress(tasks); + } + return; + } + const list = document.getElementById("task-list"); + if (!tasks.length) { + if (!list.querySelector(".empty")) { list.innerHTML = '
暂无任务，选择视频开始吧
'; } + return; + } + + // 已存在的任务卡片（避免重建列表导致滚动位置丢失） + const existingIds = new Set(); + for (const t of tasks) { + existingIds.add("task-" + t.id); + let card = document.getElementById("task-" + t.id); + if (card) { + // 增量更新：只更新卡片内容，保留 DOM 节点（滚动位置稳定） + card.innerHTML = buildTaskCard(t); + } else { + card = document.createElement("div"); + card.className = "task"; + card.id = "task-" + t.id; + card.innerHTML = buildTaskCard(t); + list.appendChild(card); + } + } + // 移除已消失的任务卡片 + list.querySelectorAll(".task").forEach(c => { + if (!existingIds.has(c.id)) c.remove(); + }); + // 移除空状态 + const empty = list.querySelector(".empty"); + if (empty) empty.remove(); +} + +function buildTaskCard(t) { + const name = t.source.split(/[\\/]/).pop(); + const barCls = t.status === "done" ? "done" : (t.status === "error" ? "error" : ""); + let filesHtml = ""; + if (t.files && t.files.length) { + filesHtml = '
' + t.files.map(f => + `⬇ ${f.split(/[\\/]/).pop()}` + ).join("") + "
"; + } + let actions = ""; + if (t.status === "error") { + actions = '
'; + } else if (t.status === "pending") { + actions = '
'; + } + + // 子任务（多视频并发）渲染 - 完成时保留进度条 + let subtasksHtml = ""; + if (t.subtasks && Object.keys(t.subtasks).length > 1) { + const now = Date.now() / 1000; + const subRows = Object.entries(t.subtasks).map(([v, st]) => { + const vname = v.split(/[\\/]/).pop(); + const sCls = st.status === "done" ? "done" : (st.status === "error" ? "error" : ""); + // 耗时：运行中的用本地时钟实时估算（后端广播的 elapsed + 本地增量），完成/失败用固定值 + let stDur = st.elapsed ? fmtDur(st.elapsed) : ""; + if ((st.status === "transcribing" || st.status === "fixing" || st.status === "translating" || st.status === "srt") + && st.started_at) { + stDur = fmtDur(now - st.started_at); + } + const stTime = st.status === "done" || st.status === "error" + ? `${stDur}` + : `${stDur}`; + return ` +
+
+ ${vname} + ${STATUS_LABEL[st.status] || st.status} + ${stTime} + +
+
+
${st.message || ""}
+
`; + }).join(""); + subtasksHtml = `
${subRows}
`; + } + + // 总体进度标签（文件夹：已完成/总数）+ 总耗时 + let overall = ""; + let overallTime = ""; + if (t.total_videos > 1) { + const doneTotal = (t.processed_count || 0) + (t.failed_count || 0); + overall = `${doneTotal}/${t.total_videos} 个视频`; + } + if (t.elapsed) { + overallTime = `耗时 ${fmtDur(t.elapsed)}`; + } else if (t.started_at && t.status !== "done" && t.status !== "error") { + // 运行中：本地时钟实时估算 + overallTime = `耗时 ${fmtDur(Date.now()/1000 - t.started_at)}`; + } + + return ` +
+ ${name} + ${STATUS_LABEL[t.status] || t.status} + ${overall} + ${overallTime} + ${new Date(t.created * 1000).toLocaleTimeString()} +
+
+
${t.message || ""}
+ ${t.error ? `
${t.error}
` : ""} + ${subtasksHtml} + ${filesHtml} + ${actions} + `; +} + +async function retryTask(id) { await fetch(`/api/tasks/${id}/retry`, { method: "POST" }); } +async function cancelTask(id) { await fetch(`/api/tasks/${id}/cancel`, { method: "POST" }); } + +// 重新生成单个视频（清中间产物 + SRT + history，重跑） +async function regenerateVideo(videoPath) { + if (!confirm(`重新生成该视频？\n将删除中间产物和已生成的字幕，然后重新处理。\n\n${videoPath}`)) return; + const route = document.querySelector('input[name="route"]:checked')?.value || "direct"; + try { + const r = await fetch("/api/video/regenerate", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ source: videoPath, route, language: "ja" }), + }); + const d = await r.json(); + if (d.error) { showToast(d.error, "err"); return; } + showToast(`已开始重新生成（清理 ${(d.removed || []).length} 个文件）`, "ok"); + // 切回真实任务视图查看进度 + currentView = "tasks"; + setTimeout(() => fetch("/api/tasks").then(r => r.json()).then(renderTasks), 2000); + } catch (e) { showToast("重新生成失败", "err"); } +} + +// ---- SSE ---- +function connectSSE() { + const es = new EventSource("/api/events"); + es.onmessage = (ev) => { + try { + const data = JSON.parse(ev.data); + if (data.type === "snapshot") { renderTasks(data.tasks); } + else if (data.type === "task_update") { + const t = data.task; + const prev = tasksMap[t.id]; + // 去重：仅当 status/progress/message/subtasks 都相同才跳过 + if (prev && prev.status === t.status && prev.progress === t.progress && + prev.message === t.message && JSON.stringify(prev.subtasks) === JSON.stringify(t.subtasks)) return; + tasksMap[t.id] = t; + // 刷新列表 + fetch("/api/tasks").then(r => r.json()).then(renderTasks); + } + } catch (e) { /* 忽略 keepalive */ } + }; + es.onerror = () => { es.close(); setTimeout(connectSSE, 3000); }; +} + +loadDrives(); +connectSSE(); +setInterval(() => { fetch("/api/tasks").then(r => r.json()).then(renderTasks); }, 5000); + +// ---- 并发数设置 ---- +async function loadConcurrency() { + try { + const r = await fetch("/api/concurrency"); + const d = await r.json(); + document.getElementById("llm-conc-input").value = d.llm; + document.getElementById("tr-conc-input").value = d.transcribe; + document.getElementById("conc-status").textContent = `转录并发 ${d.transcribe}（GPU 模型共享），LLM 并发 ${d.llm}`; + } catch (e) {} +} + +async function saveConcurrency() { + const llm = parseInt(document.getElementById("llm-conc-input").value, 10); + const tr = parseInt(document.getElementById("tr-conc-input").value, 10); + if (!llm || llm < 1 || llm > 20) { showToast("LLM 并发需在 1-20 之间", "warn"); return; } + if (!tr || tr < 1 || tr > 4) { showToast("转录并发需在 1-4 之间", "warn"); return; } + try { + const r = await fetch("/api/concurrency", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ llm, transcribe: tr }), + }); + const d = await r.json(); + showToast(d.notice || "并发设置已保存", "ok"); + loadConcurrency(); + } catch (e) { showToast("保存失败", "err"); } +} + +loadConcurrency(); diff --git a/templates/index.html b/templates/index.html new file mode 100644 index 0000000..7e612a0 --- /dev/null +++ b/templates/index.html @@ -0,0 +1,145 @@ + + + + + +字幕生成系统 subtitle-studio + + + +
+

字幕生成系统

+ +
+ ▶① 选择视频 +
+
+
+ + + + +
+
+
未选择
+
+
+
+ +
+ ▶② 选择翻译路线 +
+
+ + + +
+
+ + + 文件夹会批量处理其中所有视频（转录串行，LLM 并发） +
+
+ 转录并发： + + LLM 并发（修正/翻译）： + + + +
+
+
+ +
+ ▶③ 任务列表 +
+
+
+
+
+ + + +