chore: rebase到develop + 格式化代码

- 字幕领域模型：SubtitleWord/SubtitleSegment/SubtitleTimeline，支持合并短片段/拆分长片段
- ASR服务抽象层：Port接口 + Mock实现，可扩展Whisper/云厂商ASR
- 字幕生成器：时间轴 → ASS字幕，支持自动换行、自定义样式
- SubtitleConfig扩展：auto_generated/language/max_chars_per_line/min_chars_per_segment
- 渲染管道接入：UnifiedRenderService支持ASR自动字幕，失败降级不阻断
- Worker端集成：asr_service_factory根据ASR_PROVIDER环境变量初始化
- 向后兼容：默认关闭，不传ASR服务时完全不影响现有渲染
- 42个新增单测 + 84个渲染测试全绿，无回归

apps/worker/services/asr_service_factory.py

@@ -0,0 +1,47 @@
+"""ASR 服务工厂 — 根据环境配置创建对应 ASR 服务实例。
+
+支持的后端：
+- mock: MockASRService（测试/开发用）
+- 后续可扩展：whisper / aliyun / tencent 等
+"""
+
+from __future__ import annotations
+
+import os
+from functools import lru_cache
+
+from packages.ports.asr_service import ASRService
+
+
+@lru_cache(maxsize=1)
+def get_asr_service() -> ASRService | None:
+    """获取全局 ASR 服务实例（单例）。
+
+    根据环境变量 ASR_PROVIDER 决定使用哪个后端：
+    - mock / 空 / 未设置: 返回 None（不启用 ASR）
+    - mock: 使用 MockASRService
+
+    Returns:
+        ASRService 实例，未配置或不启用时返回 None
+    """
+    provider = os.environ.get("ASR_PROVIDER", "").lower().strip()
+
+    if not provider:
+        return None
+
+    if provider == "mock":
+        from packages.adapters.asr.mock_asr_service import MockASRService
+
+        return MockASRService()
+
+    # 未知 provider，记录日志并返回 None（不启用 ASR，不阻断主流程）
+    import logging
+
+    logger = logging.getLogger(__name__)
+    logger.warning("未知的 ASR provider: %s，ASR 自动字幕功能未启用", provider)
+    return None
+
+
+def reset_asr_service_cache() -> None:
+    """重置 ASR 服务缓存（测试用）。"""
+    get_asr_service.cache_clear()

apps/worker/video_processing/subtitle_generator.py

