GitHub主页(欢迎star❤):
https://github.com/xunbu/docutranslate
交流QQ群: 1047781902
当前没有任务，点击“新建任务”开始吧！
等待提交...
点击或拖拽文件到此处
GitHub主页(欢迎star❤):
https://github.com/xunbu/docutranslate
交流QQ群: 1047781902
当前没有任务，点击“新建任务”开始吧！
等待提交...
点击或拖拽文件到此处
+ {{ sub.index }}
+ {{ sub.start }} --> {{ sub.end }}
+ {{ sub.content | safe }}
+
没有字幕内容可显示。
+ {% endfor %} + + + \ No newline at end of file diff --git a/docutranslate/translator/ai_translator/srt_translator.py b/docutranslate/translator/ai_translator/srt_translator.py new file mode 100644 index 0000000..5606538 --- /dev/null +++ b/docutranslate/translator/ai_translator/srt_translator.py @@ -0,0 +1,137 @@ +import asyncio +from dataclasses import dataclass +from typing import Self, Literal + +import srt # 导入srt库来处理字幕文件 + +from docutranslate.agents.segments_agent import SegmentsTranslateAgentConfig, SegmentsTranslateAgent +from docutranslate.ir.document import Document +from docutranslate.translator.ai_translator.base import AiTranslatorConfig +from docutranslate.translator.base import Translator + + +@dataclass +class SrtTranslatorConfig(AiTranslatorConfig): + insert_mode: Literal["replace", "append", "prepend"] = "replace" + separator: str = "\n" + + +class SrtTranslator(Translator): + """ + 一个用于翻译 SRT (.srt) 字幕文件的翻译器。 + 它会提取每个字幕块的文本内容，进行翻译，然后根据配置将译文写回。 + """ + + def __init__(self, config: SrtTranslatorConfig): + super().__init__(config=config) + self.chunk_size = config.chunk_size + agent_config = SegmentsTranslateAgentConfig( + custom_prompt=config.custom_prompt, + to_lang=config.to_lang, + baseurl=config.base_url, + key=config.api_key, + model_id=config.model_id, + system_prompt=None, + temperature=config.temperature, + thinking=config.thinking, + max_concurrent=config.concurrent, + timeout=config.timeout, + logger=self.logger + ) + self.translate_agent = SegmentsTranslateAgent(agent_config) + self.insert_mode = config.insert_mode + self.separator = config.separator + + def _pre_translate(self, document: Document): + """ + 预处理步骤：解析SRT文件，提取所有字幕文本。 + + Returns: + tuple: (解析后的字幕对象列表, 待翻译的原文文本列表) + """ + try: + # 使用 utf-8-sig 解码以处理可能存在的BOM (Byte Order Mark) + srt_content = document.content.decode('utf-8-sig') + except (UnicodeDecodeError, AttributeError) as e: + self.logger.error(f"无法解码SRT文件内容，请确保文件编码为UTF-8: {e}") + return [], [] + + # 使用 srt 库解析内容 + try: + subtitles = list(srt.parse(srt_content)) + except srt.SRTParseError as e: + self.logger.error(f"解析SRT文件失败: {e}") + return [], [] + + # 提取所有原文文本，准备进行批量翻译 + original_texts = [sub.content for sub in subtitles] + + return subtitles, original_texts + + def _after_translate(self, subtitles: list[srt.Subtitle], translated_texts: list[str], + original_texts: list[str]) -> bytes: + """ + 翻译后处理步骤：将译文根据配置模式写回字幕对象，并生成新的SRT文件内容。 + + Returns: + bytes: 新的SRT文件内容的字节流。 + """ + for i, sub in enumerate(subtitles): + translated_text = translated_texts[i] + original_text = original_texts[i] + + # 根据插入模式更新字幕内容 + if self.insert_mode == "replace": + sub.content = translated_text + elif self.insert_mode == "append": + # strip() 避免在原文和译文间产生多余的空白 + sub.content = original_text.strip() + self.separator + translated_text.strip() + elif self.insert_mode == "prepend": + sub.content = translated_text.strip() + self.separator + original_text.strip() + else: + self.logger.error(f"不正确的SrtTranslatorConfig参数: insert_mode='{self.insert_mode}'") + # 默认回退到替换模式，避免程序中断 + sub.content = translated_text + + # 使用 srt 库将修改后的字幕对象列表重新合成为SRT格式的字符串 + new_srt_content_str = srt.compose(subtitles) + + # 返回UTF-8编码的字节流 + return new_srt_content_str.encode('utf-8') + + def translate(self, document: Document) -> Self: + """ + 同步翻译SRT文档。 + """ + subtitles, original_texts = self._pre_translate(document) + + if not original_texts: + self.logger.info("\n文件中没有找到需要翻译的字幕内容。") + return self + + # --- 步骤 2: 调用翻译Agent --- + translated_texts = self.translate_agent.send_segments(original_texts, self.chunk_size) + + # --- 步骤 3: 后处理并更新文档内容 --- + document.content = self._after_translate(subtitles, translated_texts, original_texts) + return self + + async def translate_async(self, document: Document) -> Self: + """ + 异步翻译SRT文档。 + """ + # I/O密集型操作在线程中运行 + subtitles, original_texts = await asyncio.to_thread(self._pre_translate, document) + + if not original_texts: + self.logger.info("\n文件中没有找到需要翻译的字幕内容。") + return self + + # --- 步骤 2: 调用翻译Agent (异步) --- + translated_texts = await self.translate_agent.send_segments_async(original_texts, self.chunk_size) + + # --- 步骤 3: 后处理并更新文档内容 (I/O密集型) --- + document.content = await asyncio.to_thread( + self._after_translate, subtitles, translated_texts, original_texts + ) + return self \ No newline at end of file diff --git a/docutranslate/workflow/interfaces.py b/docutranslate/workflow/interfaces.py index a971765..745eee7 100644 --- a/docutranslate/workflow/interfaces.py +++ b/docutranslate/workflow/interfaces.py @@ -70,4 +70,12 @@ class DocxExportable(Protocol[T_ExporterConfig]): ... def save_as_docx(self, name: str, output_dir: Path | str, config: T_ExporterConfig | None = None) -> Self: + ... + +@runtime_checkable +class SrtExportable(Protocol[T_ExporterConfig]): + def export_to_srt(self, config: T_ExporterConfig | None = None) -> str: + ... + + def save_as_srt(self, name: str, output_dir: Path | str, config: T_ExporterConfig | None = None) -> Self: ... \ No newline at end of file diff --git a/docutranslate/workflow/srt_workflow.py b/docutranslate/workflow/srt_workflow.py new file mode 100644 index 0000000..8afeca6 --- /dev/null +++ b/docutranslate/workflow/srt_workflow.py @@ -0,0 +1,66 @@ +from dataclasses import dataclass +from pathlib import Path +from typing import Self + +from docutranslate.exporter.base import ExporterConfig +from docutranslate.exporter.srt.srt2html_exporter import Srt2HTMLExporterConfig, Srt2HTMLExporter +from docutranslate.exporter.srt.srt2srt_exporter import Srt2SrtExporter +from docutranslate.ir.document import Document +from docutranslate.translator.ai_translator.srt_translator import SrtTranslatorConfig, SrtTranslator +from docutranslate.workflow.base import Workflow, WorkflowConfig +from docutranslate.workflow.interfaces import HTMLExportable, SrtExportable + + +@dataclass(kw_only=True) +class SrtWorkflowConfig(WorkflowConfig): + translator_config: SrtTranslatorConfig + html_exporter_config: Srt2HTMLExporterConfig + + +class SrtWorkflow(Workflow[SrtWorkflowConfig, Document, Document], HTMLExportable[Srt2HTMLExporterConfig], + SrtExportable[ExporterConfig]): + def __init__(self, config: SrtWorkflowConfig): + super().__init__(config=config) + if config.logger: + for sub_config in [self.config.translator_config]: + if sub_config: + sub_config.logger = config.logger + + def _pre_translate(self,document_original:Document): + document = document_original.copy() + translate_config = self.config.translator_config + translator = SrtTranslator(translate_config) + return document,translator + + + def translate(self) -> Self: + document, translator=self._pre_translate(self.document_original) + translator.translate(document) + self.document_translated = document + return self + + async def translate_async(self) -> Self: + document, translator = self._pre_translate(self.document_original) + await translator.translate_async(document) + self.document_translated = document + return self + + def export_to_html(self, config: Srt2HTMLExporterConfig = None) -> str: + config = config or self.config.html_exporter_config + docu = self._export(Srt2HTMLExporter(config)) + return docu.content.decode() + + def export_to_srt(self, _: ExporterConfig | None = None) -> str: + docu = self._export(Srt2SrtExporter()) + return docu.content.decode() + + def save_as_html(self, name: str = None, output_dir: Path | str = "./output", + config: Srt2HTMLExporterConfig | None = None) -> Self: + config = config or self.config.html_exporter_config + self._save(exporter=Srt2HTMLExporter(config), name=name, output_dir=output_dir) + return self + + def save_as_srt(self, name: str = None, output_dir: Path | str = "./output", + _: ExporterConfig | None = None) -> Self: + self._save(exporter=Srt2SrtExporter(), name=name, output_dir=output_dir) + return self diff --git a/pyproject.toml b/pyproject.toml index 71a83c2..b8284bf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,6 +12,7 @@ dependencies = [ "xlsx2html>=0.6.2", "json-repair>=0.48.0", "mammoth>=1.10.0", + "srt>=3.5.3", ] dynamic = ["version"] diff --git a/uv.lock b/uv.lock index 0f22180..e742f61 100644 --- a/uv.lock +++ b/uv.lock @@ -322,6 +322,7 @@ dependencies = [ { name = "mammoth" }, { name = "markdown2" }, { name = "openpyxl" }, + { name = "srt" }, { name = "xlsx2html" }, ] @@ -349,6 +350,7 @@ requires-dist = [ { name = "markdown2", specifier = ">=2.5.3" }, { name = "opencv-python", marker = "extra == 'docling'", specifier = ">=4.11.0.86" }, { name = "openpyxl", specifier = ">=3.1.5" }, + { name = "srt", specifier = ">=3.5.3" }, { name = "xlsx2html", specifier = ">=0.6.2" }, ] provides-extras = ["docling"] @@ -2302,6 +2304,12 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/e7/9c/0e6afc12c269578be5c0c1c9f4b49a8d32770a080260c333ac04cc1c832d/soupsieve-2.7-py3-none-any.whl", hash = "sha256:6e60cc5c1ffaf1cebcc12e8188320b72071e922c2e897f737cadce79ad5d30c4", size = 36677 }, ] +[[package]] +name = "srt" +version = "3.5.3" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/66/b7/4a1bc231e0681ebf339337b0cd05b91dc6a0d701fa852bb812e244b7a030/srt-3.5.3.tar.gz", hash = "sha256:4884315043a4f0740fd1f878ed6caa376ac06d70e135f306a6dc44632eed0cc0", size = 28296 } + [[package]] name = "starlette" version = "0.47.2"