first commit

.gitignore

@@ -0,0 +1,27 @@
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+
+# Environment variables
+.env
+
+# Local models
+models/
+models/*
+
+# Streamlit config (optional, if you want to ignore local server config)
+#.streamlit/
+
+# Video processing temp files
+*.mp4
+*.mp3
+*.wav
+*.srt
+*.mkv
+*.avi
+*.mov
+
+# IDE
+.vscode/
+.idea/

.streamlit/config.toml

@@ -0,0 +1,3 @@
+[server]
+maxUploadSize = 10240
+maxMessageSize = 10240

README.md

@@ -1,11 +1,103 @@
----
-title: Translator
-emoji: 📊
-colorFrom: pink
-colorTo: red
-sdk: docker
-pinned: false
-short_description: 音视频翻译
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# 🎬 AI 媒体翻译专家 (Media Translator)
+
+这是一个基于 Python 的视频/音频翻译工具，可以自动提取媒体音频、识别语音生成字幕、通过 AI 翻译字幕，并最终将字幕压制回视频中（仅视频）。支持本地 CPU 运行语音识别，翻译部分支持对接各种主流 AI API。
+
+## ✨ 功能特点
+
+- **全媒体支持**：支持视频（mp4, mkv, avi, mov）和音频（mp3, wav, m4a, flac, aac）文件。
+- **本地语音识别 (STT)**：使用 `faster-whisper` 模型，在本地即可完成语音转文字，支持自动检测语言。
+- **直接字幕翻译**：支持直接上传已有的 `.srt` 字幕文件进行翻译，无需重新处理视频。
+- **AI 智能翻译**：兼容 OpenAI 格式的 API（如 SiliconFlow、Zenmux、DeepSeek 等），支持自定义 API 地址和模型。
+- **视频硬压字幕**：自动将翻译后的字幕嵌入到视频中。
+- **多种导出格式**：支持导出原始 SRT 字幕、翻译后的 SRT 字幕以及压制好的 MP4 视频。
+- **低硬件门槛**：经过优化，即使在没有显卡（仅 CPU）的普通电脑上也能流畅运行。
+
+## 🛠️ 环境准备
+
+在运行本项目之前，请确保已安装以下工具：
+
+### 1. FFmpeg (核心依赖)
+FFmpeg 负责音频提取和视频合成，是必须安装的。
+- **Windows**:
+  1. 下载 [FFmpeg 编译版](https://www.gyan.dev/ffmpeg/builds/ffmpeg-git-full.7z)。
+  2. 解压并将 `bin` 目录路径添加到系统的 **环境变量 (PATH)** 中。(在打开的“编辑环境变量”窗口中，窗口的下半部分“**系统变量**”区域，找到并选中名为 **`Path`** 的变量。点击“新建”,将你电脑上 FFmpeg 的 bin 文件夹的完整路径粘贴进去。例如：D:\ffmpeg\bin 或 C:\Tools\ffmpeg-6.1-full_build\bin)
+  3. 验证：在终端输入 `ffmpeg -version` 确认是否有输出。
+- **macOS**: `brew install ffmpeg`
+- **Linux**: `sudo apt install ffmpeg`
+
+### 2. Python 3.8+
+建议使用 Python 3.10 或更高版本。
+
+## 🚀 安装与运行
+
+1. **克隆或下载本项目**到本地。
+
+2. **安装依赖库**：
+   ```bash
+   pip install -r requirements.txt
+   ```
+
+3. **启动程序**：
+   ```bash
+   streamlit run app.py
+   ```
+   启动后，浏览器会自动打开 `http://localhost:8501`。
+
+   **注意**：为了支持大视频上传，本项目已在 `.streamlit/config.toml` 中配置了最大 10GB 的上传限制。
+
+## ⚙️ 自动保存配置 (.env)
+
+为了避免每次运行都要手动输入 API Key，你可以创建一个名为 `.env` 的文件（可以参考项目中的 `.env.example`）：
+
+1.  在项目根目录下新建文件 `.env`。
+2.  填写以下内容：
+    ```env
+    API_KEY=你的_API_KEY
+    BASE_URL=https://api.siliconflow.cn/v1
+    MODEL_NAME=THUDM/glm-4-9b-chat
+    STT_MODEL_SIZE=base
+    TARGET_LANG=中文
+    ```
+3.  下次启动程序时，这些值将自动填充到界面中。
+
+## 📖 使用指南
+
+1. **配置 API**：
+   - 在左侧边栏输入你的 **API Key**。
+   - 如果使用 SiliconFlow，默认地址为 `https://api.siliconflow.cn/v1`。
+   - 输入你想要使用的模型名称（例如 `THUDM/glm-4-9b-chat` 或 `deepseek-chat`）。
+
+2. **上传媒体文件**：
+   - 点击或拖拽视频（mp4, mkv, avi, mov）或音频（mp3, wav, m4a 等）文件到上传区。
+
+3. **开始处理**：
+   - 选择本地 STT 模型（建议 CPU 使用 `base` 或 `small`）。
+   - 选择目标翻译语言（默认为中文）。
+   - 点击 **“开始处理”**。
+
+4. **获取结果**：
+   - 处理完成后，你可以分别下载：原始字幕、翻译字幕、带字幕的视频。
+
+## � 文件存储与清理
+
+为了确保处理流程的稳定性，程序在运行过程中会产生一些临时文件：
+
+- **存储位置**：所有上传的文件和中间产物（音频、临时字幕、合成视频）都保存在**操作系统的默认临时目录**中。
+  - **Windows**: `C:\Users\你的用户名\AppData\Local\Temp`
+  - **macOS/Linux**: `/tmp` 或 `$TMPDIR`
+- **清理机制**：程序**不会自动删除**这些临时文件。
+- **手动清理**：如果处理的文件较多或较大，建议定期手动清理系统临时文件夹中以 `tmp` 开头的文件，以释放磁盘空间。
+
+## �� 项目结构
+
+- `app.py`: Streamlit Web 界面主程序。
+- `stt_module.py`: 封装了 `faster-whisper` 的语音识别逻辑。
+- `translator.py`: 封装了基于 OpenAI 协议的 AI 翻译逻辑。
+- `utils.py`: 封装了 FFmpeg 相关的视频与音频处理工具。
+- `requirements.txt`: 项目所需的 Python 依赖列表。
+
+## ⚠️ 注意事项
+
+- **首次运行**：第一次使用某种 STT 模型（如 `base`）时，程序会自动从 HuggingFace 下���模型文件，请确保网络通畅。
+- **性能**：如果电脑配置较低，建议使用 `tiny` 或 `base` 模型以加快识别速度。
+- **翻译额度**：翻译功能依赖外部 API，请确保你的 API 账户有足够余额或额度。

app.py

@@ -0,0 +1,308 @@
+import streamlit as st
+import os
+from utils import VideoUtils
+from stt_module import STTManager
+from translator import Translator
+import tempfile
+from dotenv import load_dotenv
+
+# 加载 .env 文件中的配置
+load_dotenv()
+
+# 取消 Streamlit 上传限制（虽然主要通过命令行配置，但在脚本中提醒用户）
+# 实际上 Streamlit 的服务器配置需要在运行命令时指定，或者写在 config.toml 中
+# 我们这里先通过 UI 提醒用户
+
+st.set_page_config(page_title="AI 媒体翻译专家", layout="wide")
+
+st.title("🎬 AI 媒体字幕提取与翻译")
+
+# 侧边栏配置
+with st.sidebar:
+    st.header("⚙️ 配置参数")
+    
+    # 从环境变量读取默认值，如果没有则为空字符串或预设值
+    default_api_key = os.getenv("API_KEY", "")
+    default_base_url = os.getenv("BASE_URL", "https://api.siliconflow.cn/v1")
+    default_model = os.getenv("MODEL_NAME", "THUDM/glm-4-9b-chat")
+    default_stt_size = os.getenv("STT_MODEL_SIZE", "base")
+    default_lang = os.getenv("TARGET_LANG", "中文")
+    default_device = os.getenv("DEVICE", "cpu")
+
+    api_key = st.text_input("API Key", value=default_api_key, type="password", help="输入 API Key")
+    base_url = st.text_input("API Base URL", value=default_base_url)
+    model_name = st.text_input("模型名称", value=default_model)
+    
+    st.divider()
+    
+    # 设备选择
+    cuda_available = STTManager.is_cuda_available()
+    device_options = ["cpu", "cuda"] if cuda_available else ["cpu"]
+    device_index = device_options.index(default_device) if default_device in device_options else 0
+    device = st.selectbox(
+        "计算设备 (Device)", 
+        device_options, 
+        index=device_index,
+        help="如果有 NVIDIA 显卡且安装了 CUDA，选择 'cuda' 会大幅提升速度。"
+    )
+    
+    # 根据设备推荐精度
+    default_compute_type = "float16" if device == "cuda" else "int8"
+    compute_type = st.selectbox(
+        "计算精度 (Compute Type)",
+        ["int8", "float16", "int8_float16"],
+        index=1 if device == "cuda" else 0,
+        help="CPU 推荐 int8，GPU 推荐 float16。"
+    )
+
+    stt_options = ["tiny", "base", "small", "medium", "large-v3"]
+    # 获取已下载模型
+    downloaded_models = STTManager.get_downloaded_models()
+    
+    # 构建选项显示名称
+    option_labels = []
+    for opt in stt_options:
+        label = f"{opt} (已下载)" if opt in downloaded_models else opt
+        option_labels.append(label)
+
+    stt_index = stt_options.index(default_stt_size) if default_stt_size in stt_options else 1
+    selected_option = st.selectbox(
+        "本地 STT 模型大小", 
+        option_labels, 
+        index=stt_index, 
+        help="越大越准，但速度越慢。首次使用未下载的模型时会自动下载。"
+    )
+    # 从选项中提取真实模型名
+    model_size = selected_option.split(" ")[0]
+    
+    lang_options = ["中文", "English", "日本語", "Français"]
+    lang_index = lang_options.index(default_lang) if default_lang in lang_options else 0
+    target_lang = st.selectbox("目标语言", lang_options, index=lang_index)
+
+# 主界面
+tab1, tab2 = st.tabs(["📁 媒体处理", "📜 字幕翻译"])
+
+if "process_results" not in st.session_state:
+    st.session_state.process_results = {}
+if "srt_results" not in st.session_state:
+    st.session_state.srt_results = {}
+if "awaiting_synthesis" not in st.session_state:
+    st.session_state.awaiting_synthesis = False
+
+with tab1:
+    st.info("💡 提示：支持视频和音频文件。如果文件非常大，处理会很慢请耐心等待。")
+    uploaded_file = st.file_uploader("选择视频或音频文件", type=["mp4", "mkv", "avi", "mov", "mp3", "wav", "m4a", "flac", "aac"])
+
+    if uploaded_file:
+        # 检测文件类型
+        video_extensions = [".mp4", ".mkv", ".avi", ".mov"]
+        file_ext = os.path.splitext(uploaded_file.name)[1].lower()
+        is_video = file_ext in video_extensions
+        
+        # 如果上传了新文件且与 session_state 中记录的不同，则清除旧结果
+        if "last_uploaded_file" not in st.session_state or st.session_state.last_uploaded_file != uploaded_file.name:
+            st.session_state.process_results = {}
+            st.session_state.awaiting_synthesis = False
+            st.session_state.last_uploaded_file = uploaded_file.name
+
+        # 保存上传的文件到临时目录
+        tfile = tempfile.NamedTemporaryFile(delete=False, suffix=file_ext)
+        tfile.write(uploaded_file.read())
+        file_path = tfile.name
+
+        if is_video:
+            st.video(file_path)
+        else:
+            st.audio(file_path)
+
+        if st.button("开始处理", disabled=not api_key or st.session_state.awaiting_synthesis):
+            # 清除旧结果
+            st.session_state.process_results = {}
+            st.session_state.awaiting_synthesis = False
+            
+            with st.status("正在处理中...", expanded=True) as status:
+                # 1. 提取/准备音频
+                st.write("🎵 正在准备音频...")
+                audio_path = os.path.splitext(file_path)[0] + '.wav'
+                try:
+                    # 无论视频还是音频，都通过 prepare_audio (ffmpeg) 转换为标准格式，确保 STT 兼容性
+                    VideoUtils.prepare_audio(file_path, audio_path)
+                except Exception as e:
+                    st.error(str(e))
+                    status.update(label="处理出错", state="error", expanded=True)
+                    st.stop()
+
+                # 2. 本地 STT
+                if model_size not in downloaded_models:
+                    st.write(f"📥 正在下载 {model_size} 模型，请稍候...")
+                
+                st.write(f"✍️ 正在识别语音 (使用 {model_size} 模型，设备: {device})...")
+                stt_manager = STTManager(model_size=model_size, device=device, compute_type=compute_type)
+                stt_manager.load_model()
+                
+                segments_gen, info = stt_manager.transcribe(audio_path)
+                st.write(f"检测到语言: {info.language} (置信度: {info.language_probability:.2f})")
+
+                # 增量处理与展示
+                st.write("---")
+                st.write("实时识别与翻译预览：")
+                preview_container = st.empty()
+                all_segments = []
+                all_translated_segments = []
+                
+                translator = Translator(api_key, base_url, model_name)
+                
+                # 用于展示的表格数据
+                display_data = []
+                
+                for segment in segments_gen:
+                    # 1. 翻译当前段落
+                    trans_text = translator.translate_text(segment.text, target_lang)
+                    
+                    # 2. 保存原始和翻译后的段落
+                    all_segments.append(segment)
+                    new_trans_seg = type('Segment', (), {
+                        'start': segment.start,
+                        'end': segment.end,
+                        'text': trans_text
+                    })
+                    all_translated_segments.append(new_trans_seg)
+                    
+                    # 3. 更新预览界面
+                    time_str = f"{VideoUtils.format_timestamp(segment.start)} -> {VideoUtils.format_timestamp(segment.end)}"
+                    display_data.append({
+                        "时间轴": time_str,
+                        "原文": segment.text,
+                        "翻译": trans_text
+                    })
+                    # 仅显示最后 5 条，避免页面过长，但提供滚动查看全部的可能
+                    preview_container.table(display_data[-5:])
+
+                # 生成原始字幕
+                orig_srt_path = os.path.splitext(file_path)[0] + '_orig.srt'
+                VideoUtils.write_srt(all_segments, orig_srt_path)
+                
+                # 生成翻译字幕
+                trans_srt_path = os.path.splitext(file_path)[0] + '_trans.srt'
+                VideoUtils.write_srt(all_translated_segments, trans_srt_path)
+
+                status.update(label="处理完成！", state="complete", expanded=False)
+
+                # 保存结果到 session_state
+                st.session_state.process_results = {
+                    "orig_srt": orig_srt_path,
+                    "trans_srt": trans_srt_path,
+                    "output_video": None,
+                    "is_video": is_video
+                }
+
+                # 保存中间结果用于后续合成 (仅针对视频)
+                if is_video:
+                    st.session_state.temp_video_path = file_path
+                    st.session_state.temp_trans_srt_path = trans_srt_path
+                    st.session_state.temp_orig_srt_path = orig_srt_path
+                    st.session_state.awaiting_synthesis = True
+                
+                st.rerun()
+
+        # 4. 嵌入视频 (仅视频且在等待状态)
+        if st.session_state.awaiting_synthesis and st.session_state.process_results.get("is_video"):
+            st.success("✅ 字幕翻译已完成！")
+            col_synth_1, col_synth_2 = st.columns(2)
+            with col_synth_1:
+                if st.button("🚀 开始合成视频字幕", type="primary"):
+                    with st.status("🎬 正在合成视频字幕...", expanded=True) as status:
+                        v_path = st.session_state.temp_video_path
+                        s_path = st.session_state.temp_trans_srt_path
+                        output_video_path = os.path.splitext(v_path)[0] + '_translated.mp4'
+                        
+                        video_ready = False
+                        try:
+                            VideoUtils.embed_subtitles(v_path, s_path, output_video_path)
+                            video_ready = True
+                            st.write("✨ 视频合成成功！")
+                        except Exception as e:
+                            st.error(f"视频合成失败 (请确保已安装 FFmpeg): {e}")
+
+                        status.update(label="全部处理完成！", state="complete", expanded=False)
+
+                        # 保存最终结果到 session_state
+                        st.session_state.process_results["output_video"] = output_video_path if video_ready else None
+                        st.session_state.awaiting_synthesis = False
+                        st.rerun()
+            
+            with col_synth_2:
+                if st.button("📂 仅保存字幕"):
+                    st.session_state.awaiting_synthesis = False
+                    st.rerun()
+
+    # 结果展示与下载 (移出 button 缩进块，始终根据 session_state 显示)
+    if st.session_state.process_results:
+        st.divider()
+        col_title, col_clear = st.columns([5, 1])
+        with col_title:
+            st.subheader("🎉 处理结果")
+        with col_clear:
+            if st.button("🗑️ 清除结果"):
+                st.session_state.process_results = {}
+                st.session_state.awaiting_synthesis = False
+                st.rerun()
+        
+        col1, col2, col3 = st.columns(3)
+        
+        results = st.session_state.process_results
+        
+        if os.path.exists(results.get("orig_srt", "")):
+            with col1:
+                with open(results["orig_srt"], "rb") as f:
+                    st.download_button("⬇️ 下载原始字幕", f, file_name="original.srt", key="dl_orig")
+        
+        if os.path.exists(results.get("trans_srt", "")):
+            with col2:
+                with open(results["trans_srt"], "rb") as f:
+                    st.download_button("⬇️ 下载翻译字幕", f, file_name="translated.srt", key="dl_trans")
+        
+        if results.get("output_video") and os.path.exists(results["output_video"]):
+            with col3:
+                with open(results["output_video"], "rb") as f:
+                    st.download_button("⬇️ 下载翻译视频", f, file_name="video_with_subtitles.mp4", key="dl_video")
+
+with tab2:
+    st.info("如果你已经有原始语言的字幕文件（SRT），可以在这里直接进行翻译。")
+    uploaded_srt = st.file_uploader("上传原始 SRT 字幕", type=["srt"])
+    
+    if uploaded_srt:
+        # 如果上传了新字幕且与 session_state 中记录的不同，则清除旧结果
+        if "last_uploaded_srt" not in st.session_state or st.session_state.last_uploaded_srt != uploaded_srt.name:
+            st.session_state.srt_results = {}
+            st.session_state.last_uploaded_srt = uploaded_srt.name
+
+        srt_content = uploaded_srt.read().decode("utf-8")
+        st.text_area("字幕预览", srt_content, height=200)
+        
+        if st.button("开始翻译字幕", disabled=not api_key):
+            st.session_state.srt_results = {} # 清除旧结果
+            with st.spinner("正在翻译字幕..."):
+                segments = VideoUtils.parse_srt(srt_content)
+                translator = Translator(api_key, base_url, model_name)
+                translated_segments = translator.translate_segments(segments, target_lang)
+                
+                # 保存到临时文件
+                temp_trans_srt = tempfile.NamedTemporaryFile(delete=False, suffix='.srt')
+                VideoUtils.write_srt(translated_segments, temp_trans_srt.name)
+                
+                st.session_state.srt_results = {
+                    "trans_srt": temp_trans_srt.name
+                }
+                st.success("翻译完成！")
+
+    # 结果下载
+    if st.session_state.srt_results:
+        st.divider()
+        res = st.session_state.srt_results
+        if os.path.exists(res.get("trans_srt", "")):
+            with open(res["trans_srt"], "rb") as f:
+                st.download_button("⬇️ 下载翻译后的字幕", f, file_name="translated_only.srt", key="dl_srt_tab")
+
+if not api_key:
+    st.warning("请在左侧边栏输入 API Key 后开始。")

requirements.txt

@@ -0,0 +1,8 @@
+faster-whisper
+openai
+pydantic
+requests
+tqdm
+streamlit
+python-dotenv
+ffmpeg-python

stt_module.py

@@ -0,0 +1,71 @@
+from faster_whisper import WhisperModel
+import os
+
+class STTManager:
+    def __init__(self, model_size="base", device="cpu", compute_type="int8"):
+        """
+        model_size: tiny, base, small, medium, large-v3
+        device: cpu or cuda
+        compute_type: int8, float16, int8_float16 等
+        """
+        self.model_size = model_size
+        self.device = device
+        self.compute_type = compute_type
+        self.model = None
+
+    def load_model(self):
+        """延迟加载模型，方便在加载前显示提示"""
+        if self.model is None:
+            # 自动处理 compute_type，如果 GPU 不支持 float16 则回退
+            stt_compute_type = self.compute_type
+            if self.device == "cpu" and stt_compute_type == "float16":
+                stt_compute_type = "int8"
+                
+            self.model = WhisperModel(
+                self.model_size, 
+                device=self.device, 
+                compute_type=stt_compute_type,
+                download_root=os.path.join(os.getcwd(), "models")
+            )
+        return self.model
+
+    def transcribe(self, audio_path):
+        """识别音频并返回 segments (生成器) 和 info"""
+        model = self.load_model()
+        segments, info = model.transcribe(audio_path, beam_size=5)
+        return segments, info
+
+    @staticmethod
+    def is_cuda_available():
+        """检测 CUDA 是否可用"""
+        try:
+            import ctranslate2
+            return ctranslate2.get_cuda_device_count() > 0
+        except:
+            return False
+
+    @staticmethod
+    def get_downloaded_models():
+        """获取本地已下载的模型列表"""
+        model_dir = os.path.join(os.getcwd(), "models")
+        if not os.path.exists(model_dir):
+            return []
+        
+        models = []
+        for d in os.listdir(model_dir):
+            if "faster-whisper-" in d:
+                name = d.split("-")[-1]
+                models.append(name)
+            elif os.path.isdir(os.path.join(model_dir, d)):
+                models.append(d)
+        
+        valid_names = ["tiny", "base", "small", "medium", "large-v1", "large-v2", "large-v3"]
+        found = [m for m in models if any(v in m for v in valid_names)]
+        final_models = []
+        for f in found:
+            for v in valid_names:
+                if v in f:
+                    final_models.append(v)
+                    break
+        
+        return sorted(list(set(final_models)))

translator.py

@@ -0,0 +1,36 @@
+from openai import OpenAI
+
+class Translator:
+    def __init__(self, api_key, base_url, model="base"):
+        self.client = OpenAI(api_key=api_key, base_url=base_url)
+        self.model = model
+
+    def translate_text(self, text, target_lang="中文"):
+        """翻译单段文本"""
+        if not text.strip():
+            return ""
+        
+        prompt = f"你是一个专业的视频字幕翻译专家。请将以下字幕内容翻译成{target_lang}，保持原意，语言自然、简洁。只返回翻译后的内容，不要有任何解释。\n\n内容：{text}"
+        
+        try:
+            response = self.client.chat.completions.create(
+                model=self.model,
+                messages=[{"role": "user", "content": prompt}],
+                temperature=0.3
+            )
+            return response.choices[0].message.content.strip()
+        except Exception as e:
+            return f"翻译错误: {str(e)}"
+
+    def translate_segments(self, segments, target_lang="中文"):
+        """批量翻译字幕段 (为了演示，这里使用循环翻译，实际可优化为批量处理)"""
+        translated_segments = []
+        for seg in segments:
+            # 创建一个新的对象来保存翻译后的内容，保持原有的时间戳
+            new_seg = type('Segment', (), {
+                'start': seg.start,
+                'end': seg.end,
+                'text': self.translate_text(seg.text, target_lang)
+            })
+            translated_segments.append(new_seg)
+        return translated_segments

utils.py

@@ -0,0 +1,86 @@
+import subprocess
+import os
+import json
+
+class VideoUtils:
+    @staticmethod
+    def prepare_audio(input_path, output_path):
+        """从视频或音频中提取/转换音频为标准格式 (16kHz, mono, wav)"""
+        # 使用 wav 格式更通用，不需要 libmp3lame 编码器
+        cmd = [
+            'ffmpeg', '-y', '-i', input_path,
+            '-vn', '-acodec', 'pcm_s16le', '-ar', '16000', '-ac', '1',
+            output_path
+        ]
+        try:
+            result = subprocess.run(cmd, check=True, capture_output=True, text=True)
+            return result
+        except subprocess.CalledProcessError as e:
+            error_msg = f"FFmpeg 处理音频失败!\n错误代码: {e.returncode}\n错误输出: {e.stderr}"
+            print(error_msg) # 控制台打印
+            raise Exception(error_msg)
+        except FileNotFoundError:
+            raise Exception("找不到 FFmpeg 命令。请确保 FFmpeg 已安装并已添加到系统环境变量 PATH 中。")
+
+    @staticmethod
+    def embed_subtitles(video_path, srt_path, output_path):
+        """将字幕嵌入视频 (硬压)"""
+        # 注意：Windows下路径处理较复杂，ffmpeg 的 subtitles 滤镜需要特殊转义
+        abs_srt_path = os.path.abspath(srt_path).replace('\\', '/').replace(':', '\\:')
+        cmd = [
+            'ffmpeg', '-y', '-i', video_path,
+            '-vf', f"subtitles='{abs_srt_path}'",
+            '-c:a', 'copy',
+            output_path
+        ]
+        try:
+            result = subprocess.run(cmd, check=True, capture_output=True, text=True)
+            return result
+        except subprocess.CalledProcessError as e:
+            error_msg = f"FFmpeg 合成视频失败!\n错误代码: {e.returncode}\n错误输出: {e.stderr}"
+            print(error_msg)
+            raise Exception(error_msg)
+        except FileNotFoundError:
+            raise Exception("找不到 FFmpeg 命令。")
+
+    @staticmethod
+    def format_timestamp(seconds: float):
+        """将秒转换为 SRT 时间格式 00:00:00,000"""
+        td_hours = int(seconds // 3600)
+        td_mins = int((seconds % 3600) // 60)
+        td_secs = int(seconds % 60)
+        td_msecs = int((seconds - int(seconds)) * 1000)
+        return f"{td_hours:02}:{td_mins:02}:{td_secs:02},{td_msecs:03}"
+
+    @staticmethod
+    def write_srt(segments, output_path):
+        """生成 SRT 格式文件"""
+        with open(output_path, 'w', encoding='utf-8') as f:
+            for i, segment in enumerate(segments, 1):
+                start = VideoUtils.format_timestamp(segment.start)
+                end = VideoUtils.format_timestamp(segment.end)
+                f.write(f"{i}\n{start} --> {end}\n{segment.text.strip()}\n\n")
+
+    @staticmethod
+    def parse_srt(srt_content):
+        """简单的 SRT 解析器，返回 segments 列表"""
+        import re
+        segments = []
+        # 正则匹配 SRT 块
+        pattern = re.compile(r'(\d+)\n(\d{2}:\d{2}:\d{2},\d{3}) --> (\d{2}:\d{2}:\d{2},\d{3})\n(.*?)(?=\n\n|\n$|$)', re.DOTALL)
+        
+        def time_to_seconds(t_str):
+            h, m, s_ms = t_str.split(':')
+            s, ms = s_ms.split(',')
+            return int(h) * 3600 + int(m) * 60 + int(s) + int(ms) / 1000.0
+
+        matches = pattern.findall(srt_content)
+        for m in matches:
+            idx, start_t, end_t, text = m
+            seg = type('Segment', (), {
+                'start': time_to_seconds(start_t),
+                'end': time_to_seconds(end_t),
+                'text': text.strip()
+            })
+            segments.append(seg)
+        return segments