Merge pull request #2251: Ditto 使用用户视频+ffmpeg循环补长

- lipsync_ditto.py 重写驱动视频逻辑：
  - 默认使用用户上传的 job.video_url（签名后），default_video_url 仅兜底
  - 新增 _prepare_driver_video()：下载视频→ffprobe时长→
    视频<音频+2s时用 ffmpeg stream_loop -1 循环到目标时长（-c copy优先，失败回退libx264重编码）
    循环视频上传临时 OSS ditto-tmp/ 路径
  - 视频足够长直接使用原URL，不做处理
  - 预处理异常时回退直接使用（交给Ditto侧处理）
- ditto_service.py 更新注释：明确video_url优先用户视频，default仅兜底
- config/base.py 更新注释：ditto_default_video_url 为兜底用途
- 音频下载提前统一（情绪分析+视频预处理共用一次下载）
- 所有异常路径保持三层兜底（Ditto→GPU→MediaKit）

alembic/versions/107_add_language_to_viral_video_jobs.py

@@ -0,0 +1,23 @@
+"""add language column to viral_video_jobs
+
+Revision ID: 107
+Revises: 106
+Create Date: 2026-10-09
+
+"""
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "107"
+down_revision = "106"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute("ALTER TABLE viral_video_jobs " "ADD COLUMN IF NOT EXISTS language VARCHAR(20) NOT NULL DEFAULT 'zh-CN'")
+
+
+def downgrade() -> None:
+    op.execute("ALTER TABLE viral_video_jobs DROP COLUMN IF EXISTS language")

apps/api/app/api/routes/viral_video.py

@@ -142,6 +142,7 @@ def _to_response(job) -> ViralVideoJobResponse:
         copy_result=_build_copy_result(job),
         voice_id=getattr(job, "voice_id", "") or "",
         voice_source=getattr(job, "voice_source", "") or "",
+        language=getattr(job, "language", "zh-CN") or "zh-CN",
         video_ratio=getattr(job, "video_ratio", "9:16") or "9:16",
         video_model=getattr(job, "video_model", "") or "",
         intent_result=job.intent_result,
