fix: 生成任务入队失败时标记为failed，避免pending僵尸任务

根因：任务创建(DB commit)和入队(Celery send_task)是两个独立操作， send_task失败时任务卡在pending状态永远不会执行。修复： - 新增_safe_enqueue_generation_task安全入队函数 - send_task失败时自动标记任务为failed并记录错误 - 覆盖4处入口：批量创建、generation重试、task_center两级重试
2026-07-11 01:15:21 +08:00
10 changed files with 122 additions and 78 deletions
@@ -37,6 +37,43 @@ logger = logging.getLogger(__name__)
 router = APIRouter()


+def _safe_enqueue_generation_task(
+    task: Any,
+    generation_task_repository: Any,
+) -> bool:
+    """安全入队：send_task 失败时自动把任务标记为 failed，避免留下 pending 僵尸任务。
+
+    Returns:
+        True 表示入队成功，False 表示入队失败（已标记为 failed）
+    """
+    try:
+        celery_app.send_task("worker.generate_video", args=[task.id])
+        logger.info(
+            "[生成任务] 入队成功: task_id=%s, status=%s",
+            task.id,
+            task.status,
+        )
+        return True
+    except Exception as e:
+        logger.error(
+            "[生成任务] 入队失败，标记为失败: task_id=%s error=%s",
+            task.id,
+            e,
+            exc_info=True,
+        )
+        try:
+            task.mark_failed(f"任务入队失败: {e}")
+            generation_task_repository.update(task)
+        except Exception as update_err:
+            logger.error(
+                "[生成任务] 入队失败后更新状态也失败: task_id=%s error=%s",
+                task.id,
+                update_err,
+                exc_info=True,
+            )
+        return False
+
+
 def _check_project_access(project_id: str, user_id: str, project_repository) -> None:
    """检查用户是否有项目访问权限"""
    project = project_repository.find_by_id(project_id)
@@ -228,6 +265,7 @@ def create_generation_task(
    use_case = CreateGenerationTaskUseCase(generation_task_repository)
    count = request.count
    created_tasks = []
+    failed_tasks = []
    # 同批次任务共享 batch_id，用于视频查重时批次内比对
    batch_id = uuid.uuid4().hex if count > 1 else ""

@@ -249,19 +287,15 @@ def create_generation_task(
                    batch_id=batch_id,
                )
            )
-            celery_app.send_task("worker.generate_video", args=[task.id])
-            created_tasks.append(task)
-            logger.info(
-                "[生成任务] 入队成功: task_id=%s, status=%s, batch_id=%s",
-                task.id,
-                task.status,
-                batch_id,
-            )
+            if _safe_enqueue_generation_task(task, generation_task_repository):
+                created_tasks.append(task)
+            else:
+                failed_tasks.append(task)
    except Exception as e:
        logger.error("[生成任务] 创建失败: %s", e, exc_info=True)
        raise HTTPException(status_code=500, detail="创建生成任务失败，请稍后重试或查看任务日志")

-    items = [_to_generation_task_response(t) for t in created_tasks]
+    items = [_to_generation_task_response(t) for t in created_tasks + failed_tasks]
    return BatchGenerationTaskResponse(items=items, total=len(items))


@@ -347,5 +381,6 @@ def retry_generation_task(
            asset_select_mode=getattr(task, "asset_select_mode", ""),
        )
    )
-    celery_app.send_task("worker.generate_video", args=[retried.id])
+    if not _safe_enqueue_generation_task(retried, generation_task_repository):
+        logger.warning("[生成任务] 重试入队失败: task_id=%s", retried.id)
    return _to_generation_task_response(retried)
