461 lines
18 KiB
Python
Executable File
461 lines
18 KiB
Python
Executable File
"""预览生成路由 — 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.storage import get_storage_service
|
|
from app.core.task_enqueue import (
|
|
GLOBAL_PENDING_LIMIT,
|
|
USER_PENDING_LIMIT,
|
|
GlobalQueueFull,
|
|
UserPendingLimitExceeded,
|
|
safe_enqueue_generation_task,
|
|
)
|
|
from app.dependencies import (
|
|
get_asset_repository,
|
|
get_db_session,
|
|
get_generated_video_repository,
|
|
get_generation_task_repository,
|
|
)
|
|
from app.schemas.generation_task import (
|
|
CreatePreviewGenerationTaskRequest,
|
|
PreviewGenerationTaskResponse,
|
|
)
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
|
|
from packages.adapters.sqlalchemy_impl.edit_template_repository import (
|
|
SQLAlchemyEditTemplateRepository,
|
|
)
|
|
from packages.adapters.sqlalchemy_impl.template_repository import (
|
|
SQLAlchemyTemplateRepository,
|
|
)
|
|
from packages.application import (
|
|
CreateGenerationTaskCommand,
|
|
CreateGenerationTaskUseCase,
|
|
GetGenerationTaskUseCase,
|
|
ListGeneratedVideosByTaskUseCase,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
# 模板 mode → 视频比例映射
|
|
_TEMPLATE_MODE_TO_RATIO = {
|
|
"pip": "9:16",
|
|
"standard": "16:9",
|
|
"square": "1:1",
|
|
}
|
|
|
|
|
|
def _infer_video_ratio_from_template(template_id: str, db: Session, user_id: str = "") -> str:
|
|
"""从模板 mode 推断视频比例，前端未传 video_ratio 时使用。
|
|
|
|
Returns:
|
|
视频比例字符串（如 "9:16"），查询失败返回空字符串。
|
|
"""
|
|
if not template_id:
|
|
return ""
|
|
try:
|
|
repo = SQLAlchemyTemplateRepository(db)
|
|
template = repo.get(template_id, user_id)
|
|
if template:
|
|
mode = getattr(template, "mode", "") or ""
|
|
ratio = _TEMPLATE_MODE_TO_RATIO.get(mode.strip(), "")
|
|
if ratio:
|
|
logger.info(
|
|
"[预览生成] 从模板 mode=%s 推断 video_ratio=%s",
|
|
mode,
|
|
ratio,
|
|
)
|
|
return ratio
|
|
except Exception:
|
|
logger.warning(
|
|
"[预览生成] 查询模板失败，跳过 video_ratio 推断: template_id=%s",
|
|
template_id,
|
|
exc_info=True,
|
|
)
|
|
return ""
|
|
|
|
|
|
def _resolve_strategy_id_from_template(template_id: str, db: Session, user_id: str = "") -> str:
|
|
"""从模板读取 editing_mode / mode 作为 strategy_id。
|
|
|
|
优先查新模板系统（EditTemplate.editing_mode），fallback 旧模板（Template.mode）。
|
|
Worker 端使用 strategy_id 作为渲染 mode，为空则默认 one_take。
|
|
"""
|
|
if not template_id:
|
|
return ""
|
|
|
|
# 优先查新模板系统
|
|
try:
|
|
new_repo = SQLAlchemyEditTemplateRepository(db)
|
|
new_template = new_repo.get(template_id)
|
|
if new_template and getattr(new_template, "editing_mode", ""): # type: ignore[arg-type]
|
|
mode = new_template.editing_mode.strip()
|
|
if mode:
|
|
logger.info(
|
|
"[预览生成] 从新模板 editing_mode=%s (template_id=%s)",
|
|
mode,
|
|
template_id,
|
|
)
|
|
# 画中画已下线，pip/voice_pip 统一映射为 one_take
|
|
if mode in ("pip", "voice_pip"):
|
|
logger.info("[预览生成] %s → one_take (画中画已下线)", mode)
|
|
mode = "one_take"
|
|
return mode
|
|
except Exception:
|
|
logger.debug(
|
|
"[预览生成] 新模板查询失败，尝试旧模板: template_id=%s",
|
|
template_id,
|
|
exc_info=True,
|
|
)
|
|
|
|
# fallback 旧模板系统
|
|
try:
|
|
old_repo = SQLAlchemyTemplateRepository(db)
|
|
old_template = old_repo.get(template_id, user_id)
|
|
if old_template:
|
|
mode = getattr(old_template, "mode", "") or ""
|
|
mode = mode.strip()
|
|
if mode:
|
|
logger.info(
|
|
"[预览生成] 从旧模板 mode=%s (template_id=%s)",
|
|
mode,
|
|
template_id,
|
|
)
|
|
# 画中画已下线，pip/voice_pip 统一映射为 one_take
|
|
if mode in ("pip", "voice_pip"):
|
|
logger.info("[预览生成] %s → one_take (画中画已下线)", mode)
|
|
mode = "one_take"
|
|
return mode
|
|
except Exception:
|
|
logger.warning(
|
|
"[预览生成] 旧模板查询也失败，strategy_id 留空: template_id=%s",
|
|
template_id,
|
|
exc_info=True,
|
|
)
|
|
|
|
return ""
|
|
|
|
|
|
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]
|
|
raw_url = getattr(first_video, "file_url", "") or ""
|
|
# rendered/* 已配置公开读，直接用裸 URL
|
|
if raw_url.startswith("http"):
|
|
video_url = raw_url
|
|
else:
|
|
storage = get_storage_service()
|
|
video_url = storage.get_url(raw_url)
|
|
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", "") or "",
|
|
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),
|
|
db: Session = Depends(get_db_session),
|
|
asset_repo=Depends(get_asset_repository),
|
|
) -> PreviewGenerationTaskResponse:
|
|
"""创建预览生成任务。
|
|
|
|
预览渲染品质与正式生成一致（1080p, CRF 23, medium preset），确认生成时可直接复用预览产物。
|
|
|
|
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, preview_count=%d",
|
|
user_id,
|
|
request.template_id,
|
|
len(request.asset_ids),
|
|
request.preview_count,
|
|
)
|
|
|
|
# 预检查队列限流
|
|
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
|
|
|
|
# 确定视频比例：优先前端传入，否则从模板 mode 推断
|
|
video_ratio = request.video_ratio or ""
|
|
if not video_ratio and request.template_id:
|
|
video_ratio = _infer_video_ratio_from_template(request.template_id, db, user_id)
|
|
|
|
# 根据 video_ratio 计算输出分辨率（默认竖屏 1080x1920）
|
|
output_width, output_height = 1080, 1920
|
|
if video_ratio:
|
|
parts = video_ratio.split(":")
|
|
if len(parts) == 2:
|
|
try:
|
|
w, h = int(parts[0]), int(parts[1])
|
|
base = 1920
|
|
if w < h:
|
|
# 竖屏
|
|
output_width = round(base * w / h)
|
|
output_height = base
|
|
else:
|
|
# 横屏
|
|
output_width = base
|
|
output_height = round(base * h / w)
|
|
# 对齐到偶数
|
|
output_width = output_width - output_width % 2
|
|
output_height = output_height - output_height % 2
|
|
except (ValueError, ZeroDivisionError):
|
|
output_width, output_height = 1080, 1920
|
|
resolution = f"{output_width}x{output_height}"
|
|
|
|
logger.info(
|
|
"[预览生成] 分辨率: video_ratio=%s → %s (%dx%d)",
|
|
video_ratio, resolution, output_width, output_height,
|
|
)
|
|
|
|
# 从模板读取 editing_mode / mode 作为 strategy_id（渲染 pipeline 的 mode 参数）
|
|
strategy_id = _resolve_strategy_id_from_template(request.template_id, db, user_id)
|
|
|
|
title_config = request.title_config or {}
|
|
|
|
use_case = CreateGenerationTaskUseCase(generation_task_repository)
|
|
|
|
try:
|
|
task = use_case.execute(
|
|
CreateGenerationTaskCommand(
|
|
project_id="",
|
|
asset_library_id="",
|
|
strategy_id=strategy_id,
|
|
voice_library_id=request.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=request.source_edit_plan_id,
|
|
asset_select_mode="",
|
|
batch_id="",
|
|
video_title=request.video_title,
|
|
resolution=resolution,
|
|
bgm_config=request.bgm_config or {},
|
|
auto_retry_enabled=False,
|
|
auto_retry_max=0,
|
|
is_preview=True,
|
|
title_config=title_config,
|
|
output_width=output_width,
|
|
output_height=output_height,
|
|
)
|
|
)
|
|
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
|
|
|
|
# 关联编辑计划：如果前端未传 source_edit_plan_id，通过 template_id + user_id 查找
|
|
if not task.source_edit_plan_id and request.template_id:
|
|
try:
|
|
from packages.adapters.sqlalchemy_impl.edit_plan_repository import (
|
|
SQLAlchemyEditPlanRepository,
|
|
)
|
|
|
|
_plan_repo = SQLAlchemyEditPlanRepository(db)
|
|
_plans = _plan_repo.list_by_template(request.template_id, limit=20)
|
|
for _p in _plans:
|
|
if (_p.created_by_user_id or "") == user_id:
|
|
task.source_edit_plan_id = _p.id
|
|
generation_task_repository.update(task)
|
|
logger.info(
|
|
"[预览生成] 自动关联编辑计划: task_id=%s plan_id=%s",
|
|
task.id,
|
|
_p.id,
|
|
)
|
|
break
|
|
except Exception:
|
|
logger.warning(
|
|
"[预览生成] 查找关联编辑计划失败（不影响主流程）: task_id=%s",
|
|
task.id,
|
|
exc_info=True,
|
|
)
|
|
|
|
# 每条预览都关联独立克隆 plan：多预览前端为 N 次并发调用，若共用同一 plan
|
|
# 则 N 条预览片段完全相同；克隆时片段起点按持久化历史区间重算（含受控复用），
|
|
# 保证各预览版本内容不同
|
|
if task.source_edit_plan_id:
|
|
try:
|
|
from app.services.edit_plan_service import EditPlanService
|
|
|
|
_plan_svc = EditPlanService(db)
|
|
_preview_plan = _plan_svc.clone_plan_for_variant(
|
|
task.source_edit_plan_id,
|
|
created_by_user_id=user_id,
|
|
name_suffix="预览变体",
|
|
)
|
|
task.source_edit_plan_id = _preview_plan.id
|
|
generation_task_repository.update(task)
|
|
logger.info(
|
|
"[预览生成] 预览关联独立克隆 plan: task_id=%s clone_plan_id=%s",
|
|
task.id,
|
|
_preview_plan.id,
|
|
)
|
|
except Exception as clone_err:
|
|
# 不退回共用原 plan（否则多条预览内容相同，违反去重诉求）：
|
|
# 标记任务失败并中断，前端可重新发起预览
|
|
logger.error(
|
|
"[预览生成] 克隆预览变体 plan 失败，任务标记失败: task_id=%s error=%s",
|
|
task.id,
|
|
clone_err,
|
|
exc_info=True,
|
|
)
|
|
_mark_task_failed(generation_task_repository, task, "预览变体计划创建失败")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail="创建预览任务失败：无法生成独立剪辑计划，请重试",
|
|
) from clone_err
|
|
|
|
# 入队执行；若入队失败则标记任务为 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 as e:
|
|
_mark_task_failed(generation_task_repository, task, "待处理任务超限")
|
|
raise HTTPException(
|
|
status_code=429,
|
|
detail=f"您的待处理任务过多（当前 {e.pending_count - 1}/{e.limit}），请等待后再提交",
|
|
) 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)
|