Compare commits

...

6 Commits

Author SHA1 Message Date
CI Bot 1db2ee0808 fix: preserve insertion order in asset_id dedup to fix test ordering
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 38s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m37s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m38s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m47s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 2m25s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m41s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 3m28s
AI Code Review / AI Code Review (pull_request) Failing after 4m43s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 4m2s
CI/CD Pipeline / Unit Tests (pull_request) Successful in 4m29s
CI/CD Pipeline / Validate - Code Quality (pull_request) Successful in 5m44s
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / Integration Tests (pull_request) Successful in 2m33s
CI/CD Pipeline / CI Gate (pull_request) Successful in 9s
ACR Cleanup / ACR Image Cleanup (pull_request_target) Successful in 27s
Preview Cleanup / Cleanup Preview Environment (pull_request) Successful in 46s
Changed set-based dedup to order-preserving dedup in _build_asset_url_map.
Set dedup caused non-deterministic iteration order which broke
side_effect-based mock assertions in tests.
2026-08-17 18:08:26 +08:00
CI Bot ac416493e0 fix: update test mocks for batch query + add asset_id dedup
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 35s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m50s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m51s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m53s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 2m7s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 2m59s
CI/CD Pipeline / Unit Tests (pull_request) Failing after 3m7s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m48s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 3m29s
CI/CD Pipeline / Validate - Code Quality (pull_request) Successful in 6m0s
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
AI Code Review / AI Code Review (pull_request) Successful in 7m1s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 2m34s
CI/CD Pipeline / CI Gate (pull_request) Failing after 8s
- Replace find_by_id mocks with find_by_ids (returns list) in all 28 tests
- Add .id attributes to mock assets for asset_map key matching
- Dedup asset_ids in _build_asset_url_map to avoid redundant signing
- Fixes AI Code Review blocking issues #1 and #3
2026-08-17 16:37:39 +08:00
CI Bot 9e97473eec fix: batch query assets to resolve N+1 in _build_asset_url_map
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 58s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build API Image (pull_request) Failing after 0s
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
CI/CD Pipeline / PR Build Worker Image (pull_request) Failing after 0s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m46s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 2m18s
AI Code Review / AI Code Review (pull_request) Failing after 2m19s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 2m26s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 2m52s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m45s
CI/CD Pipeline / Unit Tests (pull_request) Failing after 3m50s
CI/CD Pipeline / Validate - Code Quality (pull_request) Successful in 6m39s
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled
- Add find_by_ids() to SQLAlchemyAssetRepository (single SQL IN query)
- Replace per-id find_by_id loop with single batch call
- Fixes AI Code Review blocking performance issue in PR #1404
2026-08-17 16:28:49 +08:00
CI Bot 34bd9372ce style: auto-format with black + isort + prettier [skip ci-format-check]
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 37s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m49s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m50s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 2m0s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 2m5s
AI Code Review / AI Code Review (pull_request) Failing after 2m36s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m26s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 3m24s
CI/CD Pipeline / Unit Tests (pull_request) Successful in 3m59s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 4m18s
CI/CD Pipeline / Validate - Code Quality (pull_request) Successful in 7m34s
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / Integration Tests (pull_request) Successful in 5m23s
CI/CD Pipeline / CI Gate (pull_request) Successful in 7s
2026-08-17 07:40:34 +00:00
xiaoxia 2e53c59cd7 fix: remove unused typing.Any import + add clips API response structure tests
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 47s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m35s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m45s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m49s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m58s
AI Code Review / AI Code Review (pull_request) Successful in 3m36s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m37s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 3m7s
CI/CD Pipeline / Unit Tests (pull_request) Successful in 3m15s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 3m41s
CI/CD Pipeline / Validate - Code Quality (pull_request) Has been cancelled
CI/CD Pipeline / Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled
- Remove unused 'from typing import Any' (Ruff F401)
- Add 28 unit/integration tests covering:
  - _clip_to_response: enum conversion, datetime formatting, asset_url param
  - _build_asset_url_map: empty list, storage failure, asset not found,
    no storage_key, success path, exception handling, mixed scenarios
  - Route integration: list/detail/split/merge with asset URL resolution