@@ -25,6 +25,35 @@ from packages.application import (
 router = APIRouter()


+def _safe_enqueue_generation_task(
+    task: Any,
+    generation_task_repository: Any,
+) -> bool:
+    """安全入队：send_task 失败时自动把任务标记为 failed，避免留下 pending 僵尸任务。"""
+    try:
+        celery_app.send_task("worker.generate_video", args=[task.id])
+        logger.info("[任务中心] 生成任务入队成功: task_id=%s", task.id)
+        return True
+    except Exception as e:
+        logger.error(
+            "[任务中心] 生成任务入队失败，标记为失败: task_id=%s error=%s",
+            task.id,
+            e,
+            exc_info=True,
+        )
+        try:
+            task.mark_failed(f"任务入队失败: {e}")
+            generation_task_repository.update(task)
+        except Exception as update_err:
+            logger.error(
+                "[任务中心] 入队失败后更新状态也失败: task_id=%s error=%s",
+                task.id,
+                update_err,
+                exc_info=True,
+            )
+        return False
+
+
 def _humanize_task_error(error_message: str) -> str:
    raw = (error_message or "").strip()
    if not raw:
@@ -153,7 +182,8 @@ def retry_task_by_id(
            created_by_user_id=authenticated_user.user.id,
        )
    )
-    celery_app.send_task("worker.generate_video", args=[retried.id])
+    if not _safe_enqueue_generation_task(retried, generation_task_repository):
+        logger.warning("[任务中心] 用户级重试入队失败: task_id=%s", retried.id)
    return UserTaskResponse(
        id=f"generation:{retried.id}",
        task_type="generation",
@@ -235,7 +265,8 @@ def retry_project_task(
                created_by_user_id=authenticated_user.user.id,
            )
        )
-        celery_app.send_task("worker.generate_video", args=[retried.id])
+        if not _safe_enqueue_generation_task(retried, generation_task_repository):
+            logger.warning("[任务中心] 项目级重试用队失败: task_id=%s", retried.id)
        return _generation_task_to_project_response(retried)
    if task_type == "ingest":
        job = ingest_job_repository.get(source_id)
@@ -70,21 +70,21 @@ class CosyVoiceService:
    - 音色克隆: POST /services/audio/tts/customization (model=voice-enrollment)
      - action=create_voice: 创建克隆音色，返回 voice_id（状态 DEPLOYING）
      - action=query_voice: 查询音色状态（DEPLOYING / OK / UNDEPLOYED）
-    - 语音合成: POST /services/audio/tts/SpeechSynthesizer (model=cosyvoice-v3-flash)
+    - 语音合成: POST /services/audio/tts/SpeechSynthesizer (model=cosyvoice-v3.5-plus)
      - 非流式: 同步返回音频 URL

    使用示例:
        service = CosyVoiceService(
            api_key="your-api-key",
            base_url="https://dashscope.aliyuncs.com/api/v1",
-            model="cosyvoice-v3-flash",
+            model="cosyvoice-v3.5-plus",
        )

        # 音色克隆
        result = service.clone_voice(audio_url="https://example.com/audio.mp3")

        # 语音合成
-        result = service.synthesize_speech(text="你好世界", voice_id="longxiaochun_v3")
+        result = service.synthesize_speech(text="你好世界", voice_id="longxiaochun")
    """

    # 音色状态轮询配置
@@ -589,20 +589,6 @@ class CosyVoiceService:
            "Content-Type": "application/json",
        }

-        # DEBUG: 打印完整请求信息，用于排查418错误
-        import json as json_lib
-        safe_headers = {k: v for k, v in headers.items()}
-        if "Authorization" in safe_headers:
-            token = safe_headers["Authorization"]
-            if len(token) > 20:
-                safe_headers["Authorization"] = token[:13] + "..." + token[-4:]
-        logger.info(
-            "[CosyVoice Debug] 请求详情: "
-            "method=%s, url=%s, headers=%s, body=%s",
-            method, url, safe_headers,
-            json_lib.dumps(json, ensure_ascii=False) if json else "None",
-        )
-
        last_error: Optional[Exception] = None

        for attempt in range(self.MAX_RETRIES):
@@ -615,14 +601,6 @@ class CosyVoiceService:
                    timeout=timeout,
                )

-                # DEBUG: 打印响应状态和完整响应体
-                logger.info(
-                    "[CosyVoice Debug] 响应详情: "
-                    "status=%d, body=%s",
-                    response.status_code,
-                    response.text[:2000],  # 最多2000字符，避免日志过大
-                )
-
                # 处理响应
                if response.status_code == 200:
                    return response.json()
@@ -16,7 +16,7 @@ class PresetVoice:
    """预置音色定义。

    Attributes:
