chore(ci): restore -x flag after diagnosis

- 新增 app/core/task_enqueue.py，统一 safe_enqueue_generation_task 实现
- generation_tasks.py 和 task_center.py 改为 import 公共函数
- 修复 task_center.py 缺失 logging 导入的问题
- 日志前缀通过参数传入，保持各模块语义不变

.env.example

@@ -44,9 +44,14 @@ OSS_ACCESS_KEY_SECRET=your-access-key-secret
 OSS_BUCKET_NAME=xiaoxia-autocut
 
 # ==================== CosyVoice 语音合成配置 ====================
+# 注意：base_url 只需写到 /api/v1，具体路径由代码拼接
+# 模型: cosyvoice-v3-flash (推荐，支持系统音色，性价比高)
+#       cosyvoice-v3-plus (高质量，系统音色少)
+#       cosyvoice-v3.5-flash / cosyvoice-v3.5-plus (仅支持克隆/设计音色，无系统音色)
+# 音色: v3系列系统音色带 _v3 后缀，如 longxiaochun_v3, longxiaoxia_v3, longanyang (无后缀)
 COSYVOICE_API_KEY=your-cosyvoice-api-key
-COSYVOICE_BASE_URL=https://dashscope.aliyuncs.com/api/v1/services/aigc/text2audio
-COSYVOICE_MODEL=cosyvoice-v1
-COSYVOICE_VOICE=longxiaochun
+COSYVOICE_BASE_URL=https://dashscope.aliyuncs.com/api/v1
+COSYVOICE_MODEL=cosyvoice-v3-flash
+COSYVOICE_VOICE=longxiaochun_v3
 COSYVOICE_SAMPLE_RATE=22050
 COSYVOICE_FORMAT=mp3

.env.production.example

@@ -42,10 +42,15 @@ OSS_DIRECT_UPLOAD_MAX_MB=2000
 OSS_DIRECT_UPLOAD_EXPIRE_SECONDS=900
 
 # ==================== CosyVoice 语音合成（必须配置）====================
+# 注意：base_url 只需写到 /api/v1，具体路径由代码拼接
+# 模型: cosyvoice-v3-flash (推荐，支持系统音色，性价比高)
+#       cosyvoice-v3-plus (高质量，系统音色少)
+#       cosyvoice-v3.5-flash / cosyvoice-v3.5-plus (仅支持克隆/设计音色，无系统音色)
+# 音色: v3系列系统音色带 _v3 后缀，如 longxiaochun_v3, longxiaoxia_v3, longanyang (无后缀)
 COSYVOICE_API_KEY=CHANGE_ME_COSYVOICE_API_KEY
-COSYVOICE_BASE_URL=https://dashscope.aliyuncs.com/api/v1/services/aigc/text2audio
-COSYVOICE_MODEL=cosyvoice-v1
-COSYVOICE_VOICE=longxiaochun
+COSYVOICE_BASE_URL=https://dashscope.aliyuncs.com/api/v1
+COSYVOICE_MODEL=cosyvoice-v3-flash
+COSYVOICE_VOICE=longxiaochun_v3
 COSYVOICE_SAMPLE_RATE=22050
 COSYVOICE_FORMAT=mp3
 

.flake8

@@ -2,6 +2,7 @@
 max-line-length = 120
 exclude =
     .git,
+    .cache,
     __pycache__,
     .venv,
     venv,

.gitea/workflows/ci-cd.yml

@@ -165,7 +165,7 @@ jobs:
         run: |
           set -eu
           PYTHONPATH="$PWD/apps/api:$PWD" python3 -m pytest tests/unit -q \
-            --cov=apps --cov-report=term --cov-report=xml
+            --cov=apps --cov-report=term --cov-report=xml --cov-fail-under=50
 
       - name: Start PostgreSQL for integration tests
         shell: sh

.gitignore

@@ -6,6 +6,7 @@ dist/
 coverage/
 
 # Python / backend
+.cache/
 .venv/
 venv/
 .venv-ci-root/

alembic/versions/037_add_logs_to_generation_tasks.py

@@ -0,0 +1,26 @@
+"""Add logs field to generation_tasks
+
+Revision ID: 037_generation_logs
+Revises: 036_expand_uuid_36
+Create Date: 2026-07-10
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "037_generation_logs"
+down_revision = "036_expand_uuid_36"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "generation_tasks",
+        sa.Column("logs", sa.Text(), nullable=False, server_default="[]"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("generation_tasks", "logs")

apps/api/app/api/routes/generation_tasks.py

@@ -1,10 +1,11 @@
+import logging
 import random
 import uuid
 from typing import Any
 
 from app.auth import AuthenticatedUser, get_current_user
-from app.core.celery_app import celery_app
 from app.core.storage import OSSStorageService, get_storage_service
+from app.core.task_enqueue import safe_enqueue_generation_task
 from app.dependencies import (
     get_asset_library_repository,
     get_asset_repository,
@@ -31,8 +32,9 @@ from packages.application import (
     ListGeneratedVideosByTaskUseCase,
 )
 
-router = APIRouter()
+logger = logging.getLogger(__name__)
 
+router = APIRouter()
 
 def _check_project_access(project_id: str, user_id: str, project_repository) -> None:
     """检查用户是否有项目访问权限"""
@@ -57,6 +59,7 @@ def _to_generation_task_response(task) -> GenerationTaskResponse:
         source_edit_plan_id=task.source_edit_plan_id or "",
         asset_select_mode=getattr(task, "asset_select_mode", ""),
         batch_id=getattr(task, "batch_id", ""),
+        logs=getattr(task, "logs", "[]"),
         status=task.status,
         progress=task.progress,
         result_count=task.result_count,
@@ -181,19 +184,37 @@ def create_generation_task(
     asset_library_repository: Any = Depends(get_asset_library_repository),
     asset_repository: Any = Depends(get_asset_repository),
 ) -> BatchGenerationTaskResponse:
-    project_id, asset_library_id = _resolve_project_and_library(
-        request, project_repository, asset_library_repository, asset_repository, authenticated_user
+    logger.info(
+        "[生成任务] 接收请求: user_id=%s, template_id=%s, asset_count=%d, mode=%s, count=%d",
+        authenticated_user.user.id,
+        request.template_id,
+        len(request.asset_ids),
+        request.asset_select_mode,
+        request.count,
     )
 
+    try:
+        project_id, asset_library_id = _resolve_project_and_library(
+            request, project_repository, asset_library_repository, asset_repository, authenticated_user
+        )
+    except HTTPException as e:
+        logger.warning("[生成任务] 校验失败: %s", e.detail)
+        raise
+
     # asset_library 存在性校验（仅在提供了 asset_library_id 时）
     resolved_asset_ids: list[str] = list(request.asset_ids)
     if asset_library_id:
         library = asset_library_repository.get(asset_library_id)
         if library is None or (project_id and library.project_id != project_id):
+            logger.warning("[生成任务] 素材库不存在: library_id=%s", asset_library_id)
             raise HTTPException(status_code=404, detail=f"AssetLibrary {asset_library_id} not found")
 
         assets = asset_repository.find_by_library(asset_library_id)
-        _ensure_library_has_ready_video_assets(assets)
+        try:
+            _ensure_library_has_ready_video_assets(assets)
+        except HTTPException as e:
+            logger.warning("[生成任务] 素材校验失败: %s", e.detail)
+            raise
 
         # 素材库自动匹配：当未显式指定 asset_ids 时，按模式自动选取
         if not resolved_asset_ids:
@@ -206,30 +227,37 @@ 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 ""
 
-    for _ in range(count):
-        task = use_case.execute(
-            CreateGenerationTaskCommand(
-                project_id=project_id,
-                asset_library_id=asset_library_id,
-                strategy_id=request.strategy_id,
-                voice_library_id=request.voice_library_id,
-                template_id=request.template_id,
-                asset_ids=resolved_asset_ids,
-                title_ids=request.title_ids,
-                voice_ids=request.voice_ids,
-                created_by_user_id=authenticated_user.user.id,
-                source_edit_plan_id=request.source_edit_plan_id,
-                asset_select_mode=request.asset_select_mode,
-                batch_id=batch_id,
+    try:
+        for _ in range(count):
+            task = use_case.execute(
+                CreateGenerationTaskCommand(
+                    project_id=project_id,
+                    asset_library_id=asset_library_id,
+                    strategy_id=request.strategy_id,
+                    voice_library_id=request.voice_library_id,
+                    template_id=request.template_id,
+                    asset_ids=resolved_asset_ids,
+                    title_ids=request.title_ids,
+                    voice_ids=request.voice_ids,
+                    created_by_user_id=authenticated_user.user.id,
+                    source_edit_plan_id=request.source_edit_plan_id,
+                    asset_select_mode=request.asset_select_mode,
+                    batch_id=batch_id,
+                )
             )
-        )
-        celery_app.send_task("worker.generate_video", args=[task.id])
-        created_tasks.append(task)
+            if safe_enqueue_generation_task(task, generation_task_repository, log_prefix="[生成任务]", log_task_status=True):
+                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))
 
 
@@ -315,5 +343,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, log_prefix="[生成任务]", log_task_status=True):
+        logger.warning("[生成任务] 重试入队失败: task_id=%s", retried.id)
     return _to_generation_task_response(retried)

apps/api/app/api/routes/task_center.py

@@ -1,7 +1,9 @@
+import logging
 from typing import Any
 
 from app.auth import AuthenticatedUser, get_current_user
 from app.core.celery_app import celery_app
+from app.core.task_enqueue import safe_enqueue_generation_task
 from app.dependencies import (
     get_generation_task_repository,
     get_ingest_job_repository,
@@ -22,8 +24,9 @@ from packages.application import (
     SubmitIngestJobUseCase,
 )
 
-router = APIRouter()
+logger = logging.getLogger(__name__)
 
+router = APIRouter()
 
 def _humanize_task_error(error_message: str) -> str:
     raw = (error_message or "").strip()
@@ -153,7 +156,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, log_prefix="[任务中心]"):
+        logger.warning("[任务中心] 用户级重试入队失败: task_id=%s", retried.id)
     return UserTaskResponse(
         id=f"generation:{retried.id}",
         task_type="generation",
@@ -235,7 +239,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, log_prefix="[任务中心]"):
+            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)

apps/api/app/api/routes/tts.py

@@ -7,6 +7,7 @@ from typing import Optional
 
 from app.auth import AuthenticatedUser, get_current_user
 from app.dependencies import (
+    get_audio_url_signer,
     get_cosyvoice_service,
     get_db_session,
     get_user_repository,
@@ -56,7 +57,10 @@ def _get_repository(session: Session = Depends(get_db_session)) -> SQLAlchemyTTS
     return SQLAlchemyTTSJobRepository(session)
 
 
-def _to_response(job) -> TTSJobResponse:
+def _to_response(job, sign_url=None) -> TTSJobResponse:
+    output_url = job.output_audio_url
+    if sign_url and output_url:
+        output_url = sign_url(output_url)
     return TTSJobResponse(
         id=job.id,
         user_id=job.user_id,
@@ -66,7 +70,7 @@ def _to_response(job) -> TTSJobResponse:
         project_id=job.project_id,
         voice_clone_profile_id=job.voice_clone_profile_id,
         status=job.status,
-        output_audio_url=job.output_audio_url,
+        output_audio_url=output_url,
         output_audio_key=job.output_audio_key,
         duration=job.duration,
         file_size=job.file_size,
@@ -176,6 +180,7 @@ def list_tts_jobs(
     status_filter: Optional[str] = Query(None, alias="status"),
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     repository: SQLAlchemyTTSJobRepository = Depends(_get_repository),
+    sign_url=Depends(get_audio_url_signer),
 ) -> ListTTSJobResponse:
     """列出用户的 TTS 合成任务。"""
     user_id = authenticated_user.user.id
@@ -183,7 +188,7 @@ def list_tts_jobs(
     skip = (page - 1) * page_size
     items, total = use_case.execute(user_id, status=status_filter, skip=skip, limit=page_size)
     return ListTTSJobResponse(
-        items=[_to_response(j) for j in items],
+        items=[_to_response(j, sign_url) for j in items],
         total=total,
         page=page,
         page_size=page_size,
@@ -195,6 +200,7 @@ def get_tts_job(
     job_id: str,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     repository: SQLAlchemyTTSJobRepository = Depends(_get_repository),
+    sign_url=Depends(get_audio_url_signer),
 ) -> TTSJobResponse:
     """获取 TTS 任务详情。"""
     user_id = authenticated_user.user.id
@@ -203,7 +209,7 @@ def get_tts_job(
         job = use_case.execute(job_id, user_id)
     except TTSJobNotFoundError:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="TTS job not found")
-    return _to_response(job)
+    return _to_response(job, sign_url)
 
 
 @router.get("/jobs/{job_id}/status", response_model=TTSStatusResponse)
@@ -211,6 +217,7 @@ def get_tts_job_status(
     job_id: str,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     repository: SQLAlchemyTTSJobRepository = Depends(_get_repository),
+    sign_url=Depends(get_audio_url_signer),
 ) -> TTSStatusResponse:
     """查询 TTS 合成状态（用于前端轮询）。"""
     user_id = authenticated_user.user.id
@@ -219,10 +226,13 @@ def get_tts_job_status(
         job = use_case.execute(job_id, user_id)
     except TTSJobNotFoundError:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="TTS job not found")
+    output_url = job.output_audio_url
+    if output_url:
+        output_url = sign_url(output_url)
     return TTSStatusResponse(
         id=job.id,
         status=job.status,
-        output_audio_url=job.output_audio_url,
+        output_audio_url=output_url,
         error_message=job.error_message,
         duration=job.duration,
         retry_count=job.retry_count,
@@ -258,6 +268,7 @@ def save_tts_job_to_library(
     tts_repository: SQLAlchemyTTSJobRepository = Depends(_get_repository),
     voice_library_repository: SQLAlchemyVoiceLibraryRepository = Depends(get_voice_library_repository),
     user_repository: UserRepository = Depends(get_user_repository),
+    sign_url=Depends(get_audio_url_signer),
 ) -> SaveToLibraryResponse:
     """将已完成的 TTS 合成结果保存到配音库。
 
