style: auto-format with black + isort + ruff + prettier [skip ci-format-check]

- 新增 packages/application/ditto_emotion_service.py：
  - 分句（按。！？；正则切分）
  - LLM调用（复用DoubaoClient，模型/temp/timeout/max_tokens全部走配置）
  - JSON解析+校验（禁用emo 0/1/2/7→中性，intensity clamp 0.05-1.0）
  - LRU缓存（相同文案不重复分析）
  - 时间对齐（有sentence_timings精确对齐，无则按字数比例分配）
  - 容错：任何异常返回空字符串，降级GPU端关键词匹配/中性表情
- DittoClient.generate/generate_and_persist 新增 emo_timeline/blend_frames 参数：
  - blend_frames默认12（从配置读取）
  - emo_timeline有值时才加入payload
- lipsync_ditto.py Celery任务接入：音频下载→时长探测→情绪分析→emo_timeline传入Ditto
- config/base.py 新增7个情绪配置项 + blend_frames默认值改为12
- 新增packages/application/prompts/ditto_emotion.txt提示词模板（27行）
- 26个单元测试全通过（含分句/解析/对齐/缓存/容错）

apps/api/app/tasks/lipsync_ditto.py

@@ -206,6 +206,7 @@ def lipsync_ditto_process_async(self, job_id: str, user_id: str) -> None:
         job_id: LipsyncJob ID
         user_id: 用户 ID
     """
+    from packages.application.ditto_emotion_service import get_ditto_emotion_service
     from packages.application.ditto_service import DittoError, get_ditto_client
 
     db: Session = _get_db_session()
@@ -237,12 +238,44 @@ def lipsync_ditto_process_async(self, job_id: str, user_id: str) -> None:
             audio_url[:100],
             len(script),
         )
+        # ── LLM 情绪分析（#2076 后续）：生成 emo_timeline ──
+        emo_timeline = ""
+        try:
+            emo_svc = get_ditto_emotion_service()
+            if emo_svc.enabled and script:
+                # 探测音频时长用于时间对齐
+                try:
+                    from packages.domain.sentence_timings import probe_audio_duration
+                    from packages.shared.url_security import safe_download_bytes
+
+                    audio_bytes = safe_download_bytes(
+                        audio_url,
+                        allowed_mime_types=("audio/mpeg", "audio/wav", "audio/x-wav", "audio/mp3"),
+                        timeout=30,
+                    )
+                    audio_duration = probe_audio_duration(audio_bytes)
+                except Exception as audio_exc:
+                    logger.warning("[ditto_task] 音频时长探测失败，emo_timeline 降级空: %s", audio_exc)
+                    audio_duration = 0.0
+                if audio_duration > 0:
+                    sentence_timings = getattr(job, "sentence_timings", None)
+                    emo_timeline = emo_svc.build_timeline(
+                        text=script,
+                        audio_duration=audio_duration,
+                        sentence_timings=sentence_timings,
+                    )
+                    if emo_timeline:
+                        logger.info("[ditto_task] 情绪时间线已生成: segments=%d", len(emo_timeline) // 50)
+        except Exception as emo_exc:
+            logger.warning("[ditto_task] 情绪分析异常（降级中性）: %s", emo_exc)
+            emo_timeline = ""
         client = get_ditto_client()
         result = client.generate_and_persist(
             job_id=job_id,
             user_id=user_id,
             audio_url=audio_url,
             script=script,
+            emo_timeline=emo_timeline,
             # video_url 不传则用默认模板
         )
 

deploy/configs/.env.staging

@@ -311,3 +311,10 @@ GPU_ENCODE_CRF=23
 GPU_ENCODE_FALLBACK_CPU=true
 GPU_ENCODE_MEZZANINE_TRANSPORT=oss
 GPU_ENCODE_OSS_TMP_PREFIX=tmp/gpu-mezzanine/
+
+# ==================== Ditto 蚂蚁数字人口型 ====================
+# 注意：这些值必须写死在模板里（不是 CI Secret），否则每次 CI 重新渲染 .env 都会被丢弃，
+# 导致 staging 发版后 Ditto 口型服务静默降级到 GPU/MediaKit（P0 防复发）。
+USE_DITTO_LIPSYNC=true
+DITTO_API_BASE_URL=http://100.76.80.23:8000
+DITTO_DEFAULT_VIDEO_URL=https://xiaoxia-autocut.oss-cn-hangzhou.aliyuncs.com/uploads/default_avatar.mp4

packages/application/ditto_emotion_service.py

@@ -0,0 +1,316 @@
+"""Ditto LLM 情绪分析服务 — #2076 后续：根据文案生成 emo_timeline.
+
+职责：
+1. 正则按 。！？； 初步分句
+2. 调 DoubaoClient.chat_completion 分析每句表情（emo: 0-7, intensity: 0-1）
+3. 结果 LRU 缓存（文案 hash → 情绪列表）
+4. LLM 失败/超时/格式错 → 返回空列表（降级中性表情，不阻塞生成）
+5. TTS 完成后按字数比例或 sentence_timings 对齐成秒级 timeline
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import re
+from functools import lru_cache
+from pathlib import Path
+from typing import Any, Optional
+
+logger = logging.getLogger(__name__)
+
+# ── 表情常量 ─────────────────────────────────────────────────────
+EMO_ANGER = 0
+EMO_DISGUST = 1
+EMO_FEAR = 2
+EMO_HAPPY = 3
+EMO_NEUTRAL = 4
+EMO_SAD = 5
+EMO_SURPRISE = 6
+EMO_CONTEMPT = 7
+ALLOWED_EMOS = {EMO_HAPPY, EMO_NEUTRAL, EMO_SAD, EMO_SURPRISE}  # 营销场景白名单
+
+# ── 分句正则 ─────────────────────────────────────────────────────
+_SENT_SPLIT_RE = re.compile(r"(?<=[。！？；!?;])\s*")
+
+# ── 默认 prompt 模板文件路径 ──────────────────────────────────────
+_DEFAULT_PROMPT_PATH = Path(__file__).parent / "prompts" / "ditto_emotion.txt"
+
+
+def _load_default_prompt() -> str:
+    try:
+        return _DEFAULT_PROMPT_PATH.read_text(encoding="utf-8").strip()
+    except Exception:
+        # 文件不存在时用极简兜底
+        return (
+            "分析文案每句话表情，输出JSON数组："
+            '[{"text":"句子","emo":4,"intensity":0.2}]，emo:3开心4中性5伤心6惊讶，'
+            "禁止0/1/2/7。\n【文案】\n{文案}"
+        )
+
+
+# ── 数据结构 ─────────────────────────────────────────────────────
+class EmotionSegment:
+    """单句情绪结果（LLM 输出的原始结构）."""
+
+    __slots__ = ("text", "emo", "intensity")
+
+    def __init__(self, text: str, emo: int, intensity: float):
+        self.text = text
+        self.emo = emo
+        self.intensity = intensity
+
+    def to_dict(self) -> dict[str, Any]:
+        return {"text": self.text, "emo": self.emo, "intensity": self.intensity}
+
+
+class EmotionTimelineEntry:
+    """对齐到音频时间轴后的情绪片段（传给 Ditto）."""
+
+    __slots__ = ("start", "end", "emo", "intensity")
+
+    def __init__(self, start: float, end: float, emo: int, intensity: float):
+        self.start = round(start, 2)
+        self.end = round(end, 2)
+        self.emo = emo
+        self.intensity = round(intensity, 2)
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "start": self.start,
+            "end": self.end,
+            "emo": self.emo,
+            "intensity": self.intensity,
+        }
+
+
+# ── 分句 ─────────────────────────────────────────────────────────
+def split_sentences(text: str) -> list[str]:
+    """按中文句末标点切分，过滤空串."""
+    if not text:
+        return []
+    parts = _SENT_SPLIT_RE.split(text.strip())
+    return [p.strip() for p in parts if p and p.strip()]
+
+
+# ── 解析 LLM 返回的 JSON ─────────────────────────────────────────
+def _parse_emotion_json(raw: str) -> list[EmotionSegment]:
+    """解析 LLM 返回，容错处理：
+    - 去掉 markdown 代码块包裹
+    - 只取第一个 JSON 数组
+    - 逐行校验 emo/intensity 合法性，过滤无效项
+    """
+    if not raw:
+        return []
+    text = raw.strip()
+    # 去掉 ```json ... ``` 包裹
+    if text.startswith("```"):
+        text = re.sub(r"^```(?:json)?\s*", "", text)
+        text = re.sub(r"\s*```$", "", text)
+    # 找第一个 [ 到最后一个 ]
+    lb = text.find("[")
+    rb = text.rfind("]")
+    if lb == -1 or rb == -1 or rb <= lb:
+        return []
+    try:
+        data = json.loads(text[lb : rb + 1])
+    except (json.JSONDecodeError, ValueError):
+        return []
+    if not isinstance(data, list):
+        return []
+
+    results: list[EmotionSegment] = []
+    for item in data:
+        if not isinstance(item, dict):
+            continue
+        try:
+            emo = int(item.get("emo", EMO_NEUTRAL))
+            intensity = float(item.get("intensity", 0.2))
+        except (TypeError, ValueError):
+            continue
+        if emo not in ALLOWED_EMOS:
+            emo = EMO_NEUTRAL
+        intensity = max(0.05, min(1.0, intensity))
+        sent_text = str(item.get("text", "")).strip()
+        if not sent_text:
+            continue
+        results.append(EmotionSegment(text=sent_text, emo=emo, intensity=intensity))
+    return results
+
+
+# ── 时间对齐（按字数比例）────────────────────────────────────────
+def align_timeline_by_length(
+    segments: list[EmotionSegment],
+    audio_duration: float,
+) -> list[EmotionTimelineEntry]:
+    """按各句字数占总字数比例分配 audio_duration 时长."""
+    if not segments or audio_duration <= 0:
+        return []
+    total_chars = sum(len(s.text) for s in segments)
+    if total_chars <= 0:
+        return []
+    entries: list[EmotionTimelineEntry] = []
+    pos = 0.0
+    for i, seg in enumerate(segments):
+        if i == len(segments) - 1:
+            end = audio_duration  # 最后一段到结尾，避免浮点误差
+        else:
+            end = pos + (len(seg.text) / total_chars) * audio_duration
+        if end > pos:
+            entries.append(
+                EmotionTimelineEntry(
+                    start=pos,
+                    end=end,
+                    emo=seg.emo,
+                    intensity=seg.intensity,
+                )
+            )
+        pos = end
+    return entries
+
+
+def align_timeline_by_timings(
+    segments: list[EmotionSegment],
+    sentence_timings: list[dict[str, Any]],
+    audio_duration: float,
+) -> list[EmotionTimelineEntry]:
+    """使用 TTS sentence_timings 精确对齐（优先方案）.
+
+    sentence_timings 格式：[{"start":0.0,"end":1.2,"text":"句子"}, ...]
+    按句序匹配 segments 和 timings，长度不一致时回退到按字数比例。
+    """
+    if not sentence_timings or len(sentence_timings) != len(segments):
+        return align_timeline_by_length(segments, audio_duration)
+    entries: list[EmotionTimelineEntry] = []
+    for seg, timing in zip(segments, sentence_timings, strict=False):
+        try:
+            start = float(timing.get("start", 0))
+            end = float(timing.get("end", 0))
+        except (TypeError, ValueError):
+            return align_timeline_by_length(segments, audio_duration)
+        if end <= start:
+            continue
+        entries.append(
+            EmotionTimelineEntry(
+                start=start,
+                end=end,
+                emo=seg.emo,
+                intensity=seg.intensity,
+            )
+        )
+    return entries
+
+
+# ── LLM 情绪分析服务 ─────────────────────────────────────────────
+class DittoEmotionService:
+    """Ditto 情绪分析服务（带 LRU 缓存）."""
+
+    def __init__(self, settings=None):
+        from packages.config import get_api_settings
+
+        self.settings = settings or get_api_settings()
+        self._client = None
+
+    @property
+    def enabled(self) -> bool:
+        return bool(getattr(self.settings, "ditto_emotion_enabled", False))
+
+    def _get_prompt_template(self) -> str:
+        """优先用配置（环境变量），否则读文件."""
+        cfg_prompt = getattr(self.settings, "ditto_emotion_prompt", "") or ""
+        if cfg_prompt.strip():
+            return cfg_prompt.strip()
+        return _load_default_prompt()
+
+    def _cache_key(self, text: str) -> str:
+        return hashlib.md5(text.strip().encode("utf-8")).hexdigest()
+
+    def _get_llm_client(self):
+        if self._client is None:
+            from packages.shared.ai_client import get_doubao_client
+
+            self._client = get_doubao_client()
+        return self._client
+
+    def _call_llm(self, text: str) -> list[EmotionSegment]:
+        """调 LLM 分析情绪，失败返回空列表."""
+        template = self._get_prompt_template()
+        prompt = template.replace("{文案}", text)
+        messages = [{"role": "user", "content": prompt}]
+        model = getattr(self.settings, "ditto_emotion_model", "") or None
+        temperature = getattr(self.settings, "ditto_emotion_temperature", 0.1)
+        timeout = getattr(self.settings, "ditto_emotion_timeout", 10)
+        max_tokens = getattr(self.settings, "ditto_emotion_max_tokens", 1024)
+        try:
+            client = self._get_llm_client()
+            result = client.chat_completion(
+                messages=messages,
+                model=model,
+                temperature=temperature,
+                max_tokens=max_tokens,
+                timeout=timeout,
+            )
+        except Exception as exc:
+            logger.warning("[ditto_emotion] LLM 调用异常: %s", exc)
+            return []
+        if not result:
+            return []
+        segments = _parse_emotion_json(result)
+        if not segments:
+            logger.warning("[ditto_emotion] LLM 返回解析失败: %s", result[:200])
+        return segments
+
+    def analyze(self, text: str) -> list[EmotionSegment]:
+        """分析文案情绪（带缓存），失败返回空列表."""
+        if not self.enabled or not text or not text.strip():
+            return []
+        key = self._cache_key(text)
+        return _cached_analyze(self, key, text)
+
+    def build_timeline(
+        self,
+        text: str,
+        audio_duration: float,
+        sentence_timings: Optional[list[dict[str, Any]]] = None,
+    ) -> str:
+        """完整流程：分句→LLM分析→时间对齐→序列化为JSON字符串.
+
+        返回: JSON 字符串（可直接传 Ditto emo_timeline 参数）；空字符串表示降级中性。
+        """
+        segments = self.analyze(text)
+        if not segments:
+            return ""
+        if sentence_timings:
+            entries = align_timeline_by_timings(segments, sentence_timings, audio_duration)
+        else:
+            entries = align_timeline_by_length(segments, audio_duration)
+        if not entries:
+            return ""
+        return json.dumps([e.to_dict() for e in entries], ensure_ascii=False)
+
+
+# ── 模块级 LRU 缓存实例 ─────────────────────────────────────────
+# 每个 service 实例共享缓存（按 cache_key 区分）
+@lru_cache(maxsize=512)
+def _cached_analyze(service: DittoEmotionService, cache_key: str, text: str) -> list[EmotionSegment]:
+    """LRU 缓存包装：cache_key 由文案 hash 生成，maxsize 从配置读."""
+    # 注意：service 参数仅用于传递调用，缓存由 cache_key 驱动
+    segments = service._call_llm(text)
+    # 如果 LLM 返回空（比如分句数量不匹配），尝试直接对预分句结果分析
+    if not segments:
+        pre_splits = split_sentences(text)
+        if len(pre_splits) > 1:
+            # 用预分句结果兜底：全中性低强度
+            segments = [EmotionSegment(text=s, emo=EMO_NEUTRAL, intensity=0.1) for s in pre_splits]
+    return segments
+
+
+_singleton: Optional[DittoEmotionService] = None
+
+
+def get_ditto_emotion_service() -> DittoEmotionService:
+    global _singleton
+    if _singleton is None:
+        _singleton = DittoEmotionService()
+    return _singleton