@@ -0,0 +1,183 @@
+"""字幕生成器 — 将字幕时间轴转换为 ASS 字幕文件。
+
+与 render_subtitles.py 的区别：
+- render_subtitles.py 处理静态整段标题/字幕
+- 本模块处理带时间轴的多段 ASR 字幕
+
+两者最终都输出 ASS 文件，供 FFmpeg 烧录。
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+from typing import Any
+
+from packages.domain.subtitle import SubtitleTimeline
+
+logger = logging.getLogger(__name__)
+
+
+# ── 常量 ──────────────────────────────────────────────────────────────────────
+
+DEFAULT_MAX_CHARS_PER_LINE = 20  # 每行最多字符数
+DEFAULT_MIN_CHARS_PER_SEGMENT = 8  # 每段最少字符数
+
+
+# ── ASS 工具函数 ────────────────────────────────────────────────────────────
+
+
+def _hex_to_ass_color(hex_color: str) -> str:
+    """将 HEX 颜色（#RRGGBB）转换为 ASS &HBBGGRR 格式。"""
+    hex_color = hex_color.lstrip("#")
+    if len(hex_color) != 6:
+        return "&H00FFFFFF"
+    r, g, b = hex_color[0:2], hex_color[2:4], hex_color[4:6]
+    return f"&H{b.upper()}{g.upper()}{r.upper()}"
+
+
+def _position_to_ass_alignment(position: str) -> int:
+    """将文字位置映射为 ASS \\an 对齐编号。"""
+    mapping = {
+        "top": 8,
+        "center": 5,
+        "bottom": 2,
+    }
+    return mapping.get(position, 2)
+
+
+def _format_ass_time(seconds: float) -> str:
+    """将秒数格式化为 ASS 时间格式 H:MM:SS.cc。"""
+    hours = int(seconds // 3600)
+    minutes = int((seconds % 3600) // 60)
+    secs = seconds % 60
+    return f"{hours}:{minutes:02d}:{secs:05.2f}"
+
+
+def _escape_ass_text(text: str) -> str:
+    """转义 ASS 文本中的特殊字符。"""
+    text = text.replace("\r\n", "\\N").replace("\n", "\\N").replace("\r", "\\N")
+    text = text.replace("{", "(").replace("}", ")")
+    return text
+
+
+def _wrap_text(text: str, max_chars: int) -> list[str]:
+    """将长文本按字数换行。
+
+    优先在标点处换行，没有合适标点时硬切。
+    """
+    if len(text) <= max_chars:
+        return [text]
+
+    lines: list[str] = []
+    remaining = text
+
+    while len(remaining) > max_chars:
+        # 在前 max_chars 个字符中找标点断开
+        break_point = max_chars
+        punctuations = "，。！？、；：,.;:!?"
+
+        for i in range(max_chars, max_chars // 2, -1):
+            if i < len(remaining) and remaining[i] in punctuations:
+                break_point = i + 1
+                break
+
+        lines.append(remaining[:break_point])
+        remaining = remaining[break_point:]
+
+    if remaining:
+        lines.append(remaining)
+
+    return lines
+
+
+# ── 主生成器 ─────────────────────────────────────────────────────────────────
+
+
+def generate_ass_from_timeline(
+    output_path: Path,
+    timeline: SubtitleTimeline,
+    *,
+    video_width: int,
+    video_height: int,
+    subtitle_config: dict[str, Any] | None = None,
+) -> Path:
+    """从字幕时间轴生成 ASS 字幕文件。
+
+    Args:
+        output_path: 输出 ASS 文件路径
+        timeline: 字幕时间轴
+        video_width: 视频宽度
+        video_height: 视频高度
+        subtitle_config: 字幕样式配置（同 SubtitleConfig dict）
+
+    Returns:
+        生成的 ASS 文件路径
+    """
+    subtitle_config = subtitle_config or {}
+
+    if not timeline.segments:
+        output_path.write_text("", encoding="utf-8")
+        return output_path
+
+    # 样式参数
+    font_name = subtitle_config.get("font", "思源黑体")
+    font_size = int(subtitle_config.get("size", 24))
+    color = _hex_to_ass_color(subtitle_config.get("color", "#ffffff"))
+    position = subtitle_config.get("position", "bottom")
+    alignment = _position_to_ass_alignment(position)
+    max_chars_per_line = int(subtitle_config.get("max_chars_per_line", DEFAULT_MAX_CHARS_PER_LINE))
+
+    # 描边（默认黑色描边，保证可读性）
+    outline_color = "&H00000000"
+    outline_width = 1.5
+
+    # 边距
+    margin_v = 60 if position == "bottom" else 60
+    margin_l = 40
+    margin_r = 40
+
+    # 生成样式行
+    style_line = (
+        f"Style: Default,{font_name},{font_size},{color},"
+        f"&H000000FF,{outline_color},&H00000000,"
+        f"-1,0,0,0,100,100,0,0,"
+        f"1,{outline_width},0,{alignment},"
+        f"{margin_l},{margin_r},{margin_v},1"
+    )
+
+    # 生成事件行
+    events: list[str] = []
+    for seg in timeline.segments:
+        start_time = _format_ass_time(seg.start)
+        end_time = _format_ass_time(seg.end)
+
+        # 自动换行
+        lines = _wrap_text(seg.text, max_chars_per_line)
+        display_text = "\\N".join(lines)
+
+        safe_text = _escape_ass_text(display_text)
+
+        events.append(f"Dialogue: 0,{start_time},{end_time},Default,,0,0,0,,{safe_text}")
+
+    # 组装 ASS 文件
+    ass_content = f"""[Script Info]
+ScriptType: v4.00+
+PlayResX: {video_width}
+PlayResY: {video_height}
+ScaledBorderAndShadow: yes
+WrapStyle: 2
+Encoding: UTF-8
+
+[V4+ Styles]
+Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding
+{style_line}
+
+[Events]
+Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
+{chr(10).join(events)}
+"""
+
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    output_path.write_text(ass_content, encoding="utf-8")
+    return output_path

apps/worker/video_processing/unified_render_service.py

@@ -41,6 +41,7 @@ from video_processing.ffmpeg_utils import (
 )
 from video_processing.render_audio import RenderContext, merge_audio_video, mix_audio
 from video_processing.render_subtitles import generate_ass_subtitles
+from video_processing.subtitle_generator import generate_ass_from_timeline
 
 logger = logging.getLogger(__name__)
 
@@ -158,6 +159,7 @@ class UnifiedRenderService:
         output_height: int = DEFAULT_OUTPUT_HEIGHT,
         output_fps: int = DEFAULT_FPS,
         transition_duration: float = DEFAULT_TRANSITION_DURATION,
+        asr_service: Any = None,  # ASRService 实例，用于自动生成字幕
     ):
         self.plan = plan
         self.clips = clips
@@ -167,6 +169,7 @@ class UnifiedRenderService:
         self.output_height = output_height
         self.output_fps = output_fps
         self.transition_duration = transition_duration
+        self.asr_service = asr_service
 
     def render(self) -> RenderResult:
         """执行渲染，返回 RenderResult.
@@ -341,6 +344,10 @@ class UnifiedRenderService:
     def _maybe_generate_ass(self, video_duration: float) -> Path | None:
         """根据 plan.config 生成 ASS 字幕文件。
 
+        支持两种字幕模式：
+        1. 静态字幕 — title/subtitle 配置了 text 时，生成整段静态字幕
+        2. ASR 自动字幕 — subtitle.auto_generated=true 时，从音频自动识别生成时间轴字幕
+
         Returns:
             ASS 文件路径，没有字幕时返回 None
         """
@@ -352,15 +359,46 @@ class UnifiedRenderService:
         subtitle_enabled = subtitle_cfg.get("enabled", True)
         title_text = title_cfg.get("text", "") or ""
         subtitle_text = subtitle_cfg.get("text", "") or ""
+        auto_generated = subtitle_cfg.get("auto_generated", False)
 
         has_title = title_enabled and bool(title_text.strip())
-        has_subtitle = subtitle_enabled and bool(subtitle_text.strip())
+        has_static_subtitle = subtitle_enabled and bool(subtitle_text.strip())
+        has_auto_subtitle = subtitle_enabled and auto_generated and self.asr_service is not None
 
-        if not has_title and not has_subtitle:
+        if not has_title and not has_static_subtitle and not has_auto_subtitle:
             return None
 
         ass_path = self.work_dir / f"subtitles_{self.plan.id}.ass"
 
+        # ASR 自动字幕模式
+        if has_auto_subtitle:
+            try:
+                timeline = self._generate_asr_subtitles(video_duration, subtitle_cfg)
+                if timeline and timeline.segments:
+                    generate_ass_from_timeline(
+                        ass_path,
+                        timeline,
+                        video_width=self.output_width,
+                        video_height=self.output_height,
+                        subtitle_config=subtitle_cfg,
+                    )
+                    logger.info(
+                        "ASR自动字幕生成完成: plan_id=%s segments=%d duration=%.1fs",
+                        self.plan.id,
+                        timeline.segment_count,
+                        video_duration,
+                    )
+                    return ass_path
+                else:
+                    # ASR 无结果，不生成字幕
+                    logger.info("ASR自动字幕无识别结果，跳过字幕: plan_id=%s", self.plan.id)
+                    return None
+            except Exception:
+                # ASR 失败降级：不生成字幕，不阻断主流程
+                logger.warning("ASR自动字幕生成失败，跳过字幕", exc_info=True)
+                return None
+
+        # 静态字幕模式（原有逻辑）
         generate_ass_subtitles(
             ass_path,
             video_width=self.output_width,
@@ -376,11 +414,95 @@ class UnifiedRenderService:
             "生成字幕: plan_id=%s title=%s subtitle=%s ass=%s",
             self.plan.id,
             has_title,
-            has_subtitle,
+            has_static_subtitle,
             ass_path,
         )
         return ass_path
 
+    def _generate_asr_subtitles(self, video_duration: float, subtitle_cfg: dict) -> Any:  # SubtitleTimeline
+        """从视频素材音频中自动识别生成字幕时间轴。
+
+        MVP 版本：使用第一个有音频的素材做ASR，然后按比例映射到整个视频时长。
+        后续优化：支持多片段拼接后的完整音频ASR。
+        """
+        from packages.domain.subtitle import SubtitleTimeline
+
+        # 找第一个有本地路径的素材
+        first_asset_path = None
+        for clip in self.clips:
+            asset_id = getattr(clip, "asset_id", None)
+            if asset_id and asset_id in self.asset_path_map:
+                first_asset_path = self.asset_path_map[asset_id]
+                break
+
+        if first_asset_path is None:
+            logger.warning("ASR字幕生成失败：找不到可用素材音频")
+            return SubtitleTimeline(segments=[], total_duration=video_duration)
+
+        # 提取素材音频为 wav（16kHz单声道，ASR友好格式）
+        audio_path = self.work_dir / f"asr_audio_{self.plan.id}.wav"
+        try:
+            self._extract_audio(first_asset_path, audio_path)
+        except Exception:
+            logger.warning("ASR音频提取失败", exc_info=True)
+            return SubtitleTimeline(segments=[], total_duration=video_duration)
+
+        if not audio_path.exists():
+            return SubtitleTimeline(segments=[], total_duration=video_duration)
+
+        # 调用 ASR 服务
+        language = subtitle_cfg.get("language", "") or None
+        timeline = self.asr_service.transcribe(
+            audio_path,
+            language=language,
+            with_word_timestamps=True,
+        )
+
+        # 字幕后处理：合并短片段 + 拆分长片段
+        min_chars = int(subtitle_cfg.get("min_chars_per_segment", 8))
+        max_chars = int(subtitle_cfg.get("max_chars_per_line", 20))
+
+        if timeline.segments:
+            timeline = timeline.merge_short_segments(min_chars=min_chars)
+            timeline = timeline.split_long_segments(max_chars=max_chars)
+
+        # 清理临时音频文件
+        try:
+            audio_path.unlink(missing_ok=True)
+        except Exception:
+            pass
+
+        return timeline
+
+    def _extract_audio(self, video_path: Path, output_path: Path) -> None:
+        """从视频中提取音频为16kHz单声道wav（ASR友好格式）。"""
+        import subprocess
+
+        cmd = [
+            "ffmpeg",
+            "-y",
+            "-i",
+            str(video_path),
+            "-vn",
+            "-acodec",
+            "pcm_s16le",
+            "-ar",
+            "16000",
+            "-ac",
+            "1",
+            str(output_path),
+        ]
+
+        result = subprocess.run(
+            cmd,
+            capture_output=True,
+            text=True,
+            timeout=120,
+        )
+
+        if result.returncode != 0:
+            raise RuntimeError(f"音频提取失败: {result.stderr[:200]}")
+
     def _can_use_pass_through(self, layers: list[RenderLayer]) -> bool:
         """判断是否可以走直通优化路径。
 

apps/worker/worker_app/tasks/generation.py

@@ -108,6 +108,7 @@ def _flush_logs(task_id: str, gen_task) -> None:
 
 # ── 共享工具模块导入 ──────────────────────────────────────────────────────────
 
+from services.asr_service_factory import get_asr_service
 from video_processing.dedup_helpers import create_video_record_and_dedup
 from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, run_ffmpeg
 from video_processing.oss_helpers import (
@@ -861,6 +862,7 @@ def _render_video(
             output_width=OUTPUT_WIDTH,
             output_height=OUTPUT_HEIGHT,
             output_fps=int(OUTPUT_FPS),
+            asr_service=get_asr_service(),
         )
         render_result = render_service.render()
         render_output_path = render_result.output_path

packages/adapters/asr/mock_asr_service.py

@@ -0,0 +1,113 @@
+"""Mock ASR 服务 — 用于测试和开发环境。
+
+生成模拟的字幕时间轴，不依赖真实ASR服务。
+"""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+from typing import Optional
+
+from packages.domain.subtitle import (
+    SubtitleSegment,
+    SubtitleTimeline,
+    SubtitleWord,
+)
+from packages.ports.asr_service import ASRService, ASRServiceError
+
+
+class MockASRService(ASRService):
+    """Mock ASR 服务，生成模拟字幕数据。
+
+    如果 audio_path 对应的目录下有同名 .txt 文件，
+    就读取该文件内容作为字幕文本，按时间均匀分段。
+    否则生成默认的测试字幕。
+    """
+
+    def __init__(self, mock_text: Optional[str] = None):
+        self._mock_text = mock_text
+
+    def transcribe(
+        self,
+        audio_path: Path,
+        language: Optional[str] = None,
+        with_word_timestamps: bool = True,
+    ) -> SubtitleTimeline:
+        if not audio_path.exists():
+            raise ASRServiceError(f"音频文件不存在: {audio_path}", provider="mock")
+
+        # 尝试读取同名 txt 文件作为字幕文本
+        text = self._mock_text
+        if text is None:
+            txt_path = audio_path.with_suffix(".txt")
+            if txt_path.exists():
+                text = txt_path.read_text(encoding="utf-8").strip()
+            else:
+                text = "这是一段测试字幕。它用于验证ASR自动字幕功能是否正常工作。每一句话都会被正确地分段并显示在视频底部。字幕的样式可以根据用户的喜好进行自定义调整。"
+
+        # 估算音频时长（用ffmpeg probe或者直接假设）
+        # mock模式下按字数估算，每秒4个字
+        total_duration = max(5.0, len(text) / 4.0)
+
+        segments = self._text_to_segments(text, total_duration, with_word_timestamps)
+
+        return SubtitleTimeline(
+            segments=segments,
+            language=language or "zh",
+            total_duration=total_duration,
+        )
+
+    def _text_to_segments(
+        self,
+        text: str,
+        total_duration: float,
+        with_word_timestamps: bool,
+    ) -> list[SubtitleSegment]:
+        """将文本按句切分成带时间轴的字幕片段。"""
+        # 按句末标点拆分
+        sentences = re.split(r"(?<=[。！？!?])", text)
+        sentences = [s.strip() for s in sentences if s.strip()]
+
+        if not sentences:
+            sentences = [text]
+
+        total_chars = sum(len(s) for s in sentences)
+        if total_chars == 0:
+            return []
+
+        segments = []
+        current_time = 0.0
+
+        for sentence in sentences:
+            char_count = len(sentence)
+            duration = total_duration * (char_count / total_chars)
+            end_time = current_time + duration
+
+            words: list[SubtitleWord] = []
+            if with_word_timestamps:
+                # 每个字作为一个词级单元（中文按字，英文按词）
+                word_time = current_time
+                word_duration = duration / char_count
+
+                for char in sentence:
+                    words.append(
+                        SubtitleWord(
+                            text=char,
+                            start=word_time,
+                            end=word_time + word_duration,
+                        )
+                    )
+                    word_time += word_duration
+
+            segments.append(
+                SubtitleSegment(
+                    text=sentence,
+                    start=current_time,
+                    end=end_time,
+                    words=words,
+                )
+            )
+            current_time = end_time
+
+        return segments

packages/domain/config_schemas.py

@@ -112,6 +112,11 @@ class SubtitleConfig(BaseModel):
     color: str = Field(default="#ffffff", description="文字颜色 (HEX)")
     size: int = Field(default=24, ge=12, le=60, description="字号")
     animation: TextAnimation = Field(default=TextAnimation.FADE_IN, description="入场动画")
+    # ASR 自动字幕
+    auto_generated: bool = Field(default=False, description="是否启用ASR自动生成字幕")
+    language: str = Field(default="", description="字幕语言，空字符串表示自动检测（如 zh/en/ja）")
+    max_chars_per_line: int = Field(default=20, ge=8, le=40, description="每行最多字符数")
+    min_chars_per_segment: int = Field(default=8, ge=2, le=20, description="每段最少字符数（低于则合并）")
 
 
 class BGMConfig(BaseModel):

packages/domain/subtitle.py

@@ -0,0 +1,183 @@
+"""字幕领域模型 — 带时间轴的字幕片段。"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import List
+
+
+@dataclass
+class SubtitleWord:
+    """单个词级别的字幕单元，带精确时间戳。"""
+
+    text: str
+    start: float  # 秒
+    end: float  # 秒
+
+    @property
+    def duration(self) -> float:
+        return max(0.0, self.end - self.start)
+
+
+@dataclass
+class SubtitleSegment:
+    """一段字幕（一句话），带时间轴和词级信息。"""
+
+    text: str
+    start: float  # 秒
+    end: float  # 秒
+    words: List[SubtitleWord] = field(default_factory=list)
+
+    @property
+    def duration(self) -> float:
+        return max(0.0, self.end - self.start)
+
+    @property
+    def char_count(self) -> int:
+        return len(self.text)
+
+
+@dataclass
+class SubtitleTimeline:
+    """完整的字幕时间轴，由多个片段组成。"""
+
+    segments: List[SubtitleSegment] = field(default_factory=list)
+    language: str = "zh"  # zh / en / ja 等
+    total_duration: float = 0.0  # 音频总时长（秒）
+
+    @property
+    def segment_count(self) -> int:
+        return len(self.segments)
+
+    @property
+    def total_chars(self) -> int:
+        return sum(s.char_count for s in self.segments)
+
+    def merge_short_segments(self, min_chars: int = 8) -> SubtitleTimeline:
+        """合并过短的字幕片段，避免字幕跳动太快。"""
+        if len(self.segments) <= 1:
+            return self
+
+        merged: List[SubtitleSegment] = []
+        buffer: List[SubtitleSegment] = []
+
+        for seg in self.segments:
+            buffer.append(seg)
+            total_chars = sum(s.char_count for s in buffer)
+            if total_chars >= min_chars:
+                merged.append(self._merge_segments(buffer))
+                buffer = []
+
+        # 剩余的合并到最后一个或单独成段
+        if buffer:
+            if merged and sum(s.char_count for s in buffer) < min_chars:
+                # 太少了，合并到上一段
+                last = merged.pop()
+                merged.append(self._merge_segments([last] + buffer))
+            else:
+                merged.append(self._merge_segments(buffer))
+
+        return SubtitleTimeline(
+            segments=merged,
+            language=self.language,
+            total_duration=self.total_duration,
+        )
+
+    def split_long_segments(self, max_chars: int = 20) -> SubtitleTimeline:
+        """拆分过长的字幕片段，按语义断句。"""
+        new_segments: List[SubtitleSegment] = []
+
+        for seg in self.segments:
+            if seg.char_count <= max_chars:
+                new_segments.append(seg)
+                continue
+
+            # 按标点符号拆分
+            parts = self._split_text_by_punctuation(seg.text, max_chars)
+            if len(parts) == 1:
+                new_segments.append(seg)
+                continue
+
+            # 按字数比例分配时间
+            total_chars = seg.char_count
+            current_time = seg.start
+            word_idx = 0
+            all_words = seg.words.copy()
+
+            for part in parts:
+                part_chars = len(part)
+                part_duration = seg.duration * (part_chars / total_chars)
+                part_end = min(current_time + part_duration, seg.end)
+
+                # 收集对应时间段的词
+                part_words = []
+                while word_idx < len(all_words) and all_words[word_idx].start < part_end:
+                    part_words.append(all_words[word_idx])
+                    word_idx += 1
+
+                new_segments.append(
+                    SubtitleSegment(
+                        text=part,
+                        start=current_time,
+                        end=part_end,
+                        words=part_words,
+                    )
+                )
+                current_time = part_end
+
+        return SubtitleTimeline(
+            segments=new_segments,
+            language=self.language,
+            total_duration=self.total_duration,
+        )
+
+    @staticmethod
+    def _merge_segments(segments: List[SubtitleSegment]) -> SubtitleSegment:
+        if not segments:
+            return SubtitleSegment(text="", start=0, end=0)
+        return SubtitleSegment(
+            text="".join(s.text for s in segments),
+            start=segments[0].start,
+            end=segments[-1].end,
+            words=[w for s in segments for w in s.words],
+        )
+
+    @staticmethod
+    def _split_text_by_punctuation(text: str, max_chars: int) -> List[str]:
+        """按标点符号智能拆分长文本。"""
+        # 中文常见句末标点
+        sentence_end = "。！？!?"
+        clause_pause = "，；：,;:"
+
+        parts: List[str] = []
+        current = ""
+
+        for char in text:
+            current += char
+
+            if len(current) >= max_chars:
+                # 超过长度，找最近的标点断开
+                break_idx = -1
+                for i in range(len(current) - 1, -1, -1):
+                    if current[i] in sentence_end or current[i] in clause_pause:
+                        break_idx = i + 1
+                        break
+
+                if break_idx > 0:
+                    parts.append(current[:break_idx])
+                    current = current[break_idx:]
+                else:
+                    # 没有标点，硬切
+                    parts.append(current[:max_chars])
+                    current = current[max_chars:]
+
+            elif char in sentence_end:
+                # 句末标点，如果长度够就断开
+                if len(current) >= max_chars // 2:
+                    parts.append(current)
+                    current = ""
+
+        if current:
+            parts.append(current)
+
+        return parts

packages/ports/asr_service.py

@@ -0,0 +1,47 @@
+"""ASR（语音识别）服务接口 — Port 层。"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from pathlib import Path
+from typing import Optional
+
+from packages.domain.subtitle import SubtitleTimeline
+
+
+class ASRService(ABC):
+    """ASR 服务抽象接口。
+
+    不同的 ASR 后端（Whisper、阿里云、腾讯云等）实现此接口，
+    上层业务代码只依赖接口，不依赖具体实现。
+    """
+
+    @abstractmethod
+    def transcribe(
+        self,
+        audio_path: Path,
+        language: Optional[str] = None,
+        with_word_timestamps: bool = True,
+    ) -> SubtitleTimeline:
+        """将音频文件转写为带时间轴的字幕。
+
+        Args:
+            audio_path: 音频文件路径（支持 wav/mp3/m4a 等常见格式）
+            language: 指定语言代码（zh/en/ja 等），None 表示自动检测
+            with_word_timestamps: 是否返回词级时间戳
+
+        Returns:
+            SubtitleTimeline 字幕时间轴对象
+
+        Raises:
+            ASRServiceError: 识别服务调用失败
+        """
+        ...
+
+
+class ASRServiceError(Exception):
+    """ASR 服务调用异常。"""
+
+    def __init__(self, message: str, provider: str = "unknown"):
+        self.provider = provider
+        super().__init__(f"[{provider}] {message}")

tests/unit/test_asr_subtitle_integration.py

@@ -0,0 +1,327 @@
+"""ASR 自动字幕集成测试 — 验证渲染管道接入 ASR 的完整链路。"""
+
+from __future__ import annotations
+
+import tempfile
+from dataclasses import dataclass
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+from video_processing.unified_render_service import UnifiedRenderService
+
+from packages.adapters.asr.mock_asr_service import MockASRService
+from packages.domain.subtitle import SubtitleSegment, SubtitleTimeline
+
+# ── Fixtures ──────────────────────────────────────────────────────────────────
+
+
+@dataclass
+class FakeClip:
+    """模拟 EditPlanClip。"""
+
+    id: str
+    plan_id: str = "plan_001"
+    asset_id: str = "asset_001"
+    clip_type: str = "video"
+    start_time: float = 0.0
+    duration: float = 10.0
+    layer: int = 0
+    role: str = "main"
+    config: dict = None
+
+
+@dataclass
+class FakePlan:
+    """模拟 EditPlan。"""
+
+    id: str = "plan_001"
+    config: dict = None
+
+
+@pytest.fixture
+def work_dir():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        yield Path(tmpdir)
+
+
+@pytest.fixture
+def test_video_path():
+    """用 ffmpeg 生成一个5秒的测试视频（带音频）。"""
+    import subprocess
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        video_path = Path(tmpdir) / "test.mp4"
+        cmd = [
+            "ffmpeg",
+            "-y",
+            "-f",
+            "lavfi",
+            "-i",
+            "testsrc=duration=5:size=320x240:rate=30",
+            "-f",
+            "lavfi",
+            "-i",
+            "sine=frequency=440:duration=5",
+            "-c:v",
+            "libx264",
+            "-preset",
+            "ultrafast",
+            "-c:a",
+            "aac",
+            "-shortest",
+            str(video_path),
+        ]
+        result = subprocess.run(cmd, capture_output=True, timeout=30)
+        if result.returncode != 0:
+            pytest.skip(f"ffmpeg 不可用或生成测试视频失败: {result.stderr[:200]}")
+        yield video_path
+
+
+# ── 测试：ASR 服务接入 ────────────────────────────────────────────────────────
+
+
+class TestASRServiceIntegration:
+    def test_asr_service_in_init(self):
+        """验证 asr_service 参数正确传递。"""
+        plan = FakePlan(config={})
+        service = UnifiedRenderService(
+            plan=plan,
+            clips=[],
+            asset_path_map={},
+            work_dir=Path("/tmp"),
+            asr_service=MockASRService(),
+        )
+        assert service.asr_service is not None
+        assert isinstance(service.asr_service, MockASRService)
+
+    def test_no_asr_service_default(self):
+        """验证不传 asr_service 时默认 None。"""
+        plan = FakePlan(config={})
+        service = UnifiedRenderService(
+            plan=plan,
+            clips=[],
+            asset_path_map={},
+            work_dir=Path("/tmp"),
+        )
+        assert service.asr_service is None
+
+    def test_maybe_generate_ass_auto_subtitle_with_asr(self, work_dir, test_video_path):
+        """验证 ASR 自动字幕模式：有 asr_service + auto_generated=true 时生成 ASS。"""
+        plan = FakePlan(
+            config={
+                "subtitle": {
+                    "enabled": True,
+                    "auto_generated": True,
+                    "position": "bottom",
+                }
+            }
+        )
+        clips = [
+            FakeClip(id="clip_1", asset_id="asset_1", duration=5.0),
+        ]
+        asset_map = {"asset_1": test_video_path}
+
+        service = UnifiedRenderService(
+            plan=plan,
+            clips=clips,
+            asset_path_map=asset_map,
+            work_dir=work_dir,
+            output_width=320,
+            output_height=240,
+            asr_service=MockASRService(mock_text="这是ASR自动生成的测试字幕。用来验证渲染管道是否正常接入。"),
+        )
+
+        ass_path = service._maybe_generate_ass(5.0)
+
+        assert ass_path is not None
+        assert ass_path.exists()
+        content = ass_path.read_text(encoding="utf-8")
+        assert "[Events]" in content
+        assert "Dialogue:" in content
+        assert "ASR" in content
+
+    def test_maybe_generate_ass_no_asr_service_skip_auto(self, work_dir):
+        """验证没有 asr_service 时，即使 auto_generated=true 也不生成 ASR 字幕。"""
+        plan = FakePlan(
+            config={
+                "subtitle": {
+                    "enabled": True,
+                    "auto_generated": True,
+                    "text": "",
+                }
+            }
+        )
+        service = UnifiedRenderService(
+            plan=plan,
+            clips=[],
+            asset_path_map={},
+            work_dir=work_dir,
+            asr_service=None,  # 没有 ASR 服务
+        )
+
+        ass_path = service._maybe_generate_ass(5.0)
+        # 没有 ASR 服务 + 没有静态字幕文本 → 返回 None
+        assert ass_path is None
+
+    def test_maybe_generate_ass_auto_mode_ignores_text(self, work_dir):
+        """验证 ASR 模式下即使有 text 字段也走 ASR（ASR无结果则无字幕）。"""
+        plan = FakePlan(
+            config={
+                "subtitle": {
+                    "enabled": True,
+                    "auto_generated": True,
+                    "text": "静态字幕文本",  # ASR模式下忽略此字段
+                }
+            }
+        )
+        mock_asr = MockASRService()
+        mock_asr.transcribe = MagicMock(side_effect=mock_asr.transcribe)
+
+        service = UnifiedRenderService(
+            plan=plan,
+            clips=[],
+            asset_path_map={},
+            work_dir=work_dir,
+            asr_service=mock_asr,
+        )
+
+        ass_path = service._maybe_generate_ass(5.0)
+        # ASR模式下无素材 → 无结果 → 返回None（不fallback到静态text）
+        assert ass_path is None
+
+    def test_maybe_generate_ass_title_still_works(self, work_dir):
+        """验证 ASR 模式下不影响 title 的处理（两者独立）。"""
+        plan = FakePlan(
+            config={
+                "title": {
+                    "enabled": True,
+                    "text": "视频标题",
+                    "position": "top",
+                },
+                "subtitle": {
+                    "enabled": False,  # 字幕关闭
+                    "auto_generated": True,
+                },
+            }
+        )
+        mock_asr = MockASRService()
+
+        service = UnifiedRenderService(
+            plan=plan,
+            clips=[],
+            asset_path_map={},
+            work_dir=work_dir,
+            asr_service=mock_asr,
+        )
+
+        ass_path = service._maybe_generate_ass(5.0)
+        assert ass_path is not None
+        content = ass_path.read_text(encoding="utf-8")
+        assert "视频标题" in content
+
+    def test_asr_failure_does_not_block(self, work_dir, test_video_path):
+        """验证 ASR 失败时不阻断主流程，降级为无字幕。"""
+        plan = FakePlan(
+            config={
+                "subtitle": {
+                    "enabled": True,
+                    "auto_generated": True,
+                }
+            }
+        )
+        clips = [FakeClip(id="clip_1", asset_id="asset_1", duration=5.0)]
+        asset_map = {"asset_1": test_video_path}
+
+        # ASR 服务总是抛异常
+        bad_asr = MockASRService()
+        bad_asr.transcribe = MagicMock(side_effect=RuntimeError("ASR service down"))
+
+        service = UnifiedRenderService(
+            plan=plan,
+            clips=clips,
+            asset_path_map=asset_map,
+            work_dir=work_dir,
+            output_width=320,
+            output_height=240,
+            asr_service=bad_asr,
+        )
+
+        # 应该不抛异常，返回 None（降级）
+        ass_path = service._maybe_generate_ass(5.0)
+        assert ass_path is None  # ASR 失败 → 无字幕
+
+    def test_auto_subtitle_disabled(self, work_dir):
+        """验证 subtitle.enabled=false 时即使 auto_generated=true 也不生成。"""
+        plan = FakePlan(
+            config={
+                "subtitle": {
+                    "enabled": False,
+                    "auto_generated": True,
+                }
+            }
+        )
+        mock_asr = MockASRService()
+        mock_asr.transcribe = MagicMock()
+
+        service = UnifiedRenderService(
+            plan=plan,
+            clips=[],
+            asset_path_map={},
+            work_dir=work_dir,
+            asr_service=mock_asr,
+        )
+
+        ass_path = service._maybe_generate_ass(5.0)
+        assert ass_path is None
+        mock_asr.transcribe.assert_not_called()
+
+
+# ── 测试：SubtitleConfig 扩展 ────────────────────────────────────────────────
+
+
+class TestSubtitleConfigExtension:
+    def test_config_has_auto_generated_field(self):
+        """验证 SubtitleConfig 有 auto_generated 字段。"""
+        from packages.domain.config_schemas import SubtitleConfig
+
+        config = SubtitleConfig()
+        assert hasattr(config, "auto_generated")
+        assert config.auto_generated is False  # 默认关闭
+
+    def test_config_default_values(self):
+        """验证新增字段的默认值。"""
+        from packages.domain.config_schemas import SubtitleConfig
+
+        config = SubtitleConfig()
+        assert config.auto_generated is False
+        assert config.language == ""
+        assert config.max_chars_per_line == 20
+        assert config.min_chars_per_segment == 8
+
+    def test_config_custom_values(self):
+        """验证可以自定义 ASR 相关字段。"""
+        from packages.domain.config_schemas import SubtitleConfig
+
+        config = SubtitleConfig(
+            auto_generated=True,
+            language="zh",
+            max_chars_per_line=15,
+            min_chars_per_segment=5,
+        )
+        assert config.auto_generated is True
+        assert config.language == "zh"
+        assert config.max_chars_per_line == 15
+        assert config.min_chars_per_segment == 5
+
+    def test_config_validation_max_chars(self):
+        """验证 max_chars_per_line 的范围校验。"""
+        from pydantic import ValidationError
+
+        from packages.domain.config_schemas import SubtitleConfig
+
+        with pytest.raises(ValidationError):
+            SubtitleConfig(max_chars_per_line=5)  # 小于8
+
+        with pytest.raises(ValidationError):
+            SubtitleConfig(max_chars_per_line=50)  # 大于40

tests/unit/test_subtitle_generator.py

@@ -0,0 +1,178 @@
+"""字幕生成器 + Mock ASR 单元测试。"""
+
+import tempfile
+from pathlib import Path
+
+import pytest
+
+from apps.worker.video_processing.subtitle_generator import (
+    _wrap_text,
+    generate_ass_from_timeline,
+)
+from packages.adapters.asr.mock_asr_service import MockASRService
+from packages.domain.subtitle import SubtitleSegment, SubtitleTimeline
+from packages.ports.asr_service import ASRServiceError
+
+
+class TestMockASRService:
+    def test_transcribe_with_mock_text(self):
+        service = MockASRService(mock_text="你好世界！这是一段测试语音识别的文字。用来验证Mock ASR是否正常工作。")
+
+        # 创建一个假的音频文件（mock不真的读内容）
+        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
+            f.write(b"fake audio data")
+            audio_path = Path(f.name)
+
+        try:
+            timeline = service.transcribe(audio_path, language="zh")
+            assert timeline is not None
+            assert timeline.language == "zh"
+            assert timeline.segment_count > 0
+            assert timeline.total_duration > 0
+            # 总字数应该对得上
+            assert timeline.total_chars == len("你好世界！这是一段测试语音识别的文字。用来验证Mock ASR是否正常工作。")
+        finally:
+            audio_path.unlink()
+
+    def test_transcribe_file_not_found(self):
+        service = MockASRService()
+        with pytest.raises(ASRServiceError):
+            service.transcribe(Path("/nonexistent/audio.wav"))
+
+    def test_transcribe_with_word_timestamps(self):
+        service = MockASRService(mock_text="你好世界！")
+
+        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
+            f.write(b"fake")
+            audio_path = Path(f.name)
+
+        try:
+            timeline = service.transcribe(audio_path, with_word_timestamps=True)
+            # 每段应该有词级时间戳
+            for seg in timeline.segments:
+                if seg.words:
+                    assert len(seg.words) > 0
+                    assert seg.words[0].start >= seg.start
+                    assert seg.words[-1].end <= seg.end
+        finally:
+            audio_path.unlink()
+
+    def test_auto_detect_language(self):
+        service = MockASRService(mock_text="Hello world. This is a test.")
+
+        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
+            f.write(b"fake")
+            audio_path = Path(f.name)
+
+        try:
+            timeline = service.transcribe(audio_path, language=None)
+            # None 时默认 zh
+            assert timeline.language == "zh"
+        finally:
+            audio_path.unlink()
+
+
+class TestWrapText:
+    def test_short_text_no_wrap(self):
+        result = _wrap_text("你好世界", 20)
+        assert result == ["你好世界"]
+
+    def test_wrap_at_punctuation(self):
+        result = _wrap_text("你好世界！这是一段很长的测试文字。", 10)
+        assert len(result) == 2
+        assert "！" in result[0]
+
+    def test_hard_wrap_no_punctuation(self):
+        result = _wrap_text("一二三四五六七八九十一二三四五六七八九十", 10)
+        assert len(result) == 2
+        assert len(result[0]) == 10
+        assert len(result[1]) == 10
+
+    def test_exact_length(self):
+        result = _wrap_text("一二三四五六七八九十", 10)
+        assert len(result) == 1
+
+
+class TestGenerateAssFromTimeline:
+    def test_generate_basic(self):
+        timeline = SubtitleTimeline(
+            segments=[
+                SubtitleSegment(text="你好世界", start=0.0, end=2.0),
+                SubtitleSegment(text="这是测试", start=2.0, end=4.0),
+            ],
+            language="zh",
+            total_duration=4.0,
+        )
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            output_path = Path(tmpdir) / "test.ass"
+            result = generate_ass_from_timeline(
+                output_path,
+                timeline,
+                video_width=1920,
+                video_height=1080,
+            )
+
+            assert result.exists()
+            content = result.read_text(encoding="utf-8")
+            assert "[Script Info]" in content
+            assert "[V4+ Styles]" in content
+            assert "[Events]" in content
+            assert "你好世界" in content
+            assert "这是测试" in content
+            assert "PlayResX: 1920" in content
+            assert "PlayResY: 1080" in content
+
+    def test_empty_timeline(self):
+        timeline = SubtitleTimeline(segments=[])
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            output_path = Path(tmpdir) / "empty.ass"
+            result = generate_ass_from_timeline(output_path, timeline, video_width=1920, video_height=1080)
+            assert result.exists()
+            assert result.read_text(encoding="utf-8") == ""
+
+    def test_with_custom_style(self):
+        timeline = SubtitleTimeline(
+            segments=[SubtitleSegment(text="测试", start=0, end=1)],
+            total_duration=1.0,
+        )
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            output_path = Path(tmpdir) / "style.ass"
+            generate_ass_from_timeline(
+                output_path,
+                timeline,
+                video_width=1280,
+                video_height=720,
+                subtitle_config={
+                    "font": "微软雅黑",
+                    "size": 32,
+                    "color": "#ff0000",
+                    "position": "bottom",
+                },
+            )
+
+            content = output_path.read_text(encoding="utf-8")
+            assert "微软雅黑" in content
+            assert "32" in content
+
+    def test_time_format(self):
+        timeline = SubtitleTimeline(
+            segments=[
+                SubtitleSegment(text="测试", start=0.5, end=1.25),
+                SubtitleSegment(text="长字幕", start=3661.0, end=3662.5),  # 超过1小时
+            ],
+            total_duration=3662.5,
+        )
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            output_path = Path(tmpdir) / "time.ass"
+            generate_ass_from_timeline(output_path, timeline, video_width=1920, video_height=1080)
+
+            content = output_path.read_text(encoding="utf-8")
+            # 0:00:00.50 格式
+            assert "0:00:00.50" in content
+            assert "0:00:01.25" in content
+            # 1:01:01.00 格式（3661秒 = 1小时1分1秒）
+            assert "1:01:01.00" in content

tests/unit/test_subtitle_timeline.py

@@ -0,0 +1,183 @@
+"""字幕时间轴单元测试。"""
+
+import pytest
+
+from packages.domain.subtitle import (
+    SubtitleSegment,
+    SubtitleTimeline,
+    SubtitleWord,
+)
+
+
+class TestSubtitleWord:
+    def test_duration(self):
+        word = SubtitleWord(text="你", start=1.0, end=1.5)
+        assert word.duration == pytest.approx(0.5)
+
+    def test_zero_duration(self):
+        word = SubtitleWord(text="", start=1.0, end=1.0)
+        assert word.duration == 0.0
+
+
+class TestSubtitleSegment:
+    def test_duration(self):
+        seg = SubtitleSegment(text="你好世界", start=0.0, end=2.0)
+        assert seg.duration == pytest.approx(2.0)
+
+    def test_char_count(self):
+        seg = SubtitleSegment(text="你好世界", start=0.0, end=2.0)
+        assert seg.char_count == 4
+
+
+class TestSubtitleTimeline:
+    def test_segment_count(self):
+        timeline = SubtitleTimeline(
+            segments=[
+                SubtitleSegment(text="第一段", start=0, end=1),
+                SubtitleSegment(text="第二段", start=1, end=2),
+            ]
+        )
+        assert timeline.segment_count == 2
+
+    def test_total_chars(self):
+        timeline = SubtitleTimeline(
+            segments=[
+                SubtitleSegment(text="你好", start=0, end=1),
+                SubtitleSegment(text="世界", start=1, end=2),
+            ]
+        )
+        assert timeline.total_chars == 4
+
+
+class TestMergeShortSegments:
+    def test_no_merge_when_long_enough(self):
+        timeline = SubtitleTimeline(
+            segments=[
+                SubtitleSegment(text="这是第一段测试文字", start=0, end=2),
+                SubtitleSegment(text="这是第二段测试文字", start=2, end=4),
+            ],
+            total_duration=4.0,
+        )
+        result = timeline.merge_short_segments(min_chars=8)
+        assert result.segment_count == 2
+
+    def test_merge_short_segments(self):
+        timeline = SubtitleTimeline(
+            segments=[
+                SubtitleSegment(text="你好", start=0, end=0.5),
+                SubtitleSegment(text="世界", start=0.5, end=1.0),
+                SubtitleSegment(text="这是一段长文字", start=1.0, end=3.0),
+            ],
+            total_duration=3.0,
+        )
+        result = timeline.merge_short_segments(min_chars=4)
+        # 前两段合并（共4字），第三段保留
+        assert result.segment_count == 2
+        assert result.segments[0].text == "你好世界"
+        assert result.segments[0].start == 0
+        assert result.segments[0].end == 1.0
+
+    def test_merge_remaining_to_last(self):
+        timeline = SubtitleTimeline(
+            segments=[
+                SubtitleSegment(text="这是第一段测试文字", start=0, end=2),
+                SubtitleSegment(text="你", start=2, end=2.2),
+                SubtitleSegment(text="好", start=2.2, end=2.4),
+            ],
+            total_duration=2.4,
+        )
+        result = timeline.merge_short_segments(min_chars=8)
+        # 最后两段字数不够，合并到上一段
+        assert result.segment_count == 1
+        assert result.segments[0].text == "这是第一段测试文字你好"
+
+    def test_single_segment_no_change(self):
+        timeline = SubtitleTimeline(
+            segments=[SubtitleSegment(text="你好", start=0, end=1)],
+            total_duration=1.0,
+        )
+        result = timeline.merge_short_segments(min_chars=8)
+        assert result.segment_count == 1
+        assert result.segments[0].text == "你好"
+
+    def test_empty_timeline(self):
+        timeline = SubtitleTimeline(segments=[])
+        result = timeline.merge_short_segments()
+        assert result.segment_count == 0
+
+
+class TestSplitLongSegments:
+    def test_no_split_when_short_enough(self):
+        timeline = SubtitleTimeline(
+            segments=[SubtitleSegment(text="你好世界", start=0, end=1)],
+            total_duration=1.0,
+        )
+        result = timeline.split_long_segments(max_chars=20)
+        assert result.segment_count == 1
+
+    def test_split_by_punctuation(self):
+        timeline = SubtitleTimeline(
+            segments=[
+                SubtitleSegment(
+                    text="这是第一段很长的测试文字。这是第二段很长的测试文字！这是第三段很长的测试文字？",
+                    start=0,
+                    end=6.0,
+                )
+            ],
+            total_duration=6.0,
+        )
+        result = timeline.split_long_segments(max_chars=15)
+        # 按标点拆成3段
+        assert result.segment_count == 3
+        assert "。" in result.segments[0].text
+        assert "！" in result.segments[1].text
+        assert "？" in result.segments[2].text
+
+    def test_hard_split_when_no_punctuation(self):
+        timeline = SubtitleTimeline(
+            segments=[
+                SubtitleSegment(
+                    text="一二三四五六七八九十一二三四五六七八九十一二三四五六七八九十",
+                    start=0,
+                    end=6.0,
+                )
+            ],
+            total_duration=6.0,
+        )
+        result = timeline.split_long_segments(max_chars=10)
+        assert result.segment_count == 3
+        assert len(result.segments[0].text) == 10
+
+    def test_time_proportional_split(self):
+        timeline = SubtitleTimeline(
+            segments=[
+                SubtitleSegment(
+                    text="你好世界，这是一段测试文字。用来验证时间比例是否正确。",
+                    start=0,
+                    end=10.0,
+                )
+            ],
+            total_duration=10.0,
+        )
+        result = timeline.split_long_segments(max_chars=10)
+        # 所有片段时间加起来应该等于总时长
+        total_time = sum(s.duration for s in result.segments)
+        assert total_time == pytest.approx(10.0, abs=0.1)
+
+
+class TestTextSplitByPunctuation:
+    def test_basic_split(self):
+        parts = SubtitleTimeline._split_text_by_punctuation("你好世界！这是测试。", max_chars=10)
+        assert len(parts) == 2
+        assert parts[0] == "你好世界！"
+        assert parts[1] == "这是测试。"
+
+    def test_no_punctuation_hard_split(self):
+        parts = SubtitleTimeline._split_text_by_punctuation("一二三四五六七八九十一二三四五六七八九十", max_chars=10)
+        assert len(parts) == 2
+        assert len(parts[0]) == 10
+
+    def test_short_text_no_split(self):
+        parts = SubtitleTimeline._split_text_by_punctuation("你好世界", max_chars=10)
+        assert len(parts) == 1
+        assert parts[0] == "你好世界"