@@ -328,7 +339,7 @@ def save_tts_job_to_library(
     return SaveToLibraryResponse(
         id=item.id,
         name=item.name,
-        audio_url=item.audio_url,
+        audio_url=sign_url(item.audio_url) if item.audio_url else "",
         duration=item.duration,
         voice_id=item.voice_id,
         voice_name=item.voice_name,

apps/api/app/api/routes/voice_clones.py

@@ -6,7 +6,7 @@ import logging
 from typing import Optional
 
 from app.auth import AuthenticatedUser, get_current_user
-from app.dependencies import get_cosyvoice_service, get_voice_clone_profile_repository
+from app.dependencies import get_audio_url_signer, get_cosyvoice_service, get_voice_clone_profile_repository
 from app.schemas.voice_clone import (
     CreateVoiceCloneRequest,
     ListVoiceCloneResponse,
@@ -37,13 +37,16 @@ logger = logging.getLogger(__name__)
 router = APIRouter()
 
 
-def _to_response(profile) -> VoiceCloneProfileResponse:
+def _to_response(profile, sign_url=None) -> VoiceCloneProfileResponse:
+    source_url = profile.source_audio_url
+    if sign_url and source_url:
+        source_url = sign_url(source_url)
     return VoiceCloneProfileResponse(
         id=profile.id,
         user_id=profile.user_id,
         name=profile.name,
         description=profile.description,
-        source_audio_url=profile.source_audio_url,
+        source_audio_url=source_url,
         voice_id=profile.voice_id,
         voice_model=profile.voice_model,
         language=profile.language,
@@ -74,6 +77,7 @@ def create_voice_clone(
     request: CreateVoiceCloneRequest,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     workflow: VoiceCloneWorkflowService = Depends(_get_workflow_service),
+    sign_url=Depends(get_audio_url_signer),
 ) -> VoiceCloneProfileResponse:
     """创建音色克隆任务。
 
@@ -109,7 +113,7 @@ def create_voice_clone(
             except Exception as inner_e:
                 logger.error(f"Failed to mark profile as failed after dispatch error: {inner_e}")
 
-    return _to_response(profile)
+    return _to_response(profile, sign_url)
 
 
 @router.get("", response_model=ListVoiceCloneResponse)
@@ -119,13 +123,14 @@ def list_voice_clones(
     limit: int = Query(50, ge=1, le=200),
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     repository: SQLAlchemyVoiceCloneProfileRepository = Depends(get_voice_clone_profile_repository),
+    sign_url=Depends(get_audio_url_signer),
 ) -> ListVoiceCloneResponse:
     """获取用户的音色克隆列表。"""
     user_id = authenticated_user.user.id
     use_case = ListVoiceClonesUseCase(repository)
     items, total = use_case.execute(user_id, status=status_filter, skip=skip, limit=limit)
     return ListVoiceCloneResponse(
-        items=[_to_response(p) for p in items],
+        items=[_to_response(p, sign_url) for p in items],
         total=total,
     )
 
@@ -135,6 +140,7 @@ def get_voice_clone(
     clone_id: str,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     repository: SQLAlchemyVoiceCloneProfileRepository = Depends(get_voice_clone_profile_repository),
+    sign_url=Depends(get_audio_url_signer),
 ) -> VoiceCloneProfileResponse:
     """获取音色克隆详情。"""
     user_id = authenticated_user.user.id
@@ -143,7 +149,7 @@ def get_voice_clone(
         profile = use_case.execute(clone_id, user_id)
     except VoiceCloneNotFoundError:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Voice clone not found")
-    return _to_response(profile)
+    return _to_response(profile, sign_url)
 
 
 @router.get("/{clone_id}/status", response_model=VoiceCloneStatusResponse)
@@ -192,6 +198,7 @@ def retry_voice_clone(
     clone_id: str,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     workflow: VoiceCloneWorkflowService = Depends(_get_workflow_service),
+    sign_url=Depends(get_audio_url_signer),
 ) -> VoiceCloneProfileResponse:
     """重试失败的音色克隆。
 
@@ -224,4 +231,4 @@ def retry_voice_clone(
             except Exception as inner_e:
                 logger.error(f"Failed to mark profile as failed after dispatch error: {inner_e}")
 
-    return _to_response(profile)
+    return _to_response(profile, sign_url)

apps/api/app/api/routes/voices.py

@@ -8,7 +8,7 @@ from __future__ import annotations
 from typing import Literal, Optional
 
 from app.auth import AuthenticatedUser, get_current_user
-from app.dependencies import get_db_session, get_user_repository
+from app.dependencies import get_audio_url_signer, get_db_session, get_user_repository
 from app.schemas.voice import (
     PresetVoiceItemResponse,
     PresetVoiceListResponse,
@@ -50,7 +50,10 @@ def _get_clone_profile_repository(session: Session = Depends(get_db_session)) ->
     return SQLAlchemyVoiceCloneProfileRepository(session)
 
 
-def _to_response(item) -> VoiceLibraryItemResponse:
+def _to_response(item, sign_url=None) -> VoiceLibraryItemResponse:
+    audio = item.audio_url
+    if sign_url and audio:
+        audio = sign_url(audio)
     return VoiceLibraryItemResponse(
         id=item.id,
         user_id=item.user_id,
@@ -59,7 +62,7 @@ def _to_response(item) -> VoiceLibraryItemResponse:
         voice_provider=item.voice_provider,
         voice_id=item.voice_id,
         voice_name=item.voice_name,
-        audio_url=item.audio_url,
+        audio_url=audio,
         duration=item.duration,
         file_size=item.file_size,
         status=item.status,
@@ -70,16 +73,20 @@ def _to_response(item) -> VoiceLibraryItemResponse:
     )
 
 
-def _to_unified_response(item, profile_id_map: dict | None = None) -> UnifiedVoiceItemResponse:
+def _to_unified_response(item, profile_id_map: dict | None = None, sign_url=None) -> UnifiedVoiceItemResponse:
     """将数据库音色转换为统一响应格式。
 
     Args:
         item: VoiceLibraryItem
         profile_id_map: voice_id → profile_id 映射，用于填充 voice_clone_profile_id
+        sign_url: 音频URL预签名函数
     """
     profile_id = None
     if profile_id_map and item.voice_id:
         profile_id = profile_id_map.get(item.voice_id)
+    audio = item.audio_url
+    if sign_url and audio:
+        audio = sign_url(audio)
     return UnifiedVoiceItemResponse(
         id=item.id,
         type="clone",
@@ -89,7 +96,7 @@ def _to_unified_response(item, profile_id_map: dict | None = None) -> UnifiedVoi
         language="zh-CN",
         voice_id=item.voice_id,
         voice_provider=item.voice_provider or "cosyvoice",
-        audio_url=item.audio_url,
+        audio_url=audio,
         duration=item.duration,
         file_size=item.file_size,
         status=item.status,
@@ -140,6 +147,7 @@ def list_voices_unified(
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     voice_repository: SQLAlchemyVoiceLibraryRepository = Depends(_get_voice_repository),
     clone_profile_repository: SQLAlchemyVoiceCloneProfileRepository = Depends(_get_clone_profile_repository),
+    sign_url=Depends(get_audio_url_signer),
 ) -> UnifiedVoiceListResponse:
     """获取配音列表（预置音色 + 用户克隆音色）。
 
@@ -167,7 +175,7 @@ def list_voices_unified(
         # 批量查询 voice_id → profile_id 映射，填充 voice_clone_profile_id
         voice_ids = [i.voice_id for i in clone_items_raw if i.voice_id]
         profile_id_map = clone_profile_repository.find_profile_ids_by_voice_ids(voice_ids) if voice_ids else {}
-        clone_items = [_to_unified_response(i, profile_id_map) for i in clone_items_raw]
+        clone_items = [_to_unified_response(i, profile_id_map, sign_url) for i in clone_items_raw]
 
     # 组装结果
     if type == "preset":
@@ -224,6 +232,7 @@ def list_voices_legacy(
     limit: int = Query(50, ge=1, le=200),
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     voice_repository: SQLAlchemyVoiceLibraryRepository = Depends(_get_voice_repository),
+    sign_url=Depends(get_audio_url_signer),
 ) -> ListVoiceLibraryResponse:
     """原有配音列表接口（仅返回用户克隆音色）。
 
@@ -233,7 +242,7 @@ def list_voices_legacy(
     use_case = ListVoiceLibraryUseCase(voice_repository)
     items, total = use_case.execute(user_id, status=status_filter, skip=skip, limit=limit)
     return ListVoiceLibraryResponse(
-        items=[_to_response(i) for i in items],
+        items=[_to_response(i, sign_url) for i in items],
         total=total,
     )
 
@@ -243,13 +252,14 @@ def get_voice(
     voice_id: str,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     voice_repository: SQLAlchemyVoiceLibraryRepository = Depends(_get_voice_repository),
+    sign_url=Depends(get_audio_url_signer),
 ) -> VoiceLibraryItemResponse:
     user_id = authenticated_user.user.id
     use_case = GetVoiceLibraryUseCase(voice_repository)
     item = use_case.execute(voice_id, user_id)
     if item is None:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Voice not found")
-    return _to_response(item)
+    return _to_response(item, sign_url)
 
 
 @router.post("", response_model=VoiceLibraryItemResponse, status_code=status.HTTP_201_CREATED)
@@ -258,6 +268,7 @@ def create_voice(
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     voice_repository: SQLAlchemyVoiceLibraryRepository = Depends(_get_voice_repository),
     user_repository: UserRepository = Depends(get_user_repository),
+    sign_url=Depends(get_audio_url_signer),
 ) -> VoiceLibraryItemResponse:
     user_id = authenticated_user.user.id
     plan_name = _get_user_plan(user_id, user_repository)
@@ -283,7 +294,7 @@ def create_voice(
             status_code=status.HTTP_429_TOO_MANY_REQUESTS,
             detail=f"配音库配额已满（{exc.used}/{exc.limit}），请升级套餐",
         )
-    return _to_response(item)
+    return _to_response(item, sign_url)
 
 
 @router.put("/{voice_id}", response_model=VoiceLibraryItemResponse)
@@ -292,6 +303,7 @@ def update_voice(
     request: UpdateVoiceLibraryRequest,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     voice_repository: SQLAlchemyVoiceLibraryRepository = Depends(_get_voice_repository),
+    sign_url=Depends(get_audio_url_signer),
 ) -> VoiceLibraryItemResponse:
     user_id = authenticated_user.user.id
     command = UpdateVoiceLibraryCommand(
@@ -313,7 +325,7 @@ def update_voice(
         item = use_case.execute(command)
     except NotFoundError:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Voice not found")
-    return _to_response(item)
+    return _to_response(item, sign_url)
 
 
 @router.delete("/{voice_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None)

apps/api/app/core/task_enqueue.py

@@ -0,0 +1,58 @@
+import logging
+from typing import Any
+
+from app.core.celery_app import celery_app
+
+logger = logging.getLogger(__name__)
+
+
+def safe_enqueue_generation_task(
+    task: Any,
+    generation_task_repository: Any,
+    *,
+    log_prefix: str = "[任务队列]",
+    log_task_status: bool = False,
+) -> bool:
+    """安全入队：send_task 失败时自动把任务标记为 failed，避免留下 pending 僵尸任务。
+
+    Args:
+        task: 生成任务对象，需有 id 属性和 mark_failed 方法
+        generation_task_repository: 任务仓储，用于更新状态
+        log_prefix: 日志前缀，便于区分调用来源
+        log_task_status: 成功日志中是否额外打印任务状态
+
+    Returns:
+        True 表示入队成功，False 表示入队失败（已标记为 failed）
+    """
+    try:
+        celery_app.send_task("worker.generate_video", args=[task.id])
+        if log_task_status:
+            logger.info(
+                "%s 入队成功: task_id=%s, status=%s",
+                log_prefix,
+                task.id,
+                task.status,
+            )
+        else:
+            logger.info("%s 入队成功: task_id=%s", log_prefix, task.id)
+        return True
+    except Exception as e:
+        logger.error(
+            "%s 入队失败，标记为失败: task_id=%s error=%s",
+            log_prefix,
+            task.id,
+            e,
+            exc_info=True,
+        )
+        try:
+            task.mark_failed(f"任务入队失败: {e}")
+            generation_task_repository.update(task)
+        except Exception as update_err:
+            logger.error(
+                "%s 入队失败后更新状态也失败: task_id=%s error=%s",
+                log_prefix,
+                task.id,
+                update_err,
+                exc_info=True,
+            )
+        return False

apps/api/app/dependencies.py

@@ -201,7 +201,37 @@ def get_voice_clone_profile_repository(
 
 
 def get_cosyvoice_service():
-    """Provide the CosyVoice service instance."""
+    """Provide the CosyVoice service instance.
+
+    注入 OSS 音频URL预签名函数，确保私有bucket下的参考音频
+    能被 CosyVoice 服务器下载。
+    """
+    from app.core.storage import get_storage_service
+
     from packages.application.cosyvoice_service import CosyVoiceService
 
-    return CosyVoiceService()
+    storage = get_storage_service()
+
+    def _sign_audio_url(url: str) -> str:
+        """对音频URL做预签名，私有bucket下 CosyVoice 服务器才能下载."""
+        return storage.get_download_url(url, expires_seconds=86400)
+
+    return CosyVoiceService(audio_url_signer=_sign_audio_url)
+
+
+def get_audio_url_signer():
+    """提供音频URL预签名函数（24小时有效期）。
+
+    用于所有 API 返回给前端的音频 URL，确保私有 OSS bucket 下可正常访问。
+    空 URL、非 OSS URL 直接原样返回；签名失败时回退到原始 URL。
+    """
+    from app.core.storage import get_storage_service
+
+    storage = get_storage_service()
+
+    def sign_audio_url(url: str) -> str:
+        if not url:
+            return url
+        return storage.get_download_url(url, expires_seconds=86400)
+
+    return sign_audio_url

apps/api/app/schemas/generation_task.py

@@ -1,4 +1,6 @@
-from pydantic import BaseModel, Field, model_validator
+import json
+
+from pydantic import BaseModel, Field, field_validator, model_validator
 
 
 class CreateGenerationTaskRequest(BaseModel):
@@ -62,6 +64,21 @@ class GenerationTaskResponse(BaseModel):
     progress: float
     result_count: int
     error_message: str
+    logs: list[dict] = Field(default_factory=list)
+
+    @field_validator("logs", mode="before")
+    @classmethod
+    def _parse_logs(cls, v: object) -> list[dict]:
+        """将 JSON 字符串解析为 list[dict]。"""
+        if isinstance(v, str):
+            try:
+                parsed = json.loads(v)
+                return parsed if isinstance(parsed, list) else []
+            except (json.JSONDecodeError, TypeError):
+                return []
+        if isinstance(v, list):
+            return v
+        return []
 
 
 class BatchGenerationTaskResponse(BaseModel):

apps/worker/video_processing/unified_render_service.py

@@ -314,7 +314,9 @@ class UnifiedRenderService:
             # 有效时长 = min(指定时长, 实际时长)；若均未设置则跳过
             effective_duration = 0.0
             if clip.duration > 0:
-                effective_duration = min(clip.duration, clip.actual_duration) if clip.actual_duration > 0 else clip.duration
+                effective_duration = (
+                    min(clip.duration, clip.actual_duration) if clip.actual_duration > 0 else clip.duration
+                )
             elif clip.actual_duration > 0:
                 effective_duration = clip.actual_duration
 

apps/worker/worker_app/tasks/generation.py

@@ -13,10 +13,13 @@
 
 from __future__ import annotations
 
+import json
 import logging
 import os
 import tempfile
+import time
 from dataclasses import dataclass, field
+from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Optional
 
@@ -81,6 +84,26 @@ def _update_task_status(task_id: str, status_action: str, **kwargs) -> bool:
         return False
 
 
+# ── 日志持久化辅助 ────────────────────────────────────────────────────────────
+
+
+def _flush_logs(task_id: str, gen_task) -> None:
+    """将 gen_task.logs 持久化到 DB（独立 session，失败不抛异常）。"""
+    try:
+        session = SessionLocal()
+        try:
+            from packages.adapters.sqlalchemy_impl.models import GenerationTaskModel
+
+            model = session.query(GenerationTaskModel).filter(GenerationTaskModel.id == task_id).first()
+            if model:
+                model.logs = gen_task.logs
+                session.commit()
+        finally:
+            session.close()
+    except Exception:
+        logger.warning("[task_id=%s] 日志持久化失败", task_id, exc_info=True)
+
+
 # ── 共享工具模块导入 ──────────────────────────────────────────────────────────
 
 from video_processing.dedup_helpers import create_video_record_and_dedup
@@ -320,6 +343,8 @@ def _download_library_assets(
     asset_ids: list[str] | None = None,
     video_extensions: tuple = (".mp4", ".mov", ".avi", ".mkv", ".webm"),
     strict: bool = True,
+    task_id: str = "",
+    gen_task=None,
 ) -> list[Path]:
     """下载视频素材 — 同时支持素材库模式和项目级模式。
 
@@ -419,19 +444,65 @@ def _download_library_assets(
                 storage_key = asset.file_url if asset.file_url else None
                 if not storage_key:
                     failed_assets.append(f"{asset.name}({asset.id})")
-                    logger.warning("素材缺少 file_url, 跳过: asset_id=%s name=%s", asset.id, asset.name)
+                    logger.warning(
+                        "[task_id=%s] 素材缺少 file_url, 跳过: asset_id=%s name=%s", task_id, asset.id, asset.name
+                    )
+                    if gen_task:
+                        gen_task.append_log(
+                            "下载素材",
+                            "素材缺少file_url, 跳过",
+                            level="WARN",
+                            asset_id=asset.id,
+                            asset_name=asset.name,
+                            success=False,
+                            file_size=0,
+                            duration=0.0,
+                        )
                     if strict:
                         raise RuntimeError(f"素材缺少 file_url: asset_id={asset.id}, name={asset.name}")
                     continue
 
                 ext = Path(storage_key).suffix or ".mp4"
                 local_file = temp_path / f"asset_{i:03d}_{asset.id}{ext}"
-                if download_asset(storage_key, local_file):
+                asset_start = time.monotonic()
+                download_ok = download_asset(storage_key, local_file)
+                asset_elapsed = time.monotonic() - asset_start
+
+                if download_ok:
+                    file_size = local_file.stat().st_size if local_file.exists() else 0
                     downloaded.append(local_file)
-                    logger.info("Downloaded asset: %s -> %s", asset.name, local_file)
+                    logger.info(
+                        "[task_id=%s] Downloaded asset: %s -> %s (size=%d, time=%.1fs)",
+                        task_id,
+                        asset.name,
+                        local_file,
+                        file_size,
+                        asset_elapsed,
+                    )
+                    if gen_task:
+                        gen_task.append_log(
+                            "下载素材",
+                            f"下载成功: {asset.name}",
+                            asset_id=asset.id,
+                            asset_name=asset.name,
+                            success=True,
+                            file_size=file_size,
+                            duration=round(asset_elapsed, 2),
+                        )
                 else:
                     failed_assets.append(f"{asset.name}({asset.id})")
-                    logger.warning("Failed to download asset: %s (id=%s)", asset.name, asset.id)
+                    logger.warning("[task_id=%s] Failed to download asset: %s (id=%s)", task_id, asset.name, asset.id)
+                    if gen_task:
+                        gen_task.append_log(
+                            "下载素材",
+                            f"下载失败: {asset.name}",
+                            level="WARN",
+                            asset_id=asset.id,
+                            asset_name=asset.name,
+                            success=False,
+                            file_size=0,
+                            duration=round(asset_elapsed, 2),
+                        )
                     if strict:
                         raise RuntimeError(f"素材下载失败: asset_id={asset.id}, name={asset.name}")
 
@@ -511,7 +582,7 @@ def generate_video(self, task_id: str) -> dict:
     """
     from packages.domain import EditingMode
 
-    logger.info("开始生成视频任务: task_id=%s", task_id)
+    logger.info("[task_id=%s] [接收任务] 开始生成视频任务", task_id)
 
     # 从数据库加载任务信息
     session = SessionLocal()
@@ -523,7 +594,7 @@ def generate_video(self, task_id: str) -> dict:
         task_repo = SQLAlchemyGenerationTaskRepository(session)
         gen_task = task_repo.get(task_id)
         if gen_task is None:
-            logger.error("生成任务不存在: task_id=%s", task_id)
+            logger.error("[task_id=%s] [接收任务] 任务不存在", task_id)
             return {"status": "failed", "error": f"generation task {task_id} not found"}
         project_id = gen_task.project_id
         asset_library_id = gen_task.asset_library_id
@@ -532,6 +603,16 @@ def generate_video(self, task_id: str) -> dict:
         mode = gen_task.strategy_id or "one_take"
         task_asset_ids = list(gen_task.asset_ids or [])
         batch_id = getattr(gen_task, "batch_id", "") or ""
+
+        # 记录接收任务日志
+        gen_task.append_log(
+            "接收任务",
+            f"模式={mode}, 模板={template_id}, 素材数={len(task_asset_ids)}",
+            mode=mode,
+            template_id=template_id,
+            asset_count=len(task_asset_ids),
+        )
+        _flush_logs(task_id, gen_task)
     finally:
         session.close()
 
@@ -558,12 +639,40 @@ def generate_video(self, task_id: str) -> dict:
             output_path = temp_path / output_name
 
             # 1. 从素材库/项目下载视频素材
+            logger.info("[task_id=%s] [下载素材] 开始下载视频素材", task_id)
+            download_start = time.monotonic()
             downloaded_videos = _download_library_assets(
                 temp_path,
                 asset_library_id=asset_library_id,
                 project_id=project_id,
                 asset_ids=task_asset_ids or None,
+                task_id=task_id,
+                gen_task=gen_task,
             )
+            download_elapsed = time.monotonic() - download_start
+            logger.info(
+                "[task_id=%s] [下载素材] 完成: 成功=%d个, 耗时=%.1fs",
+                task_id,
+                len(downloaded_videos),
+                download_elapsed,
+            )
+
+            # 重新加载 gen_task 以追加日志（session 已关闭）
+            _session = SessionLocal()
+            try:
+                _repo = SQLAlchemyGenerationTaskRepository(_session)
+                gen_task = _repo.get(task_id)
+            finally:
+                _session.close()
+
+            if gen_task:
+                gen_task.append_log(
+                    "下载素材",
+                    f"成功下载 {len(downloaded_videos)} 个视频素材",
+                    count=len(downloaded_videos),
+                    duration=round(download_elapsed, 2),
+                )
+                _flush_logs(task_id, gen_task)
 
             # 2. 下载配音（如有）
             audio_path: str | None = None
@@ -571,6 +680,7 @@ def generate_video(self, task_id: str) -> dict:
                 local_audio = temp_path / "voice.mp3"
                 if _download_voice_asset(voice_library_id, local_audio):
                     audio_path = str(local_audio)
+                    logger.info("[task_id=%s] [下载配音] 配音下载成功", task_id)
 
             # 3. 渲染
             if not downloaded_videos:
@@ -588,7 +698,26 @@ def generate_video(self, task_id: str) -> dict:
                 mode=editing_mode.value,
             )
 
+            total_duration = sum(c.duration for c in virtual_clips)
+            logger.info(
+                "[task_id=%s] [剪辑计划] 片段数=%d, 总时长=%.1fs",
+                task_id,
+                len(virtual_clips),
+                total_duration,
+            )
+
+            if gen_task:
+                gen_task.append_log(
+                    "剪辑计划",
+                    f"片段数={len(virtual_clips)}, 总时长={total_duration:.1f}s",
+                    segment_count=len(virtual_clips),
+                    total_duration=round(total_duration, 2),
+                )
+                _flush_logs(task_id, gen_task)
+
             # 使用 UnifiedRenderService 渲染
+            logger.info("[task_id=%s] [渲染] FFmpeg 渲染开始", task_id)
+            render_start = time.monotonic()
             render_service = UnifiedRenderService(
                 plan=virtual_plan,
                 clips=virtual_clips,
@@ -599,6 +728,20 @@ def generate_video(self, task_id: str) -> dict:
                 output_fps=int(OUTPUT_FPS),
             )
             render_result = render_service.render()
+            render_elapsed = time.monotonic() - render_start
+            logger.info(
+                "[task_id=%s] [渲染] FFmpeg 渲染完成: 耗时=%.1fs",
+                task_id,
+                render_elapsed,
+            )
+
+            if gen_task:
+                gen_task.append_log(
+                    "渲染",
+                    f"FFmpeg 渲染完成, 耗时={render_elapsed:.1f}s",
+                    duration=round(render_elapsed, 2),
+                )
+                _flush_logs(task_id, gen_task)
 
             # 4. 如有配音，后处理混音
             if audio_path:
@@ -608,7 +751,7 @@ def generate_video(self, task_id: str) -> dict:
                     # 混音成功，使用混音后的文件
                     output_path = final_path
                 except Exception as mux_err:
-                    logger.warning("音频混合失败，使用无音频版本: %s", mux_err)
+                    logger.warning("[task_id=%s] [混音] 音频混合失败，使用无音频版本: %s", task_id, mux_err)
                     output_path = render_result.output_path
             else:
                 output_path = render_result.output_path
@@ -617,9 +760,15 @@ def generate_video(self, task_id: str) -> dict:
             duration = probe_duration(output_path)
 
             # 5. 上传到 OSS — 失败必须抛异常，不能静默忽略
+            logger.info("[task_id=%s] [OSS上传] 开始上传: size=%d", task_id, file_size)
+            upload_start = time.monotonic()
             file_url = upload_to_oss(output_path, storage_key)
+            upload_elapsed = time.monotonic() - upload_start
             if not file_url:
                 # OSS 未配置或上传失败
+                if gen_task:
+                    gen_task.append_log("OSS上传", "上传失败", level="ERROR")
+                    _flush_logs(task_id, gen_task)
                 raise RuntimeError(
                     f"OSS 上传失败: task_id={task_id}, storage_key={storage_key}, " f"output_path={output_path}"
                 )
@@ -629,21 +778,39 @@ def generate_video(self, task_id: str) -> dict:
             verify_url = get_signed_download_url(file_url, expires_seconds=300) or file_url
             if not _verify_url_accessible(verify_url):
                 # 预签名 URL 也访问失败时，退一步用 object_exists 确认上传成功
-                from video_processing.oss_helpers import oss_bucket, normalize_storage_key
+                from video_processing.oss_helpers import normalize_storage_key, oss_bucket
 
                 bucket = oss_bucket()
                 key = normalize_storage_key(file_url)
                 if bucket and bucket.object_exists(key):
-                    logger.info(
-                        "URL 校验失败但 object_exists 确认文件存在，视为上传成功: storage_key=%s", key
-                    )
+                    logger.info("URL 校验失败但 object_exists 确认文件存在，视为上传成功: storage_key=%s", key)
+                    if gen_task:
+                        gen_task.append_log("OSS上传", "URL校验降级: object_exists确认存在", level="WARN")
                 else:
+                    if gen_task:
+                        gen_task.append_log("OSS上传", "上传后URL不可访问", level="ERROR", file_url=file_url)
+                        _flush_logs(task_id, gen_task)
                     raise RuntimeError(
                         f"OSS 上传后 URL 不可访问且 object_exists 失败: file_url={file_url}, "
                         f"storage_key={storage_key}"
                     )
 
-            logger.info("OSS 上传成功: file_url=%s", file_url)
+            logger.info(
+                "[task_id=%s] [OSS上传] 成功: 耗时=%.1fs, file_url=%s",
+                task_id,
+                upload_elapsed,
+                file_url,
+            )
+
+            if gen_task:
+                gen_task.append_log(
+                    "OSS上传",
+                    f"上传成功, 大小={file_size}, 耗时={upload_elapsed:.1f}s",
+                    file_size=file_size,
+                    duration=round(upload_elapsed, 2),
+                    file_url=file_url,
+                )
+                _flush_logs(task_id, gen_task)
 
             # 6. 创建 GeneratedVideo 记录 + 查重
             dedup_session = SessionLocal()
@@ -665,7 +832,23 @@ def generate_video(self, task_id: str) -> dict:
             # 7. 标记任务为 completed
             _update_task_status(task_id, "mark_completed", result_count=video_count or 1)
 
-            logger.info("视频生成完成: task_id=%s duration=%.2fs file_size=%d", task_id, duration, file_size)
+            # 记录完成日志
+            if gen_task:
+                gen_task.append_log(
+                    "任务完成",
+                    f"视频生成完成: 时长={duration:.2f}s, 大小={file_size}",
+                    duration=round(duration, 2),
+                    file_size=file_size,
+                    video_count=video_count or 1,
+                )
+                _flush_logs(task_id, gen_task)
+
+            logger.info(
+                "[task_id=%s] [任务完成] duration=%.2fs file_size=%d",
+                task_id,
+                duration,
+                file_size,
+            )
 
             return {
                 "status": "completed",
@@ -678,7 +861,27 @@ def generate_video(self, task_id: str) -> dict:
                 "mode": editing_mode.value,
             }
     except Exception as error:
-        logger.error("Video generation failed: %s", error, exc_info=True)
+        logger.error("[task_id=%s] [任务失败] %s", task_id, error, exc_info=True)
+
+        # 记录失败日志
+        try:
+            _session = SessionLocal()
+            try:
+                _repo = SQLAlchemyGenerationTaskRepository(_session)
+                gen_task = _repo.get(task_id)
+                if gen_task:
+                    gen_task.append_log(
+                        "任务失败",
+                        str(error),
+                        level="ERROR",
+                        error_type=type(error).__name__,
+                    )
+                    _flush_logs(task_id, gen_task)
+            finally:
+                _session.close()
+        except Exception:
+            logger.warning("[task_id=%s] 记录失败日志异常", task_id, exc_info=True)
+
         _update_task_status(task_id, "mark_failed", error_message=str(error))
         return {
             "status": "failed",

apps/worker/worker_app/tasks/voice_clone.py

@@ -4,6 +4,7 @@ import logging
 
 from celery import Task
 from celery.exceptions import Retry
+from video_processing.oss_helpers import get_signed_download_url
 from worker_app.celery_app import celery_app
 from worker_app.db import SessionLocal
 
@@ -48,7 +49,9 @@ def process_voice_clone(self: Task, profile_id: str) -> dict:
         repo = SQLAlchemyVoiceCloneProfileRepository(session)
         workflow = VoiceCloneWorkflowService(
             repository=repo,
-            cosyvoice_service=CosyVoiceService(),
+            cosyvoice_service=CosyVoiceService(
+                audio_url_signer=lambda url: get_signed_download_url(url, expires_seconds=86400) or url
+            ),
         )
 
         updated_profile = workflow.poll_and_process_clone(profile_id, timeout=300)

docs/schema-metadata-snapshot.json

@@ -1549,6 +1549,14 @@
           "type": "JSON",
           "unique": false
         },
+        {
+          "index": false,
+          "name": "logs",
+          "nullable": false,
+          "primary_key": false,
+          "type": "TEXT",
+          "unique": false
+        },
         {
           "index": false,
           "name": "created_at",

packages/adapters/sqlalchemy_impl/generation_task_repository.py

@@ -27,6 +27,7 @@ def _to_domain(model: GenerationTaskModel) -> GenerationTask:
         source_edit_plan_id=model.source_edit_plan_id or "",
         asset_select_mode=model.asset_select_mode or "",
         batch_id=model.batch_id or "",
+        logs=model.logs or "[]",
         created_at=model.created_at,
     )
 
@@ -56,6 +57,7 @@ class SQLAlchemyGenerationTaskRepository:
             source_edit_plan_id=task.source_edit_plan_id or None,
             asset_select_mode=task.asset_select_mode or "",
             batch_id=task.batch_id or "",
+            logs=task.logs,
             created_at=task.created_at,
         )
         self.session.add(model)
@@ -129,5 +131,6 @@ class SQLAlchemyGenerationTaskRepository:
         model.source_edit_plan_id = task.source_edit_plan_id or None
         model.asset_select_mode = task.asset_select_mode or ""
         model.batch_id = task.batch_id or ""
+        model.logs = task.logs
         self.session.commit()
         return task

packages/adapters/sqlalchemy_impl/models.py

@@ -257,6 +257,7 @@ class GenerationTaskModel(Base):
     asset_select_mode = Column(String(20), nullable=False, default="")
     batch_id = Column(String(36), nullable=False, default="", index=True)
     extra_meta = Column("metadata", JSON, nullable=False, default=dict)
+    logs = Column(Text, nullable=False, default="[]", server_default="[]")
     created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
 
 

packages/application/cosyvoice_service.py

@@ -1,11 +1,13 @@
-"""CosyVoice 语音服务 — Phase 3.
+"""CosyVoice 语音服务 — 适配阿里云百炼 DashScope API.
 
-封装阿里云 CosyVoice 语音合成 API，提供：
+封装阿里云百炼 CosyVoice 语音合成 API，提供：
 - 预置音色列表查询
-- 音色克隆（提交任务 + 轮询状态）
-- 语音合成（提交任务 + 轮询状态）
+- 音色克隆（提交 + 轮询状态）
+- 语音合成（同步非流式调用）
 
-API 文档: https://help.aliyun.com/zh/model-studio/cosyvoice
+API 文档:
+- 音色克隆: https://help.aliyun.com/document_detail/3027318.html
+- 语音合成: https://help.aliyun.com/zh/model-studio/cosyvoice-tts-http-api
 """
 
 from __future__ import annotations
@@ -60,33 +62,34 @@ class SynthesizeResult:
 
 
 class CosyVoiceService:
-    """CosyVoice 语音服务。
+    """CosyVoice 语音服务.
 
-    封装阿里云 CosyVoice API，提供音色克隆和语音合成功能。
-    支持同步和异步两种模式：
-    - 同步：API 直接返回结果
-    - 异步：API 返回 task_id，需要轮询状态
+    封装阿里云百炼 CosyVoice API，提供音色克隆和语音合成功能.
+
+    接口总览:
+    - 音色克隆: 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)
+      - 非流式: 同步返回音频 URL
 
     使用示例:
         service = CosyVoiceService(
             api_key="your-api-key",
-            base_url="https://dashscope.aliyuncs.com/api/v1/services/aigc/text2audio",
-            model="cosyvoice-v1",
+            base_url="https://dashscope.aliyuncs.com/api/v1",
+            model="cosyvoice-v3-flash",
         )
 
-        # 获取预置音色
-        voices = service.list_preset_voices()
-
         # 音色克隆
         result = service.clone_voice(audio_url="https://example.com/audio.mp3")
 
         # 语音合成
-        result = service.synthesize_speech(text="你好世界", voice_id="longxiaochun")
+        result = service.synthesize_speech(text="你好世界", voice_id="longxiaochun_v3")
     """
 
-    # 轮询配置
-    POLL_INTERVAL = 2.0  # 秒
-    MAX_POLL_ATTEMPTS = 60  # 最多轮询 60 次（2分钟）
+    # 音色状态轮询配置
+    CLONE_POLL_INTERVAL = 5.0  # 秒
+    CLONE_MAX_POLL_ATTEMPTS = 60  # 最多轮询 60 次（5分钟）
 
     # 重试配置
     MAX_RETRIES = 3
@@ -97,27 +100,70 @@ class CosyVoiceService:
         api_key: str = "",
         base_url: str = "",
         model: str = "",
+        clone_model: str = "",
         http_client: Optional[httpx.Client] = None,
+        audio_url_signer: Optional[callable] = None,
     ) -> None:
-        """初始化 CosyVoice 服务。
+        """初始化 CosyVoice 服务.
 
         Args:
-            api_key: CosyVoice API Key，为空时从配置读取
-            base_url: CosyVoice API Base URL，为空时从配置读取
-            model: CosyVoice 模型名称，为空时从配置读取
+            api_key: DashScope API Key，为空时从配置读取
+            base_url: DashScope API Base URL，为空时从配置读取
+            model: 语音合成模型名称，为空时从配置读取
+            clone_model: 音色克隆模型名称，为空时从配置读取
             http_client: 可选的 HTTP 客户端（用于测试注入）
+            audio_url_signer: 可选的音频URL预签名函数，签名式 fn(url) -> str.
+                用于私有 bucket 下，将裸 URL 转为预签名 URL，
+                确保 CosyVoice 服务器能下载参考音频.
         """
         settings = get_shared_settings()
 
         self._api_key = api_key or settings.cosyvoice_api_key
         self._base_url = base_url or settings.cosyvoice_base_url
         self._model = model or settings.cosyvoice_model
+        self._clone_model = clone_model or getattr(settings, "cosyvoice_clone_model", "voice-enrollment")
+        self._audio_url_signer = audio_url_signer
+
+        # base_url 规范化：去掉末尾的路径残留（兼容旧版配置）
+        # 旧版 .env 模板中 base_url 包含 /services/aigc/text2audio 完整路径，
+        # 新版只需 /api/v1，具体路径由代码拼接。这里自动修正，避免配置滞后导致418。
+        if "/services/aigc/text2audio" in self._base_url:
+            old_url = self._base_url
+            # 截取到 /api/v1 为止
+            idx = self._base_url.find("/api/v1")
+            if idx >= 0:
+                self._base_url = self._base_url[: idx + len("/api/v1")]
+            logger.warning(
+                "[CosyVoice Config] base_url包含旧版text2audio路径，已自动修正: " "%s -> %s",
+                old_url,
+                self._base_url,
+            )
 
         self._client = http_client or httpx.Client(
-            timeout=httpx.Timeout(30.0, connect=10.0),
+            timeout=httpx.Timeout(60.0, connect=10.0),
         )
         self._owns_client = http_client is None
 
+        # 启动时打印配置（脱敏），方便排查环境变量覆盖问题
+        if self._owns_client:
+            masked_key = ""
+            if self._api_key:
+                if len(self._api_key) > 8:
+                    masked_key = f"{self._api_key[:4]}...{self._api_key[-4:]}"
+                else:
+                    masked_key = "***"
+            logger.info(
+                "[CosyVoice Config] 初始化配置: "
+                "model=%s, base_url=%s, default_voice=%s, "
+                "sample_rate=%d, format=%s, api_key=%s",
+                self._model,
+                self._base_url,
+                getattr(settings, "cosyvoice_voice", "(unset)"),
+                settings.cosyvoice_sample_rate,
+                settings.cosyvoice_format,
+                masked_key or "(empty)",
+            )
+
     def __enter__(self) -> CosyVoiceService:
         return self
 
@@ -132,7 +178,7 @@ class CosyVoiceService:
     # ── 预置音色 ─────────────────────────────────────────
 
     def list_preset_voices(self) -> list[PresetVoice]:
-        """获取预置音色列表。
+        """获取预置音色列表.
 
         Returns:
             预置音色列表
@@ -146,20 +192,22 @@ class CosyVoiceService:
         audio_url: str,
         voice_name: str = "",
         language: str = "zh-CN",
+        target_model: str = "",
     ) -> dict:
-        """提交音色克隆任务（非阻塞）。
+        """提交音色克隆任务（非阻塞）.
 
-        只提交任务到 CosyVoice API，不轮询结果。
-        返回的 dict 包含 task_id（异步）或 voice_id（同步）。
+        调用百炼 voice-enrollment API 创建克隆音色.
+        创建后音色状态为 DEPLOYING，需通过 query_voice_status 轮询直到 OK.
 
         Args:
-            audio_url: 参考音频 URL
-            voice_name: 音色名称（可选）
-            language: 语言代码
+            audio_url: 参考音频 URL（必须公网可访问）
+            voice_name: 音色名称前缀（字母数字，最多10字符）
+            language: 语言代码（zh-CN 会转换为 zh）
+            target_model: 目标合成模型，默认使用当前 model
 
         Returns:
-            dict: {"task_id": str, "voice_id": str, "request_id": str}
-                  task_id 和 voice_id 至少有一个非空
+            dict: {"voice_id": str, "status": str, "request_id": str}
+                  voice_id 非空，status 通常为 DEPLOYING
 
         Raises:
             CosyVoiceError: API 调用失败
@@ -171,48 +219,67 @@ class CosyVoiceService:
         if not self._api_key:
             raise CosyVoiceAuthError("CosyVoice API Key 未配置")
 
+        # voice_name 作为 prefix，限制字母数字，最多10字符
+        # 不符合要求的做清洗
+        prefix = self._sanitize_prefix(voice_name) if voice_name else "clone"
+
+        # 语言转换：zh-CN → zh，保留 ISO 639-1 格式
+        lang_code = language.split("-")[0].lower() if language else "zh"
+
+        target = target_model or self._model
+
+        # 如果配置了 audio_url_signer，对音频URL做预签名
+        # （私有 bucket 下 CosyVoice 服务器无法直接访问裸 URL）
+        signed_audio_url = audio_url
+        if self._audio_url_signer:
+            try:
+                signed_audio_url = self._audio_url_signer(audio_url)
+                logger.info("音频URL已预签名: original=%s signed_prefix=%s", audio_url[:80], signed_audio_url[:80])
+            except Exception as e:
+                logger.warning("音频URL预签名失败，使用原始URL: %s", e)
+
         payload = {
-            "model": self._model,
+            "model": self._clone_model,
             "input": {
-                "audio_url": audio_url,
-            },
-            "parameters": {
-                "language": language,
+                "action": "create_voice",
+                "target_model": target,
+                "prefix": prefix,
+                "url": signed_audio_url,
+                "language_hints": [lang_code],
             },
         }
-        if voice_name:
-            payload["parameters"]["voice_name"] = voice_name
 
         response = self._call_api(
             method="POST",
-            path="/services/audio/voice-clone",
+            path="/services/audio/tts/customization",
             json=payload,
             timeout=60.0,
         )
 
         output = response.get("output", {})
-        task_id = output.get("task_id", "")
         voice_id = output.get("voice_id", "")
+        status = output.get("status", "DEPLOYING")
         request_id = response.get("request_id", "")
 
-        if not task_id and not voice_id:
-            raise CosyVoiceError(f"CosyVoice API 未返回 task_id 或 voice_id: {response}")
+        if not voice_id:
+            raise CosyVoiceError(f"CosyVoice API 未返回 voice_id: {response}")
 
         return {
-            "task_id": task_id,
             "voice_id": voice_id,
+            "status": status,
             "request_id": request_id,
         }
 
-    def check_task_status(self, task_id: str) -> dict:
-        """查询克隆任务状态（单次查询，不轮询）。
+    def query_voice_status(self, voice_id: str) -> dict:
+        """查询音色状态（单次查询，不轮询）.
 
         Args:
-            task_id: 任务 ID
+            voice_id: 音色 ID
 
         Returns:
-            dict: {"status": str, "voice_id": str, "message": str}
-                  status 为 SUCCEEDED/FAILED/PENDING/RUNNING
+            dict: {"status": str, "target_model": str, "gmt_create": str,
+                   "gmt_modified": str, "resource_link": str}
+                  status 为 DEPLOYING / OK / UNDEPLOYED
 
         Raises:
             CosyVoiceError: API 调用失败
@@ -221,40 +288,92 @@ class CosyVoiceService:
         if not self._api_key:
             raise CosyVoiceAuthError("CosyVoice API Key 未配置")
 
+        if not voice_id:
+            raise ValueError("voice_id 不能为空")
+
+        payload = {
+            "model": self._clone_model,
+            "input": {
+                "action": "query_voice",
+                "voice_id": voice_id,
+            },
+        }
+
         response = self._call_api(
-            method="GET",
-            path=f"/tasks/{task_id}",
+            method="POST",
+            path="/services/audio/tts/customization",
+            json=payload,
             timeout=30.0,
         )
 
         output = response.get("output", {})
-        status = output.get("task_status", "").upper()
-        voice_id = output.get("voice_id", "")
-        message = output.get("message", "")
-
         return {
-            "status": status,
-            "voice_id": voice_id,
-            "message": message,
+            "status": output.get("status", ""),
+            "target_model": output.get("target_model", ""),
+            "gmt_create": output.get("gmt_create", ""),
+            "gmt_modified": output.get("gmt_modified", ""),
+            "resource_link": output.get("resource_link", ""),
         }
 
-    def poll_clone_task(self, task_id: str, timeout: float = 300.0) -> dict:
-        """轮询音色克隆任务状态（公开方法）。
+    def check_task_status(self, task_id: str) -> dict:
+        """查询克隆任务状态（兼容旧接口，实际用 voice_id 查询）.
 
-        供 Celery 后台任务调用，轮询直到完成或超时。
+        为了兼容旧代码，task_id 参数名保留，但实际传的是 voice_id.
 
         Args:
-            task_id: CosyVoice 任务 ID
+            task_id: 音色 ID（兼容旧接口名）
+
+        Returns:
+            dict: {"status": str, "voice_id": str, "message": str}
+        """
+        result = self.query_voice_status(task_id)
+        return {
+            "status": result["status"],
+            "voice_id": task_id,
+            "message": "",
+        }
+
+    def poll_clone_task(self, voice_id: str, timeout: float = 300.0) -> dict:
+        """轮询音色克隆状态直到完成或超时.
+
+        供 Celery 后台任务调用，轮询直到状态变为 OK 或 UNDEPLOYED.
+
+        Args:
+            voice_id: 音色 ID
             timeout: 超时时间（秒），默认 300
 
         Returns:
             dict: {"voice_id": str}
 
         Raises:
-            CosyVoiceError: 任务失败
+            CosyVoiceError: 任务失败（状态 UNDEPLOYED）
             CosyVoiceTimeoutError: 超时
         """
-        return self._poll_clone_task(task_id, timeout=timeout)
+        start_time = time.time()
+        attempts = 0
+
+        while attempts < self.CLONE_MAX_POLL_ATTEMPTS:
+            elapsed = time.time() - start_time
+            if elapsed > timeout:
+                raise CosyVoiceTimeoutError(f"音色克隆任务超时（{timeout}秒）: voice_id={voice_id}")
+
+            result = self.query_voice_status(voice_id)
+            status = result.get("status", "").upper()
+
+            if status == "OK":
+                return {"voice_id": voice_id}
+            elif status == "UNDEPLOYED":
+                raise CosyVoiceError(f"音色克隆任务失败（审核未通过）: voice_id={voice_id}")
+            elif status in ("DEPLOYING", "PENDING", "PROCESSING", ""):
+                # 继续轮询
+                time.sleep(self.CLONE_POLL_INTERVAL)
+                attempts += 1
+            else:
+                logger.warning("未知的音色状态: %s (voice_id=%s)", status, voice_id)
+                time.sleep(self.CLONE_POLL_INTERVAL)
+                attempts += 1
+
+        raise CosyVoiceTimeoutError(f"音色克隆任务轮询次数超限: voice_id={voice_id}")
 
     def clone_voice(
         self,
@@ -262,122 +381,45 @@ class CosyVoiceService:
         voice_name: str = "",
         language: str = "zh-CN",
         timeout: float = 300.0,
+        target_model: str = "",
     ) -> CloneResult:
-        """克隆音色。
+        """克隆音色（阻塞，直到完成或超时）.
 
-        提交音色克隆任务到 CosyVoice API，并轮询直到完成或超时。
+        提交音色克隆到百炼 API，并轮询直到状态变为 OK 或超时.
 
         Args:
-            audio_url: 参考音频 URL
-            voice_name: 音色名称（可选）
+            audio_url: 参考音频 URL（必须公网可访问）
+            voice_name: 音色名称前缀
             language: 语言代码
             timeout: 超时时间（秒）
+            target_model: 目标合成模型
 
         Returns:
             CloneResult: 克隆结果，包含 voice_id
 
         Raises:
-            CosyVoiceError: API 调用失败
+            CosyVoiceError: API 调用失败或克隆失败
             CosyVoiceTimeoutError: 超时
             CosyVoiceAuthError: 认证失败
             ValueError: 参数无效
         """
-        if not audio_url:
-            raise ValueError("audio_url 不能为空")
-        if not self._api_key:
-            raise CosyVoiceAuthError("CosyVoice API Key 未配置")
-
-        # 构建请求
-        payload = {
-            "model": self._model,
-            "input": {
-                "audio_url": audio_url,
-            },
-            "parameters": {
-                "language": language,
-            },
-        }
-        if voice_name:
-            payload["parameters"]["voice_name"] = voice_name
-
-        # 调用 API
-        response = self._call_api(
-            method="POST",
-            path="/services/audio/voice-clone",
-            json=payload,
-            timeout=timeout,
+        submit_result = self.submit_clone_task(
+            audio_url=audio_url,
+            voice_name=voice_name,
+            language=language,
+            target_model=target_model,
         )
 
-        # 解析响应
-        output = response.get("output", {})
+        voice_id = submit_result["voice_id"]
+        request_id = submit_result["request_id"]
 
-        # 检查是否有 task_id（异步模式）
-        task_id = output.get("task_id")
-        voice_id = output.get("voice_id")
+        # 如果创建时已经是 OK 状态，直接返回
+        if submit_result.get("status", "").upper() == "OK":
+            return CloneResult(voice_id=voice_id, request_id=request_id)
 
-        if task_id:
-            # 异步模式：轮询任务状态
-            result = self._poll_clone_task(task_id, timeout)
-            return CloneResult(
-                voice_id=result["voice_id"],
-                request_id=response.get("request_id", ""),
-            )
-        elif voice_id:
-            # 同步模式：直接返回结果
-            return CloneResult(
-                voice_id=voice_id,
-                request_id=response.get("request_id", ""),
-            )
-        else:
-            raise CosyVoiceError(f"CosyVoice API 未返回 task_id 或 voice_id: {response}")
-
-    def _poll_clone_task(self, task_id: str, timeout: float) -> dict:
-        """轮询音色克隆任务状态。
-
-        Args:
-            task_id: 任务 ID
-            timeout: 超时时间（秒）
-
-        Returns:
-            任务结果字典
-
-        Raises:
-            CosyVoiceError: 任务失败
-            CosyVoiceTimeoutError: 超时
-        """
-        start_time = time.time()
-        attempts = 0
-
-        while attempts < self.MAX_POLL_ATTEMPTS:
-            elapsed = time.time() - start_time
-            if elapsed > timeout:
-                raise CosyVoiceTimeoutError(f"音色克隆任务超时（{timeout}秒）: task_id={task_id}")
-
-            response = self._call_api(
-                method="GET",
-                path=f"/tasks/{task_id}",
-                timeout=30.0,
-            )
-
-            output = response.get("output", {})
-            status = output.get("task_status", "").upper()
-
-            if status == "SUCCEEDED":
-                voice_id = output.get("voice_id", "")
-                if not voice_id:
-                    raise CosyVoiceError(f"音色克隆任务成功但未返回 voice_id: {response}")
-                return {"voice_id": voice_id}
-            elif status == "FAILED":
-                error_msg = output.get("message", "未知错误")
-                raise CosyVoiceError(f"音色克隆任务失败: {error_msg}")
-            elif status in ("PENDING", "RUNNING"):
-                # 继续轮询
-                time.sleep(self.POLL_INTERVAL)
-                attempts += 1
-            else:
-                raise CosyVoiceError(f"未知的任务状态: {status}")
-
-        raise CosyVoiceTimeoutError(f"音色克隆任务轮询次数超限: task_id={task_id}")
+        # 否则轮询
+        result = self.poll_clone_task(voice_id, timeout=timeout)
+        return CloneResult(voice_id=result["voice_id"], request_id=request_id)
 
     # ── 语音合成 ─────────────────────────────────────────
 
@@ -388,11 +430,12 @@ class CosyVoiceService:
         sample_rate: int = 0,
         format: str = "",
         speed: float = 1.0,
+        volume: int = 50,
     ) -> dict:
-        """提交语音合成任务（非阻塞）。
+        """提交语音合成任务（同步非流式，直接返回结果）.
 
-        只提交任务到 CosyVoice API，不轮询结果。
-        返回的 dict 包含 task_id（异步）或 audio_url（同步）。
+        CosyVoice SpeechSynthesizer 非流式接口是同步的，
+        调用后直接返回音频 URL. 此方法保持与旧接口兼容.
 
         Args:
             text: 要合成的文本
@@ -400,10 +443,11 @@ class CosyVoiceService:
             sample_rate: 采样率（Hz），0 表示使用配置默认值
             format: 输出格式（mp3/wav/pcm），空表示使用配置默认值
             speed: 语速（0.5-2.0），1.0 为正常速度
+            volume: 音量（0-100），默认 50
 
         Returns:
-            dict: {"task_id": str, "audio_url": str, "request_id": str}
-                  task_id 和 audio_url 至少有一个非空
+            dict: {"audio_url": str, "request_id": str,
+                   "duration": float, "file_size": int}
 
         Raises:
             CosyVoiceError: API 调用失败
@@ -423,55 +467,47 @@ class CosyVoiceService:
             "model": self._model,
             "input": {
                 "text": text,
-            },
-            "parameters": {
                 "voice": voice_id,
-                "sample_rate": sample_rate or settings.cosyvoice_sample_rate,
                 "format": format or settings.cosyvoice_format,
+                "sample_rate": sample_rate or settings.cosyvoice_sample_rate,
                 "rate": speed,
+                "volume": volume,
             },
         }
 
         response = self._call_api(
             method="POST",
-            path="/services/aigc/text2audio/generation",
+            path="/services/audio/tts/SpeechSynthesizer",
             json=payload,
-            timeout=60.0,
+            timeout=120.0,
         )
 
         output = response.get("output", {})
-        task_id = output.get("task_id", "")
-        audio_url = output.get("audio_url", "")
+        audio = output.get("audio", {})
+        audio_url = audio.get("url", "")
         request_id = response.get("request_id", "")
 
-        if not task_id and not audio_url:
-            raise CosyVoiceError(f"CosyVoice API 未返回 task_id 或 audio_url: {response}")
+        if not audio_url:
+            raise CosyVoiceError(f"CosyVoice API 未返回 audio_url: {response}")
 
         return {
-            "task_id": task_id,
+            "task_id": "",  # 同步接口无 task_id，兼容旧接口
             "audio_url": audio_url,
-            "duration": output.get("duration", 0.0),
-            "file_size": output.get("file_size", 0),
+            "duration": 0.0,  # 同步接口不返回 duration
+            "file_size": 0,  # 同步接口不返回 file_size
             "request_id": request_id,
         }
 
     def poll_synthesize_task(self, task_id: str, timeout: float = 120.0) -> dict:
-        """轮询语音合成任务状态（公开方法）。
+        """轮询合成任务（同步接口无需轮询，保留兼容）.
 
-        供 Celery 后台任务调用，轮询直到完成或超时。
-
-        Args:
-            task_id: CosyVoice 任务 ID
-            timeout: 超时时间（秒），默认 120
-
-        Returns:
-            dict: {"audio_url": str, "duration": float, "file_size": int}
+        CosyVoice SpeechSynthesizer 非流式接口是同步的，
+        此方法仅为保持接口兼容，实际调用时 task_id 应该为空.
 
         Raises:
-            CosyVoiceError: 任务失败
-            CosyVoiceTimeoutError: 超时
+            CosyVoiceError: 同步接口无需轮询
         """
-        return self._poll_synthesize_task(task_id, timeout=timeout)
+        raise CosyVoiceError("CosyVoice 非流式合成接口是同步的，无需轮询. " "请直接使用 submit_synthesize_task().")
 
     def synthesize_speech(
         self,
@@ -480,11 +516,13 @@ class CosyVoiceService:
         sample_rate: int = 0,
         format: str = "",
         speed: float = 1.0,
+        volume: int = 50,
         timeout: float = 120.0,
     ) -> SynthesizeResult:
-        """语音合成。
+        """语音合成（同步非流式）.
 
-        提交语音合成任务到 CosyVoice API，并轮询直到完成或超时。
+        调用百炼 CosyVoice SpeechSynthesizer 非流式接口，
+        直接返回合成音频 URL.
 
         Args:
             text: 要合成的文本
@@ -492,129 +530,53 @@ class CosyVoiceService:
             sample_rate: 采样率（Hz），0 表示使用配置默认值
             format: 输出格式（mp3/wav/pcm），空表示使用配置默认值
             speed: 语速（0.5-2.0），1.0 为正常速度
-            timeout: 超时时间（秒）
+            volume: 音量（0-100），默认 50
+            timeout: 超时时间（秒），保留参数兼容
 
         Returns:
             SynthesizeResult: 合成结果，包含 audio_url
 
         Raises:
             CosyVoiceError: API 调用失败
-            CosyVoiceTimeoutError: 超时
             CosyVoiceAuthError: 认证失败
             ValueError: 参数无效
         """
-        if not text:
-            raise ValueError("text 不能为空")
-        if not voice_id:
-            raise ValueError("voice_id 不能为空")
-        if not self._api_key:
-            raise CosyVoiceAuthError("CosyVoice API Key 未配置")
-
-        settings = get_shared_settings()
-
-        # 构建请求
-        payload = {
-            "model": self._model,
-            "input": {
-                "text": text,
-            },
-            "parameters": {
-                "voice": voice_id,
-                "sample_rate": sample_rate or settings.cosyvoice_sample_rate,
-                "format": format or settings.cosyvoice_format,
-                "rate": speed,
-            },
-        }
-
-        # 调用 API
-        response = self._call_api(
-            method="POST",
-            path="/services/aigc/text2audio/generation",
-            json=payload,
-            timeout=timeout,
+        result = self.submit_synthesize_task(
+            text=text,
+            voice_id=voice_id,
+            sample_rate=sample_rate,
+            format=format,
+            speed=speed,
+            volume=volume,
         )
 
-        # 解析响应
-        output = response.get("output", {})
-
-        # 检查是否有 task_id（异步模式）
-        task_id = output.get("task_id")
-        audio_url = output.get("audio_url")
-
-        if task_id:
-            # 异步模式：轮询任务状态
-            result = self._poll_synthesize_task(task_id, timeout)
-            return SynthesizeResult(
-                audio_url=result["audio_url"],
-                duration=result.get("duration", 0.0),
-                file_size=result.get("file_size", 0),
-                request_id=response.get("request_id", ""),
-            )
-        elif audio_url:
-            # 同步模式：直接返回结果
-            return SynthesizeResult(
-                audio_url=audio_url,
-                duration=output.get("duration", 0.0),
-                file_size=output.get("file_size", 0),
-                request_id=response.get("request_id", ""),
-            )
-        else:
-            raise CosyVoiceError(f"CosyVoice API 未返回 audio_url 或 task_id: {response}")
-
-    def _poll_synthesize_task(self, task_id: str, timeout: float) -> dict:
-        """轮询语音合成任务状态。
-
-        Args:
-            task_id: 任务 ID
-            timeout: 超时时间（秒）
-
-        Returns:
-            任务结果字典
-
-        Raises:
-            CosyVoiceError: 任务失败
-            CosyVoiceTimeoutError: 超时
-        """
-        start_time = time.time()
-        attempts = 0
-
-        while attempts < self.MAX_POLL_ATTEMPTS:
-            elapsed = time.time() - start_time
-            if elapsed > timeout:
-                raise CosyVoiceTimeoutError(f"语音合成任务超时（{timeout}秒）: task_id={task_id}")
-
-            response = self._call_api(
-                method="GET",
-                path=f"/tasks/{task_id}",
-                timeout=30.0,
-            )
-
-            output = response.get("output", {})
-            status = output.get("task_status", "").upper()
-
-            if status == "SUCCEEDED":
-                audio_url = output.get("audio_url", "")
-                if not audio_url:
-                    raise CosyVoiceError(f"语音合成任务成功但未返回 audio_url: {response}")
-                return {
-                    "audio_url": audio_url,
-                    "duration": output.get("duration", 0.0),
-                    "file_size": output.get("file_size", 0),
-                }
-            elif status == "FAILED":
-                error_msg = output.get("message", "未知错误")
-                raise CosyVoiceError(f"语音合成任务失败: {error_msg}")
-            elif status in ("PENDING", "RUNNING"):
-                # 继续轮询
-                time.sleep(self.POLL_INTERVAL)
-                attempts += 1
-            else:
-                raise CosyVoiceError(f"未知的任务状态: {status}")
-
-        raise CosyVoiceTimeoutError(f"语音合成任务轮询次数超限: task_id={task_id}")
+        return SynthesizeResult(
+            audio_url=result["audio_url"],
+            duration=result.get("duration", 0.0),
+            file_size=result.get("file_size", 0),
+            request_id=result.get("request_id", ""),
+        )
 
     # ── 内部方法 ─────────────────────────────────────────
 
+    def _sanitize_prefix(self, name: str) -> str:
+        """清洗音色名称为合法的 prefix（字母数字，最多10字符）.
+
+        Args:
+            name: 原始音色名称
+
+        Returns:
+            清洗后的 prefix
+        """
+        # 只保留字母和数字
+        cleaned = "".join(c for c in name if c.isalnum())
+        # 最多10字符
+        cleaned = cleaned[:10]
+        # 如果清洗后为空，用默认值
+        if not cleaned:
+            cleaned = "clone"
+        return cleaned
+
     def _call_api(
         self,
         method: str,
@@ -622,13 +584,13 @@ class CosyVoiceService:
         json: Optional[dict] = None,
         timeout: float = 30.0,
     ) -> dict:
-        """调用 CosyVoice API。
+        """调用 DashScope API.
 
-        支持重试和错误处理。
+        支持重试和错误处理.
 
         Args:
             method: HTTP 方法（GET/POST）
-            path: API 路径
+            path: API 路径（以 / 开头）
             json: 请求体
             timeout: 超时时间（秒）
 
@@ -646,6 +608,22 @@ 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):
@@ -658,29 +636,58 @@ 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()
                 elif response.status_code in (401, 403):
                     raise CosyVoiceAuthError(f"CosyVoice API 认证失败: HTTP {response.status_code}")
+                elif response.status_code == 400:
+                    # 客户端错误，不重试
+                    body_text = response.text
+                    try:
+                        body = response.json()
+                        code = body.get("code", "")
+                        message = body.get("message", "")
+                        raise CosyVoiceError(f"CosyVoice API 参数错误: HTTP 400, " f"code={code}, message={message}")
+                    except ValueError:
+                        raise CosyVoiceError(f"CosyVoice API 调用失败: HTTP 400, body={body_text}")
                 elif response.status_code >= 500:
                     # 服务端错误，可重试
                     last_error = CosyVoiceError(f"CosyVoice API 服务端错误: HTTP {response.status_code}")
                     logger.warning(
-                        f"CosyVoice API 失败 (尝试 {attempt + 1}/{self.MAX_RETRIES}): " f"HTTP {response.status_code}"
+                        "CosyVoice API 失败 (尝试 %d/%d): HTTP %d",
+                        attempt + 1,
+                        self.MAX_RETRIES,
+                        response.status_code,
                     )
                 else:
-                    # 客户端错误，不重试
+                    # 其他客户端错误，不重试
                     raise CosyVoiceError(
                         f"CosyVoice API 调用失败: HTTP {response.status_code}, " f"body={response.text}"
                     )
 
             except httpx.TimeoutException as e:
                 last_error = CosyVoiceTimeoutError(f"请求超时: {e}")
-                logger.warning(f"CosyVoice API 超时 (尝试 {attempt + 1}/{self.MAX_RETRIES})")
+                logger.warning(
+                    "CosyVoice API 超时 (尝试 %d/%d)",
+                    attempt + 1,
+                    self.MAX_RETRIES,
+                )
             except httpx.RequestError as e:
                 last_error = CosyVoiceError(f"请求错误: {e}")
-                logger.warning(f"CosyVoice API 请求错误 (尝试 {attempt + 1}/{self.MAX_RETRIES}): {e}")
+                logger.warning(
+                    "CosyVoice API 请求错误 (尝试 %d/%d): %s",
+                    attempt + 1,
+                    self.MAX_RETRIES,
+                    e,
+                )
 
             # 指数退避
             if attempt < self.MAX_RETRIES - 1:

packages/application/tts_job/workflow.py

@@ -14,7 +14,6 @@ import logging
 import os
 import shutil
 import tempfile
-import time
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from typing import Optional
 
@@ -190,10 +189,13 @@ class TTSWorkflowService:
         return job
 
     def poll_and_process_synthesis(self, job_id: str, timeout: float = 120.0) -> TTSJob:
-        """轮询 CosyVoice 合成任务并处理结果。
+        """轮询/检查 CosyVoice 合成任务并处理结果.
 
-        从 job.metadata 获取 task_id，调用 CosyVoiceService.poll_synthesize_task()
-        轮询状态，然后通过 process_synthesis_result / process_synthesis_failure 更新 job。
+        新 CosyVoice SpeechSynthesizer 非流式接口是同步的，
+        start_synthesis 阶段通常已经完成. 此方法用于：
+        1. job 已 completed → 直接返回（同步路径已处理）
+        2. job 仍在 processing → 重新提交合成（兜底）
+        3. 分段任务 → 检查分段状态
 
         供 Celery 后台任务调用。
         """
@@ -201,22 +203,38 @@ class TTSWorkflowService:
         if job is None:
             raise TTSJobNotFoundError(f"TTS job {job_id} not found")
 
+        # 已完成直接返回（同步路径在 start_synthesis 里已处理）
+        if job.status == TTSJobStatus.COMPLETED.value:
+            logger.info(f"TTS 任务已完成，跳过轮询: job_id={job_id}")
+            return job
+
         # 检查是否为分段合成任务
         segment_task_ids = (job.metadata or {}).get("segment_task_ids", [])
         if segment_task_ids:
             return self._poll_segment_tasks(job)
 
+        # 单段模式：同步接口下通常不会走到这里，
+        # 但如果因为异常导致仍在 processing，重新提交一次
         task_id = (job.metadata or {}).get("cosyvoice_task_id", "")
-        if not task_id:
-            raise ValueError(f"TTSJob {job_id} has no cosyvoice_task_id in metadata")
 
-        result = self.cosyvoice_service.poll_synthesize_task(task_id, timeout=timeout)
-        return self.process_synthesis_result(
-            job_id,
-            audio_url=result["audio_url"],
-            duration=result.get("duration", 0.0),
-            file_size=result.get("file_size", 0),
-        )
+        # 新接口（同步）：没有 task_id，重新合成
+        if not task_id:
+            logger.info(f"TTS 任务无 task_id，重新同步合成: job_id={job_id}")
+            return self._resynthesize_and_complete(job)
+
+        # 旧接口遗留的 task_id，尝试轮询（兼容过渡）
+        try:
+            result = self.cosyvoice_service.poll_synthesize_task(task_id, timeout=timeout)
+            return self.process_synthesis_result(
+                job_id,
+                audio_url=result["audio_url"],
+                duration=result.get("duration", 0.0),
+                file_size=result.get("file_size", 0),
+            )
+        except CosyVoiceError:
+            # 旧接口轮询失败，重新同步合成
+            logger.warning(f"旧 task_id 轮询失败，重新同步合成: job_id={job_id}, task_id={task_id}")
+            return self._resynthesize_and_complete(job)
 
     def process_synthesis_result(
         self,
@@ -257,6 +275,40 @@ class TTSWorkflowService:
         logger.info(f"TTS 合成成功: job_id={job_id}, audio_url={permanent_url}")
         return job
 
+    def _resynthesize_and_complete(self, job: TTSJob) -> TTSJob:
+        """重新同步合成并完成任务（兜底路径）.
+
+        当 poll_and_process_synthesis 发现 job 仍在 processing 且无 task_id 时，
+        重新调用同步合成接口，转存 OSS 后标记完成。
+        """
+        try:
+            # 从 metadata 读取合成参数（兼容旧数据，无则用默认值）
+            job_metadata = job.metadata or {}
+            speed = float(job_metadata.get("speed", 1.0))
+            volume = int(job_metadata.get("volume", 50))
+
+            result = self.cosyvoice_service.submit_synthesize_task(
+                text=job.input_text,
+                voice_id=job.voice_id,
+                sample_rate=job.sample_rate,
+                format=job.format,
+                speed=speed,
+                volume=volume,
+            )
+            audio_url = result.get("audio_url", "")
+            if not audio_url:
+                raise CosyVoiceError("重新合成未返回 audio_url")
+
+            return self.process_synthesis_result(
+                job.id,
+                audio_url=audio_url,
+                duration=result.get("duration", 0.0),
+                file_size=result.get("file_size", 0),
+            )
+        except Exception as e:
+            logger.error(f"重新同步合成失败: job_id={job.id}, error={e}")
+            return self.process_synthesis_failure(job.id, str(e))
+
     def process_synthesis_failure(self, job_id: str, error_message: str) -> TTSJob:
         """处理合成失败结果。
 
@@ -429,69 +481,95 @@ class TTSWorkflowService:
             shutil.rmtree(temp_dir, ignore_errors=True)
 
     def _poll_segment_tasks(self, job: TTSJob) -> TTSJob:
-        """轮询所有分段异步任务，全部完成后合并音频。"""
+        """分段任务完成检查（适配新同步接口）.
+
+        新 CosyVoice SpeechSynthesizer 非流式接口为同步接口，
+        分段任务在提交时应已同步返回 audio_url。
+        若历史任务处于 processing 且有 segment_task_ids 但缺少 audio_url，
+        则对缺失分段重新同步合成，全部完成后合并音频。
+        """
         segment_task_ids: list[str] = (job.metadata or {}).get("segment_task_ids", [])
         segment_audio_urls: list[str] = (job.metadata or {}).get("segment_audio_urls", [])
         segment_count = len(segment_task_ids)
 
-        poll_start = time.monotonic()
-        poll_timeout = 300.0  # 分段任务超时更长
-        poll_interval = 2.0
+        if segment_count == 0:
+            logger.warning(f"分段任务无 task_id: job_id={job.id}")
+            self._handle_segment_failure(job, "分段任务数据异常：无分段信息")
+            return self.repository.get(job.id)
 
-        while time.monotonic() - poll_start < poll_timeout:
-            all_done = True
-            results: list[dict | None] = [None] * segment_count
+        # 从 metadata 读取合成参数
+        job_metadata = job.metadata or {}
+        speed = float(job_metadata.get("speed", 1.0))
+        volume = int(job_metadata.get("volume", 50))
 
-            for idx, task_id in enumerate(segment_task_ids):
-                # 已经有音频的分段跳过轮询
-                if idx < len(segment_audio_urls) and segment_audio_urls[idx]:
-                    results[idx] = {
-                        "audio_url": segment_audio_urls[idx],
-                        "duration": 0.0,
-                        "file_size": 0,
-                    }
-                    continue
+        # 分段文本（用于缺失段重新合成）
+        segments = split_text(job.input_text, max_chars=_SEGMENT_THRESHOLD)
 
-                try:
-                    result = self.cosyvoice_service.poll_synthesize_task(task_id, timeout=poll_timeout)
-                    results[idx] = result
-                except Exception as e:
-                    logger.error(f"分段任务轮询失败: job_id={job.id}, " f"segment={idx}, error={e}")
-                    self._handle_segment_failure(job, f"分段 {idx + 1} 轮询失败: {e}")
-                    return self.repository.get(job.id)
+        results: list[dict | None] = [None] * segment_count
 
-                if results[idx] is None:
-                    all_done = False
+        # 已有音频的分段直接用
+        for idx in range(segment_count):
+            if idx < len(segment_audio_urls) and segment_audio_urls[idx]:
+                results[idx] = {
+                    "audio_url": segment_audio_urls[idx],
+                    "duration": 0.0,
+                    "file_size": 0,
+                }
 
-            if all_done and all(r is not None for r in results):
-                # 所有分段完成，下载合并
-                try:
-                    merged_data, total_duration = self._download_and_merge_segments(results, job)
+        # 找出缺失音频的分段索引
+        missing_indices = [i for i in range(segment_count) if results[i] is None]
 
-                    # 转存 OSS
-                    permanent_url, storage_key = self._upload_merged_to_oss(
-                        merged_data, job.user_id, job.id, job.format
+        if missing_indices:
+            logger.info(f"分段任务重新合成缺失段: job_id={job.id}, " f"缺失={len(missing_indices)}/{segment_count}")
+            # 并发重新合成缺失分段
+            max_workers = min(len(missing_indices), _MAX_SEGMENT_WORKERS)
+            with ThreadPoolExecutor(max_workers=max_workers) as executor:
+                future_to_idx = {}
+                for idx in missing_indices:
+                    segment_text = segments[idx] if idx < len(segments) else ""
+                    future = executor.submit(
+                        self.cosyvoice_service.submit_synthesize_task,
+                        text=segment_text,
+                        voice_id=job.voice_id,
+                        sample_rate=job.sample_rate,
+                        format=job.format,
+                        speed=speed,
+                        volume=volume,
                     )
+                    future_to_idx[future] = idx
 
-                    job.mark_completed(
-                        output_audio_url=permanent_url,
-                        output_audio_key=storage_key,
-                        duration=total_duration,
-                        file_size=len(merged_data),
-                    )
-                    job = self.repository.update(job)
-                    logger.info(f"分段合成轮询完成: job_id={job.id}, " f"merged_size={len(merged_data)}")
-                    return job
+                for future in as_completed(future_to_idx):
+                    idx = future_to_idx[future]
+                    try:
+                        results[idx] = future.result()
+                    except Exception as e:
+                        logger.error(f"分段重新合成失败: job_id={job.id}, " f"segment={idx}, error={e}")
+                        self._handle_segment_failure(job, f"分段 {idx + 1} 重新合成失败: {e}")
+                        return self.repository.get(job.id)
 
-                except Exception as e:
-                    self._handle_segment_failure(job, f"分段合并失败: {e}")
-                    return self.repository.get(job.id)
+        # 所有分段完成，下载合并
+        if all(r is not None for r in results):
+            try:
+                merged_data, total_duration = self._download_and_merge_segments(results, job)
 
-            # 等待后重试
-            time.sleep(poll_interval)
+                permanent_url, storage_key = self._upload_merged_to_oss(merged_data, job.user_id, job.id, job.format)
 
-        # 超时
-        self._handle_segment_failure(job, "分段合成轮询超时（300 秒）")
+                job.mark_completed(
+                    output_audio_url=permanent_url,
+                    output_audio_key=storage_key,
+                    duration=total_duration,
+                    file_size=len(merged_data),
+                )
+                job = self.repository.update(job)
+                logger.info(f"分段合成完成（重新合成路径）: job_id={job.id}, " f"merged_size={len(merged_data)}")
+                return job
+
+            except Exception as e:
+                self._handle_segment_failure(job, f"分段合并失败: {e}")
+                return self.repository.get(job.id)
+
+        # 理论上不会到这里（全部重新合成要么成功要么失败）
+        self._handle_segment_failure(job, "分段合成结果不完整")
         return self.repository.get(job.id)
 
     def _handle_segment_failure(self, job: TTSJob, error_message: str) -> None:

packages/application/voice_clone/workflow.py

@@ -115,14 +115,16 @@ class VoiceCloneWorkflowService:
                     language=language,
                 )
 
-                # 4. 保存 task_id / voice_id 到 metadata
+                # 4. 保存 voice_id / request_id 到 metadata
+                # 注意：key 保留 cosyvoice_task_id 以兼容旧数据，实际存的是 voice_id
                 task_metadata = dict(profile.metadata)
-                task_metadata["cosyvoice_task_id"] = submit_result.get("task_id", "")
+                task_metadata["cosyvoice_task_id"] = submit_result.get("voice_id", "")
                 task_metadata["cosyvoice_request_id"] = submit_result.get("request_id", "")
 
-                # 如果 CosyVoice 同步返回了 voice_id，直接标记 ready
+                # 如果 CosyVoice 直接返回了 OK 状态，直接标记 ready
                 voice_id = submit_result.get("voice_id", "")
-                if voice_id:
+                status = submit_result.get("status", "").upper()
+                if voice_id and status == "OK":
                     profile.mark_ready(voice_id)
                     profile.metadata = task_metadata
                     profile = self.repository.update(profile)
@@ -131,7 +133,9 @@ class VoiceCloneWorkflowService:
 
                 profile.metadata = task_metadata
                 profile = self.repository.update(profile)
-                logger.info(f"音色克隆任务已提交: profile_id={profile.id}, " f"task_id={submit_result.get('task_id')}")
+                logger.info(
+                    f"音色克隆任务已提交: profile_id={profile.id}, " f"voice_id={submit_result.get('voice_id')}"
+                )
 
             except (CosyVoiceError, CosyVoiceAuthError) as e:
                 # CosyVoice 提交失败，标记为 failed
@@ -248,11 +252,12 @@ class VoiceCloneWorkflowService:
                 )
 
                 task_metadata = dict(profile.metadata)
-                task_metadata["cosyvoice_task_id"] = submit_result.get("task_id", "")
+                task_metadata["cosyvoice_task_id"] = submit_result.get("voice_id", "")
                 task_metadata["cosyvoice_request_id"] = submit_result.get("request_id", "")
 
                 voice_id = submit_result.get("voice_id", "")
-                if voice_id:
+                status = submit_result.get("status", "").upper()
+                if voice_id and status == "OK":
                     profile.mark_ready(voice_id)
                     profile.metadata = task_metadata
                     profile = self.repository.update(profile)

packages/domain/generation_task.py

@@ -8,6 +8,7 @@
 
 from __future__ import annotations
 
+import json
 import sys
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
@@ -85,6 +86,7 @@ class GenerationTask:
     created_by_user_id: str = ""
     asset_select_mode: str = ""
     batch_id: str = ""
+    logs: str = "[]"
     created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
 
     @classmethod
@@ -227,6 +229,43 @@ class GenerationTask:
         self.transition_to(GenerationTaskStatus.CANCELLED)
         self.completed_at = datetime.now(timezone.utc)
 
+    # ── 日志辅助 ────────────────────────────────────────────────────────────
+
+    _MAX_LOGS = 200
+
+    def append_log(self, stage: str, message: str, level: str = "INFO", **kwargs) -> None:
+        """追加一条结构化日志到 logs 字段。
+
+        Args:
+            stage: 阶段名称（如 "接收任务"、"下载素材"、"渲染"）
+            message: 日志消息
+            level: 日志级别（INFO / WARN / ERROR）
+            **kwargs: 额外字段（如 asset_id、duration 等）
+        """
+        try:
+            entries = json.loads(self.logs) if self.logs else []
+        except (json.JSONDecodeError, TypeError):
+            entries = []
+        entry = {
+            "ts": datetime.now(timezone.utc).isoformat(),
+            "level": level,
+            "stage": stage,
+            "message": message,
+            **kwargs,
+        }
+        entries.append(entry)
+        # 限制最多保留 _MAX_LOGS 条，防止字段过大
+        if len(entries) > self._MAX_LOGS:
+            entries = entries[-self._MAX_LOGS :]
+        self.logs = json.dumps(entries, ensure_ascii=False)
+
+    def get_logs(self) -> list[dict]:
+        """解析 logs 字段为 list[dict]。"""
+        try:
+            return json.loads(self.logs) if self.logs else []
+        except (json.JSONDecodeError, TypeError):
+            return []
+
     def mark_pending_from_failed(self) -> None:
         """从失败状态重置为待处理（用于重试）。
 

packages/domain/preset_voices.py

@@ -16,7 +16,7 @@ class PresetVoice:
     """预置音色定义。
 
     Attributes:
-        voice_id: CosyVoice 模型音色名（如 longxiaochun）
+        voice_id: CosyVoice 模型音色名（如 longxiaochun_v3）
         name: 中文展示名
         description: 音色描述
         gender: 性别（male/female）
@@ -49,7 +49,7 @@ class PresetVoice:
 # 预置音色列表（阿里云 CosyVoice 真实可用音色）
 PRESET_VOICES: list[PresetVoice] = [
     PresetVoice(
-        voice_id="longxiaochun",
+        voice_id="longxiaochun_v3",
         name="龙小淳",
         description="温柔女声，适合情感类内容",
         gender="female",
@@ -57,7 +57,7 @@ PRESET_VOICES: list[PresetVoice] = [
         tags=["温柔", "女声", "情感"],
     ),
     PresetVoice(
-        voice_id="longxiaoxia",
+        voice_id="longxiaoxia_v3",
         name="龙小夏",
         description="知性女声，适合新闻播报",
         gender="female",
@@ -65,7 +65,7 @@ PRESET_VOICES: list[PresetVoice] = [
         tags=["知性", "女声", "播报"],
     ),
     PresetVoice(
-        voice_id="longxiaochen",
+        voice_id="longxiaochen_v3",
         name="龙小晨",
         description="磁性男声，适合有声书",
         gender="male",
@@ -73,7 +73,7 @@ PRESET_VOICES: list[PresetVoice] = [
         tags=["磁性", "男声", "有声书"],
     ),
     PresetVoice(
-        voice_id="longyue",
+        voice_id="longyue_v3",
         name="龙悦",
         description="甜美女声，适合广告配音",
         gender="female",
@@ -81,7 +81,7 @@ PRESET_VOICES: list[PresetVoice] = [
         tags=["甜美", "女声", "广告"],
     ),
     PresetVoice(
-        voice_id="longshu",
+        voice_id="longshu_v3",
         name="龙书",
         description="沉稳男声，适合教育讲解",
         gender="male",
@@ -89,7 +89,7 @@ PRESET_VOICES: list[PresetVoice] = [
         tags=["沉稳", "男声", "教育"],
     ),
     PresetVoice(
-        voice_id="longjing",
+        voice_id="longjing_v3",
         name="龙静",
         description="优雅女声，适合纪录片解说",
         gender="female",
@@ -97,7 +97,7 @@ PRESET_VOICES: list[PresetVoice] = [
         tags=["优雅", "女声", "纪录片"],
     ),
     PresetVoice(
-        voice_id="longbo",
+        voice_id="longbo_v3",
         name="龙博",
         description="浑厚男声，适合科技类内容",
         gender="male",
@@ -105,7 +105,7 @@ PRESET_VOICES: list[PresetVoice] = [
         tags=["浑厚", "男声", "科技"],
     ),
     PresetVoice(
-        voice_id="longtian",
+        voice_id="longtian_v3",
         name="龙甜",
         description="活泼女声，适合短视频配音",
         gender="female",

packages/shared/config.py

@@ -29,13 +29,15 @@ class SharedSettings(BaseSettings):
     oss_access_key_secret: str = ""
     oss_bucket_name: str = "xiaoxia-autocut"
 
-    # CosyVoice (阿里云语音合成)
+    # CosyVoice (阿里云百炼语音合成)
     cosyvoice_api_key: str = ""
-    cosyvoice_base_url: str = "https://dashscope.aliyuncs.com/api/v1/services/aigc/text2audio"
-    cosyvoice_model: str = "cosyvoice-v1"
-    cosyvoice_voice: str = "longxiaochun"  # 默认音色
+    cosyvoice_base_url: str = "https://dashscope.aliyuncs.com/api/v1"
+    cosyvoice_model: str = "cosyvoice-v3-flash"
+    cosyvoice_voice: str = "longxiaochun_v3"  # 默认音色（v3 系列系统音色带 _v3 后缀）
     cosyvoice_sample_rate: int = 22050
     cosyvoice_format: str = "mp3"  # 输出格式：mp3/wav/pcm
+    # 音色克隆模型名（固定为 voice-enrollment）
+    cosyvoice_clone_model: str = "voice-enrollment"
 
     # Environment
     environment: str = "development"

pyproject.toml

@@ -1,7 +1,39 @@
 [tool.black]
 line-length = 120
 target-version = ["py312"]
+extend-exclude = '''
+(
+  \.git
+  | \.cache
+  | \.pytest_cache
+  | \.mypy_cache
+  | __pycache__
+  | node_modules
+  | \.venv
+  | venv
+  | build
+  | dist
+  | \.next
+  | out
+  | coverage
+)
+'''
 
 [tool.isort]
 profile = "black"
 line_length = 120
+extend_skip_glob = [
+    ".git/**",
+    ".cache/**",
+    ".pytest_cache/**",
+    ".mypy_cache/**",
+    "__pycache__/**",
+    "node_modules/**",
+    ".venv/**",
+    "venv/**",
+    "build/**",
+    "dist/**",
+    ".next/**",
+    "out/**",
+    "coverage/**",
+]

setup.cfg

@@ -3,6 +3,7 @@ max-line-length = 120
 extend-ignore = E203,W503,E501,E302,E402,E722,W291,W293,F401,F403,F405,F841
 exclude =
     .git,
+    .cache,
     __pycache__,
     .venv,
     .venv-ci-root,

tests/integration/test_generation_api.py

@@ -241,7 +241,7 @@ def client():
 class TestCreateGenerationTask:
     """创建生成任务端点测试。"""
 
-    @patch("app.api.routes.generation_tasks.celery_app")
+    @patch("app.core.task_enqueue.celery_app")
     def test_create_task_success(self, mock_celery, client):
         """正常创建生成任务成功。"""
         mock_celery.send_task = MagicMock()
@@ -270,7 +270,7 @@ class TestCreateGenerationTask:
         assert mock_celery.send_task.called
         assert mock_celery.send_task.call_args[0][0] == "worker.generate_video"
 
-    @patch("app.api.routes.generation_tasks.celery_app")
+    @patch("app.core.task_enqueue.celery_app")
     def test_create_batch_tasks(self, mock_celery, client):
         """批量创建多个生成任务。"""
         mock_celery.send_task = MagicMock()
@@ -347,7 +347,7 @@ class TestListGenerationTasks:
 
     def _create_task(self, client, task_suffix: str = "1"):
         """辅助方法：创建一个生成任务。"""
-        with patch("app.api.routes.generation_tasks.celery_app") as mock_celery:
+        with patch("app.core.task_enqueue.celery_app") as mock_celery:
             mock_celery.send_task = MagicMock()
             resp = client.post(
                 "/api/v1/generation/tasks",
@@ -368,7 +368,7 @@ class TestListGenerationTasks:
         assert "items" in data
         assert data["items"] == []
 
-    @patch("app.api.routes.generation_tasks.celery_app")
+    @patch("app.core.task_enqueue.celery_app")
     def test_list_returns_user_tasks(self, mock_celery, client):
         """返回当前用户的生成任务列表。"""
         mock_celery.send_task = MagicMock()
@@ -406,7 +406,7 @@ class TestGetGenerationTask:
     """获取生成任务详情端点测试。"""
 
     def _create_task(self, client) -> str:
-        with patch("app.api.routes.generation_tasks.celery_app") as mock_celery:
+        with patch("app.core.task_enqueue.celery_app") as mock_celery:
             mock_celery.send_task = MagicMock()
             resp = client.post(
                 "/api/v1/generation/tasks",
@@ -449,7 +449,7 @@ class TestListGenerationResults:
     """列出生成结果端点测试。"""
 
     def _create_task(self, client) -> str:
-        with patch("app.api.routes.generation_tasks.celery_app") as mock_celery:
+        with patch("app.core.task_enqueue.celery_app") as mock_celery:
             mock_celery.send_task = MagicMock()
             resp = client.post(
                 "/api/v1/generation/tasks",
@@ -489,7 +489,7 @@ class TestRetryGenerationTask:
 
     def _create_failed_task(self, client) -> str:
         """创建一个失败状态的任务。"""
-        with patch("app.api.routes.generation_tasks.celery_app") as mock_celery:
+        with patch("app.core.task_enqueue.celery_app") as mock_celery:
             mock_celery.send_task = MagicMock()
             resp = client.post(
                 "/api/v1/generation/tasks",
@@ -509,7 +509,7 @@ class TestRetryGenerationTask:
         # 让我们直接通过 retry 测试来验证
         return task_id
 
-    @patch("app.api.routes.generation_tasks.celery_app")
+    @patch("app.core.task_enqueue.celery_app")
     def test_retry_failed_task(self, mock_celery, client):
         """重试失败的任务成功。"""
         mock_celery.send_task = MagicMock()
@@ -539,7 +539,7 @@ class TestRetryGenerationTask:
         assert resp.status_code == 404
         assert "not found" in resp.json()["detail"].lower()
 
-    @patch("app.api.routes.generation_tasks.celery_app")
+    @patch("app.core.task_enqueue.celery_app")
     def test_retry_completed_task_returns_409(self, mock_celery, client):
         """重试已完成的任务返回 409。"""
         mock_celery.send_task = MagicMock()
@@ -568,7 +568,7 @@ class TestRetryGenerationTask:
 class TestGenerationTaskFlow:
     """生成任务完整流程集成测试。"""
 
-    @patch("app.api.routes.generation_tasks.celery_app")
+    @patch("app.core.task_enqueue.celery_app")
     def test_create_list_detail_results_flow(self, mock_celery, client):
         """测试创建 → 列表 → 详情 → 结果 完整流程。"""
         mock_celery.send_task = MagicMock()

tests/integration/test_task_center_api.py

@@ -428,7 +428,7 @@ class TestRetryProjectTask:
         assert resp.status_code == 400
         assert "Unsupported" in resp.json()["detail"]
 
-    @patch("app.api.routes.task_center.celery_app")
+    @patch("app.core.task_enqueue.celery_app")
     def test_retry_failed_generation_task(self, mock_celery, client):
         """重试失败的 generation 任务成功。"""
         mock_celery.send_task = MagicMock()
@@ -581,7 +581,7 @@ class TestRetryProjectTask:
 class TestTaskCenterCrossEndpoint:
     """任务中心跨端点集成测试。"""
 
-    @patch("app.api.routes.task_center.celery_app")
+    @patch("app.core.task_enqueue.celery_app")
     def test_list_then_retry_then_list(self, mock_celery, client):
         """列出任务 → 重试失败任务 → 再列出验证新任务。"""
         mock_celery.send_task = MagicMock()

tests/unit/test_audio_url_presign.py

@@ -0,0 +1,93 @@
+"""测试音频URL预签名逻辑。
+
+验证所有 API 返回的音频 URL 都会经过 OSS 预签名（24小时有效期），
+确保私有 bucket 下的音频文件前端可正常访问。
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+class TestAudioUrlSigner:
+    """测试音频URL签名函数的行为。"""
+
+    def _make_signer(self, mock_storage):
+        """构造一个签名函数（模拟 get_audio_url_signer 的逻辑）。"""
+
+        def sign_audio_url(url: str) -> str:
+            if not url:
+                return url
+            return mock_storage.get_download_url(url, expires_seconds=86400)
+
+        return sign_audio_url
+
+    def test_empty_url_returns_empty(self):
+        """空URL直接返回，不调用签名。"""
+        mock_storage = MagicMock()
+        signer = self._make_signer(mock_storage)
+
+        result = signer("")
+        assert result == ""
+        mock_storage.get_download_url.assert_not_called()
+
+    def test_none_url_returns_none(self):
+        """None URL直接返回（有些字段可能为None）。"""
+        mock_storage = MagicMock()
+        signer = self._make_signer(mock_storage)
+
+        result = signer(None)  # type: ignore
+        assert result is None
+        mock_storage.get_download_url.assert_not_called()
+
+    def test_valid_url_gets_signed_24h(self):
+        """有效URL会调用 storage.get_download_url，有效期24小时（86400秒）。"""
+        mock_storage = MagicMock()
+        mock_storage.get_download_url.return_value = (
+            "https://bucket.oss-cn-hangzhou.aliyuncs.com/audio/test.mp3?signature=xxx"
+        )
+        signer = self._make_signer(mock_storage)
+
+        result = signer("https://bucket.oss-cn-hangzhou.aliyuncs.com/audio/test.mp3")
+
+        assert "signature=xxx" in result
+        mock_storage.get_download_url.assert_called_once_with(
+            "https://bucket.oss-cn-hangzhou.aliyuncs.com/audio/test.mp3",
+            expires_seconds=86400,
+        )
+
+    def test_storage_key_format_also_works(self):
+        """纯 storage key 格式也能正常签名（storage内部会处理）。"""
+        mock_storage = MagicMock()
+        mock_storage.get_download_url.return_value = "https://signed-url/audio.mp3?sig=xxx"
+        signer = self._make_signer(mock_storage)
+
+        result = signer("audio/test.mp3")
+
+        assert result == "https://signed-url/audio.mp3?sig=xxx"
+        mock_storage.get_download_url.assert_called_once_with(
+            "audio/test.mp3",
+            expires_seconds=86400,
+        )
+
+    def test_signer_via_dependencies_module(self):
+        """通过 dependencies 模块获取 signer，验证集成正确。"""
+        from app.core.storage import OSSStorageService
+
+        mock_svc = MagicMock(spec=OSSStorageService)
+        mock_svc.get_download_url.return_value = "https://signed/a.mp3?sig=123"
+
+        # 替换全局单例
+        with patch("app.core.storage._storage_service", mock_svc):
+            from app.dependencies import get_audio_url_signer
+
+            signer = get_audio_url_signer()
+            result = signer("test/audio.mp3")
+
+            assert result == "https://signed/a.mp3?sig=123"
+            mock_svc.get_download_url.assert_called_once_with(
+                "test/audio.mp3",
+                expires_seconds=86400,
+            )

tests/unit/test_generation_logging.py

@@ -0,0 +1,244 @@
+"""
+一键生成链路日志最小集 单元测试
+
+覆盖：
+- GenerationTask.append_log() 正确追加结构化日志
+- GenerationTask.append_log() 超过 200 条时截断
+- GenerationTask.get_logs() 正确解析 JSON
+- GenerationTask.get_logs() 异常 JSON 不抛异常
+- GenerationTaskResponse logs 字段 validator 解析 JSON 字符串
+- GenerationTaskResponse logs 字段 validator 处理非法输入
+- Worker 日志格式 [task_id=xxx] [阶段] 消息
+- _flush_logs 异常不抛出
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
+
+from app.schemas.generation_task import GenerationTaskResponse
+
+from packages.domain.generation_task import GenerationTask, GenerationTaskStatus
+
+
+def _make_task(**kwargs) -> GenerationTask:
+    """创建测试用 GenerationTask。"""
+    defaults = {
+        "id": "task-001",
+        "project_id": "proj-001",
+        "asset_library_id": "lib-001",
+        "strategy_id": "one_take",
+        "voice_library_id": "",
+        "template_id": "tpl-001",
+        "asset_ids": ["asset-1", "asset-2"],
+        "title_ids": [],
+        "voice_ids": [],
+        "status": GenerationTaskStatus.PENDING,
+        "progress": 0.0,
+        "result_count": 0,
+        "error_message": "",
+        "created_by_user_id": "user-001",
+        "source_edit_plan_id": "",
+        "asset_select_mode": "all",
+        "batch_id": "",
+    }
+    defaults.update(kwargs)
+    return GenerationTask(**defaults)
+
+
+class TestAppendLog:
+    """GenerationTask.append_log() 单元测试。"""
+
+    def test_append_single_log(self):
+        task = _make_task()
+        task.append_log("接收任务", "任务开始", mode="one_take")
+
+        logs = task.get_logs()
+        assert len(logs) == 1
+        entry = logs[0]
+        assert entry["level"] == "INFO"
+        assert entry["stage"] == "接收任务"
+        assert entry["message"] == "任务开始"
+        assert entry["mode"] == "one_take"
+        assert "ts" in entry
+
+    def test_append_multiple_logs(self):
+        task = _make_task()
+        task.append_log("接收任务", "任务开始")
+        task.append_log("下载素材", "下载完成", count=3)
+        task.append_log("渲染", "渲染完成", duration=12.5)
+
+        logs = task.get_logs()
+        assert len(logs) == 3
+        assert logs[0]["stage"] == "接收任务"
+        assert logs[1]["stage"] == "下载素材"
+        assert logs[1]["count"] == 3
+        assert logs[2]["stage"] == "渲染"
+        assert logs[2]["duration"] == 12.5
+
+    def test_append_log_with_error_level(self):
+        task = _make_task()
+        task.append_log("任务失败", "OSS上传失败", level="ERROR", error_type="RuntimeError")
+
+        logs = task.get_logs()
+        assert len(logs) == 1
+        assert logs[0]["level"] == "ERROR"
+        assert logs[0]["error_type"] == "RuntimeError"
+
+    def test_append_log_truncates_at_200(self):
+        task = _make_task()
+        for i in range(250):
+            task.append_log("阶段", f"消息{i}")
+
+        logs = task.get_logs()
+        assert len(logs) == 200
+        # 保留最后 200 条
+        assert logs[0]["message"] == "消息50"
+        assert logs[-1]["message"] == "消息249"
+
+    def test_append_log_handles_corrupted_json(self):
+        task = _make_task(logs="not-valid-json")
+        task.append_log("接收任务", "任务开始")
+
+        logs = task.get_logs()
+        assert len(logs) == 1
+        assert logs[0]["message"] == "任务开始"
+
+    def test_append_log_handles_empty_string(self):
+        task = _make_task(logs="")
+        task.append_log("接收任务", "任务开始")
+
+        logs = task.get_logs()
+        assert len(logs) == 1
+
+
+class TestGetLogs:
+    """GenerationTask.get_logs() 单元测试。"""
+
+    def test_get_logs_empty(self):
+        task = _make_task()
+        assert task.get_logs() == []
+
+    def test_get_logs_parses_json(self):
+        entries = [{"ts": "2026-01-01T00:00:00Z", "level": "INFO", "stage": "test", "message": "hello"}]
+        task = _make_task(logs=json.dumps(entries, ensure_ascii=False))
+        logs = task.get_logs()
+        assert len(logs) == 1
+        assert logs[0]["message"] == "hello"
+
+    def test_get_logs_handles_invalid_json(self):
+        task = _make_task(logs="{broken")
+        assert task.get_logs() == []
+
+    def test_get_logs_handles_none(self):
+        task = _make_task(logs=None)
+        assert task.get_logs() == []
+
+
+class TestGenerationTaskResponseLogs:
+    """GenerationTaskResponse logs 字段 validator 测试。"""
+
+    def _make_response_data(self, logs_value) -> dict:
+        return {
+            "id": "task-001",
+            "project_id": "proj-001",
+            "asset_library_id": "lib-001",
+            "strategy_id": "one_take",
+            "voice_library_id": "",
+            "template_id": "",
+            "asset_ids": [],
+            "title_ids": [],
+            "voice_ids": [],
+            "source_edit_plan_id": "",
+            "asset_select_mode": "all",
+            "batch_id": "",
+            "status": "completed",
+            "progress": 1.0,
+            "result_count": 1,
+            "error_message": "",
+            "logs": logs_value,
+        }
+
+    def test_logs_json_string_parsed(self):
+        entries = [{"ts": "2026-01-01T00:00:00Z", "level": "INFO", "stage": "test", "message": "ok"}]
+        data = self._make_response_data(json.dumps(entries, ensure_ascii=False))
+        resp = GenerationTaskResponse(**data)
+        assert isinstance(resp.logs, list)
+        assert len(resp.logs) == 1
+        assert resp.logs[0]["message"] == "ok"
+
+    def test_logs_list_passthrough(self):
+        entries = [{"ts": "2026-01-01T00:00:00Z", "level": "INFO", "stage": "test", "message": "ok"}]
+        data = self._make_response_data(entries)
+        resp = GenerationTaskResponse(**data)
+        assert resp.logs == entries
+
+    def test_logs_invalid_json_returns_empty(self):
+        data = self._make_response_data("{broken")
+        resp = GenerationTaskResponse(**data)
+        assert resp.logs == []
+
+    def test_logs_empty_string_returns_empty(self):
+        data = self._make_response_data("")
+        resp = GenerationTaskResponse(**data)
+        assert resp.logs == []
+
+    def test_logs_default_empty(self):
+        data = self._make_response_data("[]")
+        resp = GenerationTaskResponse(**data)
+        assert resp.logs == []
+
+
+class TestWorkerLogFormat:
+    """Worker 日志格式 [task_id=xxx] [阶段] 消息 测试。"""
+
+    def test_log_format_pattern(self):
+        """验证日志格式匹配 [task_id=xxx] [阶段] 消息。"""
+        import re
+
+        task_id = "abc123"
+        stage = "下载素材"
+        message = "完成: 成功=3个, 耗时=1.5s"
+        formatted = f"[task_id={task_id}] [{stage}] {message}"
+
+        pattern = r"^\[task_id=[\w-]+\] \[.+\] .+$"
+        assert re.match(pattern, formatted)
+
+    def test_log_entries_contain_required_fields(self):
+        """验证 append_log 生成的条目包含所有必需字段。"""
+        task = _make_task()
+        task.append_log("OSS上传", "上传成功", file_size=1024000, duration=2.5)
+
+        logs = task.get_logs()
+        entry = logs[0]
+        assert "ts" in entry
+        assert "level" in entry
+        assert "stage" in entry
+        assert "message" in entry
+        assert entry["file_size"] == 1024000
+        assert entry["duration"] == 2.5
+
+
+class TestFlushLogs:
+    """_flush_logs 异常安全测试。"""
+
+    def test_flush_logs_exception_not_raised(self):
+        """_flush_logs 在 DB 异常时不应抛出。"""
+        # 模拟 worker 环境
+        sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
+
+        from worker_app.tasks.generation import _flush_logs
+
+        task = _make_task()
+        task.append_log("测试", "消息")
+
+        # Mock SessionLocal 抛异常
+        with patch("worker_app.tasks.generation.SessionLocal", side_effect=RuntimeError("DB error")):
+            # 不应抛出
+            _flush_logs("task-001", task)

tests/unit/test_p02_p03_fixes.py

@@ -15,7 +15,6 @@ from video_processing.ffmpeg_utils import build_xfade_filter_chain
 # ── P0-3: build_xfade_filter_chain 安全钳制 ──────────────────────────────────
 
 
-
 class TestBuildXfadeFilterChainSafetyClamp:
     """验证 xfade 滤镜链的安全钳制逻辑，防止 exit 234。"""
 
@@ -124,9 +123,7 @@ class TestBuildXfadeFilterChainSafetyClamp:
             durations_found.append(float(m.group(1)))
 
         # 第一个 xfade: td 必须 ≤ 0.3 (第二个输入 clip_durations[1]=0.3)
-        assert durations_found[0] <= 0.3 + 0.001, (
-            f"第一个 xfade td={durations_found[0]} 超过 clip_durations[1]=0.3"
-        )
+        assert durations_found[0] <= 0.3 + 0.001, f"第一个 xfade td={durations_found[0]} 超过 clip_durations[1]=0.3"
         # 第二个 xfade: td 可以 = 0.5 (clip_durations[2]=5.0)
         assert durations_found[1] <= 0.5 + 0.001
         assert dur > 0
@@ -173,9 +170,9 @@ class TestBuildXfadeFilterChainSafetyClamp:
                 assert dur_val >= 0.001  # 至少 1ms
                 # P1 修复验证: td 不能超过第二个输入片段时长
                 second_input_idx = xfade_idx + 1
-                assert dur_val <= durations[second_input_idx] + 0.001, (
-                    f"td={dur_val} > clip_durations[{second_input_idx}]={durations[second_input_idx]}"
-                )
+                assert (
+                    dur_val <= durations[second_input_idx] + 0.001
+                ), f"td={dur_val} > clip_durations[{second_input_idx}]={durations[second_input_idx]}"
                 xfade_idx += 1
 
 

tests/unit/test_p02_worker_oss_fix.py

@@ -13,7 +13,6 @@ from unittest.mock import MagicMock, patch
 
 import pytest
 
-
 # ── oss_bucket endpoint scheme 修复 ──────────────────────────────────────────
 
 
@@ -25,17 +24,19 @@ class TestOSSBucketEndpointScheme:
         from video_processing.oss_helpers import oss_bucket
 
         mock_bucket_instance = MagicMock()
-        with patch.dict(
-            os.environ,
-            {
-                "OSS_ACCESS_KEY_ID": "test-key",
-                "OSS_ACCESS_KEY_SECRET": "test-secret",
-                "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                "OSS_BUCKET_NAME": "test-bucket",
-            },
-        ), patch("video_processing.oss_helpers.oss2.Auth") as mock_auth, patch(
-            "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance
-        ) as mock_bucket_cls:
+        with (
+            patch.dict(
+                os.environ,
+                {
+                    "OSS_ACCESS_KEY_ID": "test-key",
+                    "OSS_ACCESS_KEY_SECRET": "test-secret",
+                    "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                    "OSS_BUCKET_NAME": "test-bucket",
+                },
+            ),
+            patch("video_processing.oss_helpers.oss2.Auth") as mock_auth,
+            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls,
+        ):
             # 清除缓存，确保重新创建
             import video_processing.oss_helpers as oss_mod
 
@@ -45,9 +46,7 @@ class TestOSSBucketEndpointScheme:
             # 验证 endpoint 传的是带 https:// 的
             call_args = mock_bucket_cls.call_args
             endpoint_arg = call_args[0][1]  # 第 2 个位置参数是 endpoint
-            assert endpoint_arg.startswith("https://"), (
-                f"endpoint 应该带 https:// 前缀，实际为: {endpoint_arg}"
-            )
+            assert endpoint_arg.startswith("https://"), f"endpoint 应该带 https:// 前缀，实际为: {endpoint_arg}"
             assert "oss-cn-hangzhou.aliyuncs.com" in endpoint_arg
 
     def test_endpoint_with_https_keeps_as_is(self):
@@ -55,17 +54,19 @@ class TestOSSBucketEndpointScheme:
         from video_processing.oss_helpers import oss_bucket
 
         mock_bucket_instance = MagicMock()
-        with patch.dict(
-            os.environ,
-            {
-                "OSS_ACCESS_KEY_ID": "test-key",
-                "OSS_ACCESS_KEY_SECRET": "test-secret",
-                "OSS_ENDPOINT": "https://oss-cn-hangzhou.aliyuncs.com",
-                "OSS_BUCKET_NAME": "test-bucket",
-            },
-        ), patch("video_processing.oss_helpers.oss2.Auth"), patch(
-            "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance
-        ) as mock_bucket_cls:
+        with (
+            patch.dict(
+                os.environ,
+                {
+                    "OSS_ACCESS_KEY_ID": "test-key",
+                    "OSS_ACCESS_KEY_SECRET": "test-secret",
+                    "OSS_ENDPOINT": "https://oss-cn-hangzhou.aliyuncs.com",
+                    "OSS_BUCKET_NAME": "test-bucket",
+                },
+            ),
+            patch("video_processing.oss_helpers.oss2.Auth"),
+            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls,
+        ):
             import video_processing.oss_helpers as oss_mod
 
             bucket = oss_bucket()
@@ -81,17 +82,19 @@ class TestOSSBucketEndpointScheme:
         from video_processing.oss_helpers import oss_bucket
 
         mock_bucket_instance = MagicMock()
-        with patch.dict(
-            os.environ,
-            {
-                "OSS_ACCESS_KEY_ID": "test-key",
-                "OSS_ACCESS_KEY_SECRET": "test-secret",
-                "OSS_ENDPOINT": "http://oss-cn-hangzhou.aliyuncs.com",
-                "OSS_BUCKET_NAME": "test-bucket",
-            },
-        ), patch("video_processing.oss_helpers.oss2.Auth"), patch(
-            "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance
-        ) as mock_bucket_cls:
+        with (
+            patch.dict(
+                os.environ,
+                {
+                    "OSS_ACCESS_KEY_ID": "test-key",
+                    "OSS_ACCESS_KEY_SECRET": "test-secret",
+                    "OSS_ENDPOINT": "http://oss-cn-hangzhou.aliyuncs.com",
+                    "OSS_BUCKET_NAME": "test-bucket",
+                },
+            ),
+            patch("video_processing.oss_helpers.oss2.Auth"),
+            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance) as mock_bucket_cls,
+        ):
             import video_processing.oss_helpers as oss_mod
 
             bucket = oss_bucket()
@@ -133,16 +136,18 @@ class TestGetSignedDownloadUrl:
         mock_bucket = MagicMock()
         mock_bucket.sign_url.return_value = "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4?OSSAccessKeyId=xxx&Expires=xxx&Signature=xxx"
 
-        with patch.dict(
-            os.environ,
-            {
-                "OSS_ACCESS_KEY_ID": "test-key",
-                "OSS_ACCESS_KEY_SECRET": "test-secret",
-                "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                "OSS_BUCKET_NAME": "test-bucket",
-            },
-        ), patch("video_processing.oss_helpers.oss2.Auth"), patch(
-            "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket
+        with (
+            patch.dict(
+                os.environ,
+                {
+                    "OSS_ACCESS_KEY_ID": "test-key",
+                    "OSS_ACCESS_KEY_SECRET": "test-secret",
+                    "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                    "OSS_BUCKET_NAME": "test-bucket",
+                },
+            ),
+            patch("video_processing.oss_helpers.oss2.Auth"),
+            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
         ):
             result = get_signed_download_url("generated/test.mp4", expires_seconds=3600)
 
@@ -155,22 +160,24 @@ class TestGetSignedDownloadUrl:
         from video_processing.oss_helpers import get_signed_download_url
 
         mock_bucket = MagicMock()
-        mock_bucket.sign_url.return_value = "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4?sign=xxx"
+        mock_bucket.sign_url.return_value = (
+            "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4?sign=xxx"
+        )
 
-        with patch.dict(
-            os.environ,
-            {
-                "OSS_ACCESS_KEY_ID": "test-key",
-                "OSS_ACCESS_KEY_SECRET": "test-secret",
-                "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                "OSS_BUCKET_NAME": "test-bucket",
-            },
-        ), patch("video_processing.oss_helpers.oss2.Auth"), patch(
-            "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket
+        with (
+            patch.dict(
+                os.environ,
+                {
+                    "OSS_ACCESS_KEY_ID": "test-key",
+                    "OSS_ACCESS_KEY_SECRET": "test-secret",
+                    "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                    "OSS_BUCKET_NAME": "test-bucket",
+                },
+            ),
+            patch("video_processing.oss_helpers.oss2.Auth"),
+            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
         ):
-            result = get_signed_download_url(
-                "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4"
-            )
+            result = get_signed_download_url("https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4")
 
             mock_bucket.sign_url.assert_called_once()
             # 验证传给 sign_url 的是纯 storage key，不是完整 URL
@@ -193,16 +200,18 @@ class TestGetSignedDownloadUrl:
         mock_bucket = MagicMock()
         mock_bucket.sign_url.side_effect = Exception("sign failed")
 
-        with patch.dict(
-            os.environ,
-            {
-                "OSS_ACCESS_KEY_ID": "test-key",
-                "OSS_ACCESS_KEY_SECRET": "test-secret",
-                "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                "OSS_BUCKET_NAME": "test-bucket",
-            },
-        ), patch("video_processing.oss_helpers.oss2.Auth"), patch(
-            "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket
+        with (
+            patch.dict(
+                os.environ,
+                {
+                    "OSS_ACCESS_KEY_ID": "test-key",
+                    "OSS_ACCESS_KEY_SECRET": "test-secret",
+                    "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                    "OSS_BUCKET_NAME": "test-bucket",
+                },
+            ),
+            patch("video_processing.oss_helpers.oss2.Auth"),
+            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
         ):
             result = get_signed_download_url("generated/test.mp4")
             assert result is None
@@ -223,16 +232,18 @@ class TestUploadToOSSReturnsHTTPS:
 
         from pathlib import Path
 
-        with patch.dict(
-            os.environ,
-            {
-                "OSS_ACCESS_KEY_ID": "test-key",
-                "OSS_ACCESS_KEY_SECRET": "test-secret",
-                "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
-                "OSS_BUCKET_NAME": "test-bucket",
-            },
-        ), patch("video_processing.oss_helpers.oss2.Auth"), patch(
-            "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket
+        with (
+            patch.dict(
+                os.environ,
+                {
+                    "OSS_ACCESS_KEY_ID": "test-key",
+                    "OSS_ACCESS_KEY_SECRET": "test-secret",
+                    "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                    "OSS_BUCKET_NAME": "test-bucket",
+                },
+            ),
+            patch("video_processing.oss_helpers.oss2.Auth"),
+            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
         ):
             result = upload_to_oss(Path("/tmp/test.mp4"), "generated/test.mp4")
 
@@ -249,16 +260,18 @@ class TestUploadToOSSReturnsHTTPS:
 
         from pathlib import Path
 
-        with patch.dict(
-            os.environ,
-            {
-                "OSS_ACCESS_KEY_ID": "test-key",
-                "OSS_ACCESS_KEY_SECRET": "test-secret",
-                "OSS_ENDPOINT": "https://oss-cn-hangzhou.aliyuncs.com",
-                "OSS_BUCKET_NAME": "test-bucket",
-            },
-        ), patch("video_processing.oss_helpers.oss2.Auth"), patch(
-            "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket
+        with (
+            patch.dict(
+                os.environ,
+                {
+                    "OSS_ACCESS_KEY_ID": "test-key",
+                    "OSS_ACCESS_KEY_SECRET": "test-secret",
+                    "OSS_ENDPOINT": "https://oss-cn-hangzhou.aliyuncs.com",
+                    "OSS_BUCKET_NAME": "test-bucket",
+                },
+            ),
+            patch("video_processing.oss_helpers.oss2.Auth"),
+            patch("video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket),
         ):
             result = upload_to_oss(Path("/tmp/test.mp4"), "generated/test.mp4")
 

tests/unit/test_preset_voices.py

@@ -64,7 +64,7 @@ class TestPresetVoice:
     def test_preset_voice_to_dict(self) -> None:
         """序列化。"""
         voice = PresetVoice(
-            voice_id="longxiaochun",
+            voice_id="longxiaochun_v3",
             name="龙小淳",
             description="温柔女声",
             gender="female",
@@ -73,7 +73,7 @@ class TestPresetVoice:
 
         result = voice.to_dict()
 
-        assert result["voice_id"] == "longxiaochun"
+        assert result["voice_id"] == "longxiaochun_v3"
         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",
-            "longxiaoxia",
-            "longxiaochen",
-            "longyue",
-            "longshu",
-            "longjing",
-            "longbo",
-            "longtian",
+            "longxiaochun_v3",
+            "longxiaoxia_v3",
+            "longxiaochen_v3",
+            "longyue_v3",
+            "longshu_v3",
+            "longjing_v3",
+            "longbo_v3",
+            "longtian_v3",
         }
         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")
+        voice = get_preset_voice_by_id("longxiaochun_v3")
         assert voice is not None
         assert voice.name == "龙小淳"
-        assert voice.voice_id == "longxiaochun"
+        assert voice.voice_id == "longxiaochun_v3"
 
     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") is True
-        assert is_preset_voice("longxiaoxia") is True
-        assert is_preset_voice("longbo") is 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
 
     def test_is_preset_voice_false(self) -> None:
         """判断非预置音色返回 False。"""

tests/unit/test_tts_job.py

@@ -15,7 +15,7 @@ class TestTTSJobCreate:
         job = TTSJob.create(
             user_id="user_001",
             input_text="这是一段测试文本",
-            voice_id="longxiaochun",
+            voice_id="longxiaochun_v3",
             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"
+        assert job.voice_id == "longxiaochun_v3"
         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",
+            voice_id="longxiaochun_v3",
             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"
+        assert result["voice_id"] == "longxiaochun_v3"
         assert result["voice_model"] == "cosyvoice-v1"
         assert result["project_id"] == "project_001"
         assert result["voice_clone_profile_id"] == "profile_001"

tests/unit/test_tts_segment_synthesis.py

@@ -353,16 +353,11 @@ class TestHandleSegmentFailure:
 
 
 class TestPollSegmentTasks:
-    """测试 _poll_segment_tasks 异步轮询。"""
+    """测试 _poll_segment_tasks 分段缺失重新合成（适配同步接口）。"""
 
-    @patch("packages.application.tts_job.workflow.time")
     @patch("packages.application.tts_job.workflow.httpx")
-    def test_all_segments_done(self, mock_httpx: MagicMock, mock_time: MagicMock) -> None:
-        """所有分段完成后合并并标记完成。"""
-        # Mock time.monotonic 让循环只执行一次
-        mock_time.monotonic.side_effect = [0.0, 1.0, 2.0]
-        mock_time.sleep = MagicMock()
-
+    def test_all_segments_done(self, mock_httpx: MagicMock) -> None:
+        """所有分段缺少 audio_url 时重新同步合成，合并后标记完成。"""
         # Mock 下载分段音频
         mock_resp = MagicMock()
         mock_resp.content = b"seg audio"
@@ -370,7 +365,7 @@ class TestPollSegmentTasks:
         mock_httpx.get.return_value = mock_resp
 
         service = MagicMock(spec=CosyVoiceService)
-        service.poll_synthesize_task.side_effect = [
+        service.submit_synthesize_task.side_effect = [
             {"audio_url": "https://temp.com/seg1.mp3", "duration": 2.0, "file_size": 100},
             {"audio_url": "https://temp.com/seg2.mp3", "duration": 3.0, "file_size": 200},
         ]
@@ -381,6 +376,8 @@ class TestPollSegmentTasks:
         repo = MagicMock()
         job = _make_job(
             status=TTSJobStatus.PROCESSING,
+            # 长文本触发分段，用于重新合成时切分
+            input_text="这是一段很长的测试文本。" * 30,
             metadata={
                 "segment_task_ids": ["task_1", "task_2"],
                 "segment_audio_urls": ["", ""],
@@ -400,19 +397,18 @@ class TestPollSegmentTasks:
             result = workflow._poll_segment_tasks(job)
 
         assert result.status == TTSJobStatus.COMPLETED
+        # 两个缺失分段都重新合成了
+        assert service.submit_synthesize_task.call_count == 2
 
-    @patch("packages.application.tts_job.workflow.time")
-    def test_segment_poll_failure(self, mock_time: MagicMock) -> None:
-        """分段轮询失败时标记 job failed。"""
-        mock_time.monotonic.side_effect = [0.0, 1.0]
-        mock_time.sleep = MagicMock()
-
+    def test_segment_resynthesis_failure(self) -> None:
+        """分段重新合成失败时标记 job failed。"""
         service = MagicMock(spec=CosyVoiceService)
-        service.poll_synthesize_task.side_effect = CosyVoiceError("Poll failed")
+        service.submit_synthesize_task.side_effect = CosyVoiceError("Synthesis failed")
 
         repo = MagicMock()
         job = _make_job(
             status=TTSJobStatus.PROCESSING,
+            input_text="这是一段很长的测试文本。" * 30,
             metadata={
                 "segment_task_ids": ["task_1"],
                 "segment_audio_urls": [""],
@@ -426,6 +422,51 @@ class TestPollSegmentTasks:
 
         assert result.status == TTSJobStatus.FAILED
 
+    @patch("packages.application.tts_job.workflow.httpx")
+    def test_partial_audio_urls_reuse_existing(self, mock_httpx: MagicMock) -> None:
+        """部分分段已有 audio_url 时直接复用，缺失的重新合成。"""
+        mock_resp = MagicMock()
+        mock_resp.content = b"seg audio"
+        mock_resp.raise_for_status.return_value = None
+        mock_httpx.get.return_value = mock_resp
+
+        service = MagicMock(spec=CosyVoiceService)
+        # 只有 1 个分段需要重新合成
+        service.submit_synthesize_task.return_value = {
+            "audio_url": "https://temp.com/seg2.mp3",
+            "duration": 3.0,
+            "file_size": 200,
+        }
+
+        storage = MagicMock()
+        storage.upload_file.return_value = "https://oss.example.com/merged.mp3"
+
+        repo = MagicMock()
+        job = _make_job(
+            status=TTSJobStatus.PROCESSING,
+            input_text="这是一段很长的测试文本。" * 30,
+            metadata={
+                "segment_task_ids": ["task_1", "task_2"],
+                "segment_audio_urls": ["https://temp.com/seg1.mp3", ""],
+                "segment_format": "mp3",
+            },
+        )
+        repo.get.return_value = job
+        repo.update.side_effect = lambda j: j
+
+        workflow = _make_workflow(cosyvoice_service=service, repo=repo, storage=storage)
+
+        with patch("packages.application.tts_job.workflow.AudioMerger") as MockMerger:
+            mock_merger = MagicMock()
+            mock_merger.merge.return_value = b"merged data"
+            MockMerger.return_value = mock_merger
+
+            result = workflow._poll_segment_tasks(job)
+
+        assert result.status == TTSJobStatus.COMPLETED
+        # 只有 1 个缺失分段被重新合成
+        assert service.submit_synthesize_task.call_count == 1
+
 
 class TestPollAndProcessSynthesisSegmentDetection:
     """测试 poll_and_process_synthesis 正确识别分段任务。"""

tests/unit/test_unified_render_service.py

@@ -435,8 +435,7 @@ class TestBuildFilterComplex:
         last_setpts = max(setpts_positions)
         first_fps = min(fps_positions)
         assert last_setpts < first_fps, (
-            f"单视频: setpts(position={last_setpts}) 应该在 fps(position={first_fps}) 之前。"
-            f"滤镜链: {chain_str}"
+            f"单视频: setpts(position={last_setpts}) 应该在 fps(position={first_fps}) 之前。" f"滤镜链: {chain_str}"
         )
 
     def test_empty_layers_raises(self):

tests/unit/test_voice_clone_workflow.py

@@ -57,15 +57,15 @@ def _make_service(
 class TestStartClone:
     """测试 start_clone 方法。"""
 
-    def test_start_clone_with_async_task(self) -> None:
-        """异步模式：提交任务后返回 processing 状态的 profile。"""
+    def test_start_clone_with_deploying(self) -> None:
+        """提交克隆后返回 DEPLOYING 状态，profile 保持 processing。"""
         mock_repo = MagicMock()
         mock_cosyvoice = MagicMock(spec=CosyVoiceService)
 
-        # CosyVoice 返回 task_id（异步模式）
+        # CosyVoice 返回 voice_id + DEPLOYING 状态（需轮询）
         mock_cosyvoice.submit_clone_task.return_value = {
-            "task_id": "task-abc",
-            "voice_id": "",
+            "voice_id": "voice-abc",
+            "status": "DEPLOYING",
             "request_id": "req-123",
         }
 
@@ -81,21 +81,21 @@ class TestStartClone:
         )
 
         assert profile.status == VoiceCloneStatus.PROCESSING
-        assert profile.metadata["cosyvoice_task_id"] == "task-abc"
+        assert profile.metadata["cosyvoice_task_id"] == "voice-abc"
         assert profile.metadata["cosyvoice_request_id"] == "req-123"
         mock_cosyvoice.submit_clone_task.assert_called_once()
         assert mock_repo.create.call_count == 1
-        # update 至少调用 2 次：mark_processing + 保存 task_id
+        # update 至少调用 2 次：mark_processing + 保存 voice_id
         assert mock_repo.update.call_count >= 2
 
-    def test_start_clone_with_sync_result(self) -> None:
-        """同步模式：CosyVoice 直接返回 voice_id，profile 变为 ready。"""
+    def test_start_clone_with_ok_status(self) -> None:
+        """CosyVoice 直接返回 OK 状态，profile 变为 ready。"""
         mock_repo = MagicMock()
         mock_cosyvoice = MagicMock(spec=CosyVoiceService)
 
         mock_cosyvoice.submit_clone_task.return_value = {
-            "task_id": "",
             "voice_id": "voice-sync-123",
+            "status": "OK",
             "request_id": "req-456",
         }
 
@@ -244,8 +244,8 @@ class TestRetryClone:
         mock_repo.update.side_effect = lambda p: p
 
         mock_cosyvoice.submit_clone_task.return_value = {
-            "task_id": "task-retry",
-            "voice_id": "",
+            "voice_id": "voice-retry",
+            "status": "DEPLOYING",
             "request_id": "req-retry",
         }
 
@@ -253,11 +253,11 @@ class TestRetryClone:
         result = service.retry_clone(profile.id, "user-123")
 
         assert result.status == VoiceCloneStatus.PROCESSING
-        assert result.metadata["cosyvoice_task_id"] == "task-retry"
+        assert result.metadata["cosyvoice_task_id"] == "voice-retry"
         assert result.retry_count == 2  # prepare_retry 增加了一次
 
-    def test_retry_clone_with_sync_result(self) -> None:
-        """重试成功，同步模式。"""
+    def test_retry_clone_with_ok_status(self) -> None:
+        """重试成功，直接返回 OK 状态。"""
         mock_repo = MagicMock()
         mock_cosyvoice = MagicMock(spec=CosyVoiceService)
 
@@ -266,8 +266,8 @@ class TestRetryClone:
         mock_repo.update.side_effect = lambda p: p
 
         mock_cosyvoice.submit_clone_task.return_value = {
-            "task_id": "",
             "voice_id": "voice-retry-sync",
+            "status": "OK",
             "request_id": "req-retry",
         }