fix: _download_video_assets 允许仅传 asset_ids（模板预览模式）

预览生成（PR #1207）是模板模式，只传 template_id + asset_ids，
不传 project_id/asset_library_id。asset_ids 已有明确的素材ID列表，
查询逻辑也支持直接按ID查，不应被校验挡住。

改动：
- 校验条件从 (not library and not project) 改为 (not library and not project and not asset_ids)
- mode_desc 兜底：三者都空时显示"指定素材"而非"项目 None"
- 向后兼容：原有传 project_id 或 asset_library_id 的路径不受影响

alembic/versions/053_generation_task_is_preview.py

@@ -0,0 +1,61 @@
+"""#1197 - 预览生成：generation_tasks 表新增 is_preview 字段
+
+Revision ID: 053
+Revises: 052
+Create Date: 2026-08-15
+
+Changes:
+1. generation_tasks 表新增 is_preview 字段，标记是否为预览生成任务（低清 480p）
+2. 默认 False，与现有正式生成任务兼容
+3. 加索引以支持按预览/正式任务筛选
+"""
+
+import sqlalchemy as sa
+
+from alembic import context, op
+
+revision = "053_generation_task_is_preview"
+down_revision = "052_generation_task_bgm_config"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    if context.get_context().dialect.name == "postgresql":
+        # 检查列是否已存在（幂等）
+        result = conn.execute(
+            sa.text(
+                "SELECT column_name FROM information_schema.columns "
+                "WHERE table_name = 'generation_tasks' AND column_name = 'is_preview'"
+            )
+        )
+        if result.scalar() is not None:
+            return
+
+    op.add_column(
+        "generation_tasks",
+        sa.Column("is_preview", sa.Boolean, nullable=False, server_default=sa.text("false")),
+    )
+    # 加索引
+    op.create_index(
+        "ix_generation_tasks_is_preview",
+        "generation_tasks",
+        ["is_preview"],
+    )
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    if context.get_context().dialect.name == "postgresql":
+        result = conn.execute(
+            sa.text(
+                "SELECT column_name FROM information_schema.columns "
+                "WHERE table_name = 'generation_tasks' AND column_name = 'is_preview'"
+            )
+        )
+        if result.scalar() is None:
+            return
+
+    op.drop_index("ix_generation_tasks_is_preview", table_name="generation_tasks")
+    op.drop_column("generation_tasks", "is_preview")

apps/api/app/api/router.py

@@ -8,6 +8,7 @@ from app.api.routes.classification_jobs import router as classification_jobs_rou
 from app.api.routes.duplication import router as duplication_router
 from app.api.routes.feature_flags import router as feature_flags_router
 from app.api.routes.generation_tasks import router as generation_tasks_router
+from app.api.routes.generation_preview import router as generation_preview_router
 from app.api.routes.health import router as health_check_router
 from app.api.routes.ingest_jobs import router as ingest_jobs_router
 from app.api.routes.internal_render import router as internal_render_router
@@ -87,6 +88,11 @@ api_router.include_router(
     prefix="/generation",
     tags=["Generation"],
 )