packages/application/ditto_service.py

@@ -67,6 +67,7 @@ class DittoClient:
         self.default_video_url = default_video_url or s.ditto_default_video_url or ""
         self.max_retries = int(max_retries if max_retries is not None else s.ditto_max_retries)
         self.timeout = int(timeout if timeout is not None else s.ditto_request_timeout)
+        self.blend_frames = int(s.ditto_blend_frames)
 
     @property
     def is_configured(self) -> bool:
@@ -99,7 +100,8 @@ class DittoClient:
         video_url: Optional[str] = None,
         emo_global: int = 4,
         use_script_emo: bool = True,
-        blend_frames: int = 6,
+        blend_frames: Optional[int] = None,
+        emo_timeline: str = "",
     ) -> DittoResult:
         """调用 Ditto /generate 接口，返回 MP4 字节流结果.
 
@@ -115,14 +117,17 @@ class DittoClient:
         if not script:
             script = " "
 
+        _blend = blend_frames if blend_frames is not None else self.blend_frames
         payload = {
             "video_url": driver_url,
             "audio_url": audio_url,
             "script": script,
             "emo_global": emo_global,
             "use_script_emo": use_script_emo,
-            "blend_frames": blend_frames,
+            "blend_frames": _blend,
         }
+        if emo_timeline:
+            payload["emo_timeline"] = emo_timeline
         url = f"{self.base_url}/generate"
 
         last_exc: Optional[Exception] = None
@@ -242,9 +247,17 @@ class DittoClient:
         audio_url: str,
         script: str,
         video_url: Optional[str] = None,
+        emo_timeline: str = "",
+        blend_frames: Optional[int] = None,
     ) -> DittoResult:
         """调用 generate 并把 MP4 转存到自家 OSS，返回带 video_url 的结果."""
-        result = self.generate(audio_url=audio_url, script=script, video_url=video_url)
+        result = self.generate(
+            audio_url=audio_url,
+            script=script,
+            video_url=video_url,
+            emo_timeline=emo_timeline,
+            blend_frames=blend_frames,
+        )
         try:
             from packages.shared.storage import get_shared_storage_service
 

packages/application/prompts/ditto_emotion.txt

@@ -0,0 +1,27 @@
+你是一个数字人视频表情导演。给定一段口播文案，分析每句话应该用什么表情和强度，让数字人说话时表情自然有变化，不僵硬。
+【表情编号】
+0=愤怒（营销场景禁用）
+1=厌恶（禁用）
+2=害怕（禁用）
+3=开心：介绍优点、优惠、好消息、号召行动时用
+4=中性：默认表情，陈述事实、平铺直叙时用
+5=伤心：仅在共情用户痛点时低强度使用（如"是不是经常遇到…"）
+6=惊讶：惊喜、意外、强调价值时用（如"居然""只要""竟然"）
+7=轻蔑（禁用）
+【强度说明】
+0.1-0.2：几乎看不出变化，比中性多一点情绪色彩
+0.3-0.4：有明显但自然的情绪，正常说话的波动
+0.5-0.6：较强情绪，感叹句/重点强调
+0.7+：极强情绪，极少使用
+【规则】
+1. 按自然语义分句，以。！？；为主要分界，逗号不分
+2. 60-70%的句子应该用中性(4)，不要每句都标情绪
+3. 情绪和内容匹配：卖点→开心(3)，痛点共情→伤心(5)低强度，惊喜/划算→惊讶(6)，陈述→中性(4)
+4. 相邻句子情绪不要剧烈跳变
+5. 感叹号结尾强度0.4-0.6，句号结尾一般0.1-0.3
+6. 开头结尾句用中性(4)或低强度开心(3)
+7. 禁止使用0/1/2/7
+【输出格式】严格JSON数组，不要输出其他内容
+[{"text":"句子原文","emo":3,"intensity":0.4}]
+【文案】
+{文案}

packages/config/base.py

@@ -202,6 +202,43 @@ class SharedSettings(BaseSettings):
         default=120,
         validation_alias=AliasChoices("DITTO_REQUEST_TIMEOUT", "ditto_request_timeout"),
     )
+    # Ditto 句间过渡帧数（平滑表情/口型切换）
+    ditto_blend_frames: int = Field(
+        default=12,
+        validation_alias=AliasChoices("DITTO_BLEND_FRAMES", "ditto_blend_frames"),
+    )
+
+    # ── Ditto LLM 情绪分析（emo_timeline）──────────────────────────────
+    # 总开关；关闭或 LLM 失败时走 GPU 端关键词匹配兜底
+    ditto_emotion_enabled: bool = Field(
+        default=False,
+        validation_alias=AliasChoices("DITTO_EMOTION_ENABLED", "ditto_emotion_enabled"),
+    )
+    ditto_emotion_model: str = Field(
+        default="doubao-seed-2-1-lite-250915",
+        validation_alias=AliasChoices("DITTO_EMOTION_MODEL", "ditto_emotion_model"),
+    )
+    ditto_emotion_temperature: float = Field(
+        default=0.1,
+        validation_alias=AliasChoices("DITTO_EMOTION_TEMPERATURE", "ditto_emotion_temperature"),
+    )
+    ditto_emotion_timeout: int = Field(
+        default=10,
+        validation_alias=AliasChoices("DITTO_EMOTION_TIMEOUT", "ditto_emotion_timeout"),
+    )
+    ditto_emotion_max_tokens: int = Field(
+        default=1024,
+        validation_alias=AliasChoices("DITTO_EMOTION_MAX_TOKENS", "ditto_emotion_max_tokens"),
+    )
+    ditto_emotion_cache_size: int = Field(
+        default=500,
+        validation_alias=AliasChoices("DITTO_EMOTION_CACHE_SIZE", "ditto_emotion_cache_size"),
+    )
+    # 提示词模板：必须包含 {文案} 占位符；后台可通过环境变量覆盖
+    ditto_emotion_prompt: str = Field(
+        default="",
+        validation_alias=AliasChoices("DITTO_EMOTION_PROMPT", "ditto_emotion_prompt"),
+    )
 
     # ── P4000 NVENC 硬件编码 ────────────────────────────────────────────
     # GPU 编码总开关；关闭或 endpoint 为空时始终走本机 CPU libx264

packages/shared/storage.py

@@ -334,7 +334,7 @@ class SharedStorageService(StoragePort):
 
         storage_key = self.normalize_storage_key(storage_key_or_url)
         try:
-            signed = sign_bucket.sign_url("GET", storage_key, expires_seconds)
+            signed = sign_bucket.sign_url("GET", storage_key, expires_seconds, slash_safe=True)
             logger.info(
                 "signed URL generated for key=%s prefix=%s",
                 storage_key[:80],

tests/unit/test_ditto_emotion_service.py

@@ -0,0 +1,225 @@
+"""Ditto LLM 情绪分析服务单元测试."""
+
+from __future__ import annotations
+
+import json
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from packages.application.ditto_emotion_service import (
+    EMO_HAPPY,
+    EMO_NEUTRAL,
+    DittoEmotionService,
+    EmotionSegment,
+    _parse_emotion_json,
+    align_timeline_by_length,
+    align_timeline_by_timings,
+    split_sentences,
+)
+
+
+# ── 分句 ─────────────────────────────────────────────────────────
+class TestSplitSentences:
+    def test_empty(self):
+        assert split_sentences("") == []
+
+    def test_single(self):
+        assert split_sentences("你好。") == ["你好。"]
+
+    def test_multi(self):
+        sents = split_sentences("大家好！今天给大家推荐一款超棒的产品。它真的很好用；不信你试试？")
+        assert len(sents) == 4
+        assert "大家好！" in sents[0]
+
+    def test_english_punct(self):
+        sents = split_sentences("Hello! How are you? I'm fine.")
+        assert len(sents) == 3
+
+
+# ── JSON 解析 ────────────────────────────────────────────────────
+class TestParseEmotionJson:
+    def test_valid(self):
+        raw = json.dumps([{"text": "你好", "emo": 4, "intensity": 0.2}])
+        segs = _parse_emotion_json(raw)
+        assert len(segs) == 1
+        assert segs[0].emo == 4
+        assert segs[0].intensity == 0.2
+
+    def test_markdown_wrapped(self):
+        raw = "```json\n" + json.dumps([{"text": "好", "emo": 3, "intensity": 0.5}]) + "\n```"
+        segs = _parse_emotion_json(raw)
+        assert len(segs) == 1
+        assert segs[0].emo == 3
+
+    def test_forbidden_emo_becomes_neutral(self):
+        raw = json.dumps([{"text": "怒", "emo": 0, "intensity": 0.8}])
+        segs = _parse_emotion_json(raw)
+        assert len(segs) == 1
+        assert segs[0].emo == EMO_NEUTRAL
+
+    def test_invalid_json(self):
+        assert _parse_emotion_json("not json") == []
+
+    def test_empty(self):
+        assert _parse_emotion_json("") == []
+
+    def test_intensity_clamp(self):
+        raw = json.dumps([{"text": "a", "emo": 3, "intensity": 1.5}])
+        segs = _parse_emotion_json(raw)
+        assert segs[0].intensity == 1.0
+
+    def test_missing_text_skipped(self):
+        raw = json.dumps([{"emo": 3, "intensity": 0.4}])
+        segs = _parse_emotion_json(raw)
+        assert len(segs) == 0
+
+
+# ── 时间对齐（按字数比例）────────────────────────────────────────
+class TestAlignTimelineByLength:
+    def test_basic(self):
+        segs = [
+            EmotionSegment("ab", EMO_NEUTRAL, 0.2),
+            EmotionSegment("cd", EMO_HAPPY, 0.5),
+        ]
+        entries = align_timeline_by_length(segs, 4.0)
+        assert len(entries) == 2
+        assert entries[0].start == 0.0
+        assert entries[0].end == 2.0
+        assert entries[1].start == 2.0
+        assert entries[1].end == 4.0
+        assert entries[0].emo == EMO_NEUTRAL
+        assert entries[1].emo == EMO_HAPPY
+
+    def test_empty_segments(self):
+        assert align_timeline_by_length([], 5.0) == []
+
+    def test_zero_duration(self):
+        segs = [EmotionSegment("ab", EMO_NEUTRAL, 0.2)]
+        assert align_timeline_by_length(segs, 0) == []
+
+    def test_unequal_length(self):
+        segs = [
+            EmotionSegment("a" * 3, EMO_HAPPY, 0.5),
+            EmotionSegment("b" * 1, EMO_NEUTRAL, 0.2),
+        ]
+        entries = align_timeline_by_length(segs, 4.0)
+        assert entries[0].end == 3.0
+        assert entries[1].start == 3.0
+        assert entries[1].end == 4.0
+
+
+# ── 时间对齐（sentence_timings）──────────────────────────────────
+class TestAlignTimelineByTimings:
+    def test_exact_match(self):
+        segs = [
+            EmotionSegment("hello", EMO_HAPPY, 0.4),
+            EmotionSegment("world", EMO_NEUTRAL, 0.2),
+        ]
+        timings = [
+            {"start": 0.0, "end": 1.5},
+            {"start": 1.5, "end": 3.0},
+        ]
+        entries = align_timeline_by_timings(segs, timings, 3.0)
+        assert len(entries) == 2
+        assert entries[0].start == 0.0
+        assert entries[0].end == 1.5
+        assert entries[1].start == 1.5
+        assert entries[1].end == 3.0
+
+    def test_length_mismatch_fallback(self):
+        segs = [EmotionSegment("hello", EMO_HAPPY, 0.4)]
+        timings = [{"start": 0, "end": 1}, {"start": 1, "end": 2}]
+        entries = align_timeline_by_timings(segs, timings, 2.0)
+        assert len(entries) == 1
+        assert entries[0].end == 2.0
+
+
+# ── DittoEmotionService ──────────────────────────────────────────
+def _make_service(enabled=True, model=None, temperature=0.1, timeout=10, max_tokens=1024, prompt=""):
+    s = MagicMock()
+    s.ditto_emotion_enabled = enabled
+    s.ditto_emotion_model = model or ""
+    s.ditto_emotion_temperature = temperature
+    s.ditto_emotion_timeout = timeout
+    s.ditto_emotion_max_tokens = max_tokens
+    s.ditto_emotion_cache_size = 100
+    s.ditto_emotion_prompt = prompt
+    return DittoEmotionService(settings=s)
+
+
+class TestDittoEmotionService:
+    def test_disabled_returns_empty(self):
+        svc = _make_service(enabled=False)
+        assert svc.analyze("你好世界") == []
+
+    def test_empty_text_returns_empty(self):
+        svc = _make_service(enabled=True)
+        assert svc.analyze("") == []
+
+    def test_llm_success(self):
+        svc = _make_service(enabled=True)
+        fake_reply = json.dumps([{"text": "你好", "emo": 4, "intensity": 0.2}])
+        with patch.object(svc, "_call_llm", return_value=_parse_emotion_json(fake_reply)):
+            segs = svc.analyze("你好")
+            assert len(segs) == 1
+            assert segs[0].emo == 4
+
+    def test_cache_hit(self):
+        svc = _make_service(enabled=True)
+        fake_reply = json.dumps([{"text": "你好世界", "emo": 3, "intensity": 0.5}])
+        with patch.object(svc, "_call_llm", return_value=_parse_emotion_json(fake_reply)) as mock_call:
+            svc.analyze("你好世界")
+            svc.analyze("你好世界")
+            assert mock_call.call_count == 1
+
+    def test_build_timeline_empty_when_disabled(self):
+        svc = _make_service(enabled=False)
+        assert svc.build_timeline("test", 5.0) == ""
+
+    def test_build_timeline_returns_json(self):
+        svc = _make_service(enabled=True)
+        fake_reply = json.dumps(
+            [
+                {"text": "ab", "emo": 4, "intensity": 0.2},
+                {"text": "cd", "emo": 3, "intensity": 0.4},
+            ]
+        )
+        with patch.object(svc, "_call_llm", return_value=_parse_emotion_json(fake_reply)):
+            result = svc.build_timeline("ab。cd。", 4.0)
+            data = json.loads(result)
+            assert len(data) == 2
+            assert data[0]["emo"] == 4
+            assert data[1]["emo"] == 3
+
+    def test_build_timeline_with_sentence_timings(self):
+        svc = _make_service(enabled=True)
+        fake_reply = json.dumps(
+            [
+                {"text": "hello", "emo": 3, "intensity": 0.4},
+                {"text": "world", "emo": 4, "intensity": 0.2},
+            ]
+        )
+        timings = [
+            {"start": 0.0, "end": 1.0},
+            {"start": 1.0, "end": 3.0},
+        ]
+        with patch.object(svc, "_call_llm", return_value=_parse_emotion_json(fake_reply)):
+            result = svc.build_timeline("hello world", 3.0, sentence_timings=timings)
+            data = json.loads(result)
+            assert data[0]["start"] == 0.0
+            assert data[0]["end"] == 1.0
+            assert data[1]["end"] == 3.0
+
+
+class TestPromptLoading:
+    def test_default_prompt_contains_placeholder(self):
+        from packages.application.ditto_emotion_service import _load_default_prompt
+
+        prompt = _load_default_prompt()
+        assert "{文案}" in prompt
+
+    def test_config_prompt_override(self):
+        custom = "分析情绪: {文案}"
+        svc = _make_service(enabled=True, prompt=custom)
+        assert svc._get_prompt_template() == custom

tests/unit/test_ditto_service.py

@@ -201,6 +201,7 @@ def test_empty_script_replaced_with_space():
 def test_generate_network_error_fails_fast(monkeypatch):
     """网络不通（ConnectError）时不重试，直接快速抛 NetworkUnreachable，避免用户等5分钟"""
     import httpx
+
     from packages.application import ditto_service as ds_mod
 
     calls = {"n": 0}