cat-shark
4746e0363f
为视频生成 VR 双眼字幕的单体实现：FastAPI 后端、调度器与全部节点 （提音/转写/翻译/ASS/抽帧/OCR/LLM 过滤）在单进程内运行。 - 节点协议（wov_sdk 数据模型）与分布式版保持一致，预留回退桥梁 - 工作流即数据：DAG 存于 workflows/*.json，模型/链路改动只改数据 - 调度器：拓扑顺序执行、断点续跑（产物重建）、任务暂停/继续 - 抽帧按帧间隔（select 按帧号精确取帧），VLM OCR 与 LLM 过滤使用 自适应线程池弹性并发，并打印数据处理速度进度日志 - 100% 行覆盖率（pytest --cov-fail-under=100）
104 lines
4.6 KiB
Python
Executable File
104 lines
4.6 KiB
Python
Executable File
"""LLM 翻译节点。
|
|
|
|
单体版中作为进程内节点模块，由调度器直接调用。接收 SRT，提取纯文本行
|
|
分批调用 LLM，再把译文回填到原 SRT 结构并输出 cn.srt。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
from wov_sdk.models import InvokeRequest, InvokeResponse
|
|
# 单次 LLM 请求携带的字幕行数；过大会超出模型上下文，过小则请求次数过多。
|
|
CHUNK_SIZE = 20
|
|
|
|
|
|
def translate_lines(lines: list[str], params: dict) -> list[str]:
|
|
"""分批调用 LLM 翻译纯文本行，返回顺序一致的译文列表。"""
|
|
# 接口地址、Key 和模型均可通过环境变量配置（.env 自动加载），
|
|
# 默认指向 SiliconFlow 兼容接口，模型为 DeepSeek-V4-Flash。
|
|
api_base = os.getenv(
|
|
"LLM_API_BASE",
|
|
"https://api.siliconflow.cn/v1/chat/completions",
|
|
)
|
|
api_key = os.getenv("LLM_API_KEY", "")
|
|
# 单次请求超时可配置，长文本翻译场景下需要放宽。
|
|
request_timeout = float(os.getenv("LLM_TIMEOUT_SECONDS", "600"))
|
|
model = str(params.get("model") or os.getenv("LLM_MODEL", "Qwen/Qwen3.6-35B-A3B"))
|
|
target_language = str(params.get("target_language", "zh-CN"))
|
|
# 系统提示词约束模型只输出译文，保证行数和顺序可回填。
|
|
system_prompt = (
|
|
"你是专业字幕翻译。将用户提供的日文字幕翻译为"
|
|
f"{target_language}。只返回译文，保持行数和顺序，不要添加解释。"
|
|
)
|
|
translated: list[str] = []
|
|
# 按 CHUNK_SIZE 分批发送，避免单次请求超过模型上下文限制。
|
|
for start in range(0, len(lines), CHUNK_SIZE):
|
|
chunk = lines[start : start + CHUNK_SIZE]
|
|
body = {
|
|
"model": model,
|
|
"messages": [
|
|
{"role": "system", "content": system_prompt},
|
|
{"role": "user", "content": "\n".join(chunk)},
|
|
],
|
|
# 关闭推理模型的思考模式：Qwen3 等模型默认会把推理过程写入
|
|
# reasoning_content，导致 content 为空或截断译文；关闭后直接输出译文。
|
|
"enable_thinking": False,
|
|
# 放宽输出上限，避免长批次翻译被模型默认 max_tokens 截断。
|
|
"max_tokens": 8192,
|
|
}
|
|
headers = {"Content-Type": "application/json"}
|
|
# 配置了 Key 时附带 Bearer 鉴权头。
|
|
if api_key:
|
|
headers["Authorization"] = f"Bearer {api_key}"
|
|
request = urllib.request.Request(
|
|
api_base,
|
|
data=json.dumps(body).encode("utf-8"),
|
|
headers=headers,
|
|
method="POST",
|
|
)
|
|
with urllib.request.urlopen(request, timeout=request_timeout) as response:
|
|
payload = json.loads(response.read().decode("utf-8"))
|
|
# 兼容 OpenAI Chat Completions 响应格式，取第一条消息内容。
|
|
content = payload["choices"][0]["message"]["content"]
|
|
# 忽略空行，保证译文列表与输入行一一对应。
|
|
translated.extend(
|
|
[line.strip() for line in content.splitlines() if line.strip()]
|
|
)
|
|
return translated
|
|
|
|
|
|
def invoke(request: InvokeRequest) -> InvokeResponse:
|
|
"""翻译 SRT 文件中的字幕文本，输出 cn.srt。"""
|
|
srt_uri = request.inputs.get("srt_uri")
|
|
if not srt_uri:
|
|
return InvokeResponse(status="failed", error="srt_uri is required")
|
|
|
|
srt_path = Path(srt_uri)
|
|
if not srt_path.is_file():
|
|
return InvokeResponse(status="failed", error="srt file not found")
|
|
|
|
# 标准 SRT 每 4 行一组：序号、时间轴、文本、空行；文本位于第 3 行。
|
|
lines = srt_path.read_text(encoding="utf-8").splitlines()
|
|
text_indices = list(range(2, len(lines), 4))
|
|
source_lines = [lines[index] for index in text_indices]
|
|
translated_lines = translate_lines(source_lines, request.params)
|
|
# 防止模型返回行数偏差：多出的截断，缺少的用空串补齐。
|
|
translated_lines = translated_lines[: len(source_lines)]
|
|
translated_lines += [""] * max(0, len(source_lines) - len(translated_lines))
|
|
# 只替换文本行，序号、时间轴和空行保持不变。
|
|
for index, text_index in enumerate(text_indices):
|
|
lines[text_index] = translated_lines[index]
|
|
|
|
output_dir = Path(request.output_dir)
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
output_path = output_dir / "cn.srt"
|
|
# 末尾补一个换行，让文件满足常见文本工具习惯。
|
|
output_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
return InvokeResponse(status="completed", outputs={"cn_srt_uri": str(output_path)})
|
|
|