-        voice_id: CosyVoice 模型音色名（如 longxiaochun_v3）
+        voice_id: CosyVoice 模型音色名（如 longxiaochun）
        name: 中文展示名
        description: 音色描述
        gender: 性别（male/female）
@@ -49,7 +49,7 @@ class PresetVoice:
 # 预置音色列表（阿里云 CosyVoice 真实可用音色）
 PRESET_VOICES: list[PresetVoice] = [
    PresetVoice(
-        voice_id="longxiaochun_v3",
+        voice_id="longxiaochun",
        name="龙小淳",
        description="温柔女声，适合情感类内容",
        gender="female",
@@ -57,7 +57,7 @@ PRESET_VOICES: list[PresetVoice] = [
        tags=["温柔", "女声", "情感"],
    ),
    PresetVoice(
-        voice_id="longxiaoxia_v3",
+        voice_id="longxiaoxia",
        name="龙小夏",
        description="知性女声，适合新闻播报",
        gender="female",
@@ -65,7 +65,7 @@ PRESET_VOICES: list[PresetVoice] = [
        tags=["知性", "女声", "播报"],
    ),
    PresetVoice(
-        voice_id="longxiaochen_v3",
+        voice_id="longxiaochen",
        name="龙小晨",
        description="磁性男声，适合有声书",
        gender="male",
@@ -73,7 +73,7 @@ PRESET_VOICES: list[PresetVoice] = [
        tags=["磁性", "男声", "有声书"],
    ),
    PresetVoice(
-        voice_id="longyue_v3",
+        voice_id="longyue",
        name="龙悦",
        description="甜美女声，适合广告配音",
        gender="female",
@@ -81,7 +81,7 @@ PRESET_VOICES: list[PresetVoice] = [
        tags=["甜美", "女声", "广告"],
    ),
    PresetVoice(
-        voice_id="longshu_v3",
+        voice_id="longshu",
        name="龙书",
        description="沉稳男声，适合教育讲解",
        gender="male",
@@ -89,7 +89,7 @@ PRESET_VOICES: list[PresetVoice] = [
        tags=["沉稳", "男声", "教育"],
    ),
    PresetVoice(
-        voice_id="longjing_v3",
+        voice_id="longjing",
        name="龙静",
        description="优雅女声，适合纪录片解说",
        gender="female",
@@ -97,7 +97,7 @@ PRESET_VOICES: list[PresetVoice] = [
        tags=["优雅", "女声", "纪录片"],
    ),
    PresetVoice(
-        voice_id="longbo_v3",
+        voice_id="longbo",
        name="龙博",
        description="浑厚男声，适合科技类内容",
        gender="male",
@@ -105,7 +105,7 @@ PRESET_VOICES: list[PresetVoice] = [
        tags=["浑厚", "男声", "科技"],
    ),
    PresetVoice(
-        voice_id="longtian_v3",
+        voice_id="longtian",
        name="龙甜",
        description="活泼女声，适合短视频配音",
        gender="female",
@@ -32,8 +32,8 @@ class SharedSettings(BaseSettings):
    # CosyVoice (阿里云百炼语音合成)
    cosyvoice_api_key: str = ""
    cosyvoice_base_url: str = "https://dashscope.aliyuncs.com/api/v1"
-    cosyvoice_model: str = "cosyvoice-v3-flash"
-    cosyvoice_voice: str = "longxiaochun_v3"  # 默认音色（v3 系列系统音色带 _v3 后缀）
+    cosyvoice_model: str = "cosyvoice-v3.5-plus"
+    cosyvoice_voice: str = "longxiaochun"  # 默认音色
    cosyvoice_sample_rate: int = 22050
    cosyvoice_format: str = "mp3"  # 输出格式：mp3/wav/pcm
    # 音色克隆模型名（固定为 voice-enrollment）
@@ -23,7 +23,7 @@ def _make_service(
    *,
    api_key: str = "test-api-key",
    base_url: str = "https://dashscope.aliyuncs.com/api/v1",
-    model: str = "cosyvoice-v3-flash",
+    model: str = "cosyvoice-v3.5-plus",
    clone_model: str = "voice-enrollment",
    http_client: httpx.Client | None = None,
    audio_url_signer=None,
@@ -78,7 +78,7 @@ class TestSubmitCloneTask:
            200,
            {
                "output": {
-                    "voice_id": "cosyvoice-v3-flash-clone-abc123",
+                    "voice_id": "cosyvoice-v3.5-plus-clone-abc123",
                    "status": "DEPLOYING",
                },
                "usage": {"count": 1},
@@ -92,7 +92,7 @@ class TestSubmitCloneTask:
            voice_name="myvoice",
        )

-        assert result["voice_id"] == "cosyvoice-v3-flash-clone-abc123"
+        assert result["voice_id"] == "cosyvoice-v3.5-plus-clone-abc123"
        assert result["status"] == "DEPLOYING"
        assert result["request_id"] == "req-001"

@@ -104,7 +104,7 @@ class TestSubmitCloneTask:
        payload = call_args.kwargs["json"]
        assert payload["model"] == "voice-enrollment"
        assert payload["input"]["action"] == "create_voice"
-        assert payload["input"]["target_model"] == "cosyvoice-v3-flash"
+        assert payload["input"]["target_model"] == "cosyvoice-v3.5-plus"
        assert payload["input"]["prefix"] == "myvoice"
        assert payload["input"]["url"] == "https://example.com/audio.wav"
        assert payload["input"]["language_hints"] == ["zh"]
@@ -229,7 +229,7 @@ class TestQueryVoiceStatus:
            {
                "output": {
                    "status": "DEPLOYING",
-                    "target_model": "cosyvoice-v3-flash",
+                    "target_model": "cosyvoice-v3.5-plus",
                    "gmt_create": "2026-01-01T00:00:00Z",
                    "gmt_modified": "2026-01-01T00:01:00Z",
                    "resource_link": "https://...",
@@ -242,7 +242,7 @@ class TestQueryVoiceStatus:
        result = service.query_voice_status("voice-123")

        assert result["status"] == "DEPLOYING"
-        assert result["target_model"] == "cosyvoice-v3-flash"
+        assert result["target_model"] == "cosyvoice-v3.5-plus"

        # 验证请求
        payload = mock_client.request.call_args.kwargs["json"]
@@ -253,7 +253,7 @@ class TestQueryVoiceStatus:
    def test_query_ok_status(self) -> None:
        mock_client = MagicMock()
        mock_client.request.return_value = _mock_response(
-            200, {"output": {"status": "OK", "target_model": "cosyvoice-v3-flash"}}
+            200, {"output": {"status": "OK", "target_model": "cosyvoice-v3.5-plus"}}
        )

        service = _make_service(http_client=mock_client)
@@ -278,7 +278,7 @@ class TestPollCloneTask:
    def test_poll_ok_on_first_check(self) -> None:
        mock_client = MagicMock()
        mock_client.request.return_value = _mock_response(
-            200, {"output": {"status": "OK", "target_model": "cosyvoice-v3-flash"}}
+            200, {"output": {"status": "OK", "target_model": "cosyvoice-v3.5-plus"}}
        )

        service = _make_service(http_client=mock_client)
@@ -397,7 +397,7 @@ class TestSynthesizeSpeech:

        service = _make_service(http_client=mock_client)
        result = service.synthesize_speech(
-            text="你好世界", voice_id="longxiaochun_v3"
+            text="你好世界", voice_id="longxiaochun"
        )

        assert isinstance(result, SynthesizeResult)
@@ -409,9 +409,9 @@ class TestSynthesizeSpeech:
        assert "/services/audio/tts/SpeechSynthesizer" in call_args.kwargs["url"]

        payload = call_args.kwargs["json"]
-        assert payload["model"] == "cosyvoice-v3-flash"
+        assert payload["model"] == "cosyvoice-v3.5-plus"
        assert payload["input"]["text"] == "你好世界"
-        assert payload["input"]["voice"] == "longxiaochun_v3"
+        assert payload["input"]["voice"] == "longxiaochun"
        assert payload["input"]["format"] == "mp3"
        assert payload["input"]["sample_rate"] == 22050
        assert payload["input"]["rate"] == 1.0
@@ -64,7 +64,7 @@ class TestPresetVoice:
    def test_preset_voice_to_dict(self) -> None:
        """序列化。"""
        voice = PresetVoice(
-            voice_id="longxiaochun_v3",
+            voice_id="longxiaochun",
            name="龙小淳",
            description="温柔女声",
            gender="female",
@@ -73,7 +73,7 @@ class TestPresetVoice:

        result = voice.to_dict()

-        assert result["voice_id"] == "longxiaochun_v3"
+        assert result["voice_id"] == "longxiaochun"
        assert result["name"] == "龙小淳"
        assert result["description"] == "温柔女声"
        assert result["gender"] == "female"
@@ -127,14 +127,14 @@ class TestPresetVoicesConfig:
    def test_cosyvoice_voice_ids(self) -> None:
        """音色 ID 应为 CosyVoice 真实可用的音色名。"""
        expected_ids = {
-            "longxiaochun_v3",
-            "longxiaoxia_v3",
-            "longxiaochen_v3",
-            "longyue_v3",
-            "longshu_v3",
-            "longjing_v3",
-            "longbo_v3",
-            "longtian_v3",
+            "longxiaochun",
+            "longxiaoxia",
+            "longxiaochen",
+            "longyue",
+            "longshu",
+            "longjing",
+            "longbo",
+            "longtian",
        }
        actual_ids = {v.voice_id for v in PRESET_VOICES}
        assert actual_ids == expected_ids
@@ -164,10 +164,10 @@ class TestPresetVoiceHelpers:

    def test_get_preset_voice_by_id_found(self) -> None:
        """按 ID 查找存在的音色。"""
-        voice = get_preset_voice_by_id("longxiaochun_v3")
+        voice = get_preset_voice_by_id("longxiaochun")
        assert voice is not None
        assert voice.name == "龙小淳"
-        assert voice.voice_id == "longxiaochun_v3"
+        assert voice.voice_id == "longxiaochun"

    def test_get_preset_voice_by_id_not_found(self) -> None:
        """按 ID 查找不存在的音色。"""
@@ -176,9 +176,9 @@ class TestPresetVoiceHelpers:

    def test_is_preset_voice_true(self) -> None:
        """判断预置音色返回 True。"""
-        assert is_preset_voice("longxiaochun_v3") is True
-        assert is_preset_voice("longxiaoxia_v3") is True
-        assert is_preset_voice("longbo_v3") is True
+        assert is_preset_voice("longxiaochun") is True
+        assert is_preset_voice("longxiaoxia") is True
+        assert is_preset_voice("longbo") is True

    def test_is_preset_voice_false(self) -> None:
        """判断非预置音色返回 False。"""
@@ -15,7 +15,7 @@ class TestTTSJobCreate:
        job = TTSJob.create(
            user_id="user_001",
            input_text="这是一段测试文本",
-            voice_id="longxiaochun_v3",
+            voice_id="longxiaochun",
            voice_model="cosyvoice-v1",
            project_id="project_001",
            voice_clone_profile_id="profile_001",
@@ -26,7 +26,7 @@ class TestTTSJobCreate:
        assert job.id
        assert job.user_id == "user_001"
        assert job.input_text == "这是一段测试文本"
-        assert job.voice_id == "longxiaochun_v3"
+        assert job.voice_id == "longxiaochun"
        assert job.voice_model == "cosyvoice-v1"
        assert job.project_id == "project_001"
        assert job.voice_clone_profile_id == "profile_001"
@@ -291,7 +291,7 @@ class TestTTSJobToDict:
        job = TTSJob.create(
            user_id="user_001",
            input_text="测试文本",
-            voice_id="longxiaochun_v3",
+            voice_id="longxiaochun",
            voice_model="cosyvoice-v1",
            project_id="project_001",
            voice_clone_profile_id="profile_001",
@@ -306,7 +306,7 @@ class TestTTSJobToDict:
        assert result["id"] == job.id
        assert result["user_id"] == "user_001"
        assert result["input_text"] == "测试文本"
-        assert result["voice_id"] == "longxiaochun_v3"
+        assert result["voice_id"] == "longxiaochun"
        assert result["voice_model"] == "cosyvoice-v1"
        assert result["project_id"] == "project_001"
        assert result["voice_clone_profile_id"] == "profile_001"