@@ -243,6 +244,7 @@ def analyze_images(
         style_strength=request.style_strength or "medium",
         voice_id=request.voice_id or "",
         voice_source=request.voice_source or "",
+        language=getattr(request, "language", "zh-CN") or "zh-CN",
         video_ratio=request.video_ratio or "9:16",
         video_model=request.video_model or "",
         video_resolution=getattr(request, "video_resolution", "720p") or "720p",
@@ -314,6 +316,7 @@ def generate_copy(
         job.style_guide = request.style_guide
     job.voice_id = request.voice_id or job.voice_id
     job.voice_source = request.voice_source or job.voice_source
+    job.language = getattr(request, "language", "") or job.language or "zh-CN"
     job.video_ratio = request.video_ratio or job.video_ratio or "9:16"
     job.video_model = request.video_model or job.video_model or ""
     job.video_resolution = getattr(request, "video_resolution", "") or job.video_resolution or "720p"
@@ -366,6 +369,12 @@ def confirm_copy(
         job.video_ratio = request.video_ratio
     if request.video_model is not None:
         job.video_model = request.video_model
+    if request.voice_id is not None:
+        job.voice_id = request.voice_id
+    if request.voice_source is not None:
+        job.voice_source = request.voice_source
+    if getattr(request, "language", None) is not None:
+        job.language = request.language
 
     param_changed = (
         (request.duration is not None and int(request.duration) != old_duration)
@@ -422,7 +431,9 @@ def confirm_copy(
                     )
                 job.credits_prepaid = new_est
                 job.credits_transaction_id = res.get("transaction_id", "") or ""
-            logger.info("[爆款视频][confirm-copy] 参数变更，积分重算: old=%d new=%d job_id=%s", old_est, new_est, job.id)
+            logger.info(
+                "[爆款视频][confirm-copy] 参数变更，积分重算: old=%d new=%d job_id=%s", old_est, new_est, job.id
+            )
         elif not already_paid:
             from packages.domain.points_rules import calculate_viral_video_credits, resolve_video_dimensions
             from packages.domain.points_service import PointsService

apps/api/app/schemas/viral_video.py

@@ -87,6 +87,7 @@ class CreateViralVideoRequest(BaseModel):
     style_template_id: str = ""
     voice_id: str = ""
     voice_source: str = ""
+    language: str = "zh-CN"
     video_ratio: str = "9:16"
     video_model: str = ""
     video_resolution: str = "720p"
@@ -121,6 +122,7 @@ class AnalyzeImagesRequest(BaseModel):
     video_model: str = ""
     video_resolution: str = "720p"
     duration: int = Field(default=15, ge=5, le=30)
+    language: str = "zh-CN"
 
 
 class GenerateCopyRequest(BaseModel):
@@ -142,6 +144,7 @@ class GenerateCopyRequest(BaseModel):
     style_guide: dict | None = None
     voice_id: str = ""
     voice_source: str = ""
+    language: str = "zh-CN"
     video_ratio: str = "9:16"
     video_model: str = ""
     video_resolution: str = "720p"
@@ -171,6 +174,9 @@ class ConfirmCopyRequest(BaseModel):
     video_resolution: str | None = Field(default=None, description="用户选定的分辨率（confirm时可选）")
     video_ratio: str | None = Field(default=None, description="用户选定的比例（confirm时可选）")
     duration: int | None = Field(default=None, ge=5, le=30, description="用户选定的时长秒数（confirm时可选，5~30）")
+    voice_id: str | None = Field(default=None, description="用户选定的音色ID（confirm时可选）")
+    voice_source: str | None = Field(default=None, description="用户选定的音色来源（confirm时可选）")
+    language: str | None = Field(default=None, description="用户选定的语言（confirm时可选）")
 
 
 class ConfirmIntentRequest(BaseModel):
@@ -222,6 +228,7 @@ class ViralVideoJobResponse(BaseModel):
     # 音色/视频参数
     voice_id: str = ""
     voice_source: str = ""
+    language: str = "zh-CN"
     video_ratio: str = "9:16"
     video_model: str = ""
     intent_result: dict | None = None

apps/api/app/tasks/lipsync_ditto.py

@@ -2,19 +2,25 @@
 
 把 Ditto 同步 HTTP 调用（30-120s）从 API 请求移到 Celery 后台执行：
 1. 加载 LipsyncJob
-2. 调 DittoClient.generate_and_persist(video_url=默认模板, audio_url=job.audio_url, script=job.script_text)
-3. 成功：标记 completed，写入 output_video_url（Ditto 输出自带音频，无需二次混流/超分）
-4. 失败：回退 GPU MuseTalk → 再失败回退 MediaKit
+2. 确定驱动视频：用户上传的 video_url 优先，无则用 settings.ditto_default_video_url 兜底
+3. 视频时长对齐：若视频 < 音频+2s，用 ffmpeg 循环视频到足够长度后上传临时文件
+4. 调 DittoClient.generate_and_persist
+5. 成功：标记 completed，写入 output_video_url（Ditto 输出自带音频，无需二次混流/超分）
+6. 失败：回退 GPU MuseTalk → 再失败回退 MediaKit
 
 注意：
 - 保留 MuseTalk 代码不动；Ditto 优先，失败按原链路兜底
-- Ditto 使用预置的人物模板视频（settings.ditto_default_video_url），不用用户上传的 video_url
-- 不传 GFPGAN 超分，不需要 ffmpeg 音视频混流
+- Ditto 默认使用用户上传的视频作为驱动模板，default_video_url 仅作兜底
+- 用户视频短于音频时，ffmpeg stream_loop 循环到音频时长+2s余量
+- 不传 GFPGAN 超分，不需要 ffmpeg 音视频混流（Ditto 输出已带音视频）
 """
 
 from __future__ import annotations
 
 import logging
+import os
+import subprocess
+import tempfile
 from datetime import UTC, datetime
 from typing import TYPE_CHECKING, Optional
 
@@ -27,6 +33,8 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 _DITTO_URL_TTL_SECONDS = 7 * 24 * 3600  # Ditto 结果 OSS URL 7 天有效
+_VIDEO_LOOP_MARGIN_SECONDS = 2.0  # 循环视频时比音频多留 2 秒余量
+_MAX_VIDEO_PREPROCESS_SIZE = 200 * 1024 * 1024  # 用户视频最大 200MB
 
 
 def _get_db_session() -> Session:
@@ -59,37 +67,215 @@ def _sign_media_url(url: str) -> str:
         return url
 
 
-def _probe_video_duration(video_bytes: bytes) -> float:
-    """用 ffprobe 探测视频时长（秒）；失败返回 0。"""
-    try:
-        import os
-        import subprocess
-        import tempfile
+def _probe_media_duration(path_or_bytes, *, is_bytes: bool = False) -> float:
+    """用 ffprobe 探测视频/音频时长（秒）；失败返回 0。
 
-        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmp:
-            tmp.write(video_bytes)
-            tmp_path = tmp.name
-        try:
-            out = subprocess.check_output(
-                [
-                    "ffprobe",
-                    "-v",
-                    "error",
-                    "-show_entries",
-                    "format=duration",
-                    "-of",
-                    "default=noprint_wrappers=1:nokey=1",
-                    tmp_path,
-                ],
-                stderr=subprocess.STDOUT,
-                timeout=10,
-            )
-            return float(out.decode().strip() or 0)
-        finally:
-            os.unlink(tmp_path)
+    Args:
+        path_or_bytes: 文件路径(str) 或 字节数据(bytes)
+        is_bytes: 传入的是 bytes 还是文件路径
+    """
+    tmp_path = None
+    try:
+        if is_bytes:
+            with tempfile.NamedTemporaryFile(suffix=".bin", delete=False) as tmp:
+                tmp.write(path_or_bytes)
+                tmp_path = tmp.name
+            target = tmp_path
+        else:
+            target = path_or_bytes
+        out = subprocess.check_output(
+            [
+                "ffprobe",
+                "-v",
+                "error",
+                "-show_entries",
+                "format=duration",
+                "-of",
+                "default=noprint_wrappers=1:nokey=1",
+                target,
+            ],
+            stderr=subprocess.STDOUT,
+            timeout=15,
+        )
+        return float(out.decode().strip() or 0)
     except Exception as exc:
         logger.warning("[ditto_task] ffprobe 失败: %s", exc)
         return 0.0
+    finally:
+        if tmp_path and os.path.exists(tmp_path):
+            try:
+                os.unlink(tmp_path)
+            except Exception:
+                pass
+
+
+def _probe_video_duration(video_bytes: bytes) -> float:
+    """用 ffprobe 探测视频时长（秒）；失败返回 0。"""
+    return _probe_media_duration(video_bytes, is_bytes=True)
+
+
+def _prepare_driver_video(
+    *,
+    user_video_url: str,
+    default_video_url: str,
+    audio_duration: float,
+    job_id: str,
+    user_id: str,
+) -> tuple[str, bool]:
+    """准备传给 Ditto 的驱动视频 URL。
+
+    逻辑：
+    1. 优先使用用户上传的视频（user_video_url），无则用 default_video_url 兜底
+    2. 下载视频，探测时长
+    3. 若视频时长 >= 音频时长+2s 余量：直接用原 URL（签名后）
+    4. 若视频时长 < 音频时长+2s：ffmpeg stream_loop 循环到目标时长，上传临时 OSS，返回临时 URL
+    5. 任何异常：回退到 default_video_url（兜底）
+
+    Returns:
+        (video_url_for_ditto, is_temporary) — is_temporary=True 表示 URL 是本次临时生成的
+    """
+    from packages.shared.storage import get_shared_storage_service
+    from packages.shared.url_security import safe_download_bytes
+
+    storage = get_shared_storage_service()
+
+    # 1. 选择源 URL（用户视频优先）
+    source_url = user_video_url or default_video_url
+    source_label = "user" if user_video_url else "default"
+    if not source_url:
+        raise RuntimeError("无可用驱动视频（用户视频和默认模板都为空）")
+
+    # 2. 下载视频
+    try:
+        # 用户视频需要签名才能下载
+        signed_source = _sign_media_url(source_url) if user_video_url else source_url
+        video_bytes = safe_download_bytes(
+            signed_source,
+            purpose="ditto-driver-video",
+            max_size=_MAX_VIDEO_PREPROCESS_SIZE,
+            allowed_mime_types=("video/mp4", "video/quicktime", "video/x-msvideo", "video/webm"),
+            timeout=60,
+        )
+    except Exception as dl_exc:
+        logger.warning(
+            "[ditto_task] 下载驱动视频失败（%s），回退默认模板: job=%s err=%s",
+            source_label,
+            job_id,
+            dl_exc,
+        )
+        if user_video_url and default_video_url:
+            return default_video_url, False
+        raise
+
+    # 3. 探测视频时长
+    video_duration = _probe_media_duration(video_bytes, is_bytes=True)
+    target_duration = audio_duration + _VIDEO_LOOP_MARGIN_SECONDS
+
+    # 4. 视频足够长：直接用签名后的原 URL
+    if video_duration >= target_duration:
+        logger.info(
+            "[ditto_task] 驱动视频足够长（%.2fs >= %.2fs），直接使用: job=%s source=%s",
+            video_duration,
+            target_duration,
+            job_id,
+            source_label,
+        )
+        return _sign_media_url(source_url) if user_video_url else source_url, False
+
+    # 5. 视频不够长：ffmpeg 循环到目标时长
+    logger.info(
+        "[ditto_task] 驱动视频不够长（%.2fs < %.2fs），ffmpeg 循环延长: job=%s",
+        video_duration,
+        target_duration,
+        job_id,
+    )
+    looped_path = None
+    try:
+        # 写临时文件
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as src_tmp:
+            src_tmp.write(video_bytes)
+            src_path = src_tmp.name
+        looped_fd, looped_path = tempfile.mkstemp(suffix=".mp4")
+        os.close(looped_fd)
+
+        # ffmpeg: stream_loop -1 循环输入，-t 截到目标时长
+        # 使用 -c copy 快速复制流（不重编码），速度快无画质损失
+        cmd = [
+            "ffmpeg",
+            "-y",
+            "-stream_loop",
+            "-1",
+            "-i",
+            src_path,
+            "-t",
+            str(target_duration),
+            "-c",
+            "copy",
+            "-movflags",
+            "+faststart",
+            looped_path,
+        ]
+        try:
+            subprocess.run(cmd, check=True, capture_output=True, timeout=60)
+        except subprocess.CalledProcessError:
+            # copy 模式失败（编码不兼容），回退重编码
+            logger.warning("[ditto_task] stream_loop copy 失败，回退重编码: job=%s", job_id)
+            cmd = [
+                "ffmpeg",
+                "-y",
+                "-stream_loop",
+                "-1",
+                "-i",
+                src_path,
+                "-t",
+                str(target_duration),
+                "-c:v",
+                "libx264",
+                "-preset",
+                "veryfast",
+                "-crf",
+                "23",
+                "-c:a",
+                "aac",
+                "-movflags",
+                "+faststart",
+                looped_path,
+            ]
+            subprocess.run(cmd, check=True, capture_output=True, timeout=120)
+
+        # 上传临时 OSS
+        looped_key = f"ditto-tmp/{user_id}/{job_id}_looped.mp4"
+        with open(looped_path, "rb") as f:
+            tmp_url = storage.upload_file(
+                f,
+                looped_key,
+                content_type="video/mp4",
+            )
+        # 临时文件需要签名（private bucket）
+        signed_tmp_url = _sign_media_url(tmp_url)
+        logger.info(
+            "[ditto_task] 循环视频已上传: job=%s key=%s dur=%.2fs",
+            job_id,
+            looped_key,
+            target_duration,
+        )
+        return signed_tmp_url, True
+
+    except Exception as loop_exc:
+        logger.warning(
+            "[ditto_task] 视频循环处理失败，回退直接使用（Ditto 侧处理）: job=%s err=%s",
+            job_id,
+            loop_exc,
+        )
+        # 兜底：直接用原视频（交给 Ditto 侧处理时长不一致）
+        return _sign_media_url(source_url) if user_video_url else source_url, False
+    finally:
+        for p in (locals().get("src_path"), looped_path):
+            if p and os.path.exists(p):
+                try:
+                    os.unlink(p)
+                except Exception:
+                    pass
 
 
 def _refund_lip_sync(db: Session, job: "LipsyncJobModel") -> None:
@@ -125,7 +311,6 @@ def _fallback_to_gpu_then_mediakit(db: Session, job: "LipsyncJobModel") -> None:
         gpu_svc = GpuLipsyncService(db)
         if gpu_svc.has_available_worker():
             logger.info("[ditto_task] 回退 GPU MuseTalk: job_id=%s", job.id)
-            # 复用 lipsync_service._submit_to_gpu_create 逻辑
             from app.services.lipsync_service import LipsyncService
 
             svc = LipsyncService(db)
@@ -208,9 +393,13 @@ def lipsync_ditto_process_async(self, job_id: str, user_id: str) -> None:
     """
     from packages.application.ditto_emotion_service import get_ditto_emotion_service
     from packages.application.ditto_service import DittoError, get_ditto_client
+    from packages.config import get_api_settings
+    from packages.domain.sentence_timings import probe_audio_duration
+    from packages.shared.url_security import safe_download_bytes
 
     db: Session = _get_db_session()
     job: Optional[LipsyncJobModel] = None
+    prepared_video_url: str = ""
     try:
         from packages.adapters.sqlalchemy_impl.models import LipsyncJobModel
 
@@ -229,54 +418,70 @@ def lipsync_ditto_process_async(self, job_id: str, user_id: str) -> None:
 
         audio_url = job.audio_url or ""
         script = job.script_text or ""
+        user_video_url = job.video_url or ""
         if not audio_url:
             raise DittoError("job.audio_url 为空，无法调用 Ditto", code="InvalidParam")
 
+        settings = get_api_settings()
+        default_video_url = settings.ditto_default_video_url or ""
+
         logger.info(
-            "[ditto_task] 开始 Ditto 生成: job_id=%s audio=%s script_len=%d",
+            "[ditto_task] 开始 Ditto 生成: job_id=%s has_user_video=%s script_len=%d",
             job_id,
-            audio_url[:100],
+            bool(user_video_url),
             len(script),
         )
-        # ── LLM 情绪分析（#2076 后续）：生成 emo_timeline ──
+
+        # ── 0. 探测音频时长（情绪分析 + 视频循环都需要）──
+        audio_duration = 0.0
+        audio_bytes_for_probe = None
+        try:
+            audio_bytes_for_probe = 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_for_probe)
+            logger.info("[ditto_task] 音频时长: %.2fs", audio_duration)
+        except Exception as audio_exc:
+            logger.warning("[ditto_task] 音频时长探测失败: %s", audio_exc)
+            audio_duration = 0.0
+
+        # ── 1. LLM 情绪分析（生成 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)
+            if emo_svc.enabled and script and 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 = ""
+
+        # ── 2. 准备驱动视频（用户视频优先，必要时循环延长）──
+        prepared_video_url, _is_tmp = _prepare_driver_video(
+            user_video_url=user_video_url,
+            default_video_url=default_video_url,
+            audio_duration=audio_duration if audio_duration > 0 else 10.0,  # 探测失败时按10s估
+            job_id=job_id,
+            user_id=user_id,
+        )
+
+        # ── 3. 调用 Ditto ──
         client = get_ditto_client()
         result = client.generate_and_persist(
             job_id=job_id,
             user_id=user_id,
             audio_url=audio_url,
             script=script,
+            video_url=prepared_video_url,
             emo_timeline=emo_timeline,
-            # video_url 不传则用默认模板
         )
 
         # Ditto 返回的 MP4 自带音频，签名 OSS URL（7天有效）后标记完成
@@ -284,17 +489,14 @@ def lipsync_ditto_process_async(self, job_id: str, user_id: str) -> None:
         # 探测时长（用于计费）
         duration = _probe_video_duration(result.video_bytes)
         if duration <= 0:
-            # 兜底：按音频时长估算（1秒≈1秒）
-            try:
-                from packages.domain.sentence_timings import probe_audio_duration
-                from packages.shared.url_security import safe_download_bytes
-
-                audio_data = safe_download_bytes(
-                    audio_url, allowed_mime_types=("audio/mpeg", "audio/wav", "audio/x-wav"), timeout=30
-                )
-                duration = probe_audio_duration(audio_data)
-            except Exception:
-                duration = 0.0
+            # 兜底：按音频时长估算
+            if audio_bytes_for_probe is not None:
+                try:
+                    duration = probe_audio_duration(audio_bytes_for_probe)
+                except Exception:
+                    duration = 0.0
+            if duration <= 0 and audio_duration > 0:
+                duration = audio_duration
         job.output_duration = duration
         job.status = "completed"
         job.completed_at = datetime.now(UTC)
@@ -316,7 +518,6 @@ def lipsync_ditto_process_async(self, job_id: str, user_id: str) -> None:
             try:
                 db.rollback()
                 job = db.query(type(job)).filter_by(id=job_id).first() if hasattr(job, "id") else job
-                # 回退 GPU/MediaKit
                 _fallback_to_gpu_then_mediakit(db, job)
             except Exception as fallback_exc:
                 logger.exception("[ditto_task] 回退也失败 job_id=%s err=%s", job_id, fallback_exc)

apps/worker/worker_app/tasks/viral_video.py

@@ -484,6 +484,14 @@ _PERSONA_STYLE_GUIDE: dict[str, str] = {
     "店主": "热情实在，像当面招呼客人，突出靠谱和实在优惠",
     "专业顾问": "专业可信，讲清原理和效果，用事实打消顾虑",
     "年轻达人": "活泼有网感，节奏轻快，金句和梗自然不尬",
+    "老板型IP": "以老板第一人称出镜，真诚接地气，像招呼街坊邻居一样分享，突出创业初心和靠谱",
+    "知识博主": "条理清晰、数据说话，干货密度高，语气专业但不枯燥",
+    "生活美学": "画面感强，注重氛围和质感描述，语速偏慢，文字有诗意",
+    "健身教练": "energetic、鼓励式口吻，强调动作要领和效果变化",
+    "美妆达人": "细腻讲质地和妆效，像闺蜜安利，语气亲切有感染力",
+    "美食博主": "色香味描述丰富，口语化带馋感，节奏轻快",
+    "穿搭博主": "讲搭配逻辑和场景适配，时尚但不高冷，像朋友建议",
+    "育儿师": "科学育儿角度，温柔坚定，给具体可操作的建议",
 }
 
 
@@ -498,6 +506,47 @@ def _persona_style_hint(persona_id: str) -> str:
     return "【人设风格：未指定】亲切自然、像朋友分享好物"
 
 
+_VIRAL_STRUCTURE_GUIDE: dict[str, str] = {
+    "反差破局+亮明观点+还原现状": "开头3秒用反差/痛点钩子抓注意力，中段亮出核心卖点或观点，结尾还原真实到店/使用场景引导行动",
+    "痛点切入+方案展示+效果对比": "开头直击用户痛点场景，中间展示产品/服务解决方案，结尾用前后对比强化效果",
+    "故事引入+产品种草+行动引导": "用一个真实小故事/案例引入，自然过渡到产品种草，结尾明确引导用户下一步行动",
+    "场景展示+价值输出+信任背书": "开头展示使用场景让用户代入，中间输出核心价值主张，结尾用客户评价/数据等信任背书收尾",
+    "悬念开场+层层递进+高潮转化": "开头制造悬念引发好奇，内容层层推进保持张力，高潮处给出转化钩子",
+}
+
+
+def _viral_structure_hint(structure: str) -> str:
+    """根据 viral_structure 映射具体写作指导；未命中返回通用提示。"""
+    s = (structure or "").strip()
+    if s in _VIRAL_STRUCTURE_GUIDE:
+        return f"【爆款结构：{s}】{_VIRAL_STRUCTURE_GUIDE[s]}"
+    if s:
+        return f"【爆款结构：{s}】按该结构编排内容节奏和叙事逻辑"
+    return "【爆款结构：未指定】自由组织，保证开头有钩子、中段有卖点、结尾有行动引导"
+
+
+def _language_hint(language: str) -> str:
+    """根据 language 代码返回语言提示。"""
+    lang = (language or "zh-CN").strip().lower()
+    mapping = {
+        "zh-cn": "使用标准普通话，口语化表达",
+        "zh-tw": "使用台湾腔中文，语气温柔亲切",
+        "zh-hk": "使用粤语风格中文表达",
+        "en-us": "使用美式英语，自然口语化",
+        "en-gb": "使用英式英语",
+        "ja-jp": "使用日语，自然口语化",
+        "ko-kr": "使用韩语，亲切自然",
+    }
+    hint = mapping.get(lang, "")
+    if hint:
+        return f"【语言：{lang}】{hint}"
+    if lang.startswith("zh"):
+        return f"【语言：{lang}】使用中文，口语化表达，可带方言特色"
+    if lang.startswith("en"):
+        return f"【语言：{lang}】使用英语，自然口语化"
+    return f"【语言：{lang}】按该语言习惯组织口播内容"
+
+
 def _determine_theme(image_analysis: dict | None, marketing_purpose: str = "") -> str:
     """根据图片类型分布和营销目的推断默认主题。"""
     images = _images_of(image_analysis)
@@ -774,15 +823,24 @@ def _step_script_generation(job: ViralVideoJob, image_analysis: dict) -> dict:
             + "</reference_video_style>"
         )
 
+    persona_hint = _persona_style_hint(getattr(job, "persona_id", ""))
+    viral_structure_hint = _viral_structure_hint(getattr(job, "viral_structure", ""))
+    language_hint = _language_hint(getattr(job, "language", "zh-CN"))
+    industry = getattr(job, "industry", "") or "通用"
+
     user = render_user_prompt(
         template,
         marketing_purpose=marketing_purpose,
+        industry=industry,
         image_summary=images_summary,
         theme_hint=theme_hint,
-        dur=str(dur),
+        duration=str(dur),
         aspect_ratio=getattr(job, "video_ratio", None) or "9:16",
         tone=getattr(job, "tone", "") or "亲切自然",
-        target_audience=getattr(job, "target_audience", "") or "未指定",
+        target_audience=getattr(job, "target_customer", "") or "未指定",
+        persona_hint=persona_hint,
+        viral_structure_hint=viral_structure_hint,
+        language_hint=language_hint,
         extra_requirements=job.user_copy_text or "（未提供额外要求，由 AI 创作）",
         video_style_section=style_section,
     )
@@ -1078,17 +1136,22 @@ def _step_tts(job: ViralVideoJob, voiceover_script: str):
         if not text:
             logger.warning("[爆款视频] voiceover_script 为空，跳过 TTS")
             return None
+        language = getattr(job, "language", "zh-CN") or "zh-CN"
         try:
             result = tts_service.synthesize(
                 text=text,
                 voice_id=voice_id,
                 format="mp3",
+                language=language,
             )
         except TypeError:
             try:
-                result = tts_service.synthesize(text=text, voice_id=voice_id)
+                result = tts_service.synthesize(text=text, voice_id=voice_id, language=language)
             except TypeError:
-                result = tts_service.synthesize(text=text)
+                try:
+                    result = tts_service.synthesize(text=text, voice_id=voice_id)
+                except TypeError:
+                    result = tts_service.synthesize(text=text)
         if result is None:
             return None
         p = _Path(result) if not isinstance(result, _Path) else result

packages/adapters/sqlalchemy_impl/models.py

@@ -961,6 +961,7 @@ class ViralVideoJobModel(Base):
     # v1.5 音频/视频参数
     voice_id = Column(String(200), nullable=False, default="")
     voice_source = Column(String(20), nullable=False, default="")
+    language = Column(String(20), nullable=False, default="zh-CN")
     video_ratio = Column(String(10), nullable=False, default="9:16")
     video_model = Column(String(100), nullable=False, default="")
     # 结果与状态

packages/adapters/sqlalchemy_impl/viral_video_repository.py

@@ -49,6 +49,7 @@ def _to_domain(model: ViralVideoJobModel) -> ViralVideoJob:
         style_template_id=getattr(model, "style_template_id", "") or "",
         voice_id=getattr(model, "voice_id", "") or "",
         voice_source=getattr(model, "voice_source", "") or "",
+        language=getattr(model, "language", "zh-CN") or "zh-CN",
         video_ratio=getattr(model, "video_ratio", "9:16") or "9:16",
         video_model=getattr(model, "video_model", "") or "",
         status=ViralVideoStatus(model.status) if model.status else ViralVideoStatus.PENDING,
@@ -102,6 +103,7 @@ class SQLAlchemyViralVideoJobRepository:
             style_template_id=job.style_template_id,
             voice_id=job.voice_id,
             voice_source=job.voice_source,
+            language=getattr(job, "language", "zh-CN") or "zh-CN",
             video_ratio=job.video_ratio,
             video_model=job.video_model,
             status=job.status,
@@ -169,6 +171,7 @@ class SQLAlchemyViralVideoJobRepository:
         model.style_template_id = job.style_template_id
         model.voice_id = job.voice_id or ""
         model.voice_source = job.voice_source or ""
+        model.language = getattr(job, "language", "zh-CN") or "zh-CN"
         model.video_ratio = job.video_ratio or "9:16"
         model.video_model = job.video_model or ""
         model.updated_at = datetime.now(timezone.utc)

packages/application/ditto_service.py

@@ -5,15 +5,17 @@
 - POST /generate             生成口型视频（同步返回 MP4 流）
 
 关键特性：
-- 入参：video_url（人物模板视频 URL） + audio_url（TTS 音频 URL） + script（文案原文）
+- 入参：video_url（人物驱动视频 URL，用户上传优先；未传时用 default_video_url 兜底）
+        + audio_url（TTS 音频 URL） + script（文案原文）
+        + emo_timeline（可选，LLM 情绪时间线 JSON 字符串）
 - 出参：直接返回 video/mp4 字节流（自带音频，无需二次混流）
 - 429 时指数退避重试（最多 ditto_max_retries 次）
-- 500/超时视为失败
+- 500/超时/网络不可达视为失败
 - 输出 MP4 字节流转存到自家 OSS，返回公网 URL
 
 注意：
 - 保留 MuseTalk/GPU 路径不变；本服务作为更高优先级的第三条口型路径
-- 不传 emotion/表情精细控制，使用默认 emo_global=4（中性）+ use_script_emo=true（关键词驱动表情）
+- video_url 时长对齐（视频<音频时循环延长）在 Celery 任务层用 ffmpeg 预处理
 - Ditto 输出自带音视频，不需要 GFPGAN 超分，不需要 ffmpeg 音视频混流
 """
 
@@ -76,7 +78,11 @@ class DittoClient:
 
     @property
     def is_configured(self) -> bool:
-        """配置是否完整（base_url + 默认模板视频都有值）."""
+        """配置是否完整（base_url 必填 + 默认模板视频兜底 URL 有值）。
+
+        注意：即使 is_configured=True，实际生成时优先使用用户上传的 video_url；
+        default_video_url 仅作为用户未上传视频时的兜底。
+        """
         return bool(self.base_url) and bool(self.default_video_url)
 
     def health(self) -> bool:

packages/application/system_config_service.py

@@ -12,7 +12,7 @@ import logging
 import threading
 from typing import Any
 
-from packages.adapters.sqlalchemy_impl.session import SessionLocal
+from packages.adapters.sqlalchemy_impl import session as db_session
 from packages.adapters.sqlalchemy_impl.system_setting_repository import (
     SQLAlchemySystemSettingRepository,
 )
@@ -35,7 +35,9 @@ class SystemConfigService:
 
     # ── 会话 ─────────────────────────────────────────────────────
     def _get_session_factory(self):
-        factory = self._session_factory or SessionLocal
+        # 必须运行时读取模块属性：模块导入时 SessionLocal 还是 None，
+        # initialize_database() 之后才被赋值，import 时绑定会拿到旧值。
+        factory = self._session_factory or db_session.SessionLocal
         if factory is None:
             raise RuntimeError("数据库会话工厂未初始化")
         return factory

packages/application/viral_video/prompts.py

@@ -146,10 +146,12 @@ _STORYBOARD_SYSTEM = (
    - 最后一个 clip 必须结束于 total_duration 秒
    - 相邻 clip 首尾相接，不能有间隙也不能重叠
    - 每个 clip 的时长 = Y - X，必须 >= 2 秒
-10. 每个 clip 必须分配一个 reference_image_index（从 0 开始的图片序号），没有合适图片填 -1"""
+10. 每个 clip 必须分配一个 reference_image_index（从 0 开始的图片序号），没有合适图片填 -1
+11. 必须严格按<marketing_purpose><target_audience><persona><viral_structure><language><industry>指定的参数写文案和分镜，不能忽略任何一项用户参数"""
 )
 
 _STORYBOARD_USER = """<marketing_purpose>{marketing_purpose}</marketing_purpose>
+<industry>{industry}</industry>
 <image_analysis>
 {image_summary}
 </image_analysis>
@@ -159,6 +161,9 @@ _STORYBOARD_USER = """<marketing_purpose>{marketing_purpose}</marketing_purpose>
 <aspect_ratio>{aspect_ratio}</aspect_ratio>
 <tone>{tone}</tone>
 <target_audience>{target_audience}</target_audience>
+<persona>{persona_hint}</persona>
+<viral_structure>{viral_structure_hint}</viral_structure>
+<language>{language_hint}</language>
 <extra_requirements>{extra_requirements}</extra_requirements>
 </user_parameters>
 {video_style_section}

packages/config/base.py

@@ -185,8 +185,8 @@ class SharedSettings(BaseSettings):
         default="",
         validation_alias=AliasChoices("DITTO_API_BASE_URL", "ditto_api_base_url"),
     )
-    # 默认人物模板视频 URL（正面 5-10 秒循环、光线均匀、半身）。Ditto 模式下忽略
-    # 用户上传的驱动视频/图片，统一用该模板；后续可扩展为多模板让用户选择。
+    # 默认人物模板视频 URL（兜底用：用户未上传视频时使用，或视频预处理失败时回退）。
+    # 正面 5-10 秒、光线均匀、半身 1080x1920 竖版；正常流程下 Ditto 优先使用用户上传的 video_url。
     ditto_default_video_url: str = Field(
         default="",
         validation_alias=AliasChoices("DITTO_DEFAULT_VIDEO_URL", "ditto_default_video_url"),

packages/domain/viral_video.py

@@ -107,6 +107,7 @@ class ViralVideoJob:
     # v1.5.1 音频/视频参数
     voice_id: str = ""
     voice_source: str = ""
+    language: str = "zh-CN"
     video_ratio: str = "9:16"
     video_model: str = ""
     # v1.4+ 产物

tests/unit/test_system_config.py

@@ -91,6 +91,37 @@ class TestConfigService:
     def test_default_when_missing(self, config_service):
         assert config_service.get_config("not_exist", "fallback") == "fallback"
 
+    def test_uses_module_session_factory_assigned_after_import(self):
+        # 回归：模块导入时 session.SessionLocal 为 None，initialize_database()
+        # 之后才赋值；服务必须运行时读取模块属性，而不是 import 时绑定旧值。
+        from packages.adapters.sqlalchemy_impl import session as db_session
+
+        engine = create_engine("sqlite://")
+        Base.metadata.create_all(engine)
+        factory = sessionmaker(bind=engine)
+        writer = SystemConfigService(session_factory=factory)
+        writer.set_config("late_key", 7, setting_type=SETTING_TYPE_INT)
+
+        reader = SystemConfigService()  # 不注入工厂，依赖模块级 SessionLocal
+        old = db_session.SessionLocal
+        try:
+            db_session.SessionLocal = factory
+            assert reader.get_config("late_key", 0) == 7
+        finally:
+            db_session.SessionLocal = old
+
+    def test_raises_when_no_session_factory(self):
+        from packages.adapters.sqlalchemy_impl import session as db_session
+
+        svc = SystemConfigService()
+        old = db_session.SessionLocal
+        try:
+            db_session.SessionLocal = None
+            with pytest.raises(RuntimeError):
+                svc.get_config("anything", 1)
+        finally:
+            db_session.SessionLocal = old
+
     def test_db_overrides_default(self, config_service):
         config_service.set_config("k", 20, setting_type=SETTING_TYPE_INT)
         # 再次读取应命中 DB 值，而非传入的默认

tests/unit/test_viral_video_v8_bugs.py

@@ -907,3 +907,205 @@ class TestWanNativeAudioSkipTTS:
         assert "第3个镜头[10-15秒]" in prompt
         assert "配音" in prompt
         assert "推荐大家来" in prompt
+
+
+# ═══════════════════════════════════════════════════════════════
+# Issue #2249: 用户参数未生效修复验证
+# ═══════════════════════════════════════════════════════════════
+
+
+class TestIssue2249UserParamsFix:
+    """Issue #2249: 爆款视频用户参数未生效 6 项修复验证"""
+
+    # ── Bug1: target_customer 字段名修正 ──
+
+    def test_bug1_target_customer_field_used_in_prompt(self):
+        """Bug1: worker 应使用 job.target_customer 而非 job.target_audience"""
+        from unittest.mock import MagicMock, patch
+
+        from apps.worker.worker_app.tasks import viral_video as vv
+
+        job = MagicMock()
+        job.target_customer = "25-35岁女性"
+        job.persona_id = ""
+        job.viral_structure = ""
+        job.industry = "美容"
+        job.language = "zh-CN"
+        job.marketing_purpose = "引流到店"
+        job.user_copy_text = ""
+        job.video_ratio = "9:16"
+        job.tone = "亲切自然"
+        job.style_guide = None
+        job.duration = 15
+
+        # Verify the code uses target_customer
+        source = open(f"{BASE}/apps/worker/worker_app/tasks/viral_video.py").read()
+        assert 'getattr(job, "target_customer"' in source
+        assert 'getattr(job, "target_audience"' not in source
+
+    # ── Bug2: persona_id 进入 prompt ──
+
+    def test_bug2_persona_style_guide_covers_boss_ip(self):
+        """Bug2: _PERSONA_STYLE_GUIDE 必须覆盖'老板型IP'"""
+        from apps.worker.worker_app.tasks.viral_video import _PERSONA_STYLE_GUIDE
+
+        assert "老板型IP" in _PERSONA_STYLE_GUIDE
+        assert "老板" in _PERSONA_STYLE_GUIDE["老板型IP"] or "第一人称" in _PERSONA_STYLE_GUIDE["老板型IP"]
+
+    def test_bug2_persona_style_hint_called_and_returns_hint(self):
+        """Bug2: _persona_style_hint 返回有意义的提示"""
+        from apps.worker.worker_app.tasks.viral_video import _persona_style_hint
+
+        hint = _persona_style_hint("老板型IP")
+        assert "老板型IP" in hint
+        assert "人设风格" in hint
+
+    def test_bug2_persona_style_hint_unknown_value(self):
+        """Bug2: 未知人设值返回通用提示而非空"""
+        from apps.worker.worker_app.tasks.viral_video import _persona_style_hint
+
+        hint = _persona_style_hint("自由职业者")
+        assert "自由职业者" in hint  # 自由值照直提示
+
+    def test_bug2_persona_style_hint_empty(self):
+        """Bug2: 空人设返回默认提示"""
+        from apps.worker.worker_app.tasks.viral_video import _persona_style_hint
+
+        hint = _persona_style_hint("")
+        assert "未指定" in hint
+
+    def test_bug2_persona_in_storyboard_template(self):
+        """Bug2: _STORYBOARD_USER 模板包含 <persona> 标签"""
+        from packages.application.viral_video.prompts import _STORYBOARD_USER
+
+        assert "<persona>" in _STORYBOARD_USER
+        assert "{persona_hint}" in _STORYBOARD_USER
+
+    # ── Bug3: viral_structure 进入 prompt ──
+
+    def test_bug3_viral_structure_hint_known(self):
+        """Bug3: 已知结构返回具体写作指导"""
+        from apps.worker.worker_app.tasks.viral_video import _viral_structure_hint
+
+        hint = _viral_structure_hint("反差破局+亮明观点+还原现状")
+        assert "反差破局" in hint
+        assert "开头" in hint or "钩子" in hint
+
+    def test_bug3_viral_structure_hint_unknown(self):
+        """Bug3: 未知结构返回通用提示"""
+        from apps.worker.worker_app.tasks.viral_video import _viral_structure_hint
+
+        hint = _viral_structure_hint("自定义结构")
+        assert "自定义结构" in hint
+
+    def test_bug3_viral_structure_hint_empty(self):
+        """Bug3: 空结构返回默认提示"""
+        from apps.worker.worker_app.tasks.viral_video import _viral_structure_hint
+
+        hint = _viral_structure_hint("")
+        assert "未指定" in hint
+
+    def test_bug3_viral_structure_in_template(self):
+        """Bug3: _STORYBOARD_USER 模板包含 <viral_structure> 标签"""
+        from packages.application.viral_video.prompts import _STORYBOARD_USER
+
+        assert "<viral_structure>" in _STORYBOARD_USER
+        assert "{viral_structure_hint}" in _STORYBOARD_USER
+
+    # ── Bug4: industry 进入 prompt ──
+
+    def test_bug4_industry_in_template(self):
+        """Bug4: _STORYBOARD_USER 模板包含 <industry> 标签"""
+        from packages.application.viral_video.prompts import _STORYBOARD_USER
+
+        assert "<industry>" in _STORYBOARD_USER
+        assert "{industry}" in _STORYBOARD_USER
+
+    # ── Bug5: language 全链路 ──
+
+    def test_bug5_language_in_domain_model(self):
+        """Bug5: ViralVideoJob 领域模型有 language 字段"""
+        from packages.domain.viral_video import ViralVideoJob
+
+        job = ViralVideoJob(user_id="test")
+        assert hasattr(job, "language")
+        assert job.language == "zh-CN"
+
+    def test_bug5_language_in_create_request(self):
+        """Bug5: CreateViralVideoRequest 接受 language"""
+        from apps.api.app.schemas.viral_video import CreateViralVideoRequest
+
+        req = CreateViralVideoRequest(images=["http://a.jpg"], language="en-US")
+        assert req.language == "en-US"
+
+    def test_bug5_language_in_generate_copy_request(self):
+        """Bug5: GenerateCopyRequest 接受 language"""
+        from apps.api.app.schemas.viral_video import GenerateCopyRequest
+
+        req = GenerateCopyRequest(language="zh-TW")
+        assert req.language == "zh-TW"
+
+    def test_bug5_language_in_confirm_copy_request(self):
+        """Bug5+6: ConfirmCopyRequest 接受 language"""
+        from apps.api.app.schemas.viral_video import ConfirmCopyRequest
+
+        req = ConfirmCopyRequest(language="en-US")
+        assert req.language == "en-US"
+
+    def test_bug5_language_hint_function(self):
+        """Bug5: _language_hint 返回正确提示"""
+        from apps.worker.worker_app.tasks.viral_video import _language_hint
+
+        assert "普通话" in _language_hint("zh-CN")
+        assert (
+            "英语" in _language_hint("en-US")
+            or "English" in _language_hint("en-US").lower()
+            or "英语" in _language_hint("en-us")
+        )
+
+    def test_bug5_language_in_template(self):
+        """Bug5: _STORYBOARD_USER 模板包含 <language> 标签"""
+        from packages.application.viral_video.prompts import _STORYBOARD_USER
+
+        assert "<language>" in _STORYBOARD_USER
+        assert "{language_hint}" in _STORYBOARD_USER
+
+    def test_bug5_language_in_db_model(self):
+        """Bug5: DB model 有 language 列"""
+        from packages.adapters.sqlalchemy_impl.models import ViralVideoJobModel
+
+        assert hasattr(ViralVideoJobModel, "language")
+
+    def test_bug5_migration_exists(self):
+        """Bug5: alembic migration 107 存在"""
+        from pathlib import Path
+
+        migration = Path(f"{BASE}/alembic/versions/107_add_language_to_viral_video_jobs.py")
+        assert migration.exists()
+
+    # ── Bug6: voice 选择持久化 ──
+
+    def test_bug6_confirm_copy_request_has_voice_fields(self):
+        """Bug6: ConfirmCopyRequest 有 voice_id 和 voice_source 字段"""
+        from apps.api.app.schemas.viral_video import ConfirmCopyRequest
+
+        req = ConfirmCopyRequest(voice_id="longxiaochun_v3", voice_source="preset")
+        assert req.voice_id == "longxiaochun_v3"
+        assert req.voice_source == "preset"
+
+    def test_bug6_confirm_copy_request_voice_defaults_none(self):
+        """Bug6: ConfirmCopyRequest voice 字段默认 None（可选）"""
+        from apps.api.app.schemas.viral_video import ConfirmCopyRequest
+
+        req = ConfirmCopyRequest()
+        assert req.voice_id is None
+        assert req.voice_source is None
+
+    # ── System prompt 规则 ──
+
+    def test_system_prompt_has_param_enforcement_rule(self):
+        """系统 prompt 包含参数必须遵守的规则"""
+        from packages.application.viral_video.prompts import _STORYBOARD_SYSTEM
+
+        assert "marketing_purpose" in _STORYBOARD_SYSTEM or "persona" in _STORYBOARD_SYSTEM
+        assert "不能忽略" in _STORYBOARD_SYSTEM or "必须" in _STORYBOARD_SYSTEM