+api_router.include_router(
+    generation_preview_router,
+    prefix="/generation",
+    tags=["Generation"],
+)
 api_router.include_router(
     titles_router,
     prefix="/titles",

apps/api/app/api/routes/generation_preview.py

@@ -0,0 +1,257 @@
+"""预览生成路由 — Phase 1：单版本预览接口（创建 + 查询）。
+
+路径前缀：/api/v1/generation/preview（与 /generation/tasks 同体系）
+"""
+
+from __future__ import annotations
+
+import logging
+
+from app.auth import AuthenticatedUser, get_current_user
+from app.core.task_enqueue import (
+    GLOBAL_PENDING_LIMIT,
+    USER_PENDING_LIMIT,
+    GlobalQueueFull,
+    UserPendingLimitExceeded,
+    safe_enqueue_generation_task,
+)
+from app.dependencies import (
+    get_generated_video_repository,
+    get_generation_task_repository,
+)
+from app.schemas.generation_task import (
+    CreatePreviewGenerationTaskRequest,
+    PreviewGenerationTaskResponse,
+)
+from fastapi import APIRouter, Depends, HTTPException
+
+from packages.application import (
+    CreateGenerationTaskCommand,
+    CreateGenerationTaskUseCase,
+    GetGenerationTaskUseCase,
+    ListGeneratedVideosByTaskUseCase,
+)
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+PREVIEW_RESOLUTION = "854x480"
+
+
+def _calc_preview_resolution(video_ratio: str = "") -> str:
+    """根据视频比例计算预览分辨率（短边 480，长边按比例）。
+
+    支持的比例：16:9, 9:16, 1:1, 4:3, 3:4, 其他默认 16:9。
+    """
+    ratio_map = {
+        "16:9": "854x480",
+        "9:16": "480x854",
+        "1:1": "480x480",
+        "4:3": "640x480",
+        "3:4": "480x640",
+    }
+    return ratio_map.get(video_ratio.strip(), PREVIEW_RESOLUTION)
+
+
+def _mark_task_failed(repo, task, reason: str) -> None:
+    """入队失败时将任务标记为 failed，避免产生僵尸 pending 数据。"""
+    try:
+        task.mark_failed(error_message=f"入队失败：{reason}")
+        repo.update(task)
+    except Exception:
+        logger.exception("[预览生成] 标记任务失败时异常: task_id=%s", task.id)
+
+
+def _to_preview_response(task, generated_videos: list | None = None) -> PreviewGenerationTaskResponse:
+    """将领域任务对象转换为预览响应 DTO。
+
+    Args:
+        task: GenerationTask 领域对象
+        generated_videos: 生成的视频列表（可选），取第一个作为 video_url
+
+    Returns:
+        PreviewGenerationTaskResponse
+    """
+    video_url = ""
+    duration = 0.0
+    file_size = 0
+    if generated_videos:
+        first_video = generated_videos[0]
+        video_url = getattr(first_video, "file_url", "") or ""
+        duration = float(getattr(first_video, "duration", 0.0) or 0.0)
+        file_size = int(getattr(first_video, "file_size", 0) or 0)
+
+    # 从 extra_meta / metadata 中提取统计信息（如果有）
+    extra_meta = getattr(task, "extra_meta", {}) or {}
+    clip_count = int(extra_meta.get("clip_count", len(getattr(task, "asset_ids", [])) or 0))
+    transition_count = int(extra_meta.get("transition_count", max(0, clip_count - 1)))
+    material_usage = extra_meta.get("material_usage", {}) or {}
+
+    # 计算生成耗时
+    generate_duration = 0.0
+    started_at = getattr(task, "started_at", None)
+    completed_at = getattr(task, "completed_at", None)
+    if started_at and completed_at:
+        generate_duration = (completed_at - started_at).total_seconds()
+
+    return PreviewGenerationTaskResponse(
+        task_id=task.id,
+        status=task.status.value if hasattr(task.status, "value") else str(task.status),
+        progress=float(task.progress or 0.0),
+        is_preview=bool(getattr(task, "is_preview", True)),
+        resolution=getattr(task, "resolution", PREVIEW_RESOLUTION) or PREVIEW_RESOLUTION,
+        video_url=video_url,
+        duration=duration,
+        file_size=file_size,
+        clip_count=clip_count,
+        transition_count=transition_count,
+        material_usage=material_usage,
+        error_message=task.error_message or "",
+        created_at=task.created_at,
+        started_at=started_at,
+        finished_at=completed_at,
+        generate_duration=generate_duration,
+    )
+
+
+@router.post("/preview", response_model=PreviewGenerationTaskResponse, status_code=201)
+def create_preview_generation_task(
+    request: CreatePreviewGenerationTaskRequest,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    generation_task_repository=Depends(get_generation_task_repository),
+) -> PreviewGenerationTaskResponse:
+    """创建预览生成任务。
+
+    预览为完整时长的低清版（480p + 低码率），效果与正式生成一致，仅清晰度降低。
+
+    Args:
+        request: 预览任务创建请求（template_id + asset_ids 等）
+
+    Returns:
+        201 + 预览任务详情
+    """
+    user_id = authenticated_user.user.id
+    logger.info(
+        "[预览生成] 接收请求: user_id=%s, template_id=%s, asset_count=%d",
+        user_id,
+        request.template_id,
+        len(request.asset_ids),
+    )
+
+    # 预检查队列限流
+    try:
+        user_pending = generation_task_repository.count_pending_by_user(user_id)
+        global_pending = generation_task_repository.count_pending_total()
+        if user_pending + 1 > USER_PENDING_LIMIT:
+            raise UserPendingLimitExceeded(user_id=user_id, pending_count=user_pending + 1, limit=USER_PENDING_LIMIT)
+        if global_pending + 1 > GLOBAL_PENDING_LIMIT:
+            raise GlobalQueueFull(pending_count=global_pending + 1, limit=GLOBAL_PENDING_LIMIT)
+    except UserPendingLimitExceeded as e:
+        raise HTTPException(
+            status_code=429,
+            detail=f"您的待处理任务过多（当前 {e.pending_count - 1}/{e.limit}），请等待完成后再提交",
+        ) from e
+    except GlobalQueueFull as e:
+        raise HTTPException(
+            status_code=503,
+            detail="系统繁忙，请稍后再试",
+        ) from e
+
+    use_case = CreateGenerationTaskUseCase(generation_task_repository)
+
+    try:
+        task = use_case.execute(
+            CreateGenerationTaskCommand(
+                project_id="",
+                asset_library_id="",
+                strategy_id="",
+                voice_library_id="",
+                template_id=request.template_id,
+                asset_ids=list(request.asset_ids),
+                title_ids=list(request.title_ids),
+                voice_ids=list(request.voice_ids),
+                created_by_user_id=user_id,
+                source_edit_plan_id="",
+                asset_select_mode="",
+                batch_id="",
+                video_title=request.video_title,
+                resolution=_calc_preview_resolution(request.video_ratio),
+                bgm_config=request.bgm_config or {},
+                auto_retry_enabled=False,
+                auto_retry_max=0,
+                is_preview=True,
+            )
+        )
+    except ValueError as e:
+        logger.warning("[预览生成] 创建失败: %s", e)
+        raise HTTPException(status_code=400, detail=str(e)) from e
+    except Exception as e:
+        logger.error("[预览生成] 创建失败: %s", e, exc_info=True)
+        raise HTTPException(status_code=500, detail="创建预览生成任务失败，请稍后重试") from e
+
+    # 入队执行；若入队失败则标记任务为 failed 避免僵尸数据
+    try:
+        if not safe_enqueue_generation_task(
+            task,
+            generation_task_repository,
+            user_id=user_id,
+            log_prefix="[预览生成]",
+            log_task_status=True,
+        ):
+            logger.warning("[预览生成] 任务入队失败: task_id=%s", task.id)
+            _mark_task_failed(generation_task_repository, task, "任务入队失败")
+            raise HTTPException(status_code=500, detail="任务入队失败，请稍后重试")
+    except UserPendingLimitExceeded:
+        _mark_task_failed(generation_task_repository, task, "待处理任务超限")
+        raise HTTPException(
+            status_code=429,
+            detail="您的待处理任务过多，请等待完成后再提交",
+        ) from None
+    except GlobalQueueFull:
+        _mark_task_failed(generation_task_repository, task, "系统队列已满")
+        raise HTTPException(
+            status_code=503,
+            detail="系统繁忙，请稍后再试",
+        ) from None
+
+    return _to_preview_response(task)
+
+
+@router.get("/preview/{task_id}", response_model=PreviewGenerationTaskResponse)
+def get_preview_generation_task(
+    task_id: str,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    generation_task_repository=Depends(get_generation_task_repository),
+    generated_video_repository=Depends(get_generated_video_repository),
+) -> PreviewGenerationTaskResponse:
+    """查询预览生成任务状态。
+
+    Args:
+        task_id: 任务 ID
+
+    Returns:
+        预览任务详情（含状态、进度、结果 URL 等）
+    """
+    use_case = GetGenerationTaskUseCase(generation_task_repository)
+    task = use_case.execute(task_id)
+    if task is None:
+        raise HTTPException(status_code=404, detail=f"预览任务 {task_id} 不存在")
+
+    # 权限校验：任务必须属于当前用户（统一转 str 比较，避免 UUID/str 类型差异）
+    task_user_id = str(getattr(task, "created_by_user_id", "") or "")
+    if not task_user_id or task_user_id != str(authenticated_user.user.id):
+        raise HTTPException(status_code=403, detail="无权访问该任务")
+
+    # 校验是否为预览任务
+    if not getattr(task, "is_preview", False):
+        raise HTTPException(status_code=404, detail=f"预览任务 {task_id} 不存在")
+
+    # 查询生成的视频（取第一个）
+    generated_videos = []
+    status_val = task.status.value if hasattr(task.status, "value") else str(task.status)
+    if status_val == "completed":
+        list_use_case = ListGeneratedVideosByTaskUseCase(generated_video_repository)
+        generated_videos = list_use_case.execute(task_id)
+
+    return _to_preview_response(task, generated_videos=generated_videos)

apps/api/app/schemas/generation_task.py

@@ -1,4 +1,5 @@
 import json
+from datetime import datetime
 
 from pydantic import BaseModel, Field, field_validator, model_validator
 
@@ -122,3 +123,62 @@ class ListGenerationTasksResponse(BaseModel):
     """用户级生成任务列表响应（跨 project）。"""
 
     items: list[GenerationTaskResponse]
+
+
+# ── 预览生成（Phase 1） ───────────────────────────────────────────────────────
+
+
+class CreatePreviewGenerationTaskRequest(BaseModel):
+    """创建预览生成任务请求。
+
+    仅支持模板模式：template_id + asset_ids 等素材 ID 列表。
+    预览为完整时长低清版（480p + 低码率）。
+    """
+
+    template_id: str
+    asset_ids: list[str] = Field(default_factory=list)
+    title_ids: list[str] = Field(default_factory=list)
+    voice_ids: list[str] = Field(default_factory=list)
+    video_title: str = Field(default="", description="生成视频的标题/名称，为空则使用默认命名")
+    duration: float = Field(default=0.0, ge=0, description="期望视频时长（秒），0 表示由模板决定")
+    video_ratio: str = Field(default="", description="视频比例，如 16:9 / 9:16，为空使用模板默认")
+    bgm_config: dict = Field(
+        default_factory=dict,
+        description="自定义BGM配置，覆盖模板BGM设置。支持 enabled/source/asset_id/preset_id/audio_url/volume 等字段",
+    )
+
+    @model_validator(mode="after")
+    def _check_template_id(self) -> "CreatePreviewGenerationTaskRequest":
+        if not self.template_id.strip():
+            raise ValueError("template_id 不能为空")
+        return self
+
+    @model_validator(mode="after")
+    def _check_asset_ids(self) -> "CreatePreviewGenerationTaskRequest":
+        if not self.asset_ids and not self.title_ids and not self.voice_ids:
+            raise ValueError("asset_ids/title_ids/voice_ids 至少需要提供一个")
+        return self
+
+
+class PreviewGenerationTaskResponse(BaseModel):
+    """预览生成任务响应。
+
+    包含任务状态、进度、分辨率、生成结果 URL 等关键字段。
+    """
+
+    task_id: str
+    status: str
+    progress: float
+    is_preview: bool = True
+    resolution: str = ""
+    video_url: str = ""
+    duration: float = 0.0
+    file_size: int = 0
+    clip_count: int = 0
+    transition_count: int = 0
+    material_usage: dict = Field(default_factory=dict)
+    error_message: str = ""
+    created_at: datetime | None = None
+    started_at: datetime | None = None
+    finished_at: datetime | None = None
+    generate_duration: float = 0.0

apps/worker/worker_app/tasks/generation.py

@@ -600,11 +600,11 @@ def _download_library_assets(
         下载成功的视频文件 Path 列表
 
     Raises:
-        ValueError: 当 asset_library_id 和 project_id 都为空时
+        ValueError: 当 asset_library_id、project_id 和 asset_ids 都为空时
         RuntimeError: strict=True 时任何下载失败；或指定了 asset_ids 但全部下载失败
     """
-    if not asset_library_id and not project_id:
-        raise ValueError("asset_library_id 和 project_id 至少需要提供一个")
+    if not asset_library_id and not project_id and not asset_ids:
+        raise ValueError("asset_library_id、project_id 或 asset_ids 至少需要提供一个")
 
     try:
         from packages.adapters.sqlalchemy_impl.models import AssetModel
@@ -646,7 +646,7 @@ def _download_library_assets(
             assets = query.order_by(AssetModel.created_at).all()
 
             if not assets:
-                mode_desc = f"素材库 {asset_library_id}" if asset_library_id else f"项目 {project_id}"
+                mode_desc = f"素材库 {asset_library_id}" if asset_library_id else (f"项目 {project_id}" if project_id else "指定素材")
                 msg = f"未找到视频素材: {mode_desc}, asset_ids={asset_ids or 'all'}"
                 logger.error(msg)
                 raise RuntimeError(msg)
@@ -943,6 +943,7 @@ def _load_task_info(task_id: str) -> dict | None:
             "video_title": getattr(gen_task, "video_title", "") or "",
             "resolution": getattr(gen_task, "resolution", "") or "",
             "bgm_config": dict(getattr(gen_task, "bgm_config", {}) or {}),
+            "is_preview": bool(getattr(gen_task, "is_preview", False)),
         }
     finally:
         session.close()
@@ -1003,11 +1004,15 @@ def _render_video(
     output_name: str,
     resolution: str = "",
     bgm_config: dict | None = None,
+    is_preview: bool = False,
 ) -> tuple[Path, float]:
     """渲染视频（含配音混音）。
 
     使用 RenderAdapter 统一渲染入口，复用 BGM/ASR/分辨率/缩略图逻辑。
 
+    Args:
+        is_preview: 是否为预览生成，若是则强制 480p + 低码率
+
     Returns:
         (output_path, render_duration)
     """
@@ -1050,13 +1055,26 @@ def _render_video(
 
     # 确保输出分辨率配置存在
     # 优先级：用户指定 > 模板配置 > 默认 1280x720
-    plan_cfg = virtual_plan.config or {}
-    export_cfg = plan_cfg.get("export", {}) or {}
-    if resolution:
+    # 预览模式：强制 854x480 + 低码率
+    # 注意：必须拷贝字典，避免预览模式修改污染源对象（模板配置）
+    plan_cfg = dict(virtual_plan.config or {})
+    export_cfg = dict(plan_cfg.get("export", {}) or {})
+    if is_preview:
+        # 预览模式强制 480p + 低码率
+        export_cfg["resolution"] = "854x480"
+        export_cfg["bitrate"] = "1M"
+        logger.info(
+            "[task_id=%s] [渲染] 预览模式：强制分辨率=%s, 码率=%s",
+            task_id,
+            "854x480",
+            "1M",
+        )
+    elif resolution:
         # 用户在 API 调用时指定的分辨率优先级最高
         export_cfg["resolution"] = resolution
     elif not export_cfg.get("resolution"):
         export_cfg["resolution"] = f"{OUTPUT_WIDTH}x{OUTPUT_HEIGHT}"
+    # 将修改后的配置写回 virtual_plan（拷贝后的副本，不影响原始数据源）
     plan_cfg["export"] = export_cfg
     virtual_plan.config = plan_cfg
 
@@ -1301,6 +1319,7 @@ def generate_video(self, task_id: str) -> dict:
                 output_name=output_name,
                 resolution=task_info.get("resolution", ""),
                 bgm_config=task_info.get("bgm_config", {}),
+                is_preview=task_info.get("is_preview", False),
             )
 
             if gen_task:

packages/adapters/sqlalchemy_impl/generation_task_repository.py

@@ -36,6 +36,7 @@ def _to_domain(model: GenerationTaskModel) -> GenerationTask:
         video_title=getattr(model, "video_title", "") or "",
         resolution=getattr(model, "resolution", "") or "",
         bgm_config=dict(getattr(model, "bgm_config", {}) or {}),
+        is_preview=bool(getattr(model, "is_preview", False)),
         logs=model.logs or "[]",
         created_at=model.created_at,
         updated_at=model.updated_at,
@@ -74,6 +75,7 @@ class SQLAlchemyGenerationTaskRepository:
             video_title=task.video_title or "",
             resolution=task.resolution or "",
             bgm_config=task.bgm_config or {},
+            is_preview=task.is_preview or False,
             logs=task.logs,
             created_at=task.created_at,
             updated_at=task.updated_at,
@@ -238,6 +240,8 @@ class SQLAlchemyGenerationTaskRepository:
             model.resolution = task.resolution or ""
         if hasattr(model, "bgm_config"):
             model.bgm_config = task.bgm_config or {}
+        if hasattr(model, "is_preview"):
+            model.is_preview = task.is_preview or False
         model.logs = task.logs
         self.session.commit()
         return task

packages/adapters/sqlalchemy_impl/models.py

@@ -292,6 +292,7 @@ class GenerationTaskModel(Base):
     batch_id = Column(String(36), nullable=False, default="", index=True)
     video_title = Column(String(255), nullable=False, default="")
     resolution = Column(String(20), nullable=False, default="")
+    is_preview = Column(Boolean, nullable=False, default=False, index=True)
     bgm_config = Column(JSON, nullable=False, default=dict)
     extra_meta = Column("metadata", JSON, nullable=False, default=dict)
     logs = Column(Text, nullable=False, default="[]", server_default="[]")

packages/application/generation_tasks.py

@@ -26,6 +26,7 @@ class CreateGenerationTaskCommand:
     bgm_config: dict = field(default_factory=dict)
     auto_retry_enabled: bool = False
     auto_retry_max: int = 0
+    is_preview: bool = False
 
 
 class CreateGenerationTaskUseCase:
@@ -56,6 +57,7 @@ class CreateGenerationTaskUseCase:
             bgm_config=command.bgm_config,
             auto_retry_enabled=command.auto_retry_enabled,
             auto_retry_max=command.auto_retry_max,
+            is_preview=command.is_preview,
         )
         return self.generation_task_repository.create(task)
 

packages/domain/generation_task.py

@@ -115,6 +115,7 @@ class GenerationTask:
     video_title: str = ""
     resolution: str = ""
     bgm_config: dict = field(default_factory=dict)
+    is_preview: bool = False
     logs: str = "[]"
     created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
     updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
@@ -140,6 +141,7 @@ class GenerationTask:
         bgm_config: dict | None = None,
         auto_retry_enabled: bool = False,
         auto_retry_max: int = 0,
+        is_preview: bool = False,
     ) -> "GenerationTask":
         if not project_id.strip() and not template_id.strip():
             raise ValueError("project_id 或 template_id 至少需要提供一个")
@@ -164,6 +166,7 @@ class GenerationTask:
             bgm_config=dict(bgm_config) if bgm_config else {},
             auto_retry_enabled=auto_retry_enabled,
             auto_retry_max=auto_retry_max,
+            is_preview=is_preview,
         )
 
     # ── 状态查询 ────────────────────────────────────────────────────────────