- Incremental coverage: 100% (95/95 added lines covered)
2026-08-17 15:35:46 +08:00
CI Bot 9aa0c69b26 fix(clips): 全面修复 Clips API 响应结构
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 49s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
CI/CD Pipeline / PR Build Worker Image (pull_request) Failing after 0s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m51s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 2m20s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 2m22s
AI Code Review / AI Code Review (pull_request) Failing after 2m26s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 2m56s
CI/CD Pipeline / Validate - Code Quality (pull_request) Failing after 3m36s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 2m51s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m53s
CI/CD Pipeline / Unit Tests (pull_request) Failing after 3m33s
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / Integration Tests (pull_request) Successful in 2m35s
CI/CD Pipeline / CI Gate (pull_request) Failing after 30s
P0: EditorClipResponse 补全所有数据库字段
- 新增 start_time, transition_duration, asset_id, status,
  template_clip_config_id, created_at, updated_at
- 新增 asset_url (OSS签名URL, 1小时有效)

P1: split/merge 端点统一使用 EditorClipResponse
- split 返回 {left_clip, right_clip} 均为 EditorClipResponse
- merge 返回 {merged_clip, deleted_clip_ids}
- 移除 dict[str, Any] 响应类型

P1: asset_url 签名URL解析
- 批量查询 Asset 表获取 storage_key
- 使用 StorageService.get_download_url() 签名
- 列表/详情/split/merge 端点均支持

P2: ClipsFromAssetsResponse 补上 plan_id 字段
2026-08-17 14:42:58 +08:00
4 changed files with 626 additions and 36 deletions
+110 -35
View File
@@ -16,13 +16,16 @@
from __future__ import annotations
import logging
from typing import Any
from app.auth import AuthenticatedUser, get_current_user
from app.core.storage import get_storage_service
from app.dependencies import get_asset_repository
from app.services.edit_plan_service import EditPlanService
from app.services.edit_template_service import EditTemplateService
from fastapi import APIRouter, Depends, HTTPException, Query, status
from packages.adapters.sqlalchemy_impl.asset_repository import SQLAlchemyAssetRepository
from .dependencies import get_draft_plan_id, get_editor_services
from .schemas import (
ClipBatchDeleteRequest,
@@ -43,30 +46,100 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Template Editor"])
def _clip_to_response(clip) -> EditorClipResponse:
"""统一构造片段响应"""
def _clip_to_response(clip, asset_url: str | None = None) -> EditorClipResponse:
"""统一构造片段响应 — 与 edit_plan_clips 表字段完全对齐"""
def _enum_str(val) -> str:
return val.value if hasattr(val, "value") else str(val)
def _fmt_dt(val) -> str:
if val is None:
return ""
if hasattr(val, "isoformat"):
return val.isoformat()
return str(val)
return EditorClipResponse(
id=clip.id,
plan_id=clip.plan_id,
clip_type=clip.clip_type.value
if hasattr(clip.clip_type, "value")
else str(clip.clip_type),
clip_type=_enum_str(getattr(clip, "clip_type", "")),
order=clip.order,
duration=clip.duration,
start_time=getattr(clip, "start_time", 0.0) or 0.0,
text_content=clip.text_content or "",
transition_effect=clip.transition_effect.value
if hasattr(clip.transition_effect, "value")
else str(clip.transition_effect),
transition_effect=_enum_str(getattr(clip, "transition_effect", "cut")),
transition_duration=getattr(clip, "transition_duration", 0.0) or 0.0,
playback_speed=clip.playback_speed or 1.0,
asset_id=getattr(clip, "asset_id", "") or "",
asset_url=asset_url,
status=getattr(clip, "status", "pending") or "pending",
template_clip_config_id=getattr(clip, "template_clip_config_id", "") or "",
config=clip.config or {},
created_at=_fmt_dt(getattr(clip, "created_at", None)),
updated_at=_fmt_dt(getattr(clip, "updated_at", None)),
)
def _build_asset_url_map(
asset_ids: list[str],
asset_repo: SQLAlchemyAssetRepository,
) -> dict[str, str | None]:
"""批量查询素材并生成签名URL映射.
Returns:
{asset_id: signed_url_or_None}
"""
if not asset_ids:
return {}
# 去重:多个 clip 可能引用同一个素材
# 去重并保持顺序
seen: set[str] = set()
unique_ids = []
for aid in asset_ids:
if aid and aid not in seen:
seen.add(aid)
unique_ids.append(aid)
result: dict[str, str | None] = {}
try:
storage = get_storage_service()
except Exception:
logger.warning("获取存储服务失败,跳过asset_url生成")
return {aid: None for aid in asset_ids}
# 批量查询所有 Asset(单次 SQL IN 查询,避免 N+1)
try:
assets = asset_repo.find_by_ids(unique_ids)
asset_map = {a.id: a for a in assets}
except Exception:
logger.warning("批量查询素材失败: asset_ids=%s", asset_ids, exc_info=True)
return {aid: None for aid in asset_ids if aid}
for aid in unique_ids:
try:
asset = asset_map.get(aid)
if asset is None:
result[aid] = None
continue
storage_key = getattr(asset, "storage_key", None) or ""
if not storage_key:
result[aid] = None
continue
result[aid] = storage.get_download_url(storage_key, expires_seconds=3600)
except Exception:
logger.warning("生成素材签名URL失败: asset_id=%s", aid, exc_info=True)
result[aid] = None
return result
@router.get("/clips", response_model=EditorClipListResponse)
def list_draft_clips(
template_id: str,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
asset_repo: SQLAlchemyAssetRepository = Depends(get_asset_repository),
skip: int = Query(default=0, ge=0),
limit: int = Query(default=100, ge=1, le=500),
_: AuthenticatedUser = Depends(get_current_user),
@@ -75,8 +148,17 @@ def list_draft_clips(
_, plan_svc = services
clips = plan_svc.list_clips(plan_id, skip=skip, limit=limit)
total = plan_svc.count_clips(plan_id)
# 批量解析素材签名URL
asset_ids = [getattr(c, "asset_id", "") or "" for c in clips]
asset_ids = [aid for aid in asset_ids if aid]
url_map = _build_asset_url_map(asset_ids, asset_repo)
return EditorClipListResponse(
items=[_clip_to_response(c) for c in clips],
items=[
_clip_to_response(c, asset_url=url_map.get(getattr(c, "asset_id", "") or ""))
for c in clips
],
total=total,
)
@@ -156,6 +238,7 @@ def get_draft_clip_detail(
clip_id: str,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
asset_repo: SQLAlchemyAssetRepository = Depends(get_asset_repository),
_: AuthenticatedUser = Depends(get_current_user),
):
"""获取草稿中的片段详情"""
@@ -165,16 +248,20 @@ def get_draft_clip_detail(
raise HTTPException(status_code=404, detail="片段不存在")
if clip.plan_id != plan_id:
raise HTTPException(status_code=404, detail="片段不存在")
return _clip_to_response(clip)
asset_id = getattr(clip, "asset_id", "") or ""
url_map = _build_asset_url_map([asset_id], asset_repo) if asset_id else {}
return _clip_to_response(clip, asset_url=url_map.get(asset_id))
@router.post("/clips/{clip_id}/split", response_model=dict[str, Any], status_code=status.HTTP_200_OK)
@router.post("/clips/{clip_id}/split", status_code=status.HTTP_200_OK)
def split_draft_clip(
template_id: str,
clip_id: str,
body: SplitClipRequest,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
asset_repo: SQLAlchemyAssetRepository = Depends(get_asset_repository),
_: AuthenticatedUser = Depends(get_current_user),
):
"""将一个片段从指定时间点分割为两个片段"""
@@ -190,32 +277,22 @@ def split_draft_clip(
) from exc
left = result["left_clip"]
right = result["right_clip"]
asset_ids = [getattr(left, "asset_id", "") or "", getattr(right, "asset_id", "") or ""]
asset_ids = [a for a in asset_ids if a]
url_map = _build_asset_url_map(asset_ids, asset_repo)
return {
"left_clip": {
"id": left.id,
"plan_id": left.plan_id,
"clip_type": left.clip_type,
"order": left.order,
"duration": left.duration,
"start_time": left.start_time,
},
"right_clip": {
"id": right.id,
"plan_id": right.plan_id,
"clip_type": right.clip_type,
"order": right.order,
"duration": right.duration,
"start_time": right.start_time,
},
"left_clip": _clip_to_response(left, asset_url=url_map.get(getattr(left, "asset_id", "") or "")),
"right_clip": _clip_to_response(right, asset_url=url_map.get(getattr(right, "asset_id", "") or "")),
}
@router.post("/clips/merge", response_model=dict[str, Any], status_code=status.HTTP_200_OK)
@router.post("/clips/merge", status_code=status.HTTP_200_OK)
def merge_draft_clips(
template_id: str,
body: MergeClipsRequest,
plan_id: str = Depends(get_draft_plan_id),
services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
asset_repo: SQLAlchemyAssetRepository = Depends(get_asset_repository),
_: AuthenticatedUser = Depends(get_current_user),
):
"""将多个连续的同类型片段合并为一个片段"""
@@ -230,13 +307,11 @@ def merge_draft_clips(
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)
) from exc
asset_id = getattr(merged, "asset_id", "") or ""
url_map = _build_asset_url_map([asset_id], asset_repo) if asset_id else {}
return {
"id": merged.id,
"plan_id": merged.plan_id,
"clip_type": merged.clip_type,
"order": merged.order,
"duration": merged.duration,
"text_content": merged.text_content,
"merged_clip": _clip_to_response(merged, asset_url=url_map.get(asset_id)),
"deleted_clip_ids": body.clip_ids,
}
@@ -227,6 +227,7 @@ class ClipsFromAssetsResponse(BaseModel):
success: bool = True
created_count: int
plan_id: str = ""
message: str = ""
clip_ids: List[str] = Field(default_factory=list, description="创建的片段ID列表")
@@ -439,17 +440,28 @@ class EditorUpdateRequest(BaseModel):
class EditorClipResponse(BaseModel):
"""片段响应"""
"""片段响应 — 与数据库 edit_plan_clips 表字段对齐"""
id: str
plan_id: str
clip_type: str
order: int
duration: float
start_time: float = 0.0
text_content: str = ""
transition_effect: str = "cut"
transition_duration: float = 0.0
playback_speed: float = 1.0
asset_id: str = ""
asset_url: str | None = Field(
default=None,
description="素材视频签名URL(1小时有效),用于前端预览播放",
)
status: str = "pending"
template_clip_config_id: str = ""
config: dict[str, Any] = Field(default_factory=dict)
created_at: str = ""
updated_at: str = ""
class EditorClipListResponse(BaseModel):
@@ -101,6 +101,13 @@ class SQLAlchemyAssetRepository:
return None
return self._to_domain(model)
def find_by_ids(self, asset_ids: list[str]) -> list[Asset]:
"""批量查询素材(单次 SQL IN 查询,避免 N+1)。"""
if not asset_ids:
return []
models = self.session.query(AssetModel).filter(AssetModel.id.in_(asset_ids)).all()
return [self._to_domain(m) for m in models]
def get(self, asset_id: str) -> Asset | None:
return self.find_by_id(asset_id)
@@ -0,0 +1,496 @@
"""片段管理路由 clips.py 增量覆盖率测试.
覆盖 PR fix/clips-api-response-structure 新增代码:
- _clip_to_response: 枚举转换、日期格式化、asset_url 参数
- _build_asset_url_map: 批量素材 URL 解析(空列表/异常/正常路径)
- 路由层 asset_repo 注入与 URL 拼接逻辑
"""
from __future__ import annotations
import os
import sys
from enum import Enum
from pathlib import Path
from unittest.mock import MagicMock, patch
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
# ---------------------------------------------------------------------------
# 常量与工厂
# ---------------------------------------------------------------------------
TEST_TEMPLATE_ID = "tmpl-test-001"
TEST_PLAN_ID = "plan-draft-001"
TEST_USER_ID = "user-001"
def _auth_user():
u = MagicMock()
u.user.id = TEST_USER_ID
u.user_id = TEST_USER_ID
return u
def _clip(**overrides):
"""构造 mock clip,支持 Enum 类型字段"""
c = MagicMock()
c.id = overrides.get("id", "clip-001")
c.plan_id = overrides.get("plan_id", TEST_PLAN_ID)
c.clip_type = overrides.get("clip_type", "video")
c.order = overrides.get("order", 0)
c.duration = overrides.get("duration", 10.0)
c.start_time = overrides.get("start_time", 0.0)
c.text_content = overrides.get("text_content", "")
c.transition_effect = overrides.get("transition_effect", "cut")
c.transition_duration = overrides.get("transition_duration", 0.0)
c.playback_speed = overrides.get("playback_speed", 1.0)
c.asset_id = overrides.get("asset_id", "")
c.status = overrides.get("status", "ready")
c.template_clip_config_id = overrides.get("template_clip_config_id", "")
c.config = overrides.get("config", {})
c.created_at = overrides.get("created_at", None)
c.updated_at = overrides.get("updated_at", None)
return c
def _services(plan_svc_overrides=None):
tpl = MagicMock()
plan = MagicMock()
if plan_svc_overrides:
for k, v in plan_svc_overrides.items():
setattr(plan, k, v)
return tpl, plan
# ---------------------------------------------------------------------------
# 单元测试: _clip_to_response
# ---------------------------------------------------------------------------
class TestClipToResponse:
"""_clip_to_response 纯函数测试 — 覆盖行 53-80"""
def test_basic_fields(self):
from app.api.routes.templates_editor.clips import _clip_to_response
c = _clip(id="c1", order=3, duration=5.5, text_content="hello")
resp = _clip_to_response(c)
assert resp.id == "c1"
assert resp.order == 3
assert resp.duration == 5.5
assert resp.text_content == "hello"
assert resp.asset_url is None
def test_enum_clip_type(self):
"""Enum 值应被 .value 解包"""
from app.api.routes.templates_editor.clips import _clip_to_response
class ClipType(str, Enum):
VIDEO = "video"
AUDIO = "audio"
c = _clip(clip_type=ClipType.VIDEO)
resp = _clip_to_response(c)
assert resp.clip_type == "video"
def test_plain_string_clip_type(self):
"""非 Enum 字符串直接用 str()"""
from app.api.routes.templates_editor.clips import _clip_to_response
c = _clip(clip_type="main")
resp = _clip_to_response(c)
assert resp.clip_type == "main"
def test_enum_transition_effect(self):
from app.api.routes.templates_editor.clips import _clip_to_response
class Transition(str, Enum):
FADE = "fade"
c = _clip(transition_effect=Transition.FADE)
resp = _clip_to_response(c)
assert resp.transition_effect == "fade"
def test_default_transition_when_none(self):
"""transition_effect 缺失时默认 cut"""
from app.api.routes.templates_editor.clips import _clip_to_response
c = _clip()
del c.transition_effect # 触发 getattr default
resp = _clip_to_response(c)
assert resp.transition_effect == "cut"
def test_asset_url_passed(self):
"""asset_url 参数应透传到响应"""
from app.api.routes.templates_editor.clips import _clip_to_response
c = _clip(asset_id="a1")
resp = _clip_to_response(c, asset_url="https://signed-url.example.com/video.mp4")
assert resp.asset_url == "https://signed-url.example.com/video.mp4"
def test_asset_url_none_by_default(self):
from app.api.routes.templates_editor.clips import _clip_to_response
c = _clip()
resp = _clip_to_response(c)
assert resp.asset_url is None
def test_datetime_isoformat(self):
"""datetime 对象应被 isoformat()"""
from datetime import datetime
from app.api.routes.templates_editor.clips import _clip_to_response
dt = datetime(2026, 8, 17, 12, 0, 0)
c = _clip(created_at=dt, updated_at=dt)
resp = _clip_to_response(c)
assert "2026-08-17" in resp.created_at
assert "2026-08-17" in resp.updated_at
def test_none_datetime_empty_string(self):
"""None 日期应格式化为空字符串"""
from app.api.routes.templates_editor.clips import _clip_to_response
c = _clip(created_at=None, updated_at=None)
resp = _clip_to_response(c)
assert resp.created_at == ""
assert resp.updated_at == ""
def test_string_datetime_passthrough(self):
"""已经是字符串的日期直接 str()"""
from app.api.routes.templates_editor.clips import _clip_to_response
c = _clip(created_at="2026-08-17T00:00:00")
resp = _clip_to_response(c)
assert resp.created_at == "2026-08-17T00:00:00"
def test_none_defaults_for_optional_fields(self):
"""None/缺失字段的默认值"""
from app.api.routes.templates_editor.clips import _clip_to_response
c = _clip(asset_id=None, status=None, template_clip_config_id=None)
resp = _clip_to_response(c)
assert resp.asset_id == ""
assert resp.status == "pending"
assert resp.template_clip_config_id == ""
def test_zero_duration_fallback(self):
"""duration=0 → playback_speed 默认 1.0"""
from app.api.routes.templates_editor.clips import _clip_to_response
c = _clip(playback_speed=None)
resp = _clip_to_response(c)
assert resp.playback_speed == 1.0
# ---------------------------------------------------------------------------
# 单元测试: _build_asset_url_map
# ---------------------------------------------------------------------------
class TestBuildAssetUrlMap:
"""_build_asset_url_map 测试 — 覆盖行 93-118"""
def test_empty_list(self):
"""空 asset_ids 直接返回空 dict"""
from app.api.routes.templates_editor.clips import _build_asset_url_map
repo = MagicMock()
result = _build_asset_url_map([], repo)
assert result == {}
@patch("app.api.routes.templates_editor.clips.get_storage_service")
def test_storage_service_failure(self, mock_get_storage):
"""存储服务获取失败时返回全 None"""
from app.api.routes.templates_editor.clips import _build_asset_url_map
mock_get_storage.side_effect = RuntimeError("storage unavailable")
repo = MagicMock()
result = _build_asset_url_map(["a1", "a2"], repo)
assert result == {"a1": None, "a2": None}
@patch("app.api.routes.templates_editor.clips.get_storage_service")
def test_asset_not_found(self, mock_get_storage):
"""asset_id 找不到对应素材 → None"""
from app.api.routes.templates_editor.clips import _build_asset_url_map
storage = MagicMock()
mock_get_storage.return_value = storage
repo = MagicMock()
repo.find_by_ids.return_value = []
result = _build_asset_url_map(["missing-id"], repo)
assert result == {"missing-id": None}
@patch("app.api.routes.templates_editor.clips.get_storage_service")
def test_no_storage_key(self, mock_get_storage):
"""素材没有 storage_key → None"""
from app.api.routes.templates_editor.clips import _build_asset_url_map
storage = MagicMock()
mock_get_storage.return_value = storage
repo = MagicMock()
asset = MagicMock()
asset.id = "a1"
asset.storage_key = ""
repo.find_by_ids.return_value = [asset]
result = _build_asset_url_map(["a1"], repo)
assert result == {"a1": None}
@patch("app.api.routes.templates_editor.clips.get_storage_service")
def test_successful_url_generation(self, mock_get_storage):
"""正常路径:返回签名 URL"""
from app.api.routes.templates_editor.clips import _build_asset_url_map
storage = MagicMock()
storage.get_download_url.return_value = "https://cdn.example.com/signed.mp4"
mock_get_storage.return_value = storage
repo = MagicMock()
asset = MagicMock()
asset.id = "a1"
asset.storage_key = "videos/test.mp4"
repo.find_by_ids.return_value = [asset]
result = _build_asset_url_map(["a1"], repo)
assert result == {"a1": "https://cdn.example.com/signed.mp4"}
storage.get_download_url.assert_called_once_with("videos/test.mp4", expires_seconds=3600)
@patch("app.api.routes.templates_editor.clips.get_storage_service")
def test_exception_during_url_generation(self, mock_get_storage):
"""单个 asset 生成 URL 异常 → None,不影响其他"""
from app.api.routes.templates_editor.clips import _build_asset_url_map
storage = MagicMock()
storage.get_download_url.side_effect = [Exception("boom"), "https://ok.com/v2"]
mock_get_storage.return_value = storage
repo = MagicMock()
asset1 = MagicMock()
asset1.id = "a1"
asset1.storage_key = "v1.mp4"
asset2 = MagicMock()
asset2.id = "a2"
asset2.storage_key = "v2.mp4"
repo.find_by_ids.return_value = [asset1, asset2]
result = _build_asset_url_map(["a1", "a2"], repo)
assert result["a1"] is None
assert result["a2"] == "https://ok.com/v2"
@patch("app.api.routes.templates_editor.clips.get_storage_service")
def test_skip_empty_asset_id(self, mock_get_storage):
"""空字符串 asset_id 被跳过"""
from app.api.routes.templates_editor.clips import _build_asset_url_map
storage = MagicMock()
mock_get_storage.return_value = storage
repo = MagicMock()
result = _build_asset_url_map(["", "a1"], repo)
# "" not in result because it's skipped by `if not aid: continue`
assert "" not in result
@patch("app.api.routes.templates_editor.clips.get_storage_service")
def test_multiple_assets_mixed(self, mock_get_storage):
"""混合场景:正常+异常+缺失"""
from app.api.routes.templates_editor.clips import _build_asset_url_map
storage = MagicMock()
storage.get_download_url.return_value = "https://cdn.com/ok.mp4"
mock_get_storage.return_value = storage
repo = MagicMock()
good_asset = MagicMock()
good_asset.id = "a1"
good_asset.storage_key = "good.mp4"
# a1=good, a2=not found, a3=good
good_asset2 = MagicMock()
good_asset2.id = "a3"
good_asset2.storage_key = "good.mp4"
repo.find_by_ids.return_value = [good_asset, good_asset2]
result = _build_asset_url_map(["a1", "a2", "a3"], repo)
assert result["a1"] == "https://cdn.com/ok.mp4"
assert result["a2"] is None
assert result["a3"] == "https://cdn.com/ok.mp4"
# ---------------------------------------------------------------------------
# 集成测试: 路由层 asset_repo 注入
# ---------------------------------------------------------------------------
class TestClipRoutesAssetIntegration:
"""路由层测试 — 覆盖 asset_url 在 list/detail/split/merge 中的拼接逻辑"""
def _create_app(self, plan_svc_config=None):
from app.api.routes import templates_editor as editor_module
from app.dependencies import get_asset_repository
mock_clip_1 = _clip(id="c1", asset_id="asset-001")
mock_clip_2 = _clip(id="c2", asset_id="")
mock_tpl_svc = MagicMock()
mock_plan_svc = MagicMock()
mock_plan_svc.list_clips.return_value = [mock_clip_1, mock_clip_2]
mock_plan_svc.count_clips.return_value = 2
mock_plan_svc.get_clip.return_value = mock_clip_1
mock_plan_svc.create_clip.return_value = _clip(id="c-new", asset_id="")
mock_plan_svc.update_clip.return_value = _clip(id="c1", duration=15.0)
mock_plan_svc.delete_clip.return_value = True
mock_plan_svc.split_clip.return_value = {
"left_clip": _clip(id="c-left", asset_id="asset-L"),
"right_clip": _clip(id="c-right", asset_id="asset-R"),
}
mock_plan_svc.merge_clips.return_value = _clip(id="c-merged", asset_id="asset-M")
if plan_svc_config:
for k, v in plan_svc_config.items():
setattr(mock_plan_svc, k, v)
def _deps():
return mock_tpl_svc, mock_plan_svc
mock_asset_repo = MagicMock()
app = FastAPI()
app.include_router(
editor_module.router,
prefix="/api/v1/templates/{template_id}/editor",
)
app.dependency_overrides[editor_module.get_current_user] = _auth_user
app.dependency_overrides[editor_module.get_draft_plan_id] = lambda: TEST_PLAN_ID
app.dependency_overrides[editor_module.get_editor_services] = _deps
app.dependency_overrides[get_asset_repository] = lambda: mock_asset_repo
return TestClient(app), mock_plan_svc, mock_asset_repo
@patch("app.api.routes.templates_editor.clips.get_storage_service")
def test_list_clips_includes_asset_urls(self, mock_get_storage):
"""GET /clips 应为有 asset_id 的片段返回签名 URL"""
storage = MagicMock()
storage.get_download_url.return_value = "https://cdn.com/c1.mp4"
mock_get_storage.return_value = storage
client, _, asset_repo = self._create_app()
asset = MagicMock()
asset.id = "asset-001"
asset.storage_key = "videos/c1.mp4"
asset_repo.find_by_ids.return_value = [asset]
resp = client.get(f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor/clips")
assert resp.status_code == 200
data = resp.json()
items = data["items"]
assert len(items) == 2
# c1 has asset_id → should have url
assert items[0]["asset_url"] == "https://cdn.com/c1.mp4"
# c2 has empty asset_id → None
assert items[1]["asset_url"] is None
@patch("app.api.routes.templates_editor.clips.get_storage_service")
def test_get_clip_detail_with_asset_url(self, mock_get_storage):
"""GET /clips/{clip_id} 应返回素材签名 URL"""
storage = MagicMock()
storage.get_download_url.return_value = "https://cdn.com/detail.mp4"
mock_get_storage.return_value = storage
client, _, asset_repo = self._create_app()
asset = MagicMock()
asset.id = "asset-001"
asset.storage_key = "videos/detail.mp4"
asset_repo.find_by_ids.return_value = [asset]
resp = client.get(f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor/clips/clip-001")
assert resp.status_code == 200
assert resp.json()["asset_url"] == "https://cdn.com/detail.mp4"
def test_get_clip_detail_no_asset(self):
"""片段没有 asset_id 时不应调用 URL 解析"""
client, plan_svc, asset_repo = self._create_app()
# 返回没有 asset_id 的片段
plan_svc.get_clip.return_value = _clip(id="c-no-asset", asset_id="")
resp = client.get(f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor/clips/c-no-asset")
assert resp.status_code == 200
assert resp.json()["asset_url"] is None
@patch("app.api.routes.templates_editor.clips.get_storage_service")
def test_split_clip_returns_asset_urls(self, mock_get_storage):
"""POST /clips/{clip_id}/split 返回的左右片段应带签名 URL"""
storage = MagicMock()
storage.get_download_url.side_effect = ["https://cdn.com/L.mp4", "https://cdn.com/R.mp4"]
mock_get_storage.return_value = storage
client, _, asset_repo = self._create_app()
asset_l = MagicMock()
asset_l.storage_key = "videos/L.mp4"
asset_r = MagicMock()
asset_r.storage_key = "videos/R.mp4"
asset_l.id = "asset-L"
asset_r.id = "asset-R"
asset_repo.find_by_ids.return_value = [asset_l, asset_r]
resp = client.post(
f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor/clips/clip-001/split",
json={"split_time": 5.0},
)
assert resp.status_code == 200
data = resp.json()
assert data["left_clip"]["asset_url"] == "https://cdn.com/L.mp4"
assert data["right_clip"]["asset_url"] == "https://cdn.com/R.mp4"
@patch("app.api.routes.templates_editor.clips.get_storage_service")
def test_merge_clips_returns_asset_url(self, mock_get_storage):
"""POST /clips/merge 返回的合并片段应带签名 URL"""
storage = MagicMock()
storage.get_download_url.return_value = "https://cdn.com/M.mp4"
mock_get_storage.return_value = storage
client, _, asset_repo = self._create_app()
asset = MagicMock()
asset.id = "asset-M"
asset.storage_key = "videos/M.mp4"
asset_repo.find_by_ids.return_value = [asset]
resp = client.post(
f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor/clips/merge",
json={"clip_ids": ["c1", "c2"]},
)
assert resp.status_code == 200
data = resp.json()
assert data["merged_clip"]["asset_url"] == "https://cdn.com/M.mp4"
assert data["deleted_clip_ids"] == ["c1", "c2"]
def test_merge_clips_not_found(self):
"""merge 时某片段不存在应返回 404"""
client, plan_svc, _ = self._create_app()
plan_svc.get_clip.return_value = None
resp = client.post(
f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor/clips/merge",
json={"clip_ids": ["nonexistent-1", "nonexistent-2"]},
)
assert resp.status_code == 404
def test_delete_clip_success(self):
"""DELETE /clips/{clip_id} 成功返回 204"""
client, _, _ = self._create_app()
resp = client.delete(f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor/clips/clip-001")
assert resp.status_code == 204
def test_delete_clip_not_found(self):
"""DELETE 片段不存在返回 404"""
client, plan_svc, _ = self._create_app()
plan_svc.delete_clip.return_value = False
resp = client.delete(f"/api/v1/templates/{TEST_TEMPLATE_ID}/editor/clips/bad-id")
assert resp.status_code == 404