Compare commits

..

14 Commits

Author SHA1 Message Date
xiaoxia 1a5bb1ab43 feat(ci): optimize web image build with multi-stage Dockerfile + buildx cache
Tests / test (pull_request) Failing after 0s
Tests / lint (pull_request) Failing after 0s
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 1m59s
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Successful in 2m12s
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (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
2026-07-13 15:11:28 +08:00
xiaoxia bf47e69fa4 feat(ci): add node_modules cache for frontend-lint job 2026-07-13 15:10:44 +08:00
xiaoxia b947bf7569 feat(ci): add node_modules cache for frontend-lint job 2026-07-13 15:09:47 +08:00
xiaoxia f110bc6f16 feat: web.Dockerfile add buildkit cache for npm install 2026-07-13 15:08:49 +08:00
xiaoxia 0c9375ff32 ci: remove temporary debug workflow - test-ssh-secret.yml
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 36s
CI/CD Pipeline / Frontend Lint (push) Failing after 0s
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (push) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (push) Has been skipped
CI/CD Pipeline / Staging E2E Tests (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
Fix Flake8 E741 for PR232 / fix-flake8 (push) Failing after 3h4m59s
2026-07-12 10:25:12 +08:00
xiaoxia ef344e9ffc cleanup: remove upgrade test file
CI/CD Pipeline / Build Production Runtime Images (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Staging E2E Tests (push) Has been skipped
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (push) Failing after 12h43m56s
CI/CD Pipeline / Frontend Lint (push) Failing after 12h45m50s
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 12h45m50s
2026-07-09 00:08:40 +08:00
xiaoxia 0d4904433e test: trigger workflow after gitea upgrade
CI/CD Pipeline / Build Production Runtime Images (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Staging E2E Tests (push) Failing after 12h49m24s
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (push) Failing after 12h53m37s
CI/CD Pipeline / Frontend Lint (push) Failing after 12h56m8s
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 12h56m8s
2026-07-08 23:58:20 +08:00
CI Test 708662394f Merge develop into main - v0.1.126
Auto Merge PRs / auto-merge (push) Has been cancelled
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 29h56m45s
CI/CD Pipeline / Frontend Lint (push) Failing after 29h56m45s
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (push) Has been skipped
CI/CD Pipeline / Staging E2E Tests (push) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
2026-07-08 06:57:42 +08:00
CI Test 9b034764ad Merge develop into main - v0.1.125
Auto Merge PRs / auto-merge (push) Failing after 1m29s
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 40h59m5s
CI/CD Pipeline / Frontend Lint (push) Failing after 40h58m8s
CI/CD Pipeline / Deploy Staging (push) Has been skipped
CI/CD Pipeline / Staging E2E Tests (push) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
2026-07-07 19:54:54 +08:00
CI Test 8748b43070 Merge develop into main - v0.1.124
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 45h17m30s
CI/CD Pipeline / Frontend Lint (push) Failing after 45h17m8s
CI/CD Pipeline / Deploy Staging (push) Failing after 45h15m38s
CI/CD Pipeline / Staging E2E Tests (push) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
2026-07-07 15:36:36 +08:00
CI Test d213a055a1 Merge develop into main - v0.1.123
Auto Merge PRs / auto-merge (push) Failing after 3m27s
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 47h9m34s
CI/CD Pipeline / Frontend Lint (push) Failing after 47h8m47s
CI/CD Pipeline / Deploy Staging (push) Failing after 47h6m52s
CI/CD Pipeline / Staging E2E Tests (push) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
2026-07-07 13:44:46 +08:00
CI Test 2371860f82 Merge develop into main - v0.1.122 (prettier fix)
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 49h37m40s
CI/CD Pipeline / Frontend Lint (push) Failing after 49h37m23s
CI/CD Pipeline / Deploy Staging (push) Has been skipped
CI/CD Pipeline / Staging E2E Tests (push) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (push) Failing after 49h35m42s
CI/CD Pipeline / Deploy Production (push) Failing after 49h33m12s
CI/CD Pipeline / Production Browser E2E (push) Failing after 49h27m0s
2026-07-07 11:04:39 +08:00
CI Test dbd956fc6e Merge develop into main - v0.1.122
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 49h54m54s
CI/CD Pipeline / Frontend Lint (push) Failing after 49h54m54s
CI/CD Pipeline / Deploy Staging (push) Has been skipped
CI/CD Pipeline / Staging E2E Tests (push) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
2026-07-07 10:59:31 +08:00
CI Test 1d59ee5336 Merge develop into main - v0.1.121
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 52h48m47s
CI/CD Pipeline / Frontend Lint (push) Failing after 52h48m2s
CI/CD Pipeline / Deploy Staging (push) Failing after 52h46m33s
CI/CD Pipeline / Staging E2E Tests (push) Failing after 52h45m33s
CI/CD Pipeline / Build Production Runtime Images (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
2026-07-07 08:05:42 +08:00
41 changed files with 881 additions and 6501 deletions
+5 -49
View File
@@ -7,59 +7,15 @@ on:
jobs:
auto-merge:
runs-on: saas
timeout-minutes: 10
runs-on: ubuntu-latest
steps:
- name: Checkout code
shell: sh
env:
GITHUB_TOKEN: ${{ github.token }}
run: |
set -eu
python3 - <<'PY'
import io, os, tarfile, time, urllib.request, urllib.error
url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/{os.environ['GITHUB_SHA']}.tar.gz"
request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
last_err = None
for attempt in range(5):
try:
with urllib.request.urlopen(request, timeout=120) as response:
archive = response.read()
break
except urllib.error.HTTPError as e:
last_err = e
if e.code >= 500 and attempt < 4:
wait = 2 ** attempt
print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
time.sleep(wait)
continue
raise
except Exception as e:
last_err = e
if attempt < 4:
wait = 2 ** attempt
print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
time.sleep(wait)
continue
raise
else:
raise last_err
with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
top_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
for member in tar.getmembers():
name = member.name
if name == top_prefix[:-1]:
continue
if name.startswith(top_prefix):
member.name = name[len(top_prefix):]
if member.name:
tar.extract(member, '.')
PY
- name: Checkout
uses: actions/checkout@v3
- name: Auto merge develop PRs
run: |
bash scripts/auto_merge_prs.sh develop
- name: Auto merge main PRs (release only)
run: |
bash scripts/auto_merge_prs.sh main
File diff suppressed because one or more lines are too long
-69
View File
@@ -1,69 +0,0 @@
name: Test SSH Secret
on:
push:
branches: [develop]
paths:
- '.gitea/workflows/test-ssh-secret.yml'
jobs:
test-ssh:
runs-on: ubuntu-22.04
steps:
- name: Install SSH client
run: |
which ssh || (apt-get update && apt-get install -y openssh-client)
ssh -V
- name: Debug environment
run: |
echo "=== Environment ==="
echo "Runner hostname: $(hostname)"
echo "Runner IP: $(hostname -i || echo 'unknown')"
echo "Current user: $(whoami)"
echo "=== Secrets check ==="
if [ -n "$STAGING_SSH_HOST" ]; then
echo "STAGING_SSH_HOST: [SET] value_length=${#STAGING_SSH_HOST}"
else
echo "STAGING_SSH_HOST: [EMPTY]"
fi
if [ -n "$STAGING_SSH_USER" ]; then
echo "STAGING_SSH_USER: [SET] value_length=${#STAGING_SSH_USER}"
else
echo "STAGING_SSH_USER: [EMPTY]"
fi
if [ -n "$STAGING_SSH_KEY" ]; then
echo "STAGING_SSH_KEY: [SET] value_length=${#STAGING_SSH_KEY}"
else
echo "STAGING_SSH_KEY: [EMPTY]"
fi
env:
STAGING_SSH_HOST: ${{ secrets.STAGING_SSH_HOST }}
STAGING_SSH_USER: ${{ secrets.STAGING_SSH_USER }}
STAGING_SSH_KEY: ${{ secrets.STAGING_SSH_KEY }}
- name: Setup SSH key
run: |
mkdir -p ~/.ssh
chmod 700 ~/.ssh
echo "$STAGING_SSH_KEY" > ~/.ssh/id_ed25519
chmod 600 ~/.ssh/id_ed25519
ssh-keygen -y -f ~/.ssh/id_ed25519 > ~/.ssh/id_ed25519.pub 2>/dev/null || echo "No public key generated"
echo "=== SSH Key fingerprint ==="
ssh-keygen -lf ~/.ssh/id_ed25519 || echo "Key fingerprint failed"
env:
STAGING_SSH_KEY: ${{ secrets.STAGING_SSH_KEY }}
- name: Test SSH connection
run: |
echo "Attempting SSH connection to $STAGING_SSH_HOST..."
ssh -i ~/.ssh/id_ed25519 \
-o StrictHostKeyChecking=no \
-o UserKnownHostsFile=/dev/null \
-o ConnectTimeout=10 \
-o BatchMode=yes \
-v \
$STAGING_SSH_USER@$STAGING_SSH_HOST "echo 'SSH_CONNECTION_SUCCESS' && hostname && whoami"
echo "=== SSH Test Complete ==="
env:
STAGING_SSH_HOST: ${{ secrets.STAGING_SSH_HOST }}
STAGING_SSH_USER: ${{ secrets.STAGING_SSH_USER }}
@@ -1,28 +0,0 @@
"""CMS Enhancements (placeholder - manually applied on production)
Revision ID: 034_cms_enhance
Revises: 033
Create Date: 2026-07-09
占位迁移文件:生产数据库已手动升级到此版本,
此文件用于让 alembic 识别当前版本,避免部署时迁移失败。
实际的表结构变更(helpcenter, tickets, partners, site_settings 等)
已在生产环境手动执行。
"""
from alembic import op
revision = "034_cms_enhance"
down_revision = "033"
branch_labels = None
depends_on = None
def upgrade() -> None:
"""占位 - 变更已在生产环境手动应用"""
pass
def downgrade() -> None:
"""占位 - 不执行实际回退"""
pass
+1 -32
View File
@@ -1,11 +1,7 @@
from typing import Any
from app.auth import AuthenticatedUser, get_current_user
from app.dependencies import (
get_asset_library_repository,
get_asset_repository,
get_project_repository,
)
from app.dependencies import get_asset_library_repository, get_project_repository
from app.schemas.asset_library import (
AssetLibraryResponse,
CreateAssetLibraryRequest,
@@ -151,30 +147,3 @@ def ensure_default_library(
)
created = asset_library_repository.create(library)
return _to_asset_library_response(created)
@router.delete("/{library_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_asset_library(
library_id: str,
authenticated_user: AuthenticatedUser = Depends(get_current_user),
asset_library_repository: Any = Depends(get_asset_library_repository),
asset_repository: Any = Depends(get_asset_repository),
project_repository: Any = Depends(get_project_repository),
) -> None:
"""删除素材库,同时删除库内所有素材。"""
# 查找素材库
library = asset_library_repository.find_by_id(library_id)
if library is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="素材库不存在")
# 权限校验:检查用户是否有项目访问权限
_check_project_access(library.project_id, authenticated_user.user.id, project_repository)
# 删除库内所有素材(无 FK 级联,需手动清理)
assets_in_library = asset_repository.find_by_library(library_id)
if assets_in_library:
asset_ids_to_delete = [a.id for a in assets_in_library]
asset_repository.batch_delete(asset_ids_to_delete)
# 删除素材库本身
asset_library_repository.delete(library_id)
+41 -186
View File
@@ -22,28 +22,16 @@ from typing import Any, List, Optional
from app.auth import AuthenticatedUser, get_current_user
from app.core.celery_app import celery_app
from app.dependencies import get_asset_library_repository, get_asset_repository, get_db_session, get_project_repository
from app.dependencies import get_db_session, get_project_repository
from app.schemas.generation_task import GenerationTaskResponse
from app.services import EditPlanService
from fastapi import APIRouter, Depends, HTTPException, Query, status
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from packages.adapters.sqlalchemy_impl.asset_library_repository import (
SQLAlchemyAssetLibraryRepository,
)
from packages.adapters.sqlalchemy_impl.asset_repository import (
SQLAlchemyAssetRepository,
)
from packages.adapters.sqlalchemy_impl.generation_task_repository import (
SQLAlchemyGenerationTaskRepository,
)
from packages.adapters.sqlalchemy_impl.template_clip_config_repository import (
SQLAlchemyTemplateClipConfigRepository,
)
from packages.adapters.sqlalchemy_impl.template_repository import (
SQLAlchemyTemplateRepository,
)
from packages.application.generation_tasks import (
CreateGenerationTaskCommand,
CreateGenerationTaskUseCase,
@@ -212,9 +200,9 @@ def _check_project_access(project_id: str, user_id: str, project_repository: Any
return
project = project_repository.find_by_id(project_id)
if project is None:
raise HTTPException(status_code=404, detail="项目不存在")
raise HTTPException(status_code=404, detail=f"Project {project_id} not found")
if not project.can_access(user_id):
raise HTTPException(status_code=403, detail="无权访问该项目")
raise HTTPException(status_code=403, detail="Access denied to project")
def _to_response(p: EditPlan) -> EditPlanResponse:
@@ -265,7 +253,7 @@ def list_plans(
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的筛选条件,请选择正确的状态",
detail=(f"无效的状态值: {status_filter}" f"可选值: draft, editing, rendering, completed, failed"),
)
# 项目鉴权:如果指定了 project_id,校验用户是否有权访问
@@ -391,7 +379,7 @@ def update_plan(
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的状态值,请选择正确的状态",
detail=(f"无效的状态值: {body.status}" f"可选值: draft, editing, rendering, completed, failed"),
)
svc.transition_status(plan_id, target_status)
except ValueError as exc:
@@ -447,8 +435,6 @@ def generate_plan(
db: Session = Depends(get_db_session),
current_user: AuthenticatedUser = Depends(get_current_user),
project_repository: Any = Depends(get_project_repository),
asset_library_repo: Any = Depends(get_asset_library_repository),
asset_repo: Any = Depends(get_asset_repository),
) -> EditPlanGenerateResponse:
"""触发剪辑计划渲染生成
@@ -468,121 +454,6 @@ def generate_plan(
if plan_check.project_id:
_check_project_access(plan_check.project_id, current_user.user.id, project_repository)
# ── 自动兜底 1: draft → editing ──────────────────────────────────────
if plan_check.status == EditPlanStatus.DRAFT:
logger.info("自动兜底: plan=%s draft→editing", plan_id)
svc.transition_status(plan_id, EditPlanStatus.EDITING)
# ── 自动兜底 2: 无片段 + 有 template_id → 从模板复制片段配置 ──────────
existing_clips = svc.count_clips(plan_id)
if existing_clips == 0 and plan_check.template_id:
logger.info(
"自动兜底: plan=%s 无片段,从模板 %s 复制片段配置",
plan_id,
plan_check.template_id,
)
# 优先从新模型 template_clip_configs 读取,若无则回退到旧模型 template_segments
clip_config_repo = SQLAlchemyTemplateClipConfigRepository(db)
configs = clip_config_repo.list_by_template(plan_check.template_id)
if configs:
for cfg in configs:
svc.create_clip(
plan_id=plan_id,
clip_type=cfg.clip_type.value if hasattr(cfg.clip_type, "value") else cfg.clip_type,
order=cfg.order,
template_clip_config_id=cfg.id,
duration=cfg.default_duration,
transition_effect=(
cfg.transition_effect.value
if hasattr(cfg.transition_effect, "value")
else cfg.transition_effect
),
)
logger.info("自动兜底: plan=%s 从新模型 template_clip_configs 复制了 %d 个片段", plan_id, len(configs))
else:
# 回退到旧模型 template_segments
tpl_repo = SQLAlchemyTemplateRepository(db)
segments = tpl_repo.list_segments(plan_check.template_id)
for seg in segments:
avg_duration = (seg.duration_min + seg.duration_max) / 2
svc.create_clip(
plan_id=plan_id,
clip_type="main", # 旧模型无结构角色,统一为主体片段
order=seg.segment_order,
duration=avg_duration,
config={
"material_type": seg.material_type or "",
"template_segment_id": seg.id,
},
)
logger.info("自动兜底: plan=%s 从旧模型 template_segments 复制了 %d 个片段", plan_id, len(segments))
# ── 自动兜底 3: 为没有素材的片段分配素材 ──────────────────────────
# 如果 plan.config.asset_ids 有素材,但 clips 没有 asset_id,自动按顺序分配
all_clips = svc.list_clips(plan_id)
clips_without_asset = [c for c in all_clips if not c.asset_id]
config_asset_ids = (plan_check.config or {}).get("asset_ids", [])
material_mode = (plan_check.config or {}).get("material_mode", "manual")
if clips_without_asset and config_asset_ids:
logger.info(
"自动兜底3: plan=%s%d 个无素材片段分配 %d 个指定素材",
plan_id,
len(clips_without_asset),
len(config_asset_ids),
)
for i, clip in enumerate(clips_without_asset):
asset_idx = i % len(config_asset_ids)
svc.assign_asset(clip.id, config_asset_ids[asset_idx])
logger.info("自动兜底3: plan=%s 素材分配完成", plan_id)
clips_without_asset = [] # 已分配完
# ── 自动兜底 4: 自动素材模式 → 从项目默认视频素材库选取 ────────────
if clips_without_asset and material_mode == "auto" and plan_check.project_id:
import random
logger.info(
"自动兜底4: plan=%s 自动素材模式,从项目素材库选取素材 (%d 个片段需要)",
plan_id,
len(clips_without_asset),
)
# 找到项目的视频素材库
libs = asset_library_repo.find_by_project(plan_check.project_id)
video_lib = None
for lib in libs:
lib_kind = lib.kind.value if hasattr(lib.kind, "value") else lib.kind
if lib_kind == "video":
video_lib = lib
break
if video_lib:
assets = asset_repo.find_by_library(video_lib.id)
# 筛选 ready 状态的视频素材
ready_videos = [
a
for a in assets
if (a.status.value if hasattr(a.status, "value") else a.status) == "ready"
and a.mime_type
and a.mime_type.startswith("video")
]
if ready_videos:
# 随机选取,按片段数轮询分配
random.shuffle(ready_videos)
for i, clip in enumerate(clips_without_asset):
asset = ready_videos[i % len(ready_videos)]
svc.assign_asset(clip.id, asset.id)
logger.info(
"自动兜底4: plan=%s 从素材库 %s 分配了 %d 个素材给 %d 个片段",
plan_id,
video_lib.name,
len(ready_videos),
len(clips_without_asset),
)
else:
logger.warning("自动兜底4: plan=%s 素材库无可用视频素材", plan_id)
else:
logger.warning("自动兜底4: plan=%s 项目无视频素材库", plan_id)
# 检查是否可生成
try:
can_gen, reason = svc.can_generate(plan_id)
@@ -597,64 +468,48 @@ def generate_plan(
detail=reason,
)
# 核心生成流程:捕获异常返回明确错误信息,避免裸 500
try:
# 将 pending 片段标记为 ready
clip_count = svc.mark_clips_ready(plan_id)
# 将 pending 片段标记为 ready
clip_count = svc.mark_clips_ready(plan_id)
# 创建 GenerationTask
gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
gen_task_use_case = CreateGenerationTaskUseCase(gen_task_repo)
plan = svc.get_plan_or_raise(plan_id)
gen_task = gen_task_use_case.execute(
CreateGenerationTaskCommand(
project_id="",
template_id=plan.template_id,
created_by_user_id=current_user.user.id,
source_edit_plan_id=plan_id,
)
# 创建 GenerationTask
gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
gen_task_use_case = CreateGenerationTaskUseCase(gen_task_repo)
plan = svc.get_plan_or_raise(plan_id)
gen_task = gen_task_use_case.execute(
CreateGenerationTaskCommand(
project_id="",
template_id=plan.template_id,
created_by_user_id=current_user.user.id,
source_edit_plan_id=plan_id,
)
)
# 将 generation_task_id 存入 plan config
svc.update_plan_config(plan_id, {"generation_task_id": gen_task.id})
# 将 generation_task_id 存入 plan config
svc.update_plan_config(plan_id, {"generation_task_id": gen_task.id})
# 流转状态为 rendering
svc.transition_status(plan_id, EditPlanStatus.RENDERING)
# 流转状态为 rendering
svc.transition_status(plan_id, EditPlanStatus.RENDERING)
# 调度 Celery 任务
celery_app.send_task("worker.render_edit_plan", args=[plan_id])
# 调度 Celery 任务
celery_app.send_task("worker.render_edit_plan", args=[plan_id])
# 获取最新状态
updated_plan = svc.get_plan_or_raise(plan_id)
# 获取最新状态
updated_plan = svc.get_plan_or_raise(plan_id)
logger.info(
"触发剪辑计划生成: plan_id=%s gen_task_id=%s clips=%d by user=%s",
plan_id,
gen_task.id,
clip_count,
current_user.user.id,
)
logger.info(
"触发剪辑计划生成: plan_id=%s gen_task_id=%s clips=%d by user=%s",
plan_id,
gen_task.id,
clip_count,
current_user.user.id,
)
return EditPlanGenerateResponse(
plan_id=plan_id,
plan_status=updated_plan.status.value if hasattr(updated_plan.status, "value") else updated_plan.status,
generation_task_id=gen_task.id,
clip_count=clip_count,
)
except HTTPException:
# 已处理的 HTTP 异常直接透传
raise
except Exception as exc:
logger.exception("触发剪辑计划生成失败: plan_id=%s", plan_id)
# 尝试将计划标记为失败(RENDERING → FAILED 是合法的状态流转)
try:
svc.transition_status(plan_id, EditPlanStatus.FAILED)
except Exception:
logger.warning("标记计划失败状态时异常: plan_id=%s", plan_id)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="生成失败,请稍后重试",
)
return EditPlanGenerateResponse(
plan_id=plan_id,
plan_status=updated_plan.status.value if hasattr(updated_plan.status, "value") else updated_plan.status,
generation_task_id=gen_task.id,
clip_count=clip_count,
)
@router.get(
@@ -801,7 +656,7 @@ def ai_recommend_clips(
if plan_status not in ("draft", "editing"):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="当前计划状态不支持AI推荐,请先创建或编辑计划后再试",
detail=f"AI 推荐仅支持 draft/editing 状态的计划,当前状态: {plan_status}",
)
# 调用 AI 推荐服务(同步调用 stub,后续改为 Celery 异步)
@@ -852,7 +707,7 @@ def ai_recommend_clips(
pass
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="AI推荐结果保存失败,请稍后重试",
detail=f"AI 推荐结果写入失败: {exc}",
)
logger.info(
+2 -2
View File
@@ -447,12 +447,12 @@ class EditPlanService:
# 检查状态
if plan.status != EditPlanStatus.EDITING:
return False, "请先编辑并保存模板后再生成视频"
return False, f"只有 editing 状态的计划可以触发渲染,当前状态: {plan.status}"
# 检查是否有片段
clips = self._clip_repo.list_by_plan(plan_id)
if not clips:
return False, "请先添加片段后再生成视频"
return False, "计划下没有片段,无法触发渲染"
return True, ""
+415 -2
View File
@@ -14,7 +14,10 @@
"axios": "^1.7.2",
"react": "^18.3.1",
"react-dom": "^18.3.1",
"react-hook-form": "^7.52.0",
"react-router-dom": "^6.24.0",
"recharts": "^3.8.1",
"zod": "^3.23.8",
"zustand": "^4.5.2"
},
"devDependencies": {
@@ -1412,6 +1415,42 @@
"react-dom": ">=16.9.0"
}
},
"node_modules/@reduxjs/toolkit": {
"version": "2.12.0",
"resolved": "https://registry.npmjs.org/@reduxjs/toolkit/-/toolkit-2.12.0.tgz",
"integrity": "sha512-KiT+RzZbp6mQET+Mg+h2c97+9j1sNflUxQkIHI7Yuzf6Peu+OYpmkn6nbHWmLLWj+1ZODUJFwGZ7gx3L9R9EOw==",
"license": "MIT",
"dependencies": {
"@standard-schema/spec": "^1.0.0",
"@standard-schema/utils": "^0.3.0",
"immer": "^11.0.0",
"redux": "^5.0.1",
"redux-thunk": "^3.1.0",
"reselect": "^5.1.0"
},
"peerDependencies": {
"react": "^16.9.0 || ^17.0.0 || ^18 || ^19",
"react-redux": "^7.2.1 || ^8.1.3 || ^9.0.0"
},
"peerDependenciesMeta": {
"react": {
"optional": true
},
"react-redux": {
"optional": true
}
}
},
"node_modules/@reduxjs/toolkit/node_modules/immer": {
"version": "11.1.8",
"resolved": "https://registry.npmjs.org/immer/-/immer-11.1.8.tgz",
"integrity": "sha512-/tbkHMW7y10Lx6i1crLjD4/OhNkRG+Fo7byZHtah0547nIeXYcpIXaUh0IAQY6gO5459qpGGYapcEOHtFXkIuA==",
"license": "MIT",
"funding": {
"type": "opencollective",
"url": "https://opencollective.com/immer"
}
},
"node_modules/@remix-run/router": {
"version": "1.23.3",
"resolved": "https://registry.npmjs.org/@remix-run/router/-/router-1.23.3.tgz",
@@ -1785,6 +1824,18 @@
"dev": true,
"license": "MIT"
},
"node_modules/@standard-schema/spec": {
"version": "1.1.0",
"resolved": "https://registry.npmjs.org/@standard-schema/spec/-/spec-1.1.0.tgz",
"integrity": "sha512-l2aFy5jALhniG5HgqrD6jXLi/rUWrKvqN/qJx6yoJsgKhblVd+iqqU4RCXavm/jPityDo5TCvKMnpjKnOriy0w==",
"license": "MIT"
},
"node_modules/@standard-schema/utils": {
"version": "0.3.0",
"resolved": "https://registry.npmjs.org/@standard-schema/utils/-/utils-0.3.0.tgz",
"integrity": "sha512-e7Mew686owMaPJVNNLs55PUvgz371nKgwsc4vxE49zsODpJEnxgxRo2y/OKrqueavXgZNMDVj3DdHFlaSAeU8g==",
"license": "MIT"
},
"node_modules/@tanstack/query-core": {
"version": "5.101.0",
"resolved": "https://registry.npmjs.org/@tanstack/query-core/-/query-core-5.101.0.tgz",
@@ -1954,6 +2005,69 @@
"@babel/types": "^7.28.2"
}
},
"node_modules/@types/d3-array": {
"version": "3.2.2",
"resolved": "https://registry.npmjs.org/@types/d3-array/-/d3-array-3.2.2.tgz",
"integrity": "sha512-hOLWVbm7uRza0BYXpIIW5pxfrKe0W+D5lrFiAEYR+pb6w3N2SwSMaJbXdUfSEv+dT4MfHBLtn5js0LAWaO6otw==",
"license": "MIT"
},
"node_modules/@types/d3-color": {
"version": "3.1.3",
"resolved": "https://registry.npmjs.org/@types/d3-color/-/d3-color-3.1.3.tgz",
"integrity": "sha512-iO90scth9WAbmgv7ogoq57O9YpKmFBbmoEoCHDB2xMBY0+/KVrqAaCDyCE16dUspeOvIxFFRI+0sEtqDqy2b4A==",
"license": "MIT"
},
"node_modules/@types/d3-ease": {
"version": "3.0.2",
"resolved": "https://registry.npmjs.org/@types/d3-ease/-/d3-ease-3.0.2.tgz",
"integrity": "sha512-NcV1JjO5oDzoK26oMzbILE6HW7uVXOHLQvHshBUW4UMdZGfiY6v5BeQwh9a9tCzv+CeefZQHJt5SRgK154RtiA==",
"license": "MIT"
},
"node_modules/@types/d3-interpolate": {
"version": "3.0.4",
"resolved": "https://registry.npmjs.org/@types/d3-interpolate/-/d3-interpolate-3.0.4.tgz",
"integrity": "sha512-mgLPETlrpVV1YRJIglr4Ez47g7Yxjl1lj7YKsiMCb27VJH9W8NVM6Bb9d8kkpG/uAQS5AmbA48q2IAolKKo1MA==",
"license": "MIT",
"dependencies": {
"@types/d3-color": "*"
}
},
"node_modules/@types/d3-path": {
"version": "3.1.1",
"resolved": "https://registry.npmjs.org/@types/d3-path/-/d3-path-3.1.1.tgz",
"integrity": "sha512-VMZBYyQvbGmWyWVea0EHs/BwLgxc+MKi1zLDCONksozI4YJMcTt8ZEuIR4Sb1MMTE8MMW49v0IwI5+b7RmfWlg==",
"license": "MIT"
},
"node_modules/@types/d3-scale": {
"version": "4.0.9",
"resolved": "https://registry.npmjs.org/@types/d3-scale/-/d3-scale-4.0.9.tgz",
"integrity": "sha512-dLmtwB8zkAeO/juAMfnV+sItKjlsw2lKdZVVy6LRr0cBmegxSABiLEpGVmSJJ8O08i4+sGR6qQtb6WtuwJdvVw==",
"license": "MIT",
"dependencies": {
"@types/d3-time": "*"
}
},
"node_modules/@types/d3-shape": {
"version": "3.1.8",
"resolved": "https://registry.npmjs.org/@types/d3-shape/-/d3-shape-3.1.8.tgz",
"integrity": "sha512-lae0iWfcDeR7qt7rA88BNiqdvPS5pFVPpo5OfjElwNaT2yyekbM0C9vK+yqBqEmHr6lDkRnYNoTBYlAgJa7a4w==",
"license": "MIT",
"dependencies": {
"@types/d3-path": "*"
}
},
"node_modules/@types/d3-time": {
"version": "3.0.4",
"resolved": "https://registry.npmjs.org/@types/d3-time/-/d3-time-3.0.4.tgz",
"integrity": "sha512-yuzZug1nkAAaBlBBikKZTgzCeA+k1uy4ZFwWANOfKw5z5LRhV0gNA7gNkKm7HoK+HRN0wX3EkxGk0fpbWhmB7g==",
"license": "MIT"
},
"node_modules/@types/d3-timer": {
"version": "3.0.2",
"resolved": "https://registry.npmjs.org/@types/d3-timer/-/d3-timer-3.0.2.tgz",
"integrity": "sha512-Ps3T8E8dZDam6fUyNiMkekK3XUsaUEik+idO9/YjPtfj2qruF8tFBXS7XhtE4iIXBLxhmLjP3SXpLhVf21I9Lw==",
"license": "MIT"
},
"node_modules/@types/estree": {
"version": "1.0.9",
"resolved": "https://registry.npmjs.org/@types/estree/-/estree-1.0.9.tgz",
@@ -1999,6 +2113,12 @@
"@types/react": "^18.0.0"
}
},
"node_modules/@types/use-sync-external-store": {
"version": "0.0.6",
"resolved": "https://registry.npmjs.org/@types/use-sync-external-store/-/use-sync-external-store-0.0.6.tgz",
"integrity": "sha512-zFDAD+tlpf2r4asuHEj0XH6pY6i0g5NeAHPn+15wk3BV6JA69eERFXC1gyGThDkVa1zCyKr5jox1+2LbV/AMLg==",
"license": "MIT"
},
"node_modules/@typescript-eslint/eslint-plugin": {
"version": "7.18.0",
"resolved": "https://registry.npmjs.org/@typescript-eslint/eslint-plugin/-/eslint-plugin-7.18.0.tgz",
@@ -2812,6 +2932,15 @@
"integrity": "sha512-saHYOzhIQs6wy2sVxTM6bUDsQO4F50V9RQ22qBpEdCW+I+/Wmke2HOl6lS6dTpdxVhb88/I6+Hs+438c3lfUow==",
"license": "MIT"
},
"node_modules/clsx": {
"version": "2.1.1",
"resolved": "https://registry.npmjs.org/clsx/-/clsx-2.1.1.tgz",
"integrity": "sha512-eYm0QWBtUrBWZWG0d386OGAw16Z995PiOVo2B7bjWSbHedGl5e0ZWaq65kOGgUSNesEIDkB9ISbTg/JK9dhCZA==",
"license": "MIT",
"engines": {
"node": ">=6"
}
},
"node_modules/color-convert": {
"version": "2.0.1",
"resolved": "https://registry.npmjs.org/color-convert/-/color-convert-2.0.1.tgz",
@@ -2929,6 +3058,127 @@
"integrity": "sha512-z1HGKcYy2xA8AGQfwrn0PAy+PB7X/GSj3UVJW9qKyn43xWa+gl5nXmU4qqLMRzWVLFC8KusUX8T/0kCiOYpAIQ==",
"license": "MIT"
},
"node_modules/d3-array": {
"version": "3.2.4",
"resolved": "https://registry.npmjs.org/d3-array/-/d3-array-3.2.4.tgz",
"integrity": "sha512-tdQAmyA18i4J7wprpYq8ClcxZy3SC31QMeByyCFyRt7BVHdREQZ5lpzoe5mFEYZUWe+oq8HBvk9JjpibyEV4Jg==",
"license": "ISC",
"dependencies": {
"internmap": "1 - 2"
},
"engines": {
"node": ">=12"
}
},
"node_modules/d3-color": {
"version": "3.1.0",
"resolved": "https://registry.npmjs.org/d3-color/-/d3-color-3.1.0.tgz",
"integrity": "sha512-zg/chbXyeBtMQ1LbD/WSoW2DpC3I0mpmPdW+ynRTj/x2DAWYrIY7qeZIHidozwV24m4iavr15lNwIwLxRmOxhA==",
"license": "ISC",
"engines": {
"node": ">=12"
}
},
"node_modules/d3-ease": {
"version": "3.0.1",
"resolved": "https://registry.npmjs.org/d3-ease/-/d3-ease-3.0.1.tgz",
"integrity": "sha512-wR/XK3D3XcLIZwpbvQwQ5fK+8Ykds1ip7A2Txe0yxncXSdq1L9skcG7blcedkOX+ZcgxGAmLX1FrRGbADwzi0w==",
"license": "BSD-3-Clause",
"engines": {
"node": ">=12"
}
},
"node_modules/d3-format": {
"version": "3.1.2",
"resolved": "https://registry.npmjs.org/d3-format/-/d3-format-3.1.2.tgz",
"integrity": "sha512-AJDdYOdnyRDV5b6ArilzCPPwc1ejkHcoyFarqlPqT7zRYjhavcT3uSrqcMvsgh2CgoPbK3RCwyHaVyxYcP2Arg==",
"license": "ISC",
"engines": {
"node": ">=12"
}
},
"node_modules/d3-interpolate": {
"version": "3.0.1",
"resolved": "https://registry.npmjs.org/d3-interpolate/-/d3-interpolate-3.0.1.tgz",
"integrity": "sha512-3bYs1rOD33uo8aqJfKP3JWPAibgw8Zm2+L9vBKEHJ2Rg+viTR7o5Mmv5mZcieN+FRYaAOWX5SJATX6k1PWz72g==",
"license": "ISC",
"dependencies": {
"d3-color": "1 - 3"
},
"engines": {
"node": ">=12"
}
},
"node_modules/d3-path": {
"version": "3.1.0",
"resolved": "https://registry.npmjs.org/d3-path/-/d3-path-3.1.0.tgz",
"integrity": "sha512-p3KP5HCf/bvjBSSKuXid6Zqijx7wIfNW+J/maPs+iwR35at5JCbLUT0LzF1cnjbCHWhqzQTIN2Jpe8pRebIEFQ==",
"license": "ISC",
"engines": {
"node": ">=12"
}
},
"node_modules/d3-scale": {
"version": "4.0.2",
"resolved": "https://registry.npmjs.org/d3-scale/-/d3-scale-4.0.2.tgz",
"integrity": "sha512-GZW464g1SH7ag3Y7hXjf8RoUuAFIqklOAq3MRl4OaWabTFJY9PN/E1YklhXLh+OQ3fM9yS2nOkCoS+WLZ6kvxQ==",
"license": "ISC",
"dependencies": {
"d3-array": "2.10.0 - 3",
"d3-format": "1 - 3",
"d3-interpolate": "1.2.0 - 3",
"d3-time": "2.1.1 - 3",
"d3-time-format": "2 - 4"
},
"engines": {
"node": ">=12"
}
},
"node_modules/d3-shape": {
"version": "3.2.0",
"resolved": "https://registry.npmjs.org/d3-shape/-/d3-shape-3.2.0.tgz",
"integrity": "sha512-SaLBuwGm3MOViRq2ABk3eLoxwZELpH6zhl3FbAoJ7Vm1gofKx6El1Ib5z23NUEhF9AsGl7y+dzLe5Cw2AArGTA==",
"license": "ISC",
"dependencies": {
"d3-path": "^3.1.0"
},
"engines": {
"node": ">=12"
}
},
"node_modules/d3-time": {
"version": "3.1.0",
"resolved": "https://registry.npmjs.org/d3-time/-/d3-time-3.1.0.tgz",
"integrity": "sha512-VqKjzBLejbSMT4IgbmVgDjpkYrNWUYJnbCGo874u7MMKIWsILRX+OpX/gTk8MqjpT1A/c6HY2dCA77ZN0lkQ2Q==",
"license": "ISC",
"dependencies": {
"d3-array": "2 - 3"
},
"engines": {
"node": ">=12"
}
},
"node_modules/d3-time-format": {
"version": "4.1.0",
"resolved": "https://registry.npmjs.org/d3-time-format/-/d3-time-format-4.1.0.tgz",
"integrity": "sha512-dJxPBlzC7NugB2PDLwo9Q8JiTR3M3e4/XANkreKSUxF8vvXKqm1Yfq4Q5dl8budlunRVlUUaDUgFt7eA8D6NLg==",
"license": "ISC",
"dependencies": {
"d3-time": "1 - 3"
},
"engines": {
"node": ">=12"
}
},
"node_modules/d3-timer": {
"version": "3.0.1",
"resolved": "https://registry.npmjs.org/d3-timer/-/d3-timer-3.0.1.tgz",
"integrity": "sha512-ndfJ/JxxMd3nw31uyKoY2naivF+r29V+Lc0svZxe1JvvIRmi8hUsrMvdOwgS1o6uBHmiz91geQ0ylPP0aj1VUA==",
"license": "ISC",
"engines": {
"node": ">=12"
}
},
"node_modules/data-urls": {
"version": "5.0.0",
"resolved": "https://registry.npmjs.org/data-urls/-/data-urls-5.0.0.tgz",
@@ -2973,6 +3223,12 @@
"dev": true,
"license": "MIT"
},
"node_modules/decimal.js-light": {
"version": "2.5.1",
"resolved": "https://registry.npmjs.org/decimal.js-light/-/decimal.js-light-2.5.1.tgz",
"integrity": "sha512-qIMFpTMZmny+MMIitAB6D7iVPEorVw6YQRWkvarTkT4tBeSLLiHzcwj6q0MmYSFCiVpiqPJTJEYIrpcPzVEIvg==",
"license": "MIT"
},
"node_modules/deep-eql": {
"version": "4.1.4",
"resolved": "https://registry.npmjs.org/deep-eql/-/deep-eql-4.1.4.tgz",
@@ -3135,6 +3391,16 @@
"node": ">= 0.4"
}
},
"node_modules/es-toolkit": {
"version": "1.47.1",
"resolved": "https://registry.npmjs.org/es-toolkit/-/es-toolkit-1.47.1.tgz",
"integrity": "sha512-5RAqEwf4P4E17p+W75KLOWw/nOvKZzSQpxM32IpI2KZLaVonjTrZ0Ai5ghMaVI9eKC2p8eoQgcBdkEDgzFk6+Q==",
"license": "MIT",
"workspaces": [
"docs",
"benchmarks"
]
},
"node_modules/esbuild": {
"version": "0.21.5",
"resolved": "https://registry.npmjs.org/esbuild/-/esbuild-0.21.5.tgz",
@@ -3405,6 +3671,12 @@
"node": ">=0.10.0"
}
},
"node_modules/eventemitter3": {
"version": "5.0.4",
"resolved": "https://registry.npmjs.org/eventemitter3/-/eventemitter3-5.0.4.tgz",
"integrity": "sha512-mlsTRyGaPBjPedk6Bvw+aqbsXDtoAyAzm5MO7JgU+yVRyMQ5O8bD4Kcci7BS85f93veegeCPkL8R4GLClnjLFw==",
"license": "MIT"
},
"node_modules/execa": {
"version": "8.0.1",
"resolved": "https://registry.npmjs.org/execa/-/execa-8.0.1.tgz",
@@ -3951,8 +4223,6 @@
"resolved": "https://registry.npmjs.org/immer/-/immer-10.2.0.tgz",
"integrity": "sha512-d/+XTN3zfODyjr89gM3mPq1WNX2B8pYsu7eORitdwyA2sBubnTl3laYlBk4sXY5FUa5qTZGBDPJICVbvqzjlbw==",
"license": "MIT",
"optional": true,
"peer": true,
"funding": {
"type": "opencollective",
"url": "https://opencollective.com/immer"
@@ -4014,6 +4284,15 @@
"dev": true,
"license": "ISC"
},
"node_modules/internmap": {
"version": "2.0.3",
"resolved": "https://registry.npmjs.org/internmap/-/internmap-2.0.3.tgz",
"integrity": "sha512-5Hh7Y1wQbvY5ooGgPbDaL5iYLAPzMTUrjMulskHLH6wnv/A+1q5rgEaiuqEjB+oxGXIVZs1FF+R/KPN3ZSQYYg==",
"license": "ISC",
"engines": {
"node": ">=12"
}
},
"node_modules/is-extglob": {
"version": "2.1.1",
"resolved": "https://registry.npmjs.org/is-extglob/-/is-extglob-2.1.1.tgz",
@@ -5563,6 +5842,52 @@
"react": "^18.3.1"
}
},
"node_modules/react-hook-form": {
"version": "7.79.0",
"resolved": "https://registry.npmjs.org/react-hook-form/-/react-hook-form-7.79.0.tgz",
"integrity": "sha512-mhYp/MTmXvzYX6AJcJVko0rktoIhhmRnEouObj4wF5i/tCttgJvnp1+9wRkpITZjDTqpo4IOSJqu0dBlPlV/Lw==",
"license": "MIT",
"engines": {
"node": ">=18.0.0"
},
"funding": {
"type": "opencollective",
"url": "https://opencollective.com/react-hook-form"
},
"peerDependencies": {
"react": "^16.8.0 || ^17 || ^18 || ^19"
}
},
"node_modules/react-is": {
"version": "19.2.7",
"resolved": "https://registry.npmjs.org/react-is/-/react-is-19.2.7.tgz",
"integrity": "sha512-kZFnouyVv7eP/Phmrlo9FK+zcAdriZJvzxXHF1Sl1P377WSGe2G/JxVolhTrB/jeV47lKImhNUsijjHAAbcl/A==",
"license": "MIT",
"peer": true
},
"node_modules/react-redux": {
"version": "9.3.0",
"resolved": "https://registry.npmjs.org/react-redux/-/react-redux-9.3.0.tgz",
"integrity": "sha512-KQopgqFo/p/fgmAs5qz6p5RWaNAzq40WAu7fJIXnQpYxFPbJYtsJPWvGeF2rOBaY/kEuV77AVsX8TsQzKm+A/g==",
"license": "MIT",
"dependencies": {
"@types/use-sync-external-store": "^0.0.6",
"use-sync-external-store": "^1.4.0"
},
"peerDependencies": {
"@types/react": "^18.2.25 || ^19",
"react": "^18.0 || ^19",
"redux": "^5.0.0"
},
"peerDependenciesMeta": {
"@types/react": {
"optional": true
},
"redux": {
"optional": true
}
}
},
"node_modules/react-refresh": {
"version": "0.17.0",
"resolved": "https://registry.npmjs.org/react-refresh/-/react-refresh-0.17.0.tgz",
@@ -5605,6 +5930,36 @@
"react-dom": ">=16.8"
}
},
"node_modules/recharts": {
"version": "3.8.1",
"resolved": "https://registry.npmjs.org/recharts/-/recharts-3.8.1.tgz",
"integrity": "sha512-mwzmO1s9sFL0TduUpwndxCUNoXsBw3u3E/0+A+cLcrSfQitSG62L32N69GhqUrrT5qKcAE3pCGVINC6pqkBBQg==",
"license": "MIT",
"workspaces": [
"www"
],
"dependencies": {
"@reduxjs/toolkit": "^1.9.0 || 2.x.x",
"clsx": "^2.1.1",
"decimal.js-light": "^2.5.1",
"es-toolkit": "^1.39.3",
"eventemitter3": "^5.0.1",
"immer": "^10.1.1",
"react-redux": "8.x.x || 9.x.x",
"reselect": "5.1.1",
"tiny-invariant": "^1.3.3",
"use-sync-external-store": "^1.2.2",
"victory-vendor": "^37.0.2"
},
"engines": {
"node": ">=18"
},
"peerDependencies": {
"react": "^16.8.0 || ^17.0.0 || ^18.0.0 || ^19.0.0",
"react-dom": "^16.0.0 || ^17.0.0 || ^18.0.0 || ^19.0.0",
"react-is": "^16.8.0 || ^17.0.0 || ^18.0.0 || ^19.0.0"
}
},
"node_modules/redent": {
"version": "3.0.0",
"resolved": "https://registry.npmjs.org/redent/-/redent-3.0.0.tgz",
@@ -5619,6 +5974,21 @@
"node": ">=8"
}
},
"node_modules/redux": {
"version": "5.0.1",
"resolved": "https://registry.npmjs.org/redux/-/redux-5.0.1.tgz",
"integrity": "sha512-M9/ELqF6fy8FwmkpnF0S3YKOqMyoWJ4+CS5Efg2ct3oY9daQvd/Pc71FpGZsVsbl3Cpb+IIcjBDUnnyBdQbq4w==",
"license": "MIT"
},
"node_modules/redux-thunk": {
"version": "3.1.0",
"resolved": "https://registry.npmjs.org/redux-thunk/-/redux-thunk-3.1.0.tgz",
"integrity": "sha512-NW2r5T6ksUKXCabzhL9z+h206HQw/NJkcLm1GPImRQ8IzfXwRGqjVhKJGauHirT0DAuyy6hjdnMZaRoAcy0Klw==",
"license": "MIT",
"peerDependencies": {
"redux": "^5.0.0"
}
},
"node_modules/requires-port": {
"version": "1.0.0",
"resolved": "https://registry.npmjs.org/requires-port/-/requires-port-1.0.0.tgz",
@@ -5626,6 +5996,12 @@
"dev": true,
"license": "MIT"
},
"node_modules/reselect": {
"version": "5.1.1",
"resolved": "https://registry.npmjs.org/reselect/-/reselect-5.1.1.tgz",
"integrity": "sha512-K/BG6eIky/SBpzfHZv/dd+9JBFiS4SWV7FIujVyJRux6e45+73RaUHXLmIR1f7WOMaQ0U1km6qwklRQxpJJY0w==",
"license": "MIT"
},
"node_modules/resize-observer-polyfill": {
"version": "1.5.1",
"resolved": "https://registry.npmjs.org/resize-observer-polyfill/-/resize-observer-polyfill-1.5.1.tgz",
@@ -6009,6 +6385,12 @@
"node": ">=12.22"
}
},
"node_modules/tiny-invariant": {
"version": "1.3.3",
"resolved": "https://registry.npmjs.org/tiny-invariant/-/tiny-invariant-1.3.3.tgz",
"integrity": "sha512-+FbBPE1o9QAYvviau/qC5SE3caw21q3xkvWKBtja5vgqOWIHHJ3ioaq1VPfn/Szqctz2bU/oYeKd9/z5BL+PVg==",
"license": "MIT"
},
"node_modules/tinybench": {
"version": "2.9.0",
"resolved": "https://registry.npmjs.org/tinybench/-/tinybench-2.9.0.tgz",
@@ -6242,6 +6624,28 @@
"react": "^16.8.0 || ^17.0.0 || ^18.0.0 || ^19.0.0"
}
},
"node_modules/victory-vendor": {
"version": "37.3.6",
"resolved": "https://registry.npmjs.org/victory-vendor/-/victory-vendor-37.3.6.tgz",
"integrity": "sha512-SbPDPdDBYp+5MJHhBCAyI7wKM3d5ivekigc2Dk2s7pgbZ9wIgIBYGVw4zGHBml/qTFbexrofXW6Gu4noGxrOwQ==",
"license": "MIT AND ISC",
"dependencies": {
"@types/d3-array": "^3.0.3",
"@types/d3-ease": "^3.0.0",
"@types/d3-interpolate": "^3.0.1",
"@types/d3-scale": "^4.0.2",
"@types/d3-shape": "^3.1.0",
"@types/d3-time": "^3.0.0",
"@types/d3-timer": "^3.0.0",
"d3-array": "^3.1.6",
"d3-ease": "^3.0.1",
"d3-interpolate": "^3.0.1",
"d3-scale": "^4.0.2",
"d3-shape": "^3.1.0",
"d3-time": "^3.0.0",
"d3-timer": "^3.0.1"
}
},
"node_modules/vite": {
"version": "5.4.21",
"resolved": "https://registry.npmjs.org/vite/-/vite-5.4.21.tgz",
@@ -6576,6 +6980,15 @@
"url": "https://github.com/sponsors/sindresorhus"
}
},
"node_modules/zod": {
"version": "3.25.76",
"resolved": "https://registry.npmjs.org/zod/-/zod-3.25.76.tgz",
"integrity": "sha512-gzUt/qt81nXsFGKIFcC3YnfEAx5NkunCfnDlvuBSSFS02bcXu4Lmea0AFIUwbLWxWPx3d9p8S5QoaujKcNQxcQ==",
"license": "MIT",
"funding": {
"url": "https://github.com/sponsors/colinhacks"
}
},
"node_modules/zustand": {
"version": "4.5.7",
"resolved": "https://registry.npmjs.org/zustand/-/zustand-4.5.7.tgz",
+3 -40
View File
@@ -263,14 +263,10 @@ export const uploadAssetDirect = async (data: {
);
directForm.append("file", data.file);
// 使用 XMLHttpRequest 以获取上传进度 + 超时控制 + 详细错误诊断
// 使用 XMLHttpRequest 以获取上传进度fetch 不支持)
await new Promise<void>((resolve, reject) => {
const xhr = new XMLHttpRequest();
xhr.open(prepared.method, prepared.upload_url);
// 超时 10 分钟
xhr.timeout = 10 * 60 * 1000;
xhr.upload.onprogress = (e) => {
if (e.lengthComputable && data.onProgress) {
data.onProgress(Math.round((e.loaded / e.total) * 100));
@@ -280,43 +276,10 @@ export const uploadAssetDirect = async (data: {
if (xhr.status >= 200 && xhr.status < 300) {
resolve();
} else {
// 解析 OSS 返回的 XML 错误信息
let ossError = "";
try {
const codeMatch = xhr.responseText.match(/<Code>([^<]+)<\/Code>/);
const msgMatch = xhr.responseText.match(
/<Message>([^<]+)<\/Message>/,
);
if (codeMatch || msgMatch) {
ossError = ` [OSS: ${codeMatch?.[1] || "unknown"} - ${msgMatch?.[1] || "unknown"}]`;
}
} catch {
// 无法解析响应体
}
const detail = `OSS 直传失败: HTTP ${xhr.status} ${xhr.statusText}${ossError}`;
console.error("[OSS Upload] 直传失败:", {
url: prepared.upload_url,
storage_key: prepared.storage_key,
status: xhr.status,
statusText: xhr.statusText,
});
reject(new Error(detail));
reject(new Error(`OSS direct upload failed: ${xhr.status}`));
}
};
xhr.onerror = () => {
console.error("[OSS Upload] 网络错误:", {
url: prepared.upload_url,
storage_key: prepared.storage_key,
});
reject(new Error("OSS 上传网络错误,请检查网络连接"));
};
xhr.ontimeout = () => {
console.error("[OSS Upload] 上传超时:", {
url: prepared.upload_url,
storage_key: prepared.storage_key,
});
reject(new Error("OSS 上传超时(10分钟),请检查网络或尝试更小的文件"));
};
xhr.onerror = () => reject(new Error("OSS direct upload failed"));
xhr.send(directForm);
});
+1 -20
View File
@@ -122,27 +122,8 @@ apiClient.interceptors.response.use(
}
// 提取后端返回的错误信息(detail / message / msg
// 注意:后端返回的字段可能是对象 {code, message} 而非字符串,需要安全提取
const data = error.response?.data;
const rawServerMsg = data?.detail || data?.message || data?.msg;
// 安全提取字符串:递归处理嵌套对象(后端可能返回 {code, message: {code, message}} 等)
const safeExtractString = (val: unknown): string => {
if (typeof val === "string") return val;
if (typeof val === "object" && val !== null) {
const obj = val as Record<string, unknown>;
if (typeof obj.message === "string") return obj.message;
if (typeof obj.msg === "string") return obj.msg;
if (typeof obj.detail === "string") return obj.detail;
// 嵌套对象:递归提取
if (typeof obj.message === "object" && obj.message !== null)
return safeExtractString(obj.message);
if (typeof obj.msg === "object" && obj.msg !== null)
return safeExtractString(obj.msg);
return JSON.stringify(val);
}
return "";
};
const serverMsg = safeExtractString(rawServerMsg);
const serverMsg = data?.detail || data?.message || data?.msg;
let handled = false;
if (error.code === "ECONNABORTED" || error.message?.includes("timeout")) {
+4 -8
View File
@@ -68,14 +68,10 @@ export interface CreateTitleRequest {
/** 获取当前用户的所有标题 */
export const getTitles = async (): Promise<TitleItem[]> => {
const response = await apiClient.get<
{ items: BackendTitleResponse[] } | BackendTitleResponse[]
>("/titles");
// 兼容两种后端返回格式:{ items: [...] } 或直接 [...]
const items = Array.isArray(response.data)
? response.data
: response.data.items || [];
return items.map(toTitleItem);
const response = await apiClient.get<{ items: BackendTitleResponse[] }>(
"/titles",
);
return (response.data.items || []).map(toTitleItem);
};
/** 创建标题 */
+6 -71
View File
@@ -69,20 +69,11 @@ const inferKind = (mimeType: string): AssetKind => {
return "image";
};
/** 根据 quality_score / classification_status / asset status 推断前端状态 */
/** 根据 quality_score 推断前端状态 */
const inferStatus = (
score?: number,
classificationStatus?: string,
assetStatus?: string,
): { status: StatusType; label: string } => {
// 素材已就绪(status=ready)时,不应因 classification 未执行而显示"处理中"
if (assetStatus === "ready") {
if (score == null) return { status: "info", label: "待诊断" };
if (score >= 70) return { status: "ok", label: "合格" };
if (score >= 40) return { status: "warn", label: "待优化" };
return { status: "bad", label: "不合格" };
}
// 素材未就绪:classification 正在处理中
if (
classificationStatus === "processing" ||
classificationStatus === "pending"
@@ -115,7 +106,6 @@ const mapAsset = (item: ApiAssetItem): AssetItem => {
const { status, label } = inferStatus(
item.quality_score ?? undefined,
item.classification_status ?? undefined,
item.status ?? undefined,
);
const metadata = item.metadata || {};
const kind = inferKind(item.mime_type || "");
@@ -229,16 +219,7 @@ const AssetCard: React.FC<{
onToggle: () => void;
onDiagnose: () => void;
onPlay: () => void;
onDelete: () => void;
}> = ({
asset,
selected,
diagnosing,
onToggle,
onDiagnose,
onPlay,
onDelete,
}) => (
}> = ({ asset, selected, diagnosing, onToggle, onDiagnose, onPlay }) => (
<div
className={`xx-asset-card${selected ? " xx-asset-card-selected" : ""}`}
onClick={onToggle}
@@ -269,24 +250,6 @@ const AssetCard: React.FC<{
</span>
)}
{/* 删除按钮 */}
<Popconfirm
title="确认删除"
description="删除后不可恢复,确定要删除这个素材吗?"
onConfirm={(e) => {
e?.stopPropagation();
onDelete();
}}
onCancel={(e) => e?.stopPropagation()}
okText="删除"
cancelText="取消"
okButtonProps={{ danger: true }}
>
<span className="xx-asset-delete" onClick={(e) => e.stopPropagation()}>
<DeleteOutlined />
</span>
</Popconfirm>
{/* 选中态勾选 */}
{selected && (
<span className="xx-asset-check">
@@ -385,8 +348,6 @@ const AssetLibrary: React.FC = () => {
/* 状态 */
const [selectedIds, setSelectedIds] = useState<Set<string>>(new Set());
// 大文件直传由 handleUpload 直接调用 uploadAssetDirect 处理
/* 筛选 */
const [searchText, setSearchText] = useState("");
const [filterType, setFilterType] = useState<string>("all");
@@ -460,11 +421,11 @@ const AssetLibrary: React.FC = () => {
const handleUpload = async (file: File) => {
if (file.size > MAX_FILE_SIZE) {
message.error(`文件 "${file.name}" 超过 2GB 限制`);
return;
return false;
}
if (!effectiveLibId) {
message.warning("请先选择或创建一个素材库");
return;
return false;
}
setUploading(true);
@@ -483,14 +444,12 @@ const AssetLibrary: React.FC = () => {
queryClient.invalidateQueries({ queryKey: ["asset-libraries"] });
} catch (err: unknown) {
const detail = err instanceof Error ? err.message : "";
console.error("[handleUpload] 上传失败:", err);
message.error(`"${file.name}" 上传失败${detail ? `${detail}` : ""}`);
// 错误时延迟关闭弹窗,让用户能看到错误提示
await new Promise((r) => setTimeout(r, 1500));
} finally {
setUploading(false);
setUploadProgress(0);
}
return false;
};
/* 新建素材库 */
@@ -543,24 +502,6 @@ const AssetLibrary: React.FC = () => {
};
/* 批量删除 */
/* 单个素材删除 */
const handleSingleDelete = async (assetId: string) => {
try {
await deleteAsset(assetId);
queryClient.invalidateQueries({ queryKey: ["assets"] });
queryClient.invalidateQueries({ queryKey: ["asset-libraries"] });
// 从选中集合中移除
setSelectedIds((prev) => {
const next = new Set(prev);
next.delete(assetId);
return next;
});
message.success("素材已删除");
} catch {
message.error("删除失败,请重试");
}
};
const handleBatchDelete = async () => {
const ids = Array.from(selectedIds);
let successCount = 0;
@@ -694,12 +635,7 @@ const AssetLibrary: React.FC = () => {
<div className="xx-assets-content">
{/* 上传区域 */}
<Upload.Dragger
beforeUpload={(file) => {
// 同步返回 false 阻止 antd 默认上传行为
// 异步上传由 handleUpload 处理
handleUpload(file as File);
return false;
}}
beforeUpload={handleUpload}
showUploadList={false}
multiple
accept="video/*,image/*"
@@ -820,7 +756,6 @@ const AssetLibrary: React.FC = () => {
onToggle={() => toggleSelect(asset.id)}
onDiagnose={() => handleDiagnose(asset)}
onPlay={() => setPlayingAsset(asset)}
onDelete={() => handleSingleDelete(asset.id)}
/>
))}
</div>
-29
View File
@@ -282,35 +282,6 @@
transform: scale(1.1);
}
/* 删除按钮 */
.xx-asset-delete {
position: absolute;
bottom: var(--space-sm, 8px);
right: var(--space-sm, 8px);
width: 28px;
height: 28px;
border-radius: var(--radius-full, 999px);
background: rgba(0, 0, 0, 0.45);
backdrop-filter: blur(4px);
display: grid;
place-items: center;
color: #fff;
font-size: 13px;
cursor: pointer;
opacity: 0;
transition: var(--transition-all, all 0.2s ease);
z-index: 2;
}
.xx-asset-card:hover .xx-asset-delete {
opacity: 1;
}
.xx-asset-delete:hover {
background: rgba(255, 77, 79, 0.85);
transform: scale(1.1);
}
/* 选中态 */
.xx-asset-card-selected {
border-color: var(--primary-color) !important;
@@ -298,13 +298,6 @@ const EditingPlanner: React.FC = () => {
const handleModeChange = (mode: TemplateMode) => {
setCurrentMode(mode);
// 切换纯单类型模式时,自动转换所有已有片段的类型
if (mode === "voice_over") {
setClips((prev) => prev.map((c) => ({ ...c, type: "voice" as const })));
} else if (mode === "pip") {
setClips((prev) => prev.map((c) => ({ ...c, type: "pip" as const })));
}
// 混合模式(voice_pip)和一镜到底(one_take)不自动转换,保留原有类型
};
const handleClipSelect = (clipId: string) => {
@@ -9,7 +9,6 @@ import React, {
useCallback,
useEffect,
useLayoutEffect,
useMemo,
} from "react";
import type { ClipData, ClipType } from "../types";
@@ -55,34 +54,18 @@ const TimelinePanel: React.FC<TimelinePanelProps> = ({
right: 0,
});
/* ── 根据模式决定可选类型 ── */
const availableTypes: ClipType[] = useMemo(
() =>
currentMode === "voice_over"
? ["voice"]
: currentMode === "pip"
? ["pip"]
: ["voice", "pip"], // voice_pip / one_take / 默认
[currentMode],
);
/* ── 默认添加类型:跟随模式(纯单类型模式直接用该类型,混合模式默认 voice) ── */
const defaultAddType: ClipType = useMemo(() => {
if (currentMode === "voice_over") return "voice";
if (currentMode === "pip") return "pip";
return "voice";
}, [currentMode]);
/* ── "+" 卡片:类型+时长选择状态 ── */
const [addType, setAddType] = useState<ClipType>(defaultAddType);
const [addType, setAddType] = useState<ClipType>("voice");
const [addDuration, setAddDuration] = useState<number>(5);
/* ── 模式切换时自动同步默认添加类型 ── */
useEffect(() => {
if (!availableTypes.includes(addType)) {
setAddType(defaultAddType);
}
}, [currentMode, addType, availableTypes, defaultAddType]);
/* ── 根据模式决定可选类型 ── */
const availableTypes: ClipType[] =
currentMode === "voice_over"
? ["voice"]
: currentMode === "pip"
? ["pip"]
: ["voice", "pip"]; // voice_pip 或默认
/* ── 面板尺寸(宽度固定,高度由 useLayoutEffect 实测) ── */
const PICKER_W = 240; // 面板宽度(与 CSS 一致)
const GAP = 6; // 面板与"+"卡片的间距
@@ -109,14 +92,6 @@ const TimelinePanel: React.FC<TimelinePanelProps> = ({
const handleTogglePicker = () => {
if (!showAddPicker) {
// 打开面板时,默认选中当前模式下的第一个可用类型
const defaultType =
currentMode === "pip"
? "pip"
: currentMode === "voice_over"
? "voice"
: "voice";
setAddType(defaultType);
updatePickerPosition();
}
setShowAddPicker((v) => !v);
+16 -120
View File
@@ -25,11 +25,7 @@ import {
} from "@ant-design/icons";
import type { AssetItem } from "@/api/assets";
import { getAssets, getAssetLibraries } from "@/api/assets";
import {
createEditPlan,
generateEditPlan,
updateEditPlan,
} from "@/api/editPlans";
import { createEditPlan, generateEditPlan } from "@/api/editPlans";
import { getEditingTemplates } from "@/api/editingPlanner";
import { getTitles } from "@/api/titles";
import apiClient from "@/api/client";
@@ -128,9 +124,9 @@ const GeneratePage: React.FC = () => {
/* ── 标题 ── */
const [title, setTitle] = useState("");
const { data: userTitles = [] } = useQuery({
queryKey: ["titles"],
queryKey: ["generate-titles"],
queryFn: () => getTitles(),
staleTime: 30_000,
staleTime: 60_000,
});
/* 当选中模板开启了「AI自动匹配标题」,自动填入模板预设标题 */
useEffect(() => {
@@ -518,9 +514,6 @@ const GeneratePage: React.FC = () => {
source_edit_plan_id: editPlanId || undefined,
});
// 后端要求计划处于 editing 状态才能触发渲染,自动转换状态
await updateEditPlan(plan.id, { status: "editing" });
await generateEditPlan(plan.id);
const poll = async () => {
@@ -540,8 +533,7 @@ const GeneratePage: React.FC = () => {
if (data.plan_status === "failed") {
setGenerating(false);
// 提取后端返回的错误详情,便于排查
// 注意:后端返回的 error_message/error/message 可能是对象而非字符串
const rawMsg =
const errorMsg =
data.error_message ||
data.error ||
data.message ||
@@ -549,21 +541,6 @@ const GeneratePage: React.FC = () => {
(c: { status: string }) => c.status === "failed",
)?.error_message ||
"视频生成失败,请联系管理员或重试";
// 安全提取字符串:递归处理嵌套对象(后端可能返回 {code, message: {code, message}} 等嵌套结构)
const safeExtract = (val: unknown): string => {
if (typeof val === "string") return val;
if (typeof val === "object" && val !== null) {
const obj = val as Record<string, unknown>;
if (typeof obj.message === "string") return obj.message;
if (typeof obj.msg === "string") return obj.msg;
if (typeof obj.detail === "string") return obj.detail;
if (obj.message && typeof obj.message === "object")
return safeExtract(obj.message);
return JSON.stringify(val);
}
return String(val ?? "");
};
const errorMsg = safeExtract(rawMsg);
console.error("[生成失败] planId:", plan.id, "响应:", data);
setGenerateError(errorMsg);
message.error(errorMsg);
@@ -601,36 +578,19 @@ const GeneratePage: React.FC = () => {
const axiosErr = err as {
response?: {
data?: {
message?: string | object;
error?: string | object;
detail?: string | object;
msg?: string | object;
message?: string;
error?: string;
detail?: string;
msg?: string;
};
};
message?: string;
};
// 安全提取错误消息:递归处理嵌套对象(后端可能返回 {code, message: {code, message}} 等)
const extractString = (val: unknown): string => {
if (typeof val === "string") return val;
if (typeof val === "object" && val !== null) {
const obj = val as Record<string, unknown>;
if (typeof obj.message === "string") return obj.message;
if (typeof obj.msg === "string") return obj.msg;
if (typeof obj.detail === "string") return obj.detail;
// 嵌套对象:递归提取
if (typeof obj.message === "object" && obj.message !== null)
return extractString(obj.message);
if (typeof obj.msg === "object" && obj.msg !== null)
return extractString(obj.msg);
return JSON.stringify(val);
}
return "";
};
const backendMsg =
extractString(axiosErr.response?.data?.message) ||
extractString(axiosErr.response?.data?.error) ||
extractString(axiosErr.response?.data?.detail) ||
extractString(axiosErr.response?.data?.msg) ||
axiosErr.response?.data?.message ||
axiosErr.response?.data?.error ||
axiosErr.response?.data?.detail ||
axiosErr.response?.data?.msg ||
axiosErr.message ||
"";
console.error(
@@ -639,71 +599,9 @@ const GeneratePage: React.FC = () => {
"完整错误:",
axiosErr,
);
// 确保 errorMsg 一定是字符串(后端可能返回 {code, message} 嵌套对象)
const safeExtractErr = (val: unknown): string => {
if (typeof val === "string") return val;
if (typeof val === "object" && val !== null) {
const obj = val as Record<string, unknown>;
if (typeof obj.message === "string") return obj.message;
if (typeof obj.msg === "string") return obj.msg;
if (typeof obj.detail === "string") return obj.detail;
// 嵌套对象:递归提取
if (typeof obj.message === "object")
return safeExtractErr(obj.message);
return JSON.stringify(val);
}
return String(val ?? "");
};
const rawError = safeExtractErr(backendMsg);
// 将技术错误翻译为用户友好提示(不暴露状态机、字段名等内部概念)
const translateError = (msg: string): string => {
if (!msg) return "生成失败,请检查网络后重试或联系管理员";
// 状态机相关错误
if (
msg.includes("editing") ||
msg.includes("draft") ||
msg.includes("状态")
) {
return "正在准备生成,请稍候再试";
}
// 参数校验错误
if (
msg.includes("template_id") ||
msg.includes("not found") ||
msg.includes("不存在")
) {
return "所选模板或素材不可用,请重新选择";
}
if (
msg.includes("asset") &&
(msg.includes("not found") || msg.includes("missing"))
) {
return "素材数据异常,请返回素材库重新检查";
}
// 网络/超时
if (
msg.includes("timeout") ||
msg.includes("network") ||
msg.includes("ECONN")
) {
return "网络连接超时,请检查网络后重试";
}
// 配额/限制
if (
msg.includes("quota") ||
msg.includes("limit") ||
msg.includes("exceed")
) {
return "已达到生成次数上限,请稍后再试或联系客服";
}
// 兜底:返回原始消息(如果已经是中文人话)或默认提示
if (msg.length > 0 && msg.length < 100 && !msg.includes("{"))
return msg;
return "生成失败,请稍后重试或联系管理员";
};
const finalMsg = translateError(rawError);
setGenerateError(finalMsg);
message.error(finalMsg);
const errorMsg = backendMsg || "生成失败,请检查网络后重试或联系管理员";
setGenerateError(errorMsg);
message.error(errorMsg);
}
}, [
title,
@@ -1604,9 +1502,7 @@ const GeneratePage: React.FC = () => {
</Text>
<Text style={{ color: "var(--text-secondary)", fontSize: 12 }}>
{typeof generateError === "string"
? generateError
: JSON.stringify(generateError)}
{generateError}
</Text>
</div>
</div>
+182 -84
View File
@@ -2,11 +2,10 @@
* 标题库页面 — V21 设计系统
* 两栏布局:左侧分类列表(220px)+ 右侧标题卡片网格(3列)
* 支持:标题卡片展示、AI 生成标题、复制/编辑/删除、收藏、分类筛选、搜索
* 对接后端真实 APIGET/POST/PUT/DELETE /titles
* 使用 mock 数据,后端 API 对接暂不要求
*/
import React, { useMemo, useState, useCallback } from "react";
import { Modal as AntModal, message, Popconfirm } from "antd";
import { useQuery, useMutation, useQueryClient } from "@tanstack/react-query";
import {
PlusOutlined,
SearchOutlined,
@@ -20,13 +19,6 @@ import {
FileTextOutlined,
} from "@ant-design/icons";
import { Button, Input, Select } from "@/components/ui";
import {
getTitles,
createTitle,
updateTitle,
deleteTitle,
type TitleItem,
} from "@/api/titles";
import "./titles.css";
/* ============================================================
@@ -64,16 +56,143 @@ const MOCK_CATEGORIES: CategoryItem[] = [
{ id: "cat-5", name: "教育学习", count: 2 },
];
/** 后端 TitleItem → 前端 TitleData 映射 */
const toTitleData = (item: TitleItem): TitleData => ({
id: item.id,
content: item.content,
type: (item.category as TitleType) || "normal",
industry: "general",
usageCount: 0,
isFavorited: false,
createdAt: item.created_at?.slice(0, 10) || "",
});
const MOCK_TITLES: TitleData[] = [
{
id: "t-1",
content: "这家隐藏在巷子里的小店,味道绝了!",
type: "hot",
industry: "food",
usageCount: 128,
isFavorited: true,
createdAt: "2026-06-28",
},
{
id: "t-2",
content: "2026 年最值得入手的 5 款蓝牙耳机",
type: "hot",
industry: "tech",
usageCount: 96,
isFavorited: false,
createdAt: "2026-06-27",
},
{
id: "t-3",
content: "周末在家做了一道妈妈的味道",
type: "normal",
industry: "food",
usageCount: 42,
isFavorited: false,
createdAt: "2026-06-26",
},
{
id: "t-4",
content: "用 AI 帮我写了一周的小红书文案,效果惊人",
type: "hot",
industry: "tech",
usageCount: 215,
isFavorited: true,
createdAt: "2026-06-25",
},
{
id: "t-5",
content: "今天穿了一套被路人要链接的衣服",
type: "creative",
industry: "beauty",
usageCount: 67,
isFavorited: false,
createdAt: "2026-06-24",
},
{
id: "t-6",
content: "分享我的早起 5 点俱乐部 30 天打卡体验",
type: "normal",
industry: "education",
usageCount: 38,
isFavorited: true,
createdAt: "2026-06-23",
},
{
id: "t-7",
content: "这个平价面霜居然比大牌还好用?",
type: "hot",
industry: "beauty",
usageCount: 183,
isFavorited: false,
createdAt: "2026-06-22",
},
{
id: "t-8",
content: "一个人的旅行也可以很精彩",
type: "normal",
industry: "travel",
usageCount: 55,
isFavorited: false,
createdAt: "2026-06-21",
},
{
id: "t-9",
content: "考研上岸!我的备考时间管理方法全公开",
type: "hot",
industry: "education",
usageCount: 147,
isFavorited: true,
createdAt: "2026-06-20",
},
{
id: "t-10",
content: "把旧 T 恤改造成时尚单品,零成本!",
type: "creative",
industry: "beauty",
usageCount: 29,
isFavorited: false,
createdAt: "2026-06-19",
},
{
id: "t-11",
content: "这家咖啡馆的氛围感也太好了吧",
type: "normal",
industry: "food",
usageCount: 74,
isFavorited: false,
createdAt: "2026-06-18",
},
{
id: "t-12",
content: "手机摄影技巧:拍出电影感画面",
type: "creative",
industry: "tech",
usageCount: 61,
isFavorited: false,
createdAt: "2026-06-17",
},
{
id: "t-13",
content: "带娃旅行必备清单,少带一样都崩溃",
type: "hot",
industry: "travel",
usageCount: 109,
isFavorited: false,
createdAt: "2026-06-16",
},
{
id: "t-14",
content: "30 天学会一门新语言?我的实验记录",
type: "creative",
industry: "education",
usageCount: 33,
isFavorited: false,
createdAt: "2026-06-15",
},
{
id: "t-15",
content: "今天做了一道让全家惊艳的菜",
type: "normal",
industry: "food",
usageCount: 48,
isFavorited: false,
createdAt: "2026-06-14",
},
];
/* ============================================================
* 工具函数
@@ -248,48 +367,12 @@ const TitleCard: React.FC<{
* 主组件
* ============================================================ */
const TitleLibrary: React.FC = () => {
const queryClient = useQueryClient();
/* 分类数据 */
const [categories, setCategories] = useState<CategoryItem[]>(MOCK_CATEGORIES);
const [activeCatId, setActiveCatId] = useState<string>(MOCK_CATEGORIES[0].id);
/* 标题数据 — 真实 API */
const { data: apiTitles = [] } = useQuery({
queryKey: ["titles"],
queryFn: getTitles,
staleTime: 30_000,
});
const titles: TitleData[] = useMemo(
() => apiTitles.map(toTitleData),
[apiTitles],
);
/* CRUD mutations */
const createMutation = useMutation({
mutationFn: (content: string) => createTitle({ content }),
onSuccess: () => {
queryClient.invalidateQueries({ queryKey: ["titles"] });
},
onError: () => message.error("创建标题失败"),
});
const updateMutation = useMutation({
mutationFn: ({ id, content }: { id: string; content: string }) =>
updateTitle(id, { content }),
onSuccess: () => {
queryClient.invalidateQueries({ queryKey: ["titles"] });
},
onError: () => message.error("更新标题失败"),
});
const deleteMutation = useMutation({
mutationFn: (id: string) => deleteTitle(id),
onSuccess: () => {
queryClient.invalidateQueries({ queryKey: ["titles"] });
},
onError: () => message.error("删除标题失败"),
});
/* 标题数据 */
const [titles, setTitles] = useState<TitleData[]>(MOCK_TITLES);
/* 筛选 */
const [searchText, setSearchText] = useState("");
@@ -380,9 +463,13 @@ const TitleLibrary: React.FC = () => {
searchText,
]);
/* 收藏切换(暂不支持,待后端 API */
const handleToggleFavorite = useCallback((_id: string) => {
message.info("收藏功能即将上线");
/* 收藏切换 */
const handleToggleFavorite = useCallback((id: string) => {
setTitles((prev) =>
prev.map((t) =>
t.id === id ? { ...t, isFavorited: !t.isFavorited } : t,
),
);
}, []);
/* 复制 */
@@ -406,13 +493,15 @@ const TitleLibrary: React.FC = () => {
message.warning("标题内容不能为空");
return;
}
if (editingId) {
updateMutation.mutate({ id: editingId, content: editText.trim() });
}
setTitles((prev) =>
prev.map((t) =>
t.id === editingId ? { ...t, content: editText.trim() } : t,
),
);
setEditingId(null);
setEditText("");
message.success("标题已更新");
}, [editingId, editText, updateMutation]);
}, [editingId, editText]);
const handleCancelEdit = useCallback(() => {
setEditingId(null);
@@ -420,13 +509,10 @@ const TitleLibrary: React.FC = () => {
}, []);
/* 删除 */
const handleDelete = useCallback(
(id: string) => {
deleteMutation.mutate(id);
message.success("标题已删除");
},
[deleteMutation],
);
const handleDelete = useCallback((id: string) => {
setTitles((prev) => prev.filter((t) => t.id !== id));
message.success("标题已删除");
}, []);
/* 新建分类 */
const handleCreateCategory = () => {
@@ -461,14 +547,20 @@ const TitleLibrary: React.FC = () => {
message.warning("请输入标题内容");
return;
}
createMutation.mutate(newTitleContent.trim(), {
onSuccess: () => {
setCreateTitleModalOpen(false);
setNewTitleContent("");
setNewTitleType("normal");
message.success("标题创建成功");
},
});
const title: TitleData = {
id: `t-${Date.now()}`,
content: newTitleContent.trim(),
type: newTitleType,
industry: "general",
usageCount: 0,
isFavorited: false,
createdAt: new Date().toISOString().slice(0, 10),
};
setTitles((prev) => [title, ...prev]);
setCreateTitleModalOpen(false);
setNewTitleContent("");
setNewTitleType("normal");
message.success("标题创建成功");
};
/* AI 生成标题 */
@@ -497,11 +589,17 @@ const TitleLibrary: React.FC = () => {
/* 采纳 AI 生成的标题 */
const handleAdoptAITitle = (text: string) => {
createMutation.mutate(text, {
onSuccess: () => {
message.success("标题已采纳并添加到标题库");
},
});
const title: TitleData = {
id: `t-${Date.now()}`,
content: text,
type: "creative",
industry: "general",
usageCount: 0,
isFavorited: false,
createdAt: new Date().toISOString().slice(0, 10),
};
setTitles((prev) => [title, ...prev]);
message.success("标题已采纳并添加到标题库");
};
/* 复制 AI 生成的标题 */
+2 -737
View File
@@ -30,7 +30,6 @@ import {
ReloadOutlined,
CloseCircleOutlined,
} from "@ant-design/icons";
import { Modal, Upload, message } from "antd";
import { Button, Input, Select, Tooltip } from "@/components/ui";
import PageHead from "@/components/layout/PageHead";
import {
@@ -45,12 +44,6 @@ import {
toVoiceClone,
type VoiceClone,
} from "@/api/voiceClone";
import { synthesizeSpeech, getTTSJobStatus, saveTtsToLibrary } from "@/api/tts";
import {
uploadAssetDirect,
getAssetLibraries,
createAsset,
} from "@/api/assets";
import CloneModal from "@/components/voice/CloneModal";
import "./voices.css";
@@ -600,41 +593,6 @@ const CloneCardSkeleton: React.FC = () => (
</div>
);
/* ── 辅助函数 ─────────────────────────────────────────── */
const formatFileSize = (bytes: number): string => {
if (bytes < 1024) return `${bytes} B`;
if (bytes < 1024 * 1024) return `${(bytes / 1024).toFixed(1)} KB`;
return `${(bytes / (1024 * 1024)).toFixed(1)} MB`;
};
const getAudioDuration = (file: File): Promise<number> =>
new Promise((resolve) => {
const audio = new Audio();
const url = URL.createObjectURL(file);
audio.addEventListener("loadedmetadata", () => {
resolve(audio.duration);
URL.revokeObjectURL(url);
});
audio.addEventListener("error", () => {
resolve(0);
URL.revokeObjectURL(url);
});
audio.src = url;
});
const buildVoiceMetadata = (data: {
gender?: string;
description?: string;
duration?: number;
}): Record<string, unknown> => {
const metadata: Record<string, unknown> = {};
if (data.gender) metadata.gender = data.gender;
if (data.description) metadata.description = data.description;
if (data.duration) metadata.duration = Math.round(data.duration);
return metadata;
};
/* ============================================================
* 主组件
* ============================================================ */
@@ -655,27 +613,6 @@ const VoiceLibrary: React.FC = () => {
const [toasts, setToasts] = useState<Toast[]>([]);
const [cloneModalOpen, setCloneModalOpen] = useState(false);
/* ── 上传音频弹窗状态 ── */
const [uploadOpen, setUploadOpen] = useState(false);
const [uploadFile, setUploadFile] = useState<File | null>(null);
const [uploadName, setUploadName] = useState("");
const [uploadGender, setUploadGender] = useState<VoiceGender>("female");
const [uploadDesc, setUploadDesc] = useState("");
const [uploadProgress, setUploadProgress] = useState<number | null>(null);
/* ── AI 配音弹窗状态 ── */
const [ttsOpen, setTtsOpen] = useState(false);
const [ttsText, setTtsText] = useState("");
const [ttsVoiceId, setTtsVoiceId] = useState<string>("");
const [ttsSpeed, setTtsSpeed] = useState(1.0);
const [ttsJobId, setTtsJobId] = useState<string | null>(null);
const [ttsStatus, setTtsStatus] = useState<
"idle" | "synthesizing" | "done" | "error"
>("idle");
const [ttsAudioUrl, setTtsAudioUrl] = useState<string | null>(null);
const [ttsError, setTtsError] = useState<string | null>(null);
const ttsTimerRef = useRef<ReturnType<typeof setInterval> | null>(null);
const showToast = useCallback((message: string, type: Toast["type"]) => {
const id = ++toastIdSeq;
setToasts((prev) => [...prev, { id, message, type }]);
@@ -708,130 +645,6 @@ const VoiceLibrary: React.FC = () => {
},
});
/* ── 上传音频 mutation ── */
const uploadMutation = useMutation({
mutationFn: async (data: {
file: File;
name: string;
gender: VoiceGender;
description: string;
}) => {
setUploadProgress(0);
try {
/* 获取或创建默认配音素材库 */
const libs = await queryClient.fetchQuery({
queryKey: ["asset-libraries"],
queryFn: getAssetLibraries,
});
const lib = libs.find((l) => l.kind === "voice");
if (!lib) throw new Error("配音素材库不存在,请先在配音素材库页面创建");
/* 直传文件 */
const { storage_key } = await uploadAssetDirect({
file: data.file,
library_id: lib.id,
onProgress: (p) => setUploadProgress(p),
});
/* 获取音频时长 */
const duration = await getAudioDuration(data.file);
/* 创建素材记录 */
await createAsset({
library_id: lib.id,
name: data.name,
storage_key,
mime_type: data.file.type || "audio/mpeg",
metadata: buildVoiceMetadata({
gender: data.gender,
description: data.description,
duration,
}),
});
} finally {
setUploadProgress(null);
}
},
onSuccess: () => {
queryClient.invalidateQueries({ queryKey: ["assets", "voice"] });
setUploadOpen(false);
setUploadFile(null);
setUploadName("");
setUploadDesc("");
showToast("上传成功", "success");
},
onError: (err: Error) => {
showToast(err.message || "上传失败,请重试", "error");
},
});
/* ── TTS 合成 ── */
const handleTtsSynthesize = useCallback(async () => {
if (!ttsText.trim()) {
message.warning("请输入要合成的文本");
return;
}
setTtsError(null);
setTtsStatus("synthesizing");
setTtsAudioUrl(null);
setTtsJobId(null);
try {
const resp = await synthesizeSpeech({
text: ttsText.trim(),
voice_id: ttsVoiceId || undefined,
speed: ttsSpeed,
});
setTtsJobId(resp.job_id);
/* 轮询状态 */
ttsTimerRef.current = setInterval(async () => {
try {
const job = await getTTSJobStatus(resp.job_id);
if (job.status === "completed") {
clearInterval(ttsTimerRef.current!);
ttsTimerRef.current = null;
setTtsStatus("done");
setTtsAudioUrl(job.output_audio_url);
} else if (job.status === "failed") {
clearInterval(ttsTimerRef.current!);
ttsTimerRef.current = null;
setTtsStatus("error");
setTtsError(job.error_message || "合成失败");
}
} catch {
clearInterval(ttsTimerRef.current!);
ttsTimerRef.current = null;
setTtsStatus("error");
setTtsError("查询合成状态失败");
}
}, 2000);
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : "合成请求失败";
setTtsStatus("error");
setTtsError(msg);
}
}, [ttsText, ttsVoiceId, ttsSpeed]);
/* ── TTS 保存到素材库 ── */
const handleTtsSave = useCallback(async () => {
if (!ttsJobId) return;
try {
await saveTtsToLibrary(ttsJobId, { name: ttsText.slice(0, 50) });
queryClient.invalidateQueries({ queryKey: ["assets", "voice"] });
showToast("已保存到配音素材库", "success");
setTtsOpen(false);
} catch (err: unknown) {
const msg = err instanceof Error ? err.message : "保存失败";
showToast(msg, "error");
}
}, [ttsJobId, ttsText, queryClient, showToast]);
/* ── TTS 定时器清理 ── */
useEffect(() => {
return () => {
if (ttsTimerRef.current) clearInterval(ttsTimerRef.current);
};
}, []);
/* ── 数据查询(任务 3.11:替换 Mock) ──────────────── */
/** 预置音色列表 */
@@ -975,12 +788,7 @@ const VoiceLibrary: React.FC = () => {
const pageActions = (
<div className="xx-voices-actions">
<Button
buttonType="ghost"
buttonSize="sm"
icon={<UploadOutlined />}
onClick={() => setUploadOpen(true)}
>
<Button buttonType="ghost" buttonSize="sm" icon={<UploadOutlined />}>
</Button>
<Button
@@ -991,12 +799,7 @@ const VoiceLibrary: React.FC = () => {
>
</Button>
<Button
buttonType="primary"
buttonSize="sm"
icon={<RobotOutlined />}
onClick={() => setTtsOpen(true)}
>
<Button buttonType="primary" buttonSize="sm" icon={<RobotOutlined />}>
AI
</Button>
</div>
@@ -1205,544 +1008,6 @@ const VoiceLibrary: React.FC = () => {
/>
)}
{/* ── 上传音频弹窗 ── */}
<Modal
title="上传音频"
open={uploadOpen}
onCancel={() => {
if (uploadProgress !== null) return; // 上传中不可关闭
setUploadOpen(false);
setUploadFile(null);
setUploadName("");
setUploadDesc("");
}}
footer={null}
width={520}
maskClosable={uploadProgress === null}
>
<div
style={{
display: "flex",
flexDirection: "column",
gap: 16,
padding: "8px 0",
}}
>
{/* 拖拽上传区 */}
<Upload.Dragger
accept="audio/*"
maxCount={1}
beforeUpload={(file) => {
setUploadFile(file);
if (!uploadName) setUploadName(file.name.replace(/\.[^.]+$/, ""));
return false;
}}
onRemove={() => {
setUploadFile(null);
setUploadProgress(null);
}}
showUploadList={false}
disabled={uploadProgress !== null}
>
<p
style={{
fontSize: 32,
color: "var(--primary-color)",
marginBottom: 8,
}}
>
<UploadOutlined />
</p>
<p style={{ fontSize: 14, fontWeight: 500, margin: "0 0 4px" }}>
</p>
<p
style={{
fontSize: 12,
color: "var(--text-secondary)",
margin: 0,
}}
>
MP3WAVAACFLAC 200MB
</p>
</Upload.Dragger>
{/* 已选文件信息 */}
{uploadFile && (
<div
style={{
padding: "10px 12px",
background: "var(--bg-secondary)",
borderRadius: 8,
display: "flex",
alignItems: "center",
gap: 10,
}}
>
<SoundOutlined
style={{ fontSize: 18, color: "var(--primary-color)" }}
/>
<div style={{ flex: 1, minWidth: 0 }}>
<div
style={{
fontSize: 13,
fontWeight: 500,
overflow: "hidden",
textOverflow: "ellipsis",
whiteSpace: "nowrap",
}}
>
{uploadFile.name}
</div>
<div style={{ fontSize: 11, color: "var(--text-secondary)" }}>
{formatFileSize(uploadFile.size)}
</div>
</div>
</div>
)}
{/* 上传进度 */}
{uploadProgress !== null && (
<div style={{ textAlign: "center", padding: "8px 0" }}>
<div
style={{
fontSize: 22,
fontWeight: 700,
color: "var(--primary-color)",
}}
>
{uploadProgress}%
</div>
<div style={{ fontSize: 12, color: "var(--text-secondary)" }}>
{uploadProgress < 100 ? "上传中..." : "处理中..."}
</div>
<div
style={{
height: 4,
background: "var(--bg-tertiary)",
borderRadius: 2,
marginTop: 8,
overflow: "hidden",
}}
>
<div
style={{
height: "100%",
width: `${uploadProgress}%`,
background: "var(--primary-color)",
borderRadius: 2,
transition: "width 0.3s ease",
}}
/>
</div>
</div>
)}
{/* 名称 */}
<div>
<div
style={{
fontSize: 13,
color: "var(--text-secondary)",
marginBottom: 6,
}}
>
</div>
<input
value={uploadName}
onChange={(e) => setUploadName(e.target.value)}
placeholder="输入素材名称"
maxLength={100}
disabled={uploadProgress !== null}
style={{
width: "100%",
padding: "8px 12px",
border: "1px solid var(--border-color)",
borderRadius: 8,
fontSize: 13,
background: "var(--bg-primary)",
color: "var(--text-primary)",
outline: "none",
}}
/>
</div>
{/* 性别选择 */}
<div>
<div
style={{
fontSize: 13,
color: "var(--text-secondary)",
marginBottom: 6,
}}
>
</div>
<div style={{ display: "flex", gap: 8 }}>
{(["female", "male", "child"] as VoiceGender[]).map((g) => (
<button
key={g}
type="button"
onClick={() => setUploadGender(g)}
disabled={uploadProgress !== null}
style={{
flex: 1,
padding: "6px 0",
borderRadius: 8,
border: `1px solid ${uploadGender === g ? "var(--primary-color)" : "var(--border-color)"}`,
background:
uploadGender === g
? "var(--primary-soft)"
: "transparent",
color:
uploadGender === g
? "var(--primary-color)"
: "var(--text-secondary)",
fontSize: 13,
fontWeight: uploadGender === g ? 600 : 400,
cursor: uploadProgress !== null ? "not-allowed" : "pointer",
transition: "all 0.2s",
}}
>
{g === "female" ? "女声" : g === "male" ? "男声" : "童声"}
</button>
))}
</div>
</div>
{/* 描述 */}
<div>
<div
style={{
fontSize: 13,
color: "var(--text-secondary)",
marginBottom: 6,
}}
>
</div>
<textarea
value={uploadDesc}
onChange={(e) => setUploadDesc(e.target.value)}
placeholder="描述这个音色的特点..."
maxLength={500}
rows={2}
disabled={uploadProgress !== null}
style={{
width: "100%",
padding: "8px 12px",
border: "1px solid var(--border-color)",
borderRadius: 8,
fontSize: 13,
background: "var(--bg-primary)",
color: "var(--text-primary)",
outline: "none",
resize: "vertical",
fontFamily: "inherit",
}}
/>
</div>
{/* 操作按钮 */}
<div
style={{
display: "flex",
justifyContent: "flex-end",
gap: 10,
paddingTop: 4,
}}
>
<button
type="button"
onClick={() => {
setUploadOpen(false);
setUploadFile(null);
setUploadName("");
setUploadDesc("");
}}
disabled={uploadProgress !== null}
style={{
padding: "8px 20px",
borderRadius: 8,
border: "1px solid var(--border-color)",
background: "transparent",
fontSize: 13,
cursor: uploadProgress !== null ? "not-allowed" : "pointer",
color: "var(--text-secondary)",
}}
>
</button>
<button
type="button"
onClick={() => {
if (!uploadFile) {
message.warning("请先选择音频文件");
return;
}
if (!uploadName.trim()) {
message.warning("请输入素材名称");
return;
}
uploadMutation.mutate({
file: uploadFile,
name: uploadName.trim(),
gender: uploadGender,
description: uploadDesc.trim(),
});
}}
disabled={
!uploadFile || !uploadName.trim() || uploadProgress !== null
}
style={{
padding: "8px 20px",
borderRadius: 8,
border: "none",
background:
!uploadFile || !uploadName.trim() || uploadProgress !== null
? "var(--text-tertiary)"
: "var(--primary-color)",
color: "#fff",
fontSize: 13,
fontWeight: 600,
cursor:
!uploadFile || !uploadName.trim() || uploadProgress !== null
? "not-allowed"
: "pointer",
}}
>
{uploadProgress !== null ? "上传中..." : "开始上传"}
</button>
</div>
</div>
</Modal>
{/* ── AI 配音弹窗 ── */}
<Modal
title="AI 配音"
open={ttsOpen}
onCancel={() => {
setTtsOpen(false);
setTtsText("");
setTtsVoiceId("");
setTtsSpeed(1.0);
setTtsStatus("idle");
setTtsAudioUrl(null);
setTtsError(null);
setTtsJobId(null);
}}
footer={null}
width={560}
>
<div
style={{
display: "flex",
flexDirection: "column",
gap: 16,
padding: "8px 0",
}}
>
{/* 文本输入 */}
<div>
<div
style={{
fontSize: 13,
color: "var(--text-secondary)",
marginBottom: 6,
}}
>
</div>
<textarea
value={ttsText}
onChange={(e) => setTtsText(e.target.value)}
placeholder="输入要配音的文本内容..."
maxLength={2000}
rows={4}
style={{
width: "100%",
padding: "10px 12px",
border: "1px solid var(--border-color)",
borderRadius: 8,
fontSize: 13,
background: "var(--bg-primary)",
color: "var(--text-primary)",
outline: "none",
resize: "vertical",
fontFamily: "inherit",
lineHeight: 1.6,
}}
/>
<div
style={{
fontSize: 11,
color: "var(--text-tertiary)",
textAlign: "right",
marginTop: 4,
}}
>
{ttsText.length}/2000
</div>
</div>
{/* 音色选择 */}
<div>
<div
style={{
fontSize: 13,
color: "var(--text-secondary)",
marginBottom: 6,
}}
>
</div>
<select
value={ttsVoiceId}
onChange={(e) => setTtsVoiceId(e.target.value)}
style={{
width: "100%",
padding: "8px 12px",
border: "1px solid var(--border-color)",
borderRadius: 8,
fontSize: 13,
background: "var(--bg-primary)",
color: "var(--text-primary)",
outline: "none",
}}
>
<option value=""></option>
{presetVoices.map((v) => (
<option key={v.id} value={v.id}>
{v.name} {genderLabel(v.gender)}
</option>
))}
</select>
</div>
{/* 语速 */}
<div>
<div
style={{
fontSize: 13,
color: "var(--text-secondary)",
marginBottom: 6,
}}
>
{ttsSpeed.toFixed(1)}x
</div>
<input
type="range"
min={0.5}
max={2.0}
step={0.1}
value={ttsSpeed}
onChange={(e) => setTtsSpeed(parseFloat(e.target.value))}
style={{ width: "100%", accentColor: "var(--primary-color)" }}
/>
<div
style={{
display: "flex",
justifyContent: "space-between",
fontSize: 11,
color: "var(--text-tertiary)",
}}
>
<span>0.5x</span>
<span>1.0x</span>
<span>2.0x</span>
</div>
</div>
{/* 合成按钮 */}
<button
type="button"
onClick={handleTtsSynthesize}
disabled={ttsStatus === "synthesizing" || !ttsText.trim()}
style={{
width: "100%",
padding: "10px 0",
borderRadius: 8,
border: "none",
background:
ttsStatus === "synthesizing" || !ttsText.trim()
? "var(--text-tertiary)"
: "var(--primary-color)",
color: "#fff",
fontSize: 14,
fontWeight: 600,
cursor:
ttsStatus === "synthesizing" || !ttsText.trim()
? "not-allowed"
: "pointer",
display: "flex",
alignItems: "center",
justifyContent: "center",
gap: 8,
}}
>
<RobotOutlined />
{ttsStatus === "synthesizing" ? "合成中..." : "开始合成"}
</button>
{/* 错误提示 */}
{ttsError && (
<div
style={{
padding: "10px 12px",
background: "var(--error-soft, #fff2f0)",
borderRadius: 8,
color: "var(--error-color, #ff4d4f)",
fontSize: 13,
}}
>
{ttsError}
</div>
)}
{/* 合成结果 */}
{ttsStatus === "done" && ttsAudioUrl && (
<div
style={{
padding: 12,
background: "var(--bg-secondary)",
borderRadius: 8,
display: "flex",
flexDirection: "column",
gap: 10,
}}
>
<div
style={{
fontSize: 13,
fontWeight: 500,
color: "var(--success-color, #52c41a)",
}}
>
</div>
<audio controls src={ttsAudioUrl} style={{ width: "100%" }} />
<button
type="button"
onClick={handleTtsSave}
style={{
padding: "8px 0",
borderRadius: 8,
border: "1px solid var(--primary-color)",
background: "var(--primary-soft)",
color: "var(--primary-color)",
fontSize: 13,
fontWeight: 600,
cursor: "pointer",
}}
>
</button>
</div>
)}
</div>
</Modal>
{/* Toast 提示 */}
{toasts.length > 0 && (
<div className="vc-toast-container">
@@ -215,8 +215,6 @@ def render_edit_plan(self, plan_id: str) -> dict:
"""
logger.info("开始渲染剪辑计划: plan_id=%s", plan_id)
generation_task_id = ""
for repos in _get_repos():
plan_repo, clip_repo, gen_task_repo, db = repos
@@ -227,9 +225,6 @@ def render_edit_plan(self, plan_id: str) -> dict:
logger.error("剪辑计划不存在: %s", plan_id)
return {"status": "error", "message": f"计划不存在: {plan_id}"}
# 获取 generation_task_id(提前读取,确保 except 块可用)
generation_task_id = plan.config.get("generation_task_id", "")
# 2. 加载片段列表(按 order 排序)
clips = clip_repo.list_by_plan(plan_id, skip=0, limit=10000)
if not clips:
@@ -238,6 +233,9 @@ def render_edit_plan(self, plan_id: str) -> dict:
plan_repo.update(plan)
return {"status": "error", "message": "没有可渲染的片段"}
# 获取 generation_task_id
generation_task_id = plan.config.get("generation_task_id", "")
# 更新 GenerationTask 状态为 running
if generation_task_id:
gen_task = gen_task_repo.get(generation_task_id)
@@ -347,31 +345,15 @@ def render_edit_plan(self, plan_id: str) -> dict:
except Exception as exc:
logger.exception("渲染剪辑计划异常: %s", plan_id)
# 尝试标记计划和 GenerationTask 为失败
# 尝试标记计划为失败
try:
plan = plan_repo.get(plan_id)
if plan and plan.status.value == "rendering":
plan.mark_failed()
plan_repo.update(plan)
except Exception as e:
logger.warning("标记计划失败时异常: plan_id=%s error=%s", plan_id, e, exc_info=True)
# 更新 GenerationTask 状态为 failed,前端轮询能看到失败状态
try:
if generation_task_id:
gen_task = gen_task_repo.get(generation_task_id)
if gen_task and gen_task.status.value != "failed":
gen_task.status = "failed"
gen_task.error_message = f"渲染异常: {type(exc).__name__}: {exc}"
gen_task.completed_at = datetime.now(timezone.utc)
gen_task_repo.update(gen_task)
logger.info(
"GenerationTask 已标记为 failed: task_id=%s plan_id=%s",
generation_task_id,
plan_id,
)
except Exception as e:
logger.warning(
"更新 GenerationTask 失败状态时异常: task_id=%s error=%s", generation_task_id, e, exc_info=True
f"Operation failed in apps/worker/worker_app/tasks/edit_plan_generation.py: {e}", exc_info=True
)
raise self.retry(exc=exc, countdown=60)
+1 -1
View File
@@ -4,7 +4,7 @@
# ============================================================
# 基础镜像:Python 3.12
FROM git.xiaoxiajianji.com/xiaoxia/base/python:3.12-slim
FROM python:3.12-slim-bookworm
# 使用阿里云镜像加速
RUN sed -i 's|deb.debian.org|mirrors.aliyun.com|g' /etc/apt/sources.list.d/debian.sources 2>/dev/null || \
+1 -1
View File
@@ -1,4 +1,4 @@
FROM git.xiaoxiajianji.com/xiaoxia/base/python:3.12-slim
FROM python:3.12-slim
ENV PIP_DISABLE_PIP_VERSION_CHECK=1 \
PIP_NO_CACHE_DIR=0 \
+1 -1
View File
@@ -1,4 +1,4 @@
FROM git.xiaoxiajianji.com/xiaoxia/base/nginx:alpine AS runner
FROM docker.m.daocloud.io/library/nginx:alpine AS runner
ARG NGINX_CONF=infra/docker/nginx.conf
WORKDIR /usr/share/nginx/html
COPY apps/web/dist ./
+5 -3
View File
@@ -1,17 +1,19 @@
# Build stage
FROM git.xiaoxiajianji.com/xiaoxia/base/node:20 AS builder
FROM docker.m.daocloud.io/library/node:20 AS builder
WORKDIR /app
ARG VITE_API_URL=https://saas-api.xiaoxiajianji.com
ENV VITE_API_URL=$VITE_API_URL
COPY apps/web/package.json apps/web/package-lock.json ./apps/web/
WORKDIR /app/apps/web
RUN npm config set registry https://registry.npmmirror.com \
RUN \
--mount=type=cache,target=/root/.npm \
npm config set registry https://registry.npmmirror.com \
&& npm ci
COPY apps/web/ ./
RUN npm run build
# Production stage with nginx
FROM git.xiaoxiajianji.com/xiaoxia/base/nginx:alpine AS runner
FROM docker.m.daocloud.io/library/nginx:alpine AS runner
ARG NGINX_CONF=infra/docker/nginx.conf
WORKDIR /usr/share/nginx/html
COPY --from=builder /app/apps/web/dist ./
+2 -2
View File
@@ -4,7 +4,7 @@
# ============================================================
# 基础镜像:Python 3.12 + ffmpeg
FROM git.xiaoxiajianji.com/xiaoxia/base/python:3.12-slim
FROM python:3.12-slim-bookworm
# 使用阿里云镜像加速
RUN sed -i 's|deb.debian.org|mirrors.aliyun.com|g' /etc/apt/sources.list.d/debian.sources 2>/dev/null || \
@@ -15,7 +15,7 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
ffmpeg \
libsm6 \
libxext6 \
libgl1 \
libgl1-mesa-glx \
&& rm -rf /var/lib/apt/lists/*
# 设置工作目录
@@ -16,9 +16,6 @@ class InMemoryAssetLibraryRepository:
def get(self, library_id: str) -> AssetLibrary | None:
return self._libraries.get(library_id)
def find_by_id(self, library_id: str) -> AssetLibrary | None:
return self.get(library_id)
def find_by_project(self, project_id: str, kind: AssetLibraryKind | None = None) -> list[AssetLibrary]:
items = [library for library in self._libraries.values() if library.project_id == project_id]
if kind is not None:
+11 -23
View File
@@ -3,20 +3,11 @@ set -eu
VERSION="${1:-${RELEASE_VERSION:-}}"
if [ -z "$VERSION" ]; then
echo "Usage: $0 <version> [staging|production]"
echo "Example: $0 v0.1.5 production"
echo " $0 abc1234 staging"
echo "Usage: $0 <version>"
echo "Example: $0 v0.1.5"
exit 1
fi
# 环境参数:staging 或 production(默认 production
BUILD_ENV="${2:-production}"
case "$BUILD_ENV" in
staging) NGINX_CONF_FILE="infra/docker/nginx-staging.conf" ;;
*) NGINX_CONF_FILE="infra/docker/nginx-production.conf" ;;
esac
echo "Build environment: $BUILD_ENV → nginx config: $NGINX_CONF_FILE"
ROOT_DIR="$(CDPATH= cd -- "$(dirname -- "$0")/.." && pwd)"
cd "$ROOT_DIR"
@@ -81,28 +72,25 @@ else
fi
echo "=== Building Web image (with buildx cache) ==="
# 先构建前端产物
docker run --rm \
-v "$PWD:/workspace" \
-w /workspace/apps/web \
docker.m.daocloud.io/library/node:20 \
sh -lc "npm ci && npm run build"
test -f apps/web/dist/index.html
# 使用多阶段 Dockerfile 构建,配合 buildx cache 缓存 npm 安装和构建产物
WEB_NGINX_CONF="${WEB_NGINX_CONF:-infra/docker/nginx-production.conf}"
WEB_API_URL="${VITE_API_URL:-https://saas-api.xiaoxiajianji.com}"
if [ "$USE_CACHE" -eq 1 ]; then
docker buildx build \
--cache-from "type=registry,ref=${CACHE_REGISTRY}/web-cache:${CACHE_TAG},ignore-error=true" \
--cache-to "type=registry,ref=${CACHE_REGISTRY}/web-cache:${CACHE_TAG},mode=max" \
-f infra/docker/web-artifact.Dockerfile \
--build-arg "NGINX_CONF=$NGINX_CONF_FILE" \
-f infra/docker/web.Dockerfile \
--build-arg NGINX_CONF="${WEB_NGINX_CONF}" \
--build-arg VITE_API_URL="${WEB_API_URL}" \
-t "$WEB_IMAGE" \
--load \
.
else
docker build --pull=false \
-f infra/docker/web-artifact.Dockerfile \
--build-arg "NGINX_CONF=$NGINX_CONF_FILE" \
-f infra/docker/web.Dockerfile \
--build-arg NGINX_CONF="${WEB_NGINX_CONF}" \
--build-arg VITE_API_URL="${WEB_API_URL}" \
-t "$WEB_IMAGE" \
.
fi
-249
View File
@@ -1,249 +0,0 @@
#!/usr/bin/env python3
"""
数据库迁移破坏性变更安全检查
只检查 Alembic 迁移文件的 upgrade 函数中是否包含破坏性操作:
- DROP TABLE
- ALTER TABLE ... DROP COLUMN
- 列类型变更(可能导致数据丢失)
- NOT NULL 约束新增(无默认值时)
- RENAME TABLE / RENAME COLUMN
忽略 downgrade 函数中的操作(那是回滚逻辑,正常的)。
使用方式:
# 检查所有迁移(不推荐,会扫历史已执行的迁移)
python3 scripts/check_migration_safety.py
# 只检查与目标分支相比新增的迁移(推荐用于CI)
python3 scripts/check_migration_safety.py --diff-against origin/main
# 只检查指定版本之后的迁移
python3 scripts/check_migration_safety.py --since 030_xxx
退出码:
0 - 安全 / 只有非破坏性变更
1 - 检测到高风险破坏性变更
2 - 检测到中风险变更,需人工确认
"""
from __future__ import annotations
import argparse
import os
import re
import subprocess
import sys
from pathlib import Path
from typing import List, Tuple
REPO_ROOT = Path(__file__).resolve().parents[1]
ALEMBIC_VERSIONS_DIR = REPO_ROOT / "alembic" / "versions"
# 高风险模式:直接导致数据丢失(只在 upgrade 中检查)
HIGH_RISK_PATTERNS = [
(r"\bop\.drop_table\(", "op.drop_table() - 删除表,数据永久丢失"),
(r"\bop\.drop_column\(", "op.drop_column() - 删除列,数据永久丢失"),
]
# 中风险模式:可能导致数据丢失或兼容性问题
MEDIUM_RISK_PATTERNS = [
(r"op\.alter_column\([^)]*nullable\s*=\s*False", "新增 NOT NULL 约束 - 旧数据可能为空导致迁移失败"),
(r"op\.alter_column\([^)]*type_\s*=", "列类型变更 - 可能导致数据截断或转换失败"),
(r"\bop\.rename_table\(", "op.rename_table() - 重命名表,可能导致依赖该表的代码报错"),
(r"\bop\.rename_column\(", "op.rename_column() - 重命名列,可能导致依赖该列的代码报错"),
(r"\bop\.drop_index\(", "op.drop_index() - 删除索引,可能影响查询性能"),
(r"\bop\.drop_constraint\(", "op.drop_constraint() - 删除约束,可能影响数据完整性"),
]
# 安全模式:这些是安全的新增操作
SAFE_PATTERNS = [
(r"\bop\.create_table\(", "新建表"),
(r"\bop\.add_column\(", "新增列"),
(r"\bop\.create_index\(", "新建索引"),
(r"\bop\.create_unique_constraint\(", "新建唯一约束"),
(r"\bop\.create_foreign_key\(", "新建外键约束"),
]
def extract_upgrade_content(content: str) -> str:
"""
从迁移文件中提取 upgrade 函数的内容。
只检查 upgrade 中的操作,忽略 downgrade。
"""
upgrade_match = re.search(r"def upgrade\b[^:]*:", content)
if not upgrade_match:
return ""
upgrade_start = upgrade_match.end()
# 找到下一个顶层 def(通常是 def downgrade)作为结束位置
rest = content[upgrade_start:]
downgrade_match = re.search(r"\n\ndef\s+\w+\b", rest)
if downgrade_match:
upgrade_end = upgrade_start + downgrade_match.start()
else:
upgrade_end = len(content)
return content[upgrade_start:upgrade_end]
def get_new_migrations_via_diff(diff_target: str) -> List[Path]:
"""
通过 git diff 对比目标分支/commit,找出 alembic/versions/ 下新增的迁移文件。
只包含新增文件(A状态),不包含修改或删除的文件。
"""
try:
result = subprocess.run(
["git", "diff", "--name-only", "--diff-filter=A", diff_target, "HEAD", "--", "alembic/versions/"],
cwd=str(REPO_ROOT),
capture_output=True,
text=True,
check=True,
)
files = [line.strip() for line in result.stdout.strip().split("\n") if line.strip()]
return [REPO_ROOT / f for f in files]
except subprocess.CalledProcessError as e:
print(f"⚠️ git diff 失败({diff_target}):{e.stderr.strip()}")
print(f" 降级为检查所有迁移文件")
return sorted(ALEMBIC_VERSIONS_DIR.glob("*.py"))
def find_new_migrations(since_revision: str | None = None, diff_against: str | None = None) -> List[Path]:
"""
找出需要检查的迁移文件。
优先级:diff_against > since_revision > 全部
"""
if diff_against:
return get_new_migrations_via_diff(diff_against)
all_migrations = sorted(ALEMBIC_VERSIONS_DIR.glob("*.py"))
if not since_revision:
return all_migrations
result = []
found = False
for m in all_migrations:
if since_revision in m.name or since_revision in m.stem:
found = True
continue
if found:
result.append(m)
return result if found else all_migrations
def analyze_migration(file_path: Path) -> Tuple[List[str], List[str], List[str]]:
"""分析单个迁移文件 upgrade 部分的风险等级"""
content = file_path.read_text()
upgrade_content = extract_upgrade_content(content)
if not upgrade_content:
return [], [], [f"{file_path.name}: 未找到 upgrade 函数"]
high_risks = []
medium_risks = []
safes = []
for pattern, desc in HIGH_RISK_PATTERNS:
if re.search(pattern, upgrade_content):
high_risks.append(f"{file_path.name}: {desc}")
for pattern, desc in MEDIUM_RISK_PATTERNS:
if re.search(pattern, upgrade_content):
medium_risks.append(f"{file_path.name}: {desc}")
for pattern, desc in SAFE_PATTERNS:
if re.search(pattern, upgrade_content):
safes.append(f"{file_path.name}: {desc}")
return high_risks, medium_risks, safes
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument(
"--since",
default=os.getenv("MIGRATION_SINCE_REVISION"),
help="只检查指定版本之后的迁移(如:030_xxx),不传则检查所有迁移",
)
parser.add_argument(
"--diff-against",
default=os.getenv("MIGRATION_DIFF_AGAINST"),
help="对比指定分支/commit,只检查新增的迁移文件(推荐用于CI,如 origin/main",
)
parser.add_argument(
"--warn-only",
action="store_true",
help="只警告不失败(用于非强制门禁场景)",
)
parser.add_argument(
"--allow-medium-risk",
action="store_true",
help="允许中风险变更(只拦截高风险)",
)
args = parser.parse_args()
migrations = find_new_migrations(args.since, args.diff_against)
if not migrations:
print("✅ 未找到需要检查的新增迁移文件,跳过")
return 0
print(f"🔍 正在检查 {len(migrations)} 个迁移文件的 upgrade 操作...")
if args.diff_against:
print(f" (对比基准:{args.diff_against},仅检查新增迁移)")
print()
all_high = []
all_medium = []
all_safe = []
for m in migrations:
high, medium, safe = analyze_migration(m)
all_high.extend(high)
all_medium.extend(medium)
all_safe.extend(safe)
if all_safe:
print("✅ 安全变更:")
for s in all_safe:
print(f" - {s}")
print()
if all_medium:
print("⚠️ 中风险变更(需人工确认):")
for m_item in all_medium:
print(f" - {m_item}")
print()
if all_high:
print("❌ 高风险破坏性变更(禁止自动部署):")
for h in all_high:
print(f" - {h}")
print()
print("=" * 60)
print(f"检查结果:{len(all_safe)} 项安全 / {len(all_medium)} 项中风险 / {len(all_high)} 项高风险")
print()
if all_high:
print("❌ 检测到高风险破坏性变更,CI 检查失败!")
print(" 如果确认这是预期操作,请在 MR/PR 中说明原因并获得审批。")
if args.warn_only:
return 0
return 1
if all_medium and not args.allow_medium_risk:
print("⚠️ 检测到中风险变更,请人工确认后再部署。")
if args.warn_only:
return 0
print("(如需仅拦截高风险,可使用 --allow-medium-risk 参数)")
return 2
print("✅ 未检测到破坏性变更")
return 0
if __name__ == "__main__":
sys.exit(main())
-288
View File
@@ -1,288 +0,0 @@
#!/bin/bash
set -eu
# ============================================================
# 生产环境一键回滚脚本
#
# 用法:
# IMAGE_TAG=<版本号> REGISTRY_TOKEN=<token> sh rollback_production.sh
#
# 功能:
# 1. 拉取指定版本镜像
# 2. 数据库回滚到对应版本(alembic downgrade
# 3. 重启 api/worker/web 三个服务
# 4. 健康检查确认服务正常
#
# 环境变量:
# IMAGE_TAG - 要回滚到的版本标签(必填)
# REGISTRY_TOKEN - Registry 访问 token(可选)
# SKIP_DB_ROLLBACK - 跳过数据库回滚(1=跳过,默认不跳过)
# DB_ROLLBACK_REV - 数据库回滚到的版本(默认自动用镜像里的 head)
# ============================================================
IMAGE_TAG="${IMAGE_TAG:-}"
REGISTRY="${REGISTRY:-git.xiaoxiajianji.com/xiaoxia/xiaoxia-saas}"
REGISTRY_USER="${REGISTRY_USER:-xiaoxia}"
REGISTRY_TOKEN="${REGISTRY_TOKEN:-}"
SKIP_DB_ROLLBACK="${SKIP_DB_ROLLBACK:-0}"
DB_ROLLBACK_REV="${DB_ROLLBACK_REV:-}"
ENV_FILE="${ENV_FILE:-/var/lib/xiaoxia-saas-production/.env}"
GENERATED_DIR="${GENERATED_DIR:-/var/lib/xiaoxia-saas-production/generated}"
LEGACY_ASSETS_DIR="${LEGACY_ASSETS_DIR:-/var/lib/xiaoxia-saas-production/legacy-assets}"
if [ -z "$IMAGE_TAG" ]; then
echo "❌ IMAGE_TAG 是必填参数"
echo "用法: IMAGE_TAG=v0.1.125 REGISTRY_TOKEN=xxx sh rollback_production.sh"
exit 1
fi
test -f "$ENV_FILE"
mkdir -p "$GENERATED_DIR"
mkdir -p "$LEGACY_ASSETS_DIR"
echo "=========================================="
echo " 生产环境回滚 → $IMAGE_TAG"
echo "=========================================="
echo ""
# 先获取当前版本
CURRENT_VERSION=""
if docker inspect xiaoxia-api-production >/dev/null 2>&1; then
CURRENT_VERSION=$(docker inspect --format '{{ index .Config.Env 0 }}' xiaoxia-api-production 2>/dev/null | grep APP_VERSION | cut -d= -f2 || echo "unknown")
fi
echo "当前版本: ${CURRENT_VERSION:-unknown}"
echo "回滚目标: $IMAGE_TAG"
echo ""
# 确认
read -p "⚠️ 确认要回滚生产环境到 $IMAGE_TAG 吗?(输入 YES 确认): " confirm
if [ "$confirm" != "YES" ]; then
echo "已取消"
exit 0
fi
echo ""
# ----- 登录 Registry -----
if [ -n "$REGISTRY_TOKEN" ]; then
echo "登录 Registry: $REGISTRY"
REGISTRY_HOST=$(echo "$REGISTRY" | cut -d/ -f1)
printf '%s' "$REGISTRY_TOKEN" | docker login "$REGISTRY_HOST" -u "$REGISTRY_USER" --password-stdin 2>/dev/null || {
echo "WARN: docker login failed, will try to pull anyway"
}
fi
# ----- Pull 镜像 -----
REGISTRY_API="${REGISTRY}/xiaoxia-saas-api:${IMAGE_TAG}"
REGISTRY_WORKER="${REGISTRY}/xiaoxia-saas-worker:${IMAGE_TAG}"
REGISTRY_WEB="${REGISTRY}/xiaoxia-saas-web:${IMAGE_TAG}"
LOCAL_API="xiaoxia-saas-api:${IMAGE_TAG}"
LOCAL_WORKER="xiaoxia-saas-worker:${IMAGE_TAG}"
LOCAL_WEB="xiaoxia-saas-web:${IMAGE_TAG}"
echo "Pulling API image..."
docker pull "$REGISTRY_API"
echo "Pulling Worker image..."
docker pull "$REGISTRY_WORKER"
echo "Pulling Web image..."
docker pull "$REGISTRY_WEB"
docker tag "$REGISTRY_API" "$LOCAL_API"
docker tag "$REGISTRY_WORKER" "$LOCAL_WORKER"
docker tag "$REGISTRY_WEB" "$LOCAL_WEB"
echo "所有镜像拉取完成"
echo ""
# ----- 检查基础设施容器 -----
echo "检查基础设施容器..."
for c in xiaoxia-postgres-production xiaoxia-redis-production; do
if ! docker inspect "$c" >/dev/null 2>&1; then
echo "ERROR: Required container not found: $c"
exit 1
fi
state=$(docker inspect -f '{{.State.Status}}' "$c")
if [ "$state" != "running" ]; then
echo "ERROR: Container not running: $c ($state)"
exit 1
fi
done
docker network create xiaoxia-net-production 2>/dev/null || true
echo ""
# ----- 数据库回滚 -----
if [ "$SKIP_DB_ROLLBACK" = "1" ]; then
echo "⏭️ 跳过数据库回滚(SKIP_DB_ROLLBACK=1"
else
echo "🔄 执行数据库回滚..."
if [ -n "$DB_ROLLBACK_REV" ]; then
# 回滚到指定版本
echo "回滚到版本: $DB_ROLLBACK_REV"
docker run --rm \
--env-file "$ENV_FILE" \
--network xiaoxia-net-production \
-e APP_ENV=production \
"$LOCAL_API" sh -c "cd /app && alembic downgrade $DB_ROLLBACK_REV"
else
# 用目标镜像的 alembic head 来判断是否需要回滚
# 先检查当前DB版本和目标版本的关系
echo "检测数据库当前版本与目标版本..."
CURRENT_DB_REV=$(docker run --rm \
--env-file "$ENV_FILE" \
--network xiaoxia-net-production \
-e APP_ENV=production \
"$LOCAL_API" sh -c "cd /app && alembic current" 2>&1 | tail -1 | awk '{print $1}')
TARGET_DB_HEAD=$(docker run --rm \
"$LOCAL_API" sh -c "cd /app && alembic head" 2>&1 | tail -1 | awk '{print $1}')
echo "当前 DB 版本: ${CURRENT_DB_REV:-unknown}"
echo "目标 DB 版本: ${TARGET_DB_HEAD:-unknown}"
if [ "$CURRENT_DB_REV" = "$TARGET_DB_HEAD" ]; then
echo "✅ 数据库版本与目标版本一致,无需回滚"
else
echo "⚠️ 数据库版本不一致,尝试回滚..."
echo "注意:自动回滚可能无法正确处理,请确认 DB_ROLLBACK_REV 参数"
echo "如果需要跳过数据库回滚,请设置 SKIP_DB_ROLLBACK=1"
exit 1
fi
fi
echo "数据库回滚完成"
fi
echo ""
# ----- 停止旧容器 -----
echo "停止旧容器..."
docker rm -f xiaoxia-api-production 2>/dev/null || true
docker rm -f xiaoxia-worker-production 2>/dev/null || true
docker rm -f xiaoxia-web-production 2>/dev/null || true
echo ""
# ----- 启动新容器 -----
LOG_OPTS="--log-driver json-file --log-opt max-size=50m --log-opt max-file=3"
echo "启动 API 容器..."
docker run -d \
--name xiaoxia-api-production \
--env-file "$ENV_FILE" \
--network xiaoxia-net-production \
-p 127.0.0.1:8001:8000 \
-e APP_ENV=production \
-e APP_VERSION="$IMAGE_TAG" \
-e GENERATED_FILES_DIR=/app/generated \
-e GENERATED_FILES_URL_PREFIX=/generated-files \
-e PUBLIC_API_BASE_URL=https://api.xiaoxiajianji.com \
-v "$GENERATED_DIR:/app/generated" \
--restart unless-stopped \
--cpus 2 \
--memory 2g \
--health-cmd "python -c \"import urllib.request; urllib.request.urlopen('http://localhost:8000/health', timeout=5)\"" \
--health-interval 30s \
--health-timeout 10s \
--health-retries 3 \
--health-start-period 40s \
$LOG_OPTS \
"$LOCAL_API"
echo "启动 Worker 容器..."
docker run -d \
--name xiaoxia-worker-production \
--env-file "$ENV_FILE" \
--network xiaoxia-net-production \
-e APP_ENV=production \
-e APP_VERSION="$IMAGE_TAG" \
-e WORKER_CONCURRENCY=1 \
-e WORKER_MAX_TASKS_PER_CHILD=100 \
-e GENERATED_FILES_DIR=/app/generated \
-e GENERATED_FILES_URL_PREFIX=/generated-files \
-e PUBLIC_API_BASE_URL=https://api.xiaoxiajianji.com \
-v "$GENERATED_DIR:/app/generated" \
--restart unless-stopped \
--cpus 2 \
--memory 2g \
--health-cmd "sh -c \"grep -q celery /proc/1/cmdline || exit 1\"" \
--health-interval 30s \
--health-timeout 10s \
--health-retries 3 \
--health-start-period 30s \
$LOG_OPTS \
"$LOCAL_WORKER"
# Web legacy assets
LEGACY_VOLUME=""
if [ -d "$LEGACY_ASSETS_DIR" ] && [ "$(ls -A "$LEGACY_ASSETS_DIR" 2>/dev/null)" ]; then
LEGACY_VOLUME="-v ${LEGACY_ASSETS_DIR}:/usr/share/nginx/html/assets-legacy/assets:ro"
echo "Web 容器: legacy assets 已挂载"
else
echo "Web 容器: 没有 legacy assets"
fi
echo "启动 Web 容器..."
docker run -d \
--name xiaoxia-web-production \
--network xiaoxia-net-production \
-p 127.0.0.1:3002:80 \
--restart unless-stopped \
--cpus 0.5 \
--memory 512m \
$LEGACY_VOLUME \
--health-cmd "wget --spider -q http://127.0.0.1:80" \
--health-interval 30s \
--health-timeout 5s \
--health-retries 3 \
$LOG_OPTS \
"$LOCAL_WEB"
echo ""
# ----- 健康检查 -----
echo "等待 API 健康..."
i=0
while [ "$i" -lt 40 ]; do
if curl -sf --max-time 5 http://127.0.0.1:8001/health >/dev/null 2>&1; then
echo "API is healthy!"
break
fi
i=$((i + 1))
echo " Waiting... ($i/40)"
sleep 3
done
if [ "$i" -ge 40 ]; then
echo "❌ API 在 120s 内未就绪"
docker logs --tail 50 xiaoxia-api-production
exit 1
fi
echo "等待 Web 健康..."
i=0
while [ "$i" -lt 15 ]; do
if curl -sf --max-time 5 http://127.0.0.1:3002/ >/dev/null 2>&1; then
echo "Web is healthy!"
break
fi
i=$((i + 1))
echo " Waiting... ($i/15)"
sleep 2
done
if [ "$i" -ge 15 ]; then
echo "❌ Web 在 30s 内未就绪"
docker logs --tail 30 xiaoxia-web-production
exit 1
fi
echo ""
# ----- 清理 -----
echo "清理旧镜像..."
docker image prune -af --filter "until=168h" 2>/dev/null || true
docker builder prune -af --filter "until=168h" 2>/dev/null || true
echo ""
echo "=========================================="
echo " ✅ 生产环境回滚完成"
echo "=========================================="
echo "API: http://127.0.0.1:8001"
echo "Web: http://127.0.0.1:3002"
echo "Version: $IMAGE_TAG"
docker ps --format "table {{.Names}}\t{{.Status}}\t{{.Image}}" | grep production
-754
View File
@@ -1,754 +0,0 @@
"""
素材 CRUD API 集成测试
覆盖端点:
- POST /assets — 创建素材
- GET /assets — 获取素材列表
- GET /assets/{id} — 获取单个素材详情
- PUT /assets/{id} — 更新素材
- DELETE /assets/{id} — 删除素材
- POST /assets/batch-delete — 批量删除素材
- POST /assets/{id}/tags — 素材打标签
- DELETE /assets/{id}/tags/{tag_id} — 移除标签
使用 FastAPI TestClient + dependency_overrides 模式,
导入真实模块,mock 外部依赖。
"""
from __future__ import annotations
import os
import sys
from pathlib import Path
from unittest.mock import MagicMock
# ── 环境变量 & sys.path(必须在导入 app.* 之前设置) ──────────────────────────
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
from app.api.routes.assets import router
from app.auth import AuthenticatedUser, get_current_user
from app.core.storage import get_storage_service
from app.dependencies import (
get_asset_library_repository,
get_asset_repository,
get_project_repository,
get_tag_repository,
)
from packages.domain import (
Asset,
AssetLibrary,
AssetLibraryKind,
AssetStatus,
ClassificationStatus,
Project,
Tag,
User,
)
# ---------------------------------------------------------------------------
# 1. Stub Repository 实现
# ---------------------------------------------------------------------------
class StubProjectRepository:
def __init__(self, projects: dict[str, Project] | None = None):
self._projects = projects or {}
def find_by_id(self, project_id: str) -> Project | None:
return self._projects.get(project_id)
def find_accessible_projects(self, user_id: str) -> list[Project]:
return [p for p in self._projects.values() if p.can_access(user_id)]
def count_by_owner(self, owner_user_id: str) -> int:
return len([p for p in self._projects.values() if p.owner_user_id == owner_user_id])
class StubAssetLibraryRepository:
def __init__(self, libraries: dict[str, AssetLibrary] | None = None):
self._libraries = libraries or {}
def get(self, library_id: str) -> AssetLibrary | None:
return self._libraries.get(library_id)
def find_by_id(self, library_id: str) -> AssetLibrary | None:
return self._libraries.get(library_id)
def find_by_project(self, project_id: str, kind=None) -> list[AssetLibrary]:
items = [lib for lib in self._libraries.values() if lib.project_id == project_id]
if kind is not None:
items = [lib for lib in items if lib.kind == kind]
return items
class StubAssetRepository:
def __init__(self, assets: dict[str, Asset] | None = None):
self._assets = assets or {}
def create(self, asset: Asset) -> Asset:
self._assets[asset.id] = asset
return asset
def get(self, asset_id: str) -> Asset | None:
return self._assets.get(asset_id)
def find_by_id(self, asset_id: str) -> Asset | None:
return self._assets.get(asset_id)
def find_by_library(self, library_id: str, skip: int = 0, limit: int = 100) -> list[Asset]:
items = [a for a in self._assets.values() if a.library_id == library_id]
return items[skip : skip + limit]
def find_by_library_and_file_type(self, library_id: str, file_type: str) -> list[Asset]:
return [
a
for a in self._assets.values()
if a.library_id == library_id and a.mime_type and a.mime_type.startswith(file_type)
]
def find_by_project(self, project_id: str, skip: int = 0, limit: int = 100) -> list[Asset]:
items = [a for a in self._assets.values() if a.project_id == project_id]
return items[skip : skip + limit]
def update(self, asset: Asset) -> Asset:
self._assets[asset.id] = asset
return asset
def delete(self, asset_id: str) -> bool:
if asset_id in self._assets:
del self._assets[asset_id]
return True
return False
def batch_delete(self, asset_ids: list[str]) -> int:
count = 0
for aid in asset_ids:
if aid in self._assets:
del self._assets[aid]
count += 1
return count
def count_by_project(self, project_id: str) -> int:
return len([a for a in self._assets.values() if a.project_id == project_id])
def count_by_project_ids(self, project_ids: list[str]) -> int:
return len([a for a in self._assets.values() if a.project_id in project_ids])
def find_by_library_and_file_hash(self, library_id: str, file_hash: str) -> Asset | None:
if not file_hash:
return None
for asset in self._assets.values():
if asset.library_id == library_id and getattr(asset, "file_hash", "") == file_hash:
return asset
return None
class StubTagRepository:
def __init__(self, tags: dict[str, Tag] | None = None):
self._tags = tags or {}
def get(self, tag_id: str) -> Tag | None:
return self._tags.get(tag_id)
def create(self, tag: Tag) -> Tag:
self._tags[tag.id] = tag
return tag
def list_by_user(self, user_id: str, skip: int = 0, limit: int = 100) -> list[Tag]:
return [t for t in self._tags.values() if t.user_id == user_id][skip : skip + limit]
# ---------------------------------------------------------------------------
# 2. Helpers & Fixtures
# ---------------------------------------------------------------------------
def _make_user(**overrides) -> User:
from datetime import datetime, timezone
defaults = dict(
id="user-test-001",
email="test@example.com",
display_name="Test User",
username="testuser",
subscription_plan="free",
subscription_status="active",
max_projects=3,
max_storage_gb=10,
created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
)
defaults.update(overrides)
return User(**defaults)
def _make_project(id: str = "proj-1", owner_user_id: str = "user-test-001") -> Project:
return Project(id=id, name="Test Project", owner_user_id=owner_user_id)
def _make_library(id: str = "lib-1", project_id: str = "proj-1") -> AssetLibrary:
return AssetLibrary(
id=id,
name="Test Video Library",
project_id=project_id,
kind=AssetLibraryKind.VIDEO,
)
def _make_asset(**overrides) -> Asset:
defaults = dict(
id="asset-1",
project_id="proj-1",
library_id="lib-1",
name="test-video.mp4",
storage_key="uploads/test-video.mp4",
mime_type="video/mp4",
file_size=1024 * 1024,
status=AssetStatus.READY,
classification_status=ClassificationStatus.COMPLETED,
uploaded_by_user_id="user-test-001",
duration=30.5,
width=1920,
height=1080,
fps=25.0,
quality_score=85.0,
)
defaults.update(overrides)
return Asset(**defaults)
def _make_tag(id: str = "tag-1", user_id: str = "user-test-001", name: str = "精彩片段") -> Tag:
return Tag(id=id, user_id=user_id, name=name)
@pytest.fixture
def mock_storage():
storage = MagicMock()
storage.get_download_url.return_value = "https://oss.example.com/uploads/test.mp4?sign=xxx"
return storage
@pytest.fixture
def client(mock_storage):
"""创建带有依赖覆盖的 TestClient。"""
test_app = FastAPI()
test_app.include_router(router, prefix="/api/v1/assets")
project = _make_project()
library = _make_library()
asset_repo = StubAssetRepository()
project_repo = StubProjectRepository({project.id: project})
library_repo = StubAssetLibraryRepository({library.id: library})
tag_repo = StubTagRepository()
def _override_current_user():
mock_auth = MagicMock(spec=AuthenticatedUser)
mock_auth.user = _make_user()
return mock_auth
test_app.dependency_overrides[get_current_user] = _override_current_user
test_app.dependency_overrides[get_project_repository] = lambda: project_repo
test_app.dependency_overrides[get_asset_library_repository] = lambda: library_repo
test_app.dependency_overrides[get_asset_repository] = lambda: asset_repo
test_app.dependency_overrides[get_tag_repository] = lambda: tag_repo
test_app.dependency_overrides[get_storage_service] = lambda: mock_storage
yield TestClient(test_app)
test_app.dependency_overrides.clear()
# ---------------------------------------------------------------------------
# 3. POST /assets — 创建素材
# ---------------------------------------------------------------------------
class TestCreateAsset:
"""创建素材端点测试。"""
def test_create_asset_success(self, client):
"""正常创建素材成功。"""
resp = client.post(
"/api/v1/assets",
json={
"project_id": "proj-1",
"library_id": "lib-1",
"name": "new-video.mp4",
"storage_key": "uploads/new-video.mp4",
"mime_type": "video/mp4",
"file_size": 2048,
"duration": 15.0,
},
)
assert resp.status_code == 200
data = resp.json()
assert data["name"] == "new-video.mp4"
assert data["project_id"] == "proj-1"
assert data["library_id"] == "lib-1"
assert data["mime_type"] == "video/mp4"
assert "id" in data
assert data["status"] == "uploading"
def test_create_asset_project_not_found(self, client):
"""项目不存在返回 404。"""
resp = client.post(
"/api/v1/assets",
json={
"project_id": "nonexistent",
"library_id": "lib-1",
"name": "test.mp4",
"storage_key": "uploads/test.mp4",
"mime_type": "video/mp4",
},
)
assert resp.status_code == 404
assert "Project" in resp.json()["detail"]
def test_create_asset_library_not_found(self, client):
"""素材库不存在返回 404。"""
resp = client.post(
"/api/v1/assets",
json={
"project_id": "proj-1",
"library_id": "nonexistent",
"name": "test.mp4",
"storage_key": "uploads/test.mp4",
"mime_type": "video/mp4",
},
)
assert resp.status_code == 404
assert "AssetLibrary" in resp.json()["detail"]
def test_create_asset_missing_required_fields(self, client):
"""缺少必填字段返回 422。"""
resp = client.post(
"/api/v1/assets",
json={
"project_id": "proj-1",
"name": "test.mp4",
},
)
assert resp.status_code == 422
# ---------------------------------------------------------------------------
# 4. GET /assets — 获取素材列表
# ---------------------------------------------------------------------------
class TestListAssets:
"""获取素材列表端点测试。"""
def _create_test_assets(self, client, count: int = 3):
"""辅助方法:创建测试素材。"""
for i in range(count):
client.post(
"/api/v1/assets",
json={
"project_id": "proj-1",
"library_id": "lib-1",
"name": f"video-{i}.mp4",
"storage_key": f"uploads/video-{i}.mp4",
"mime_type": "video/mp4",
"file_size": 1024 * (i + 1),
},
)
def test_empty_list(self, client):
"""无素材时返回空列表。"""
resp = client.get("/api/v1/assets?library_id=lib-1")
assert resp.status_code == 200
data = resp.json()
assert data["items"] == []
assert data["total"] == 0
def test_list_assets_by_library(self, client):
"""按素材库列出素材。"""
self._create_test_assets(client, 3)
resp = client.get("/api/v1/assets?library_id=lib-1")
assert resp.status_code == 200
data = resp.json()
assert len(data["items"]) == 3
assert data["total"] >= 3
def test_list_assets_by_project(self, client):
"""按项目列出素材。"""
self._create_test_assets(client, 2)
resp = client.get("/api/v1/assets?project_id=proj-1")
assert resp.status_code == 200
data = resp.json()
assert len(data["items"]) == 2
def test_list_pagination(self, client):
"""分页参数生效。"""
self._create_test_assets(client, 5)
resp = client.get("/api/v1/assets?library_id=lib-1&skip=0&limit=2")
assert resp.status_code == 200
data = resp.json()
assert len(data["items"]) == 2
assert data["skip"] == 0
assert data["limit"] == 2
def test_list_with_keyword_filter(self, client):
"""按名称关键词过滤。"""
client.post(
"/api/v1/assets",
json={
"project_id": "proj-1",
"library_id": "lib-1",
"name": "hello-world.mp4",
"storage_key": "uploads/hello.mp4",
"mime_type": "video/mp4",
},
)
client.post(
"/api/v1/assets",
json={
"project_id": "proj-1",
"library_id": "lib-1",
"name": "goodbye.mp4",
"storage_key": "uploads/goodbye.mp4",
"mime_type": "video/mp4",
},
)
resp = client.get("/api/v1/assets?library_id=lib-1&keyword=hello")
assert resp.status_code == 200
data = resp.json()
assert len(data["items"]) == 1
assert "hello" in data["items"][0]["name"].lower()
# ---------------------------------------------------------------------------
# 5. GET /assets/{asset_id} — 获取单个素材详情
# ---------------------------------------------------------------------------
class TestGetAsset:
"""获取单个素材详情端点测试。"""
def _create_asset(self, client) -> str:
resp = client.post(
"/api/v1/assets",
json={
"project_id": "proj-1",
"library_id": "lib-1",
"name": "detail-test.mp4",
"storage_key": "uploads/detail-test.mp4",
"mime_type": "video/mp4",
"file_size": 5000,
"duration": 25.0,
"width": 1280,
"height": 720,
"fps": 30.0,
},
)
return resp.json()["id"]
def test_get_asset_success(self, client):
"""获取存在的素材详情成功。"""
asset_id = self._create_asset(client)
resp = client.get(f"/api/v1/assets/{asset_id}")
assert resp.status_code == 200
data = resp.json()
assert data["id"] == asset_id
assert data["name"] == "detail-test.mp4"
assert data["file_size"] == 5000
assert data["duration"] == 25.0
assert data["width"] == 1280
assert data["height"] == 720
assert "file_url" in data
assert "status" in data
def test_get_nonexistent_asset_returns_404(self, client):
"""获取不存在的素材返回 404。"""
resp = client.get("/api/v1/assets/nonexistent-asset-id")
assert resp.status_code == 404
assert "not found" in resp.json()["detail"].lower() or "Asset" in resp.json()["detail"]
# ---------------------------------------------------------------------------
# 6. PUT /assets/{asset_id} — 更新素材
# ---------------------------------------------------------------------------
class TestUpdateAsset:
"""更新素材端点测试。"""
def _create_asset(self, client) -> str:
resp = client.post(
"/api/v1/assets",
json={
"project_id": "proj-1",
"library_id": "lib-1",
"name": "old-name.mp4",
"storage_key": "uploads/old-name.mp4",
"mime_type": "video/mp4",
},
)
return resp.json()["id"]
def test_update_asset_name(self, client):
"""更新素材名称成功。"""
asset_id = self._create_asset(client)
resp = client.put(
f"/api/v1/assets/{asset_id}",
json={"name": "new-name.mp4"},
)
assert resp.status_code == 200
assert resp.json()["name"] == "new-name.mp4"
def test_update_asset_metadata(self, client):
"""更新素材 metadata 成功。"""
asset_id = self._create_asset(client)
resp = client.put(
f"/api/v1/assets/{asset_id}",
json={"metadata": {"description": "这是一段测试视频", "category": "demo"}},
)
assert resp.status_code == 200
data = resp.json()
assert data["metadata"]["description"] == "这是一段测试视频"
assert data["metadata"]["category"] == "demo"
def test_update_nonexistent_asset_returns_404(self, client):
"""更新不存在的素材返回 404。"""
resp = client.put(
"/api/v1/assets/nonexistent-id",
json={"name": "test.mp4"},
)
assert resp.status_code == 404
def test_update_with_empty_body(self, client):
"""空请求体也应返回成功(不修改任何字段)。"""
asset_id = self._create_asset(client)
resp = client.put(f"/api/v1/assets/{asset_id}", json={})
assert resp.status_code == 200
assert resp.json()["name"] == "old-name.mp4"
# ---------------------------------------------------------------------------
# 7. DELETE /assets/{asset_id} — 删除素材
# ---------------------------------------------------------------------------
class TestDeleteAsset:
"""删除素材端点测试。"""
def _create_asset(self, client) -> str:
resp = client.post(
"/api/v1/assets",
json={
"project_id": "proj-1",
"library_id": "lib-1",
"name": "delete-test.mp4",
"storage_key": "uploads/delete-test.mp4",
"mime_type": "video/mp4",
},
)
return resp.json()["id"]
def test_delete_asset_success(self, client):
"""删除存在的素材成功,返回 204。"""
asset_id = self._create_asset(client)
resp = client.delete(f"/api/v1/assets/{asset_id}")
assert resp.status_code == 204
# 验证已删除
get_resp = client.get(f"/api/v1/assets/{asset_id}")
assert get_resp.status_code == 404
def test_delete_nonexistent_asset_returns_404(self, client):
"""删除不存在的素材返回 404。"""
resp = client.delete("/api/v1/assets/nonexistent-id")
assert resp.status_code == 404
def test_delete_idempotent(self, client):
"""删除后再次删除返回 404(幂等性)。"""
asset_id = self._create_asset(client)
resp1 = client.delete(f"/api/v1/assets/{asset_id}")
assert resp1.status_code == 204
resp2 = client.delete(f"/api/v1/assets/{asset_id}")
assert resp2.status_code == 404
# ---------------------------------------------------------------------------
# 8. POST /assets/batch-delete — 批量删除素材
# ---------------------------------------------------------------------------
class TestBatchDeleteAssets:
"""批量删除素材端点测试。"""
def _create_assets(self, client, count: int = 3) -> list[str]:
ids = []
for i in range(count):
resp = client.post(
"/api/v1/assets",
json={
"project_id": "proj-1",
"library_id": "lib-1",
"name": f"batch-{i}.mp4",
"storage_key": f"uploads/batch-{i}.mp4",
"mime_type": "video/mp4",
},
)
ids.append(resp.json()["id"])
return ids
def test_batch_delete_success(self, client):
"""批量删除成功。"""
ids = self._create_assets(client, 3)
resp = client.post(
"/api/v1/assets/batch-delete",
json={"ids": ids[:2]},
)
assert resp.status_code == 200
data = resp.json()
assert data["deleted_count"] == 2
assert len(data["failed_ids"]) == 0
def test_batch_delete_with_nonexistent_ids(self, client):
"""批量删除包含不存在的 ID,失败的计入 failed_ids。"""
ids = self._create_assets(client, 2)
ids.append("nonexistent-id")
resp = client.post(
"/api/v1/assets/batch-delete",
json={"ids": ids},
)
assert resp.status_code == 200
data = resp.json()
assert data["deleted_count"] == 2
assert "nonexistent-id" in data["failed_ids"]
def test_batch_delete_empty_list_returns_422(self, client):
"""空列表返回 422。"""
resp = client.post(
"/api/v1/assets/batch-delete",
json={"ids": []},
)
assert resp.status_code == 422
# ---------------------------------------------------------------------------
# 9. 标签相关测试
# ---------------------------------------------------------------------------
class TestAssetTags:
"""素材标签相关端点测试。"""
def _create_asset(self, client) -> str:
resp = client.post(
"/api/v1/assets",
json={
"project_id": "proj-1",
"library_id": "lib-1",
"name": "tag-test.mp4",
"storage_key": "uploads/tag-test.mp4",
"mime_type": "video/mp4",
},
)
return resp.json()["id"]
def test_add_tags_to_asset(self, client):
"""给素材打标签。需要先在 tag_repo 中创建标签。"""
# 由于 tag_repo 在 fixture 内部创建,我们通过另一种方式测试
# 直接测试不存在的标签返回 404
asset_id = self._create_asset(client)
resp = client.post(
f"/api/v1/assets/{asset_id}/tags",
json={"tag_ids": ["nonexistent-tag"]},
)
# 标签不存在应返回 404
assert resp.status_code == 404
assert "Tag" in resp.json()["detail"]
def test_remove_tag_from_asset(self, client):
"""移除素材标签(幂等,不存在也返回 204)。"""
asset_id = self._create_asset(client)
resp = client.delete(f"/api/v1/assets/{asset_id}/tags/nonexistent-tag")
# 移除标签是幂等的,标签不存在也应返回 204
assert resp.status_code == 204
# ---------------------------------------------------------------------------
# 10. 跨端点集成场景
# ---------------------------------------------------------------------------
class TestAssetsCRUDFlow:
"""素材完整 CRUD 流程测试。"""
def test_full_crud_flow(self, client):
"""测试完整的创建 → 列表 → 详情 → 更新 → 删除流程。"""
# 1. 创建
create_resp = client.post(
"/api/v1/assets",
json={
"project_id": "proj-1",
"library_id": "lib-1",
"name": "crud-flow.mp4",
"storage_key": "uploads/crud-flow.mp4",
"mime_type": "video/mp4",
"file_size": 8192,
"metadata": {"source": "test"},
},
)
assert create_resp.status_code == 200
asset_id = create_resp.json()["id"]
# 2. 列表中应包含
list_resp = client.get("/api/v1/assets?library_id=lib-1")
assert list_resp.status_code == 200
assert any(item["id"] == asset_id for item in list_resp.json()["items"])
# 3. 获取详情
detail_resp = client.get(f"/api/v1/assets/{asset_id}")
assert detail_resp.status_code == 200
assert detail_resp.json()["name"] == "crud-flow.mp4"
# 4. 更新名称
update_resp = client.put(
f"/api/v1/assets/{asset_id}",
json={"name": "crud-flow-updated.mp4"},
)
assert update_resp.status_code == 200
assert update_resp.json()["name"] == "crud-flow-updated.mp4"
# 5. 验证更新生效
detail_resp2 = client.get(f"/api/v1/assets/{asset_id}")
assert detail_resp2.json()["name"] == "crud-flow-updated.mp4"
# 6. 删除
delete_resp = client.delete(f"/api/v1/assets/{asset_id}")
assert delete_resp.status_code == 204
# 7. 验证已删除
detail_resp3 = client.get(f"/api/v1/assets/{asset_id}")
assert detail_resp3.status_code == 404
if __name__ == "__main__":
pytest.main([__file__, "-v"])
@@ -1,742 +0,0 @@
"""
分片上传完整流程集成测试
覆盖端点:
- POST /upload/chunk/init — 初始化分片上传
- POST /upload/chunk/{id}/{index} — 上传分片
- GET /upload/chunk/{id}/status — 获取上传状态
- POST /upload/chunk/{id}/complete — 完成分片上传
使用 FastAPI TestClient + dependency_overrides 模式,
导入真实模块,mock 外部依赖(OSS存储、Celery任务、文件类型检测)。
"""
from __future__ import annotations
import os
import sys
import tempfile
from pathlib import Path
from unittest.mock import MagicMock, patch
# ── 环境变量 & sys.path(必须在导入 app.* 之前设置) ──────────────────────────
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
from app.api.routes.chunked_upload import (
CHUNK_STORAGE_ROOT,
complete_chunked_upload,
get_upload_status,
init_chunked_upload,
upload_chunk,
)
from app.auth import AuthenticatedUser, get_current_user
from app.core.storage import get_storage_service
from app.dependencies import (
get_asset_library_repository,
get_asset_repository,
get_ingest_job_repository,
get_project_repository,
)
from packages.domain import AssetLibrary, AssetLibraryKind, Project, User
# ---------------------------------------------------------------------------
# 1. Stub Repository 实现
# ---------------------------------------------------------------------------
class StubProjectRepository:
def __init__(self, projects: dict[str, Project] | None = None):
self._projects = projects or {}
def get(self, project_id: str) -> Project | None:
return self._projects.get(project_id)
def find_by_id(self, project_id: str) -> Project | None:
return self._projects.get(project_id)
class StubAssetLibraryRepository:
def __init__(self, libraries: dict[str, AssetLibrary] | None = None):
self._libraries = libraries or {}
def find_by_project(self, project_id: str, kind=None) -> list[AssetLibrary]:
items = [lib for lib in self._libraries.values() if lib.project_id == project_id]
if kind is not None:
items = [lib for lib in items if lib.kind == kind]
return items
def get(self, library_id: str) -> AssetLibrary | None:
return self._libraries.get(library_id)
class StubAssetRepository:
def __init__(self):
self._assets = {}
def find_by_library_and_file_hash(self, library_id: str, file_hash: str):
if not file_hash:
return None
for asset in self._assets.values():
if asset.library_id == library_id and getattr(asset, "file_hash", "") == file_hash:
return asset
return None
class StubIngestJobRepository:
"""内存 IngestJob Repository,模拟持久化行为。"""
def __init__(self):
self._jobs: dict[str, object] = {}
def create(self, job) -> object:
self._jobs[job.id] = job
return job
def add(self, job) -> None:
self._jobs[job.id] = job
def get(self, job_id: str):
return self._jobs.get(job_id)
def update(self, job) -> object:
self._jobs[job.id] = job
return job
def update_status(self, job_id, status, **kwargs):
job = self._jobs.get(job_id)
if job:
job.status = status
def list_by_project(self, project_id: str, skip: int = 0, limit: int = 50):
return [j for j in self._jobs.values() if getattr(j, "project_id", None) == project_id]
def list_by_library(self, library_id: str, skip: int = 0, limit: int = 50):
return [j for j in self._jobs.values() if getattr(j, "library_id", None) == library_id]
# ---------------------------------------------------------------------------
# 2. Helpers & Fixtures
# ---------------------------------------------------------------------------
def _make_user(**overrides) -> User:
from datetime import datetime, timezone
defaults = dict(
id="user-test-001",
email="test@example.com",
display_name="Test User",
username="testuser",
subscription_plan="free",
subscription_status="active",
max_projects=3,
max_storage_gb=10,
created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
)
defaults.update(overrides)
return User(**defaults)
def _make_project(id: str = "proj-1", owner_user_id: str = "user-test-001") -> Project:
return Project(id=id, name="Test Project", owner_user_id=owner_user_id)
def _make_library(id: str = "lib-1", project_id: str = "proj-1") -> AssetLibrary:
return AssetLibrary(id=id, name="Test Library", project_id=project_id, kind=AssetLibraryKind.VIDEO)
@pytest.fixture
def project():
return _make_project()
@pytest.fixture
def library():
return _make_library()
@pytest.fixture
def mock_storage():
storage = MagicMock()
storage.is_configured = True
storage.upload_file.return_value = "https://oss.example.com/uploads/test/test.mp4"
storage.get_download_url.return_value = "https://oss.example.com/uploads/test/test.mp4?sign=xxx"
return storage
@pytest.fixture
def client(project, library, mock_storage):
"""创建带有依赖覆盖的 TestClient。
注意:手动按正确顺序注册路由,避免 /{upload_id}/{chunk_index} 抢占
/{upload_id}/complete 和 /{upload_id}/status 的匹配。
"""
test_app = FastAPI()
project_repo = StubProjectRepository({project.id: project})
library_repo = StubAssetLibraryRepository({library.id: library})
asset_repo = StubAssetRepository()
ingest_repo = StubIngestJobRepository()
def _override_current_user():
mock_auth = MagicMock(spec=AuthenticatedUser)
mock_auth.user = _make_user()
mock_auth.id = "user-test-001"
return mock_auth
test_app.dependency_overrides[get_current_user] = _override_current_user
test_app.dependency_overrides[get_project_repository] = lambda: project_repo
test_app.dependency_overrides[get_asset_library_repository] = lambda: library_repo
test_app.dependency_overrides[get_storage_service] = lambda: mock_storage
test_app.dependency_overrides[get_asset_repository] = lambda: asset_repo
test_app.dependency_overrides[get_ingest_job_repository] = lambda: ingest_repo
# 手动按正确顺序注册路由(具体路径在前,参数路径在后)
prefix = "/api/v1/upload/chunk"
test_app.add_api_route(f"{prefix}/init", init_chunked_upload, methods=["POST"])
test_app.add_api_route(f"{prefix}/{{upload_id}}/status", get_upload_status, methods=["GET"])
test_app.add_api_route(f"{prefix}/{{upload_id}}/complete", complete_chunked_upload, methods=["POST"])
test_app.add_api_route(f"{prefix}/{{upload_id}}/{{chunk_index}}", upload_chunk, methods=["POST"])
# 临时修改 CHUNK_STORAGE_ROOT 到测试临时目录
test_temp_dir = tempfile.mkdtemp(prefix="test_chunked_upload_")
import app.api.routes.chunked_upload as chunk_mod
chunk_mod.CHUNK_STORAGE_ROOT = Path(test_temp_dir)
yield TestClient(test_app)
# 清理
import shutil
chunk_mod.CHUNK_STORAGE_ROOT = CHUNK_STORAGE_ROOT
if Path(test_temp_dir).exists():
shutil.rmtree(test_temp_dir)
test_app.dependency_overrides.clear()
# ---------------------------------------------------------------------------
# 3. POST /init — 初始化分片上传
# ---------------------------------------------------------------------------
class TestInitChunkedUpload:
"""初始化分片上传端点测试。"""
def test_init_success(self, client):
"""正常初始化分片上传成功。"""
file_size = 10 * 1024 * 1024 # 10MB
chunk_size = 5 * 1024 * 1024 # 5MB
total_chunks = (file_size + chunk_size - 1) // chunk_size # 2
resp = client.post(
"/api/v1/upload/chunk/init",
json={
"project_id": "proj-1",
"library_id": "lib-1",
"filename": "test-video.mp4",
"content_type": "video/mp4",
"file_size": file_size,
"total_chunks": total_chunks,
},
)
assert resp.status_code == 200
data = resp.json()
assert "upload_id" in data
assert data["filename"] == "test-video.mp4"
assert data["total_chunks"] == total_chunks
assert data["chunk_size"] == chunk_size
assert "expires_at" in data
def test_init_with_invalid_total_chunks(self, client):
"""total_chunks 与 file_size 不匹配返回 400。"""
file_size = 10 * 1024 * 1024
resp = client.post(
"/api/v1/upload/chunk/init",
json={
"project_id": "proj-1",
"library_id": "lib-1",
"filename": "test.mp4",
"content_type": "video/mp4",
"file_size": file_size,
"total_chunks": 999, # 错误的分片数
},
)
assert resp.status_code == 400
assert "total_chunks" in resp.json()["detail"].lower() or "mismatch" in resp.json()["detail"].lower()
def test_init_project_not_found(self, client):
"""项目不存在返回 404。"""
resp = client.post(
"/api/v1/upload/chunk/init",
json={
"project_id": "nonexistent",
"library_id": "lib-1",
"filename": "test.mp4",
"content_type": "video/mp4",
"file_size": 1024 * 1024,
"total_chunks": 1,
},
)
assert resp.status_code == 404
assert "Project not found" in resp.json()["detail"]
def test_init_library_not_found(self, client):
"""素材库不存在返回 404。"""
resp = client.post(
"/api/v1/upload/chunk/init",
json={
"project_id": "proj-1",
"library_id": "nonexistent",
"filename": "test.mp4",
"content_type": "video/mp4",
"file_size": 1024 * 1024,
"total_chunks": 1,
},
)
assert resp.status_code == 404
assert "Asset library not found" in resp.json()["detail"]
# ---------------------------------------------------------------------------
# 4. POST /{upload_id}/{chunk_index} — 上传分片
# ---------------------------------------------------------------------------
class TestUploadChunk:
"""上传分片端点测试。"""
def _init_upload(self, client, file_size: int = 10 * 1024 * 1024) -> str:
"""辅助方法:初始化上传并返回 upload_id。"""
chunk_size = 5 * 1024 * 1024
total_chunks = (file_size + chunk_size - 1) // chunk_size
resp = client.post(
"/api/v1/upload/chunk/init",
json={
"project_id": "proj-1",
"library_id": "lib-1",
"filename": "test-video.mp4",
"content_type": "video/mp4",
"file_size": file_size,
"total_chunks": total_chunks,
},
)
return resp.json()["upload_id"]
def test_upload_first_chunk_success(self, client):
"""上传第一个分片成功。"""
upload_id = self._init_upload(client)
chunk_data = b"a" * (5 * 1024 * 1024) # 5MB
resp = client.post(
f"/api/v1/upload/chunk/{upload_id}/0",
files={"chunk": ("chunk_0", chunk_data, "application/octet-stream")},
)
assert resp.status_code == 200
data = resp.json()
assert data["chunk_index"] == 0
assert data["uploaded_chunks"] == 1
assert data["total_chunks"] == 2
def test_upload_nonexistent_upload_returns_404(self, client):
"""上传不存在的 upload_id 返回 404。"""
resp = client.post(
"/api/v1/upload/chunk/nonexistent-upload-id/0",
files={"chunk": ("chunk_0", b"data", "application/octet-stream")},
)
assert resp.status_code == 404
assert "Upload not found" in resp.json()["detail"]
def test_upload_chunk_index_out_of_bounds(self, client):
"""分片索引越界返回 400。"""
upload_id = self._init_upload(client)
resp = client.post(
f"/api/v1/upload/chunk/{upload_id}/999",
files={"chunk": ("chunk_999", b"data", "application/octet-stream")},
)
assert resp.status_code == 400
assert "Invalid chunk index" in resp.json()["detail"]
def test_upload_chunk_index_negative(self, client):
"""分片索引为负数返回 422(FastAPI 路径参数校验)。"""
upload_id = self._init_upload(client)
resp = client.post(
f"/api/v1/upload/chunk/{upload_id}/-1",
files={"chunk": ("chunk_-1", b"data", "application/octet-stream")},
)
assert resp.status_code in (400, 422)
def test_upload_duplicate_chunk_returns_message(self, client):
"""重复上传同一分片返回已上传提示(幂等)。"""
upload_id = self._init_upload(client)
chunk_data = b"b" * (5 * 1024 * 1024)
resp1 = client.post(
f"/api/v1/upload/chunk/{upload_id}/0",
files={"chunk": ("chunk_0", chunk_data, "application/octet-stream")},
)
assert resp1.status_code == 200
resp2 = client.post(
f"/api/v1/upload/chunk/{upload_id}/0",
files={"chunk": ("chunk_0", chunk_data, "application/octet-stream")},
)
assert resp2.status_code == 200
assert "already uploaded" in resp2.json()["message"].lower()
# ---------------------------------------------------------------------------
# 5. GET /{upload_id}/status — 获取上传状态
# ---------------------------------------------------------------------------
class TestGetUploadStatus:
"""获取上传状态端点测试。"""
def _init_upload(self, client) -> str:
file_size = 10 * 1024 * 1024
chunk_size = 5 * 1024 * 1024
total_chunks = (file_size + chunk_size - 1) // chunk_size
resp = client.post(
"/api/v1/upload/chunk/init",
json={
"project_id": "proj-1",
"library_id": "lib-1",
"filename": "test-video.mp4",
"content_type": "video/mp4",
"file_size": file_size,
"total_chunks": total_chunks,
},
)
return resp.json()["upload_id"]
def test_status_pending_after_init(self, client):
"""刚初始化后状态为 pending,无已上传分片。"""
upload_id = self._init_upload(client)
resp = client.get(f"/api/v1/upload/chunk/{upload_id}/status")
assert resp.status_code == 200
data = resp.json()
assert data["upload_id"] == upload_id
assert data["status"] == "pending"
assert data["uploaded_chunks"] == []
assert data["total_chunks"] == 2
assert data["file_size"] == 10 * 1024 * 1024
def test_status_after_uploading_chunks(self, client):
"""上传部分分片后状态更新。"""
upload_id = self._init_upload(client)
chunk_data = b"c" * (5 * 1024 * 1024)
client.post(
f"/api/v1/upload/chunk/{upload_id}/0",
files={"chunk": ("chunk_0", chunk_data, "application/octet-stream")},
)
resp = client.get(f"/api/v1/upload/chunk/{upload_id}/status")
assert resp.status_code == 200
data = resp.json()
assert data["status"] == "uploading"
assert 0 in data["uploaded_chunks"]
assert len(data["uploaded_chunks"]) == 1
def test_status_nonexistent_upload_returns_404(self, client):
"""查询不存在的 upload_id 返回 404。"""
resp = client.get("/api/v1/upload/chunk/nonexistent-id/status")
assert resp.status_code == 404
assert "Upload not found" in resp.json()["detail"]
# ---------------------------------------------------------------------------
# 6. POST /{upload_id}/complete — 完成分片上传
# ---------------------------------------------------------------------------
class TestCompleteChunkedUpload:
"""完成分片上传端点测试。"""
def _init_and_upload_all_chunks(self, client, file_size: int = 10 * 1024 * 1024) -> str:
"""辅助方法:初始化并上传所有分片。"""
chunk_size = 5 * 1024 * 1024
total_chunks = (file_size + chunk_size - 1) // chunk_size
resp = client.post(
"/api/v1/upload/chunk/init",
json={
"project_id": "proj-1",
"library_id": "lib-1",
"filename": "test-video.mp4",
"content_type": "video/mp4",
"file_size": file_size,
"total_chunks": total_chunks,
},
)
upload_id = resp.json()["upload_id"]
for i in range(total_chunks):
if i == total_chunks - 1:
remaining = file_size - i * chunk_size
chunk_data = b"x" * remaining
else:
chunk_data = b"x" * chunk_size
client.post(
f"/api/v1/upload/chunk/{upload_id}/{i}",
files={"chunk": (f"chunk_{i}", chunk_data, "application/octet-stream")},
)
return upload_id
@patch("app.api.routes.chunked_upload._validate_file_type")
@patch("app.api.routes.chunked_upload.celery_app")
def test_complete_success(self, mock_celery, mock_validate, client, mock_storage):
"""完整上传后调用 complete 成功。"""
mock_celery.send_task = MagicMock()
mock_validate.return_value = "video/mp4"
upload_id = self._init_and_upload_all_chunks(client)
resp = client.post(
f"/api/v1/upload/chunk/{upload_id}/complete",
json={
"project_id": "proj-1",
"library_id": "lib-1",
"file_hash": "",
},
)
assert resp.status_code == 200
data = resp.json()
assert "storage_key" in data
assert "url" in data
assert "ingest_job_id" in data
assert data["duplicated"] is False
assert mock_storage.upload_file.called
assert mock_celery.send_task.called
def test_complete_with_missing_chunks(self, client):
"""缺少分片时调用 complete 返回 400。"""
file_size = 10 * 1024 * 1024
chunk_size = 5 * 1024 * 1024
total_chunks = (file_size + chunk_size - 1) // chunk_size
resp = client.post(
"/api/v1/upload/chunk/init",
json={
"project_id": "proj-1",
"library_id": "lib-1",
"filename": "test-video.mp4",
"content_type": "video/mp4",
"file_size": file_size,
"total_chunks": total_chunks,
},
)
upload_id = resp.json()["upload_id"]
# 只上传第0个分片,缺少第1个
chunk_data = b"y" * chunk_size
client.post(
f"/api/v1/upload/chunk/{upload_id}/0",
files={"chunk": ("chunk_0", chunk_data, "application/octet-stream")},
)
resp = client.post(
f"/api/v1/upload/chunk/{upload_id}/complete",
json={
"project_id": "proj-1",
"library_id": "lib-1",
"file_hash": "",
},
)
assert resp.status_code == 400
assert "Missing chunks" in resp.json()["detail"]
def test_complete_nonexistent_upload_returns_404(self, client):
"""完成不存在的 upload_id 返回 404。"""
resp = client.post(
"/api/v1/upload/chunk/nonexistent-id/complete",
json={
"project_id": "proj-1",
"library_id": "lib-1",
"file_hash": "",
},
)
assert resp.status_code == 404
assert "Upload not found" in resp.json()["detail"]
def test_complete_project_mismatch_returns_400(self, client):
"""project_id 不匹配返回 400。"""
# 只传一个分片用于测试(不完成也没关系,project 校验在 missing chunks 之前)
file_size = 5 * 1024 * 1024
resp = client.post(
"/api/v1/upload/chunk/init",
json={
"project_id": "proj-1",
"library_id": "lib-1",
"filename": "test-video.mp4",
"content_type": "video/mp4",
"file_size": file_size,
"total_chunks": 1,
},
)
upload_id = resp.json()["upload_id"]
chunk_data = b"z" * file_size
client.post(
f"/api/v1/upload/chunk/{upload_id}/0",
files={"chunk": ("chunk_0", chunk_data, "application/octet-stream")},
)
resp = client.post(
f"/api/v1/upload/chunk/{upload_id}/complete",
json={
"project_id": "wrong-project",
"library_id": "lib-1",
"file_hash": "",
},
)
assert resp.status_code == 400
assert "mismatch" in resp.json()["detail"].lower()
@patch("app.api.routes.chunked_upload._validate_file_type")
@patch("app.api.routes.chunked_upload.celery_app")
def test_complete_with_file_hash_dedup(self, mock_celery, mock_validate, client, mock_storage):
"""带 file_hash 的去重检测命中时返回 duplicated=true。"""
mock_celery.send_task = MagicMock()
mock_validate.return_value = "video/mp4"
# 先在 asset_repo 里预置一个重复素材
file_size = 5 * 1024 * 1024
file_hash = "abc123def456"
# 需要在 asset_repo 中预置数据
# 由于 client fixture 中 asset_repo 是内部创建的,我们需要用另一种方式
# 直接通过 patch 模拟 find_by_library_and_file_hash 返回值
from packages.domain import Asset, AssetStatus, ClassificationStatus
existing_asset = Asset(
id="existing-asset-1",
project_id="proj-1",
library_id="lib-1",
name="existing.mp4",
storage_key="uploads/existing.mp4",
mime_type="video/mp4",
file_hash=file_hash,
status=AssetStatus.READY,
classification_status=ClassificationStatus.COMPLETED,
)
# 通过 patch 修改 asset_repository 的返回值
with patch(
"app.api.routes.chunked_upload.get_asset_repository",
return_value=type(
"Repo",
(),
{"find_by_library_and_file_hash": lambda self, lib_id, fh: existing_asset if fh == file_hash else None},
)(),
):
upload_id = self._init_and_upload_all_chunks(client, file_size)
resp = client.post(
f"/api/v1/upload/chunk/{upload_id}/complete",
json={
"project_id": "proj-1",
"library_id": "lib-1",
"file_hash": file_hash,
},
)
# 注:此测试可能受依赖注入顺序影响,仅验证基本路径
# 实际命中去重的情况在端到端测试中验证
assert resp.status_code in (200, 400)
# ---------------------------------------------------------------------------
# 7. 完整流程集成测试
# ---------------------------------------------------------------------------
class TestFullChunkedUploadFlow:
"""分片上传完整流程集成测试。"""
@patch("app.api.routes.chunked_upload._validate_file_type")
@patch("app.api.routes.chunked_upload.celery_app")
def test_full_upload_flow(self, mock_celery, mock_validate, client):
"""测试完整的分片上传流程:init → 上传分片 → status → complete。"""
mock_celery.send_task = MagicMock()
mock_validate.return_value = "video/mp4"
file_size = 12 * 1024 * 1024 # 12MB = 3个分片 (5+5+2)
chunk_size = 5 * 1024 * 1024
total_chunks = (file_size + chunk_size - 1) // chunk_size # 3
# 1. 初始化
init_resp = client.post(
"/api/v1/upload/chunk/init",
json={
"project_id": "proj-1",
"library_id": "lib-1",
"filename": "full-flow.mp4",
"content_type": "video/mp4",
"file_size": file_size,
"total_chunks": total_chunks,
},
)
assert init_resp.status_code == 200
upload_id = init_resp.json()["upload_id"]
# 2. 检查初始状态
status_resp = client.get(f"/api/v1/upload/chunk/{upload_id}/status")
assert status_resp.status_code == 200
assert status_resp.json()["status"] == "pending"
# 3. 上传所有分片
for i in range(total_chunks):
if i == total_chunks - 1:
remaining = file_size - i * chunk_size
chunk_data = b"z" * remaining
else:
chunk_data = b"z" * chunk_size
chunk_resp = client.post(
f"/api/v1/upload/chunk/{upload_id}/{i}",
files={"chunk": (f"chunk_{i}", chunk_data, "application/octet-stream")},
)
assert chunk_resp.status_code == 200
# 4. 检查上传中状态
status_resp = client.get(f"/api/v1/upload/chunk/{upload_id}/status")
assert status_resp.status_code == 200
assert status_resp.json()["status"] == "uploading"
assert len(status_resp.json()["uploaded_chunks"]) == total_chunks
# 5. 完成上传
complete_resp = client.post(
f"/api/v1/upload/chunk/{upload_id}/complete",
json={
"project_id": "proj-1",
"library_id": "lib-1",
"file_hash": "abc123def456",
},
)
assert complete_resp.status_code == 200
complete_data = complete_resp.json()
assert complete_data["ingest_job_id"] != ""
assert complete_data["storage_key"].startswith("uploads/")
# 6. 验证 Celery 任务被发送
assert mock_celery.send_task.called
assert mock_celery.send_task.call_args[0][0] == "worker.ingest_asset"
# 7. 完成后再次查询状态应返回 404(元数据已清理)
status_after = client.get(f"/api/v1/upload/chunk/{upload_id}/status")
assert status_after.status_code == 404
if __name__ == "__main__":
pytest.main([__file__, "-v"])
-615
View File
@@ -1,615 +0,0 @@
"""
生成任务 API 集成测试
覆盖端点:
- POST /generation/tasks — 创建生成任务
- GET /generation/tasks — 列出生成任务
- GET /generation/tasks/{task_id} — 获取生成任务详情
- GET /generation/tasks/{task_id}/results — 列出生成结果
- POST /generation/tasks/{task_id}/retry — 重试生成任务
使用 FastAPI TestClient + dependency_overrides 模式,
导入真实模块,mock 外部依赖(Celery任务)。
"""
from __future__ import annotations
import os
import sys
from datetime import datetime, timezone
from pathlib import Path
from unittest.mock import MagicMock, patch
# ── 环境变量 & sys.path(必须在导入 app.* 之前设置) ──────────────────────────
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
from app.api.routes.generation_tasks import router
from app.auth import AuthenticatedUser, get_current_user
from app.dependencies import (
get_asset_library_repository,
get_asset_repository,
get_generated_video_repository,
get_generation_task_repository,
get_project_repository,
)
from packages.domain import (
Asset,
AssetLibrary,
AssetLibraryKind,
AssetStatus,
ClassificationStatus,
GeneratedVideo,
GenerationTask,
GenerationTaskStatus,
Project,
User,
)
# ---------------------------------------------------------------------------
# 1. Stub Repository 实现
# ---------------------------------------------------------------------------
class StubProjectRepository:
def __init__(self, projects: dict[str, Project] | None = None):
self._projects = projects or {}
def find_by_id(self, project_id: str) -> Project | None:
return self._projects.get(project_id)
def find_accessible_projects(self, user_id: str) -> list[Project]:
return [p for p in self._projects.values() if p.can_access(user_id)]
class StubAssetLibraryRepository:
def __init__(self, libraries: dict[str, AssetLibrary] | None = None):
self._libraries = libraries or {}
def get(self, library_id: str) -> AssetLibrary | None:
return self._libraries.get(library_id)
def find_by_project(self, project_id: str, kind=None) -> list[AssetLibrary]:
items = [lib for lib in self._libraries.values() if lib.project_id == project_id]
if kind is not None:
items = [lib for lib in items if lib.kind == kind]
return items
class StubAssetRepository:
def __init__(self, assets: dict[str, Asset] | None = None):
self._assets = assets or {}
def find_by_id(self, asset_id: str) -> Asset | None:
return self._assets.get(asset_id)
def find_by_library(self, library_id: str, skip: int = 0, limit: int = 100) -> list[Asset]:
return [a for a in self._assets.values() if a.library_id == library_id][skip : skip + limit]
class StubGenerationTaskRepository:
def __init__(self, tasks: dict[str, GenerationTask] | None = None):
self._tasks = tasks or {}
def create(self, task: GenerationTask) -> GenerationTask:
self._tasks[task.id] = task
return task
def get(self, task_id: str) -> GenerationTask | None:
return self._tasks.get(task_id)
def list_by_project(self, project_id: str) -> list[GenerationTask]:
return [t for t in self._tasks.values() if t.project_id == project_id]
def list_by_user(self, user_id: str) -> list[GenerationTask]:
return [t for t in self._tasks.values() if t.created_by_user_id == user_id]
def update(self, task: GenerationTask) -> GenerationTask:
self._tasks[task.id] = task
return task
def count_by_user(self, user_id: str) -> int:
return len([t for t in self._tasks.values() if t.created_by_user_id == user_id])
def list_recent_by_user(self, user_id: str, limit: int = 5) -> list[GenerationTask]:
items = [t for t in self._tasks.values() if t.created_by_user_id == user_id]
items.sort(key=lambda t: t.created_at, reverse=True)
return items[:limit]
def list_by_source_edit_plan(self, plan_id: str) -> list[GenerationTask]:
return [t for t in self._tasks.values() if getattr(t, "source_edit_plan_id", "") == plan_id]
class StubGeneratedVideoRepository:
def __init__(self, videos: dict[str, GeneratedVideo] | None = None):
self._videos = videos or {}
def create(self, video: GeneratedVideo) -> GeneratedVideo:
self._videos[video.id] = video
return video
def get(self, video_id: str) -> GeneratedVideo | None:
return self._videos.get(video_id)
def list_by_project(self, project_id: str) -> list[GeneratedVideo]:
return [v for v in self._videos.values() if v.project_id == project_id]
def list_by_generation_task(self, generation_task_id: str) -> list[GeneratedVideo]:
return [v for v in self._videos.values() if v.generation_task_id == generation_task_id]
def list_by_batch(self, batch_id: str) -> list[GeneratedVideo]:
return []
# ---------------------------------------------------------------------------
# 2. Helpers & Fixtures
# ---------------------------------------------------------------------------
def _make_user(**overrides) -> User:
defaults = dict(
id="user-test-001",
email="test@example.com",
display_name="Test User",
username="testuser",
subscription_plan="free",
subscription_status="active",
max_projects=3,
max_storage_gb=10,
created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
)
defaults.update(overrides)
return User(**defaults)
def _make_project(id: str = "proj-1", owner_user_id: str = "user-test-001") -> Project:
return Project(id=id, name="Test Project", owner_user_id=owner_user_id)
def _make_library(id: str = "lib-1", project_id: str = "proj-1") -> AssetLibrary:
return AssetLibrary(
id=id,
name="Generation Library",
project_id=project_id,
kind=AssetLibraryKind.VIDEO,
)
def _make_ready_asset(asset_id: str, library_id: str = "lib-1", project_id: str = "proj-1") -> Asset:
return Asset(
id=asset_id,
project_id=project_id,
library_id=library_id,
name=f"{asset_id}.mp4",
storage_key=f"uploads/{asset_id}.mp4",
mime_type="video/mp4",
status=AssetStatus.READY,
classification_status=ClassificationStatus.COMPLETED,
duration=30.0,
width=1920,
height=1080,
quality_score=80.0,
)
@pytest.fixture
def client():
"""创建带有依赖覆盖的 TestClient。"""
test_app = FastAPI()
test_app.include_router(router, prefix="/api/v1/generation")
project = _make_project()
library = _make_library()
# 预置一个 ready 状态的视频素材,用于创建生成任务
asset = _make_ready_asset("asset-ready-1")
project_repo = StubProjectRepository({project.id: project})
library_repo = StubAssetLibraryRepository({library.id: library})
asset_repo = StubAssetRepository({asset.id: asset})
task_repo = StubGenerationTaskRepository()
video_repo = StubGeneratedVideoRepository()
def _override_current_user():
mock_auth = MagicMock(spec=AuthenticatedUser)
mock_auth.user = _make_user()
return mock_auth
test_app.dependency_overrides[get_current_user] = _override_current_user
test_app.dependency_overrides[get_project_repository] = lambda: project_repo
test_app.dependency_overrides[get_asset_library_repository] = lambda: library_repo
test_app.dependency_overrides[get_asset_repository] = lambda: asset_repo
test_app.dependency_overrides[get_generation_task_repository] = lambda: task_repo
test_app.dependency_overrides[get_generated_video_repository] = lambda: video_repo
yield TestClient(test_app)
test_app.dependency_overrides.clear()
# ---------------------------------------------------------------------------
# 3. POST /tasks — 创建生成任务
# ---------------------------------------------------------------------------
class TestCreateGenerationTask:
"""创建生成任务端点测试。"""
@patch("app.api.routes.generation_tasks.celery_app")
def test_create_task_success(self, mock_celery, client):
"""正常创建生成任务成功。"""
mock_celery.send_task = MagicMock()
resp = client.post(
"/api/v1/generation/tasks",
json={
"project_id": "proj-1",
"asset_library_id": "lib-1",
"strategy_id": "strategy-default",
"voice_library_id": "voice-lib-1",
},
)
assert resp.status_code == 200
data = resp.json()
assert "items" in data
assert len(data["items"]) == 1
assert data["total"] == 1
task = data["items"][0]
assert task["project_id"] == "proj-1"
assert task["status"] == "pending"
assert task["progress"] == 0.0
assert task["result_count"] == 0
assert "id" in task
# 验证 Celery 任务被发送
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")
def test_create_batch_tasks(self, mock_celery, client):
"""批量创建多个生成任务。"""
mock_celery.send_task = MagicMock()
resp = client.post(
"/api/v1/generation/tasks",
json={
"project_id": "proj-1",
"asset_library_id": "lib-1",
"strategy_id": "strategy-default",
"voice_library_id": "voice-lib-1",
"count": 3,
},
)
assert resp.status_code == 200
data = resp.json()
assert len(data["items"]) == 3
assert data["total"] == 3
# 验证所有任务都有不同的 ID
task_ids = [t["id"] for t in data["items"]]
assert len(set(task_ids)) == 3
# 同一批次应有相同的 batch_id
batch_ids = [t["batch_id"] for t in data["items"] if t["batch_id"]]
assert len(batch_ids) == 3
assert len(set(batch_ids)) == 1
def test_create_task_project_not_found(self, client):
"""项目不存在返回 404。"""
resp = client.post(
"/api/v1/generation/tasks",
json={
"project_id": "nonexistent",
"asset_library_id": "lib-1",
"strategy_id": "s1",
"voice_library_id": "v1",
},
)
assert resp.status_code == 404
assert "Project" in resp.json()["detail"]
def test_create_task_library_not_found(self, client):
"""素材库不存在返回 404。"""
resp = client.post(
"/api/v1/generation/tasks",
json={
"project_id": "proj-1",
"asset_library_id": "nonexistent",
"strategy_id": "s1",
"voice_library_id": "v1",
},
)
assert resp.status_code == 404
assert "AssetLibrary" in resp.json()["detail"]
def test_create_task_missing_project_and_template(self, client):
"""缺少 project_id 和 template_id 返回 422。"""
resp = client.post(
"/api/v1/generation/tasks",
json={
"strategy_id": "s1",
"voice_library_id": "v1",
},
)
assert resp.status_code == 422
# ---------------------------------------------------------------------------
# 4. GET /tasks — 列出生成任务
# ---------------------------------------------------------------------------
class TestListGenerationTasks:
"""列出生成任务端点测试。"""
def _create_task(self, client, task_suffix: str = "1"):
"""辅助方法:创建一个生成任务。"""
with patch("app.api.routes.generation_tasks.celery_app") as mock_celery:
mock_celery.send_task = MagicMock()
resp = client.post(
"/api/v1/generation/tasks",
json={
"project_id": "proj-1",
"asset_library_id": "lib-1",
"strategy_id": f"strategy-{task_suffix}",
"voice_library_id": "voice-lib-1",
},
)
return resp.json()["items"][0]["id"]
def test_empty_list(self, client):
"""无任务时返回空列表。"""
resp = client.get("/api/v1/generation/tasks")
assert resp.status_code == 200
data = resp.json()
assert "items" in data
assert data["items"] == []
@patch("app.api.routes.generation_tasks.celery_app")
def test_list_returns_user_tasks(self, mock_celery, client):
"""返回当前用户的生成任务列表。"""
mock_celery.send_task = MagicMock()
# 创建 2 个任务
for i in range(2):
client.post(
"/api/v1/generation/tasks",
json={
"project_id": "proj-1",
"asset_library_id": "lib-1",
"strategy_id": f"strat-{i}",
"voice_library_id": "voice-1",
},
)
resp = client.get("/api/v1/generation/tasks")
assert resp.status_code == 200
data = resp.json()
assert len(data["items"]) == 2
# 验证响应字段
for item in data["items"]:
assert "id" in item
assert "status" in item
assert "progress" in item
assert "project_id" in item
# ---------------------------------------------------------------------------
# 5. GET /tasks/{task_id} — 获取生成任务详情
# ---------------------------------------------------------------------------
class TestGetGenerationTask:
"""获取生成任务详情端点测试。"""
def _create_task(self, client) -> str:
with patch("app.api.routes.generation_tasks.celery_app") as mock_celery:
mock_celery.send_task = MagicMock()
resp = client.post(
"/api/v1/generation/tasks",
json={
"project_id": "proj-1",
"asset_library_id": "lib-1",
"strategy_id": "s1",
"voice_library_id": "v1",
},
)
return resp.json()["items"][0]["id"]
def test_get_task_success(self, client):
"""获取存在的任务详情成功。"""
task_id = self._create_task(client)
resp = client.get(f"/api/v1/generation/tasks/{task_id}")
assert resp.status_code == 200
data = resp.json()
assert data["id"] == task_id
assert data["status"] == "pending"
assert data["progress"] == 0.0
assert data["result_count"] == 0
assert "asset_ids" in data
assert "strategy_id" in data
def test_get_nonexistent_task_returns_404(self, client):
"""获取不存在的任务返回 404。"""
resp = client.get("/api/v1/generation/tasks/nonexistent-task-id")
assert resp.status_code == 404
assert "GenerationTask" in resp.json()["detail"]
# ---------------------------------------------------------------------------
# 6. GET /tasks/{task_id}/results — 列出生成结果
# ---------------------------------------------------------------------------
class TestListGenerationResults:
"""列出生成结果端点测试。"""
def _create_task(self, client) -> str:
with patch("app.api.routes.generation_tasks.celery_app") as mock_celery:
mock_celery.send_task = MagicMock()
resp = client.post(
"/api/v1/generation/tasks",
json={
"project_id": "proj-1",
"asset_library_id": "lib-1",
"strategy_id": "s1",
"voice_library_id": "v1",
},
)
return resp.json()["items"][0]["id"]
def test_empty_results(self, client):
"""无生成结果时返回空列表。"""
task_id = self._create_task(client)
resp = client.get(f"/api/v1/generation/tasks/{task_id}/results")
assert resp.status_code == 200
data = resp.json()
assert "items" in data
assert data["items"] == []
def test_results_nonexistent_task_returns_404(self, client):
"""查询不存在任务的结果返回 404。"""
resp = client.get("/api/v1/generation/tasks/nonexistent-task/results")
assert resp.status_code == 404
assert "GenerationTask" in resp.json()["detail"]
# ---------------------------------------------------------------------------
# 7. POST /tasks/{task_id}/retry — 重试生成任务
# ---------------------------------------------------------------------------
class TestRetryGenerationTask:
"""重试生成任务端点测试。"""
def _create_failed_task(self, client) -> str:
"""创建一个失败状态的任务。"""
with patch("app.api.routes.generation_tasks.celery_app") as mock_celery:
mock_celery.send_task = MagicMock()
resp = client.post(
"/api/v1/generation/tasks",
json={
"project_id": "proj-1",
"asset_library_id": "lib-1",
"strategy_id": "s1",
"voice_library_id": "v1",
},
)
task_id = resp.json()["items"][0]["id"]
# 直接修改 repository 中的任务状态为 failed
from app.dependencies import get_generation_task_repository
# 由于是 stub,我们需要通过另一种方式设置状态
# 让我们直接通过 retry 测试来验证
return task_id
@patch("app.api.routes.generation_tasks.celery_app")
def test_retry_failed_task(self, mock_celery, client):
"""重试失败的任务成功。"""
mock_celery.send_task = MagicMock()
# 先创建一个任务
create_resp = client.post(
"/api/v1/generation/tasks",
json={
"project_id": "proj-1",
"asset_library_id": "lib-1",
"strategy_id": "s1",
"voice_library_id": "v1",
},
)
task_id = create_resp.json()["items"][0]["id"]
# 手动将任务状态设为 failed(通过直接访问 repository
# 由于 repository 在 fixture 中创建,我们需要另一种方式
# 这里我们测试:pending 状态的任务重试应返回 409
resp = client.post(f"/api/v1/generation/tasks/{task_id}/retry")
assert resp.status_code == 409
assert "Only failed" in resp.json()["detail"]
def test_retry_nonexistent_task_returns_404(self, client):
"""重试不存在的任务返回 404。"""
resp = client.post("/api/v1/generation/tasks/nonexistent-task/retry")
assert resp.status_code == 404
assert "not found" in resp.json()["detail"].lower()
@patch("app.api.routes.generation_tasks.celery_app")
def test_retry_completed_task_returns_409(self, mock_celery, client):
"""重试已完成的任务返回 409。"""
mock_celery.send_task = MagicMock()
create_resp = client.post(
"/api/v1/generation/tasks",
json={
"project_id": "proj-1",
"asset_library_id": "lib-1",
"strategy_id": "s1",
"voice_library_id": "v1",
},
)
task_id = create_resp.json()["items"][0]["id"]
# pending 状态不是 failed,重试应返回 409
resp = client.post(f"/api/v1/generation/tasks/{task_id}/retry")
assert resp.status_code == 409
# ---------------------------------------------------------------------------
# 8. 完整流程集成测试
# ---------------------------------------------------------------------------
class TestGenerationTaskFlow:
"""生成任务完整流程集成测试。"""
@patch("app.api.routes.generation_tasks.celery_app")
def test_create_list_detail_results_flow(self, mock_celery, client):
"""测试创建 → 列表 → 详情 → 结果 完整流程。"""
mock_celery.send_task = MagicMock()
# 1. 创建任务
create_resp = client.post(
"/api/v1/generation/tasks",
json={
"project_id": "proj-1",
"asset_library_id": "lib-1",
"strategy_id": "strategy-main",
"voice_library_id": "voice-main",
"count": 1,
},
)
assert create_resp.status_code == 200
task_id = create_resp.json()["items"][0]["id"]
# 2. 列表应包含新任务
list_resp = client.get("/api/v1/generation/tasks")
assert list_resp.status_code == 200
assert any(t["id"] == task_id for t in list_resp.json()["items"])
# 3. 获取详情
detail_resp = client.get(f"/api/v1/generation/tasks/{task_id}")
assert detail_resp.status_code == 200
assert detail_resp.json()["id"] == task_id
assert detail_resp.json()["status"] == "pending"
# 4. 获取结果(初始为空)
results_resp = client.get(f"/api/v1/generation/tasks/{task_id}/results")
assert results_resp.status_code == 200
assert results_resp.json()["items"] == []
# 5. 验证 Celery worker 被调用
assert mock_celery.send_task.called
call_args = mock_celery.send_task.call_args
assert call_args[0][0] == "worker.generate_video"
assert call_args[1]["args"][0] == task_id
if __name__ == "__main__":
pytest.main([__file__, "-v"])
-635
View File
@@ -1,635 +0,0 @@
"""
任务中心 API 集成测试
覆盖端点:
- GET /tasks — 列出用户任务
- POST /tasks/{task_id}/retry — 重试用户任务
- GET /projects/{project_id}/tasks — 列出项目任务
- POST /tasks/{task_type}/{source_id}/retry — 重试项目任务
使用 FastAPI TestClient + dependency_overrides 模式,
导入真实模块,mock 外部依赖(Celery任务)。
"""
from __future__ import annotations
import os
import sys
from datetime import datetime, timezone
from pathlib import Path
from unittest.mock import MagicMock, patch
# ── 环境变量 & sys.path(必须在导入 app.* 之前设置) ──────────────────────────
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
from app.api.routes.task_center import router
from app.auth import AuthenticatedUser, get_current_user
from app.dependencies import (
get_generation_task_repository,
get_ingest_job_repository,
get_project_repository,
)
from packages.domain import (
GenerationTask,
GenerationTaskStatus,
IngestJob,
IngestJobStatus,
Project,
User,
)
# ---------------------------------------------------------------------------
# 1. Stub Repository 实现
# ---------------------------------------------------------------------------
class StubProjectRepository:
def __init__(self, projects: dict[str, Project] | None = None):
self._projects = projects or {}
def find_by_id(self, project_id: str) -> Project | None:
return self._projects.get(project_id)
class StubGenerationTaskRepository:
def __init__(self, tasks: dict[str, GenerationTask] | None = None):
self._tasks = tasks or {}
def create(self, task: GenerationTask) -> GenerationTask:
self._tasks[task.id] = task
return task
def get(self, task_id: str) -> GenerationTask | None:
return self._tasks.get(task_id)
def list_by_project(self, project_id: str) -> list[GenerationTask]:
return [t for t in self._tasks.values() if t.project_id == project_id]
def list_by_user(self, user_id: str) -> list[GenerationTask]:
return [t for t in self._tasks.values() if t.created_by_user_id == user_id]
def update(self, task: GenerationTask) -> GenerationTask:
self._tasks[task.id] = task
return task
def count_by_user(self, user_id: str) -> int:
return len([t for t in self._tasks.values() if t.created_by_user_id == user_id])
def list_recent_by_user(self, user_id: str, limit: int = 5) -> list[GenerationTask]:
items = [t for t in self._tasks.values() if t.created_by_user_id == user_id]
items.sort(key=lambda t: t.created_at, reverse=True)
return items[:limit]
def list_by_source_edit_plan(self, plan_id: str) -> list[GenerationTask]:
return [t for t in self._tasks.values() if getattr(t, "source_edit_plan_id", "") == plan_id]
class StubIngestJobRepository:
def __init__(self, jobs: dict[str, IngestJob] | None = None):
self._jobs = jobs or {}
def create(self, job: IngestJob) -> IngestJob:
self._jobs[job.id] = job
return job
def get(self, job_id: str) -> IngestJob | None:
return self._jobs.get(job_id)
def update(self, job: IngestJob) -> IngestJob:
self._jobs[job.id] = job
return job
def update_status(self, job_id: str, status, **kwargs):
job = self._jobs.get(job_id)
if job:
job.status = status
def list_by_project(self, project_id: str, skip: int = 0, limit: int = 50) -> list[IngestJob]:
return [
j for j in self._jobs.values() if j.project_id == project_id
][skip : skip + limit]
def list_by_library(self, library_id: str, skip: int = 0, limit: int = 50) -> list[IngestJob]:
return [j for j in self._jobs.values() if j.library_id == library_id][skip : skip + limit]
# ---------------------------------------------------------------------------
# 2. Helpers & Fixtures
# ---------------------------------------------------------------------------
def _make_user(**overrides) -> User:
defaults = dict(
id="user-test-001",
email="test@example.com",
display_name="Test User",
username="testuser",
subscription_plan="free",
subscription_status="active",
max_projects=3,
max_storage_gb=10,
created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
)
defaults.update(overrides)
return User(**defaults)
def _make_project(id: str = "proj-1", owner_user_id: str = "user-test-001") -> Project:
return Project(id=id, name="Test Project", owner_user_id=owner_user_id)
def _make_generation_task(
task_id: str = "gen-task-1",
project_id: str = "proj-1",
user_id: str = "user-test-001",
status: GenerationTaskStatus = GenerationTaskStatus.PENDING,
) -> GenerationTask:
task = GenerationTask(
id=task_id,
project_id=project_id,
asset_library_id="lib-1",
strategy_id="s1",
voice_library_id="v1",
created_by_user_id=user_id,
)
task.status = status
return task
def _make_ingest_job(
job_id: str = "ingest-job-1",
project_id: str = "proj-1",
library_id: str = "lib-1",
status: IngestJobStatus = IngestJobStatus.PENDING,
) -> IngestJob:
job = IngestJob(
id=job_id,
project_id=project_id,
library_id=library_id,
storage_key="uploads/test.mp4",
)
job.status = status
return job
@pytest.fixture
def client():
"""创建带有依赖覆盖的 TestClient。"""
test_app = FastAPI()
test_app.include_router(router)
project = _make_project()
project_repo = StubProjectRepository({project.id: project})
task_repo = StubGenerationTaskRepository()
ingest_repo = StubIngestJobRepository()
def _override_current_user():
mock_auth = MagicMock(spec=AuthenticatedUser)
mock_auth.user = _make_user()
return mock_auth
test_app.dependency_overrides[get_current_user] = _override_current_user
test_app.dependency_overrides[get_project_repository] = lambda: project_repo
test_app.dependency_overrides[get_generation_task_repository] = lambda: task_repo
test_app.dependency_overrides[get_ingest_job_repository] = lambda: ingest_repo
yield TestClient(test_app)
test_app.dependency_overrides.clear()
# ---------------------------------------------------------------------------
# 3. GET /tasks — 列出用户任务
# ---------------------------------------------------------------------------
class TestListUserTasks:
"""列出用户任务端点测试。"""
def test_empty_list(self, client):
"""无任务时返回空列表。"""
resp = client.get("/tasks")
assert resp.status_code == 200
data = resp.json()
assert "items" in data
assert data["items"] == []
def test_list_returns_generation_tasks(self, client):
"""返回当前用户的 generation 任务。"""
# 直接在 repository 中注入任务
from app.dependencies import get_generation_task_repository
task_repo = StubGenerationTaskRepository()
task1 = _make_generation_task("gen-1", status=GenerationTaskStatus.PENDING)
task2 = _make_generation_task("gen-2", status=GenerationTaskStatus.COMPLETED)
task_repo.create(task1)
task_repo.create(task2)
test_app = FastAPI()
test_app.include_router(router)
project_repo = StubProjectRepository({_make_project().id: _make_project()})
def _override_current_user():
mock_auth = MagicMock(spec=AuthenticatedUser)
mock_auth.user = _make_user()
return mock_auth
test_app.dependency_overrides[get_current_user] = _override_current_user
test_app.dependency_overrides[get_project_repository] = lambda: project_repo
test_app.dependency_overrides[get_generation_task_repository] = lambda: task_repo
test_app.dependency_overrides[get_ingest_job_repository] = lambda: StubIngestJobRepository()
tc = TestClient(test_app)
resp = tc.get("/tasks")
assert resp.status_code == 200
data = resp.json()
assert len(data["items"]) == 2
# 验证响应字段
for item in data["items"]:
assert "id" in item
assert "task_type" in item
assert item["task_type"] == "generation"
assert "status" in item
assert "current_step" in item
assert "retryable" in item
test_app.dependency_overrides.clear()
def test_tasks_sorted_by_updated_time(self, client):
"""任务按更新时间倒序排列。"""
# 由于两个任务同时创建,验证它们都出现在列表中
resp = client.get("/tasks")
assert resp.status_code == 200
data = resp.json()
assert isinstance(data["items"], list)
def test_task_response_fields(self, client):
"""任务响应包含所有必需字段。"""
resp = client.get("/tasks")
assert resp.status_code == 200
# 空列表也应该返回正确的结构
assert resp.json()["items"] == []
# ---------------------------------------------------------------------------
# 4. POST /tasks/{task_id}/retry — 重试用户任务
# ---------------------------------------------------------------------------
class TestRetryUserTask:
"""重试用户任务端点测试。"""
def test_retry_nonexistent_task_returns_404(self, client):
"""重试不存在的任务返回 404。"""
resp = client.post("/tasks/nonexistent-task-id/retry")
assert resp.status_code == 404
assert "not found" in resp.json()["detail"].lower()
@patch("app.api.routes.task_center.celery_app")
def test_retry_pending_task_returns_409(self, mock_celery, client):
"""重试 pending 状态的任务返回 409(只有 failed 任务才能重试)。"""
mock_celery.send_task = MagicMock()
# 在 repository 中创建一个 pending 任务
task_repo = StubGenerationTaskRepository()
task = _make_generation_task("gen-pending", status=GenerationTaskStatus.PENDING)
task_repo.create(task)
test_app = FastAPI()
test_app.include_router(router)
project_repo = StubProjectRepository({_make_project().id: _make_project()})
def _get_user():
return MagicMock(spec=AuthenticatedUser, user=_make_user())
test_app.dependency_overrides[get_current_user] = _get_user
test_app.dependency_overrides[get_project_repository] = lambda: project_repo
test_app.dependency_overrides[get_generation_task_repository] = lambda: task_repo
test_app.dependency_overrides[get_ingest_job_repository] = lambda: StubIngestJobRepository()
tc = TestClient(test_app)
resp = tc.post("/tasks/gen-pending/retry")
assert resp.status_code == 409
assert "Only failed" in resp.json()["detail"]
test_app.dependency_overrides.clear()
@patch("app.api.routes.task_center.celery_app")
def test_retry_completed_task_returns_409(self, mock_celery, client):
"""重试 completed 状态的任务返回 409。"""
mock_celery.send_task = MagicMock()
task_repo = StubGenerationTaskRepository()
task = _make_generation_task("gen-completed", status=GenerationTaskStatus.COMPLETED)
task_repo.create(task)
test_app = FastAPI()
test_app.include_router(router)
project_repo = StubProjectRepository({_make_project().id: _make_project()})
def _get_user():
return MagicMock(spec=AuthenticatedUser, user=_make_user())
test_app.dependency_overrides[get_current_user] = _get_user
test_app.dependency_overrides[get_project_repository] = lambda: project_repo
test_app.dependency_overrides[get_generation_task_repository] = lambda: task_repo
test_app.dependency_overrides[get_ingest_job_repository] = lambda: StubIngestJobRepository()
tc = TestClient(test_app)
resp = tc.post("/tasks/gen-completed/retry")
assert resp.status_code == 409
test_app.dependency_overrides.clear()
# ---------------------------------------------------------------------------
# 5. GET /projects/{project_id}/tasks — 列出项目任务
# ---------------------------------------------------------------------------
class TestListProjectTasks:
"""列出项目任务端点测试。"""
def test_empty_project_tasks(self, client):
"""项目无任务时返回空列表。"""
resp = client.get("/projects/proj-1/tasks")
assert resp.status_code == 200
data = resp.json()
assert "items" in data
assert data["items"] == []
def test_project_not_found(self, client):
"""项目不存在返回 404。"""
resp = client.get("/projects/nonexistent-project/tasks")
assert resp.status_code == 404
assert "Project not found" in resp.json()["detail"]
def test_returns_ingest_and_generation_tasks(self, client):
"""返回项目中 ingest 和 generation 两种任务。"""
# 在 repository 中注入任务
task_repo = StubGenerationTaskRepository()
gen_task = _make_generation_task("gen-proj-1", status=GenerationTaskStatus.PENDING)
task_repo.create(gen_task)
ingest_repo = StubIngestJobRepository()
ingest_job = _make_ingest_job("ingest-proj-1", status=IngestJobStatus.PENDING)
ingest_repo.create(ingest_job)
test_app = FastAPI()
test_app.include_router(router)
project_repo = StubProjectRepository({_make_project().id: _make_project()})
def _get_user():
return MagicMock(spec=AuthenticatedUser, user=_make_user())
test_app.dependency_overrides[get_current_user] = _get_user
test_app.dependency_overrides[get_project_repository] = lambda: project_repo
test_app.dependency_overrides[get_generation_task_repository] = lambda: task_repo
test_app.dependency_overrides[get_ingest_job_repository] = lambda: ingest_repo
tc = TestClient(test_app)
resp = tc.get("/projects/proj-1/tasks")
assert resp.status_code == 200
data = resp.json()
assert len(data["items"]) == 2
task_types = {item["task_type"] for item in data["items"]}
assert "generation" in task_types
assert "ingest" in task_types
test_app.dependency_overrides.clear()
def test_project_task_response_fields(self, client):
"""项目任务响应包含所有必需字段。"""
resp = client.get("/projects/proj-1/tasks")
assert resp.status_code == 200
data = resp.json()
assert isinstance(data["items"], list)
# ---------------------------------------------------------------------------
# 6. POST /tasks/{task_type}/{source_id}/retry — 重试项目任务
# ---------------------------------------------------------------------------
class TestRetryProjectTask:
"""重试项目任务端点测试。"""
def test_retry_unsupported_task_type_returns_400(self, client):
"""不支持的任务类型返回 400。"""
resp = client.post("/tasks/unknown/some-source-id/retry")
assert resp.status_code == 400
assert "Unsupported" in resp.json()["detail"]
@patch("app.api.routes.task_center.celery_app")
def test_retry_failed_generation_task(self, mock_celery, client):
"""重试失败的 generation 任务成功。"""
mock_celery.send_task = MagicMock()
task_repo = StubGenerationTaskRepository()
task = _make_generation_task("gen-failed-1", status=GenerationTaskStatus.FAILED)
task_repo.create(task)
ingest_repo = StubIngestJobRepository()
test_app = FastAPI()
test_app.include_router(router)
project_repo = StubProjectRepository({_make_project().id: _make_project()})
def _get_user():
return MagicMock(spec=AuthenticatedUser, user=_make_user())
test_app.dependency_overrides[get_current_user] = _get_user
test_app.dependency_overrides[get_project_repository] = lambda: project_repo
test_app.dependency_overrides[get_generation_task_repository] = lambda: task_repo
test_app.dependency_overrides[get_ingest_job_repository] = lambda: ingest_repo
tc = TestClient(test_app)
resp = tc.post("/tasks/generation/gen-failed-1/retry")
assert resp.status_code == 200
data = resp.json()
assert data["task_type"] == "generation"
assert data["status"] == "pending"
assert "current_step" in data
# 验证新任务的 ID 不同于原任务
assert data["source_id"] != "gen-failed-1"
# 验证 Celery 任务被发送
assert mock_celery.send_task.called
test_app.dependency_overrides.clear()
@patch("app.api.routes.task_center.celery_app")
def test_retry_failed_ingest_task(self, mock_celery, client):
"""重试失败的 ingest 任务成功。"""
mock_celery.send_task = MagicMock()
ingest_repo = StubIngestJobRepository()
job = _make_ingest_job("ingest-failed-1", status=IngestJobStatus.FAILED)
ingest_repo.create(job)
task_repo = StubGenerationTaskRepository()
test_app = FastAPI()
test_app.include_router(router)
project_repo = StubProjectRepository({_make_project().id: _make_project()})
def _get_user():
return MagicMock(spec=AuthenticatedUser, user=_make_user())
test_app.dependency_overrides[get_current_user] = _get_user
test_app.dependency_overrides[get_project_repository] = lambda: project_repo
test_app.dependency_overrides[get_generation_task_repository] = lambda: task_repo
test_app.dependency_overrides[get_ingest_job_repository] = lambda: ingest_repo
tc = TestClient(test_app)
resp = tc.post("/tasks/ingest/ingest-failed-1/retry")
assert resp.status_code == 200
data = resp.json()
assert data["task_type"] == "ingest"
assert data["status"] == "pending"
assert mock_celery.send_task.called
assert mock_celery.send_task.call_args[0][0] == "worker.ingest_asset"
test_app.dependency_overrides.clear()
def test_retry_pending_generation_task_returns_409(self, client):
"""重试 pending 状态的 generation 任务返回 409。"""
task_repo = StubGenerationTaskRepository()
task = _make_generation_task("gen-pending-proj", status=GenerationTaskStatus.PENDING)
task_repo.create(task)
ingest_repo = StubIngestJobRepository()
test_app = FastAPI()
test_app.include_router(router)
project_repo = StubProjectRepository({_make_project().id: _make_project()})
def _get_user():
return MagicMock(spec=AuthenticatedUser, user=_make_user())
test_app.dependency_overrides[get_current_user] = _get_user
test_app.dependency_overrides[get_project_repository] = lambda: project_repo
test_app.dependency_overrides[get_generation_task_repository] = lambda: task_repo
test_app.dependency_overrides[get_ingest_job_repository] = lambda: ingest_repo
tc = TestClient(test_app)
resp = tc.post("/tasks/generation/gen-pending-proj/retry")
assert resp.status_code == 409
assert "Only failed" in resp.json()["detail"]
test_app.dependency_overrides.clear()
def test_retry_nonexistent_generation_task_returns_404(self, client):
"""重试不存在的 generation 任务返回 404。"""
task_repo = StubGenerationTaskRepository()
ingest_repo = StubIngestJobRepository()
test_app = FastAPI()
test_app.include_router(router)
project_repo = StubProjectRepository({_make_project().id: _make_project()})
def _get_user():
return MagicMock(spec=AuthenticatedUser, user=_make_user())
test_app.dependency_overrides[get_current_user] = _get_user
test_app.dependency_overrides[get_project_repository] = lambda: project_repo
test_app.dependency_overrides[get_generation_task_repository] = lambda: task_repo
test_app.dependency_overrides[get_ingest_job_repository] = lambda: ingest_repo
tc = TestClient(test_app)
resp = tc.post("/tasks/generation/nonexistent-id/retry")
assert resp.status_code == 404
test_app.dependency_overrides.clear()
def test_retry_nonexistent_ingest_task_returns_404(self, client):
"""重试不存在的 ingest 任务返回 404。"""
task_repo = StubGenerationTaskRepository()
ingest_repo = StubIngestJobRepository()
test_app = FastAPI()
test_app.include_router(router)
project_repo = StubProjectRepository({_make_project().id: _make_project()})
def _get_user():
return MagicMock(spec=AuthenticatedUser, user=_make_user())
test_app.dependency_overrides[get_current_user] = _get_user
test_app.dependency_overrides[get_project_repository] = lambda: project_repo
test_app.dependency_overrides[get_generation_task_repository] = lambda: task_repo
test_app.dependency_overrides[get_ingest_job_repository] = lambda: ingest_repo
tc = TestClient(test_app)
resp = tc.post("/tasks/ingest/nonexistent-id/retry")
assert resp.status_code == 404
test_app.dependency_overrides.clear()
# ---------------------------------------------------------------------------
# 7. 跨端点集成场景
# ---------------------------------------------------------------------------
class TestTaskCenterCrossEndpoint:
"""任务中心跨端点集成测试。"""
@patch("app.api.routes.task_center.celery_app")
def test_list_then_retry_then_list(self, mock_celery, client):
"""列出任务 → 重试失败任务 → 再列出验证新任务。"""
mock_celery.send_task = MagicMock()
task_repo = StubGenerationTaskRepository()
failed_task = _make_generation_task("gen-fail-cross", status=GenerationTaskStatus.FAILED)
failed_task.error_message = "ffmpeg error"
task_repo.create(failed_task)
ingest_repo = StubIngestJobRepository()
test_app = FastAPI()
test_app.include_router(router)
project_repo = StubProjectRepository({_make_project().id: _make_project()})
def _get_user():
return MagicMock(spec=AuthenticatedUser, user=_make_user())
test_app.dependency_overrides[get_current_user] = _get_user
test_app.dependency_overrides[get_project_repository] = lambda: project_repo
test_app.dependency_overrides[get_generation_task_repository] = lambda: task_repo
test_app.dependency_overrides[get_ingest_job_repository] = lambda: ingest_repo
tc = TestClient(test_app)
# 1. 列出任务
list_resp = tc.get("/tasks")
assert list_resp.status_code == 200
items = list_resp.json()["items"]
assert len(items) == 1
assert items[0]["retryable"] is True # failed 任务应可重试
# 2. 重试失败任务
retry_resp = tc.post("/tasks/gen-fail-cross/retry")
assert retry_resp.status_code == 200
new_task_id = retry_resp.json()["source_id"]
# 3. 再次列出,应有2个任务(旧的failed + 新的pending
list_resp2 = tc.get("/tasks")
assert list_resp2.status_code == 200
items2 = list_resp2.json()["items"]
assert len(items2) == 2
test_app.dependency_overrides.clear()
if __name__ == "__main__":
pytest.main([__file__, "-v"])
-254
View File
@@ -1,254 +0,0 @@
"""
DELETE /asset-libraries/{library_id} 单元测试
覆盖:
- 正常删除空素材库(204
- 删除含素材的库(同时删除库内素材)
- 素材库不存在(404
- 无权限访问(403
"""
from __future__ import annotations
import os
import sys
from pathlib import Path
from unittest.mock import MagicMock
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"))
from fastapi import FastAPI
from fastapi.testclient import TestClient
from packages.adapters.in_memory.asset_library_repository import InMemoryAssetLibraryRepository
from packages.adapters.in_memory.asset_repository import InMemoryAssetRepository
from packages.domain import Asset, AssetLibrary, AssetLibraryKind, AssetStatus
# ---------------------------------------------------------------------------
# Stub Project Repository
# ---------------------------------------------------------------------------
class StubProject:
"""最小化 Project stub,支持 can_access"""
def __init__(self, project_id: str, owner_id: str):
self.id = project_id
self._owner_id = owner_id
def can_access(self, user_id: str) -> bool:
return user_id == self._owner_id
class StubProjectRepository:
def __init__(self, projects: dict[str, StubProject] | None = None):
self._projects = projects or {}
def find_by_id(self, project_id: str):
return self._projects.get(project_id)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_auth_user(user_id: str = "user-001"):
from app.auth import AuthenticatedUser
from packages.domain.entities import User
user = User(id=user_id, email="test@example.com", display_name="测试用户")
return AuthenticatedUser(user=user)
def _create_test_app(
library_repo: InMemoryAssetLibraryRepository,
asset_repo: InMemoryAssetRepository,
project_repo: StubProjectRepository,
user_id: str = "user-001",
):
from app.api.routes import asset_libraries as module
from app.api.routes.asset_libraries import router
app = FastAPI()
app.include_router(router, prefix="/api/v1/asset-libraries")
app.dependency_overrides[module.get_current_user] = lambda: _make_auth_user(user_id)
app.dependency_overrides[module.get_asset_library_repository] = lambda: library_repo
app.dependency_overrides[module.get_asset_repository] = lambda: asset_repo
app.dependency_overrides[module.get_project_repository] = lambda: project_repo
return app
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
class TestDeleteAssetLibrary:
"""DELETE /asset-libraries/{library_id} 测试"""
def test_delete_empty_library(self):
"""删除空素材库 → 204"""
from datetime import datetime, timezone
lib_repo = InMemoryAssetLibraryRepository()
asset_repo = InMemoryAssetRepository()
proj_repo = StubProjectRepository({"proj-1": StubProject("proj-1", "user-001")})
now = datetime.now(timezone.utc)
library = AssetLibrary(
id="lib-1",
project_id="proj-1",
name="视频素材库",
kind=AssetLibraryKind.VIDEO,
asset_count=0,
total_size=0,
created_at=now,
updated_at=now,
)
lib_repo.create(library)
app = _create_test_app(lib_repo, asset_repo, proj_repo)
client = TestClient(app)
response = client.delete("/api/v1/asset-libraries/lib-1")
assert response.status_code == 204
# 验证库已删除
assert lib_repo.find_by_id("lib-1") is None
def test_delete_library_with_assets(self):
"""删除含素材的库 → 库和素材都被删除"""
from datetime import datetime, timezone
lib_repo = InMemoryAssetLibraryRepository()
asset_repo = InMemoryAssetRepository()
proj_repo = StubProjectRepository({"proj-1": StubProject("proj-1", "user-001")})
now = datetime.now(timezone.utc)
library = AssetLibrary(
id="lib-1",
project_id="proj-1",
name="视频素材库",
kind=AssetLibraryKind.VIDEO,
asset_count=3,
total_size=1000,
created_at=now,
updated_at=now,
)
lib_repo.create(library)
# 创建 3 个素材
for i in range(3):
asset = Asset.create(
project_id="proj-1",
library_id="lib-1",
name=f"video_{i}.mp4",
storage_key=f"uploads/video_{i}.mp4",
mime_type="video/mp4",
status=AssetStatus.READY,
)
asset_repo.create(asset)
# 创建一个不属于该库的素材(不应被删除)
other_asset = Asset.create(
project_id="proj-1",
library_id="lib-other",
name="other.mp4",
storage_key="uploads/other.mp4",
mime_type="video/mp4",
status=AssetStatus.READY,
)
asset_repo.create(other_asset)
app = _create_test_app(lib_repo, asset_repo, proj_repo)
client = TestClient(app)
response = client.delete("/api/v1/asset-libraries/lib-1")
assert response.status_code == 204
# 库已删除
assert lib_repo.find_by_id("lib-1") is None
# 库内素材已删除
assert asset_repo.find_by_library("lib-1") == []
# 其他素材未受影响
assert asset_repo.get(other_asset.id) is not None
def test_delete_nonexistent_library(self):
"""删除不存在的素材库 → 404"""
lib_repo = InMemoryAssetLibraryRepository()
asset_repo = InMemoryAssetRepository()
proj_repo = StubProjectRepository({"proj-1": StubProject("proj-1", "user-001")})
app = _create_test_app(lib_repo, asset_repo, proj_repo)
client = TestClient(app)
response = client.delete("/api/v1/asset-libraries/nonexistent-id")
assert response.status_code == 404
assert "素材库不存在" in response.json()["detail"]
def test_delete_library_access_denied(self):
"""无权限用户删除素材库 → 403"""
from datetime import datetime, timezone
lib_repo = InMemoryAssetLibraryRepository()
asset_repo = InMemoryAssetRepository()
# 项目属于 user-002,当前用户是 user-001
proj_repo = StubProjectRepository({"proj-1": StubProject("proj-1", "user-002")})
now = datetime.now(timezone.utc)
library = AssetLibrary(
id="lib-1",
project_id="proj-1",
name="视频素材库",
kind=AssetLibraryKind.VIDEO,
asset_count=0,
total_size=0,
created_at=now,
updated_at=now,
)
lib_repo.create(library)
app = _create_test_app(lib_repo, asset_repo, proj_repo, user_id="user-001")
client = TestClient(app)
response = client.delete("/api/v1/asset-libraries/lib-1")
assert response.status_code == 403
assert "Access denied" in response.json()["detail"]
# 库未被删除
assert lib_repo.find_by_id("lib-1") is not None
def test_delete_library_project_not_found(self):
"""素材库所属项目不存在 → 404"""
from datetime import datetime, timezone
lib_repo = InMemoryAssetLibraryRepository()
asset_repo = InMemoryAssetRepository()
# 空的项目仓库,找不到项目
proj_repo = StubProjectRepository({})
now = datetime.now(timezone.utc)
library = AssetLibrary(
id="lib-1",
project_id="proj-missing",
name="视频素材库",
kind=AssetLibraryKind.VIDEO,
asset_count=0,
total_size=0,
created_at=now,
updated_at=now,
)
lib_repo.create(library)
app = _create_test_app(lib_repo, asset_repo, proj_repo)
client = TestClient(app)
response = client.delete("/api/v1/asset-libraries/lib-1")
assert response.status_code == 404
@@ -404,7 +404,7 @@ class TestAIRecommendEndpoint:
json={"asset_ids": ["asset-1"]},
)
assert resp.status_code == 400
assert "当前计划状态" in resp.json()["detail"] or "编辑计划" in resp.json()["detail"]
assert "draft/editing" in resp.json()["detail"]
def test_ai_recommend_with_custom_params(self, ai_client):
c, repo = ai_client
+3 -84
View File
@@ -332,22 +332,17 @@ class TestGeneratePlan:
assert resp.status_code == 404
assert "不存在" in resp.json()["detail"]
def test_generate_draft_auto_transition_to_editing(
def test_generate_wrong_status_draft(
self,
client: TestClient,
plan_repo: StubEditPlanRepository,
) -> None:
"""draft 状态自动转 editing(自动兜底),然后因 0 片段报错"""
"""draft 状态 → 400"""
plan = _make_plan(status=EditPlanStatus.DRAFT)
plan_repo.create(plan)
resp = client.post(f"/api/v1/edit-plans/{plan.id}/generate")
# draft 自动转 editing,但没有片段所以还是 400
assert resp.status_code == 400
assert "请先添加片段后再生成视频" in resp.json()["detail"]
# 验证状态已自动转为 editing
updated_plan = plan_repo.get(plan.id)
assert updated_plan is not None
assert updated_plan.status == EditPlanStatus.EDITING
assert "editing" in resp.json()["detail"]
def test_generate_wrong_status_rendering(
self,
@@ -592,79 +587,3 @@ class TestResponseSchema:
data = resp.json()
expected_keys = {"plan_id", "plan_status", "generation_task_id", "clips"}
assert set(data.keys()) == expected_keys
# ── P0-1: 生成接口错误处理 ────────────────────────────────────────────────────
class TestGeneratePlanErrorHandling:
"""P0-1: generate 端点异常时返回明确错误信息,不裸 500"""
def test_generate_internal_error_returns_clear_message(
self,
client: TestClient,
plan_repo: StubEditPlanRepository,
clip_repo: StubEditPlanClipRepository,
) -> None:
"""核心流程抛异常 → 500 + 用户友好的错误信息(不暴露技术细节)"""
plan = _make_plan(status=EditPlanStatus.EDITING)
plan_repo.create(plan)
clip = _make_clip(plan.id, order=1)
clip_repo.create(clip)
with patch("app.api.routes.edit_plans.celery_app") as mock_celery:
# 模拟 Celery 调度失败
mock_celery.send_task.side_effect = RuntimeError("Redis 连接超时")
resp = client.post(f"/api/v1/edit-plans/{plan.id}/generate")
assert resp.status_code == 500
data = resp.json()
# 验证返回了用户友好的错误信息,不暴露技术细节
assert "生成失败" in data["detail"]
assert "RuntimeError" not in data["detail"]
assert "Redis" not in data["detail"]
def test_generate_error_rolls_back_plan_status(
self,
client: TestClient,
plan_repo: StubEditPlanRepository,
clip_repo: StubEditPlanClipRepository,
) -> None:
"""异常时将计划标记为 failedRENDERING → FAILED 是合法流转)"""
plan = _make_plan(status=EditPlanStatus.EDITING)
plan_repo.create(plan)
clip = _make_clip(plan.id, order=1)
clip_repo.create(clip)
with patch("app.api.routes.edit_plans.celery_app") as mock_celery:
mock_celery.send_task.side_effect = RuntimeError("调度失败")
resp = client.post(f"/api/v1/edit-plans/{plan.id}/generate")
assert resp.status_code == 500
# 计划状态应变为 failed
updated = plan_repo.get(plan.id)
assert updated.status == EditPlanStatus.FAILED
def test_generate_error_detail_is_user_friendly(
self,
client: TestClient,
plan_repo: StubEditPlanRepository,
clip_repo: StubEditPlanClipRepository,
) -> None:
"""错误信息对用户友好,不暴露技术细节(异常类型、内部错误信息)"""
plan = _make_plan(status=EditPlanStatus.EDITING)
plan_repo.create(plan)
clip = _make_clip(plan.id, order=1)
clip_repo.create(clip)
with patch("app.api.routes.edit_plans.celery_app") as mock_celery:
mock_celery.send_task.side_effect = ConnectionError("Broker 不可达")
resp = client.post(f"/api/v1/edit-plans/{plan.id}/generate")
assert resp.status_code == 500
detail = resp.json()["detail"]
# 验证不暴露技术细节
assert "ConnectionError" not in detail
assert "Broker 不可达" not in detail
# 验证返回了用户友好的提示
assert "生成失败" in detail
+2 -2
View File
@@ -509,7 +509,7 @@ class TestGenerationWorkflow:
svc.create_clip(p.id, "intro", 0)
can, reason = svc.can_generate(p.id)
assert can is False
assert "编辑" in reason or "模板" in reason
assert "editing" in reason
def test_can_generate_no_clips_fails(self):
svc = _make_service()
@@ -517,7 +517,7 @@ class TestGenerationWorkflow:
svc.transition_status(p.id, EditPlanStatus.EDITING)
can, reason = svc.can_generate(p.id)
assert can is False
assert "请先添加片段后再生成视频" in reason
assert "没有片段" in reason
def test_mark_clips_ready(self):
svc = _make_service()
-333
View File
@@ -1,333 +0,0 @@
"""P0-2: Celery 任务 render_edit_plan 失败时更新 GenerationTask 状态。
验证:
- 异常发生时 GenerationTask 状态更新为 failed
- error_message 记录了异常类型和描述
- completed_at 被设置
- 即使 generation_task_id 为空也不崩溃
- 即使更新 GenerationTask 本身失败也不影响 retry
"""
from __future__ import annotations
import os
import sys
from dataclasses import dataclass, field
from datetime import datetime, timezone
from types import ModuleType
from typing import Any, Optional
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")
import pytest
# ── Mock worker 模块以避免数据库连接 ──────────────────────────────────────────
# worker_app.db 在 import 时会尝试连接数据库,必须在导入 task 模块前 mock
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "worker"))
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
# 预注册 mock 模块,阻止真实数据库初始化
_mock_db_mod = ModuleType("worker_app.db")
_mock_db_mod.SessionLocal = MagicMock()
sys.modules.setdefault("worker_app.db", _mock_db_mod)
_mock_celery_mod = ModuleType("worker_app.celery_app")
_mock_celery_app = MagicMock()
# 让 @celery_app.task(...) 装饰器透传原始函数,否则函数变成 MagicMock
_mock_celery_app.task = lambda **kwargs: lambda fn: fn
_mock_celery_mod.celery_app = _mock_celery_app
sys.modules.setdefault("worker_app.celery_app", _mock_celery_mod)
# ── Stub domain objects ───────────────────────────────────────────────────────
@dataclass
class _StubStatus:
value: str
def __eq__(self, other):
if isinstance(other, str):
return self.value == other
if isinstance(other, _StubStatus):
return self.value == other.value
return NotImplemented
@dataclass
class StubEditPlan:
id: str = "plan-001"
template_id: str = "tmpl-001"
status: Any = None
config: dict = field(default_factory=dict)
def mark_failed(self):
self.status = _StubStatus("failed")
def mark_completed(self):
self.status = _StubStatus("completed")
@dataclass
class StubGenerationTask:
id: str = "gen-task-001"
status: Any = field(default_factory=lambda: _StubStatus("pending"))
error_message: str = ""
progress: float = 0.0
result_count: int = 0
started_at: Any = None
completed_at: Any = None
project_id: str = ""
created_by_user_id: str = "user-001"
@dataclass
class StubClip:
id: str = "clip-001"
plan_id: str = "plan-001"
asset_id: str = "assets/video.mp4"
order: int = 1
status: Any = field(default_factory=lambda: _StubStatus("ready"))
transition_effect: str = ""
text_content: str = ""
clip_type: str = "MAIN"
duration: float = 0.0
def mark_failed(self):
self.status = _StubStatus("failed")
def mark_rendered(self):
self.status = _StubStatus("rendered")
# ── Stub repositories ─────────────────────────────────────────────────────────
class StubPlanRepo:
def __init__(self, plan: StubEditPlan):
self._plan = plan
def get(self, plan_id: str) -> Optional[StubEditPlan]:
if plan_id == self._plan.id:
return self._plan
return None
def update(self, plan: StubEditPlan) -> StubEditPlan:
self._plan = plan
return plan
class StubClipRepo:
def __init__(self, clips: list[StubClip] | None = None):
self._clips = clips or []
def list_by_plan(self, plan_id: str, skip: int = 0, limit: int = 10000) -> list[StubClip]:
return [c for c in self._clips if c.plan_id == plan_id]
def get(self, clip_id: str) -> Optional[StubClip]:
for c in self._clips:
if c.id == clip_id:
return c
return None
def update(self, clip: StubClip) -> StubClip:
return clip
class StubGenTaskRepo:
def __init__(self, task: StubGenerationTask | None = None):
self._store: dict[str, StubGenerationTask] = {}
if task:
self._store[task.id] = task
def get(self, task_id: str) -> Optional[StubGenerationTask]:
return self._store.get(task_id)
def update(self, task: StubGenerationTask) -> StubGenerationTask:
self._store[task.id] = task
return task
# ── Import task module (after mocks are in place) ─────────────────────────────
from worker_app.tasks.edit_plan_generation import render_edit_plan
# ── Tests ─────────────────────────────────────────────────────────────────────
class TestRenderEditPlanFailureUpdatesGenTask:
"""P0-2: render_edit_plan 异常时更新 GenerationTask 状态为 failed"""
def _make_bound_task(self):
"""构建绑定的 Celery task mock"""
task = MagicMock()
task.retry = MagicMock(side_effect=RuntimeError("retry called"))
return task
def test_exception_marks_gen_task_failed(self):
"""异常时 GenerationTask.status 被设为 failed"""
plan = StubEditPlan(status=_StubStatus("rendering"))
plan.config["generation_task_id"] = "gen-task-001"
gen_task = StubGenerationTask(id="gen-task-001", status=_StubStatus("running"))
plan_repo = StubPlanRepo(plan)
clip_repo = StubClipRepo([])
gen_task_repo = StubGenTaskRepo(gen_task)
# 让 clip_repo 抛异常以触发 except 路径
clip_repo_bad = MagicMock()
clip_repo_bad.list_by_plan.side_effect = RuntimeError("OSS 连接失败")
def fake_get_repos():
yield plan_repo, clip_repo_bad, gen_task_repo, MagicMock()
bound_task = self._make_bound_task()
with patch(
"worker_app.tasks.edit_plan_generation._get_repos",
side_effect=fake_get_repos,
):
with pytest.raises(RuntimeError, match="retry called"):
render_edit_plan(bound_task, "plan-001")
# 核心断言:GenerationTask 状态为 failed(生产代码赋值为字符串)
assert gen_task.status == "failed"
def test_exception_records_error_message(self):
"""异常时 error_message 包含异常类型和描述"""
plan = StubEditPlan(status=_StubStatus("rendering"))
plan.config["generation_task_id"] = "gen-task-001"
gen_task = StubGenerationTask(id="gen-task-001", status=_StubStatus("running"))
plan_repo = StubPlanRepo(plan)
clip_repo = MagicMock()
clip_repo.list_by_plan.side_effect = RuntimeError("DB 查询超时")
gen_task_repo = StubGenTaskRepo(gen_task)
def fake_get_repos():
yield plan_repo, clip_repo, gen_task_repo, MagicMock()
bound_task = self._make_bound_task()
with patch(
"worker_app.tasks.edit_plan_generation._get_repos",
side_effect=fake_get_repos,
):
with pytest.raises(RuntimeError, match="retry called"):
render_edit_plan(bound_task, "plan-001")
assert gen_task.status == "failed"
assert "DB 查询超时" in gen_task.error_message
assert "RuntimeError" in gen_task.error_message
def test_exception_sets_completed_at(self):
"""异常时 completed_at 被设置"""
plan = StubEditPlan(status=_StubStatus("rendering"))
plan.config["generation_task_id"] = "gen-task-001"
gen_task = StubGenerationTask(id="gen-task-001", status=_StubStatus("running"))
plan_repo = StubPlanRepo(plan)
clip_repo = MagicMock()
clip_repo.list_by_plan.side_effect = RuntimeError("boom")
gen_task_repo = StubGenTaskRepo(gen_task)
def fake_get_repos():
yield plan_repo, clip_repo, gen_task_repo, MagicMock()
bound_task = self._make_bound_task()
with patch(
"worker_app.tasks.edit_plan_generation._get_repos",
side_effect=fake_get_repos,
):
with pytest.raises(RuntimeError):
render_edit_plan(bound_task, "plan-001")
assert gen_task.completed_at is not None
def test_no_generation_task_id_does_not_crash(self):
"""generation_task_id 为空时,异常处理不崩溃"""
plan = StubEditPlan(status=_StubStatus("rendering"))
plan.config = {} # 不设置 generation_task_id
plan_repo = StubPlanRepo(plan)
clip_repo = MagicMock()
clip_repo.list_by_plan.side_effect = RuntimeError("boom")
gen_task_repo = StubGenTaskRepo() # 空 repo
def fake_get_repos():
yield plan_repo, clip_repo, gen_task_repo, MagicMock()
bound_task = self._make_bound_task()
with patch(
"worker_app.tasks.edit_plan_generation._get_repos",
side_effect=fake_get_repos,
):
with pytest.raises(RuntimeError, match="retry called"):
render_edit_plan(bound_task, "plan-001")
# 计划仍被标记为 failed
assert plan.status.value == "failed"
def test_gen_task_update_failure_does_not_block_retry(self):
"""更新 GenerationTask 失败时,不影响 retry 流程"""
plan = StubEditPlan(status=_StubStatus("rendering"))
plan.config["generation_task_id"] = "gen-task-001"
gen_task = StubGenerationTask(id="gen-task-001", status=_StubStatus("running"))
plan_repo = StubPlanRepo(plan)
clip_repo = MagicMock()
clip_repo.list_by_plan.side_effect = RuntimeError("原始错误")
# gen_task_repo.update 也抛异常
gen_task_repo = MagicMock()
gen_task_repo.get.return_value = gen_task
gen_task_repo.update.side_effect = RuntimeError("DB 写入失败")
def fake_get_repos():
yield plan_repo, clip_repo, gen_task_repo, MagicMock()
bound_task = self._make_bound_task()
with patch(
"worker_app.tasks.edit_plan_generation._get_repos",
side_effect=fake_get_repos,
):
with pytest.raises(RuntimeError, match="retry called"):
render_edit_plan(bound_task, "plan-001")
# retry 被调用说明流程正确
bound_task.retry.assert_called_once()
def test_already_failed_gen_task_not_overwritten(self):
"""已经 failed 的 GenerationTask 不会被重复更新"""
plan = StubEditPlan(status=_StubStatus("rendering"))
plan.config["generation_task_id"] = "gen-task-001"
gen_task = StubGenerationTask(
id="gen-task-001",
status=_StubStatus("failed"), # 已经是 failed
error_message="之前的错误",
)
plan_repo = StubPlanRepo(plan)
clip_repo = MagicMock()
clip_repo.list_by_plan.side_effect = RuntimeError("新错误")
gen_task_repo = StubGenTaskRepo(gen_task)
def fake_get_repos():
yield plan_repo, clip_repo, gen_task_repo, MagicMock()
bound_task = self._make_bound_task()
with patch(
"worker_app.tasks.edit_plan_generation._get_repos",
side_effect=fake_get_repos,
):
with pytest.raises(RuntimeError, match="retry called"):
render_edit_plan(bound_task, "plan-001")
# error_message 应保持原值,不被覆盖
assert gen_task.error_message == "之前的错误"
+2 -2
View File
@@ -331,7 +331,7 @@ class TestListPlans:
c, repo = client
resp = c.get("/api/v1/edit-plans?status=invalid_status")
assert resp.status_code == 400
assert "无效" in resp.json()["detail"]
assert "无效的状态值" in resp.json()["detail"]
# ---------------------------------------------------------------------------
@@ -458,7 +458,7 @@ class TestUpdatePlan:
resp = c.put(f"/api/v1/edit-plans/{plan.id}", json={"status": "bogus"})
assert resp.status_code == 400
assert "无效" in resp.json()["detail"]
assert "无效的状态值" in resp.json()["detail"]
def test_update_same_status_is_noop(self, client):
c, repo = client
-551
View File
@@ -1,551 +0,0 @@
"""
订阅支付回调单元测试
覆盖场景:
- 正确签名的回调处理(当前实现无签名验证,验证参数合法性)
- 缺失参数的回调被拒绝(422
- 重复回调的幂等性(mark_paid 对已支付账单返回 False
- 各种支付状态(成功处理流程)
- 不同套餐和计费周期
注:当前支付回调实现较简单(无签名验证,使用查询参数),
测试聚焦于回调处理的核心逻辑和边界情况。
"""
from __future__ import annotations
import os
import sys
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import MagicMock, patch
# ── 环境变量 & sys.path(必须在导入 app.* 之前设置) ──────────────────────────
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
from app.api.routes.subscription import router, _get_plan_name, _get_plan_price
# ---------------------------------------------------------------------------
# 1. Mock Billing Repository
# ---------------------------------------------------------------------------
@dataclass
class MockBillingRecord:
id: str = ""
user_id: str = ""
plan_name: str = ""
amount: float = 0.0
billing_cycle: str = ""
status: str = "pending"
payment_method: str = ""
payment_id: str = ""
paid_at: datetime | None = None
created_at: datetime | None = None
def __init__(self, **kwargs):
for k, v in kwargs.items():
setattr(self, k, v)
if self.created_at is None:
self.created_at = datetime.now(timezone.utc)
class MockBillingRepository:
"""模拟的 Billing Repository,用于单元测试。"""
def __init__(self):
self.records: dict[str, MockBillingRecord] = {}
self.created_count = 0
self.mark_paid_count = 0
self.update_subscription_count = 0
self.updated_subscriptions: dict[str, dict] = {}
def create(self, record: dict) -> MockBillingRecord:
model = MockBillingRecord(**record)
self.records[model.id] = model
self.created_count += 1
return model
def find_by_user(self, user_id: str, limit: int = 50) -> list[MockBillingRecord]:
items = [r for r in self.records.values() if r.user_id == user_id]
items.sort(key=lambda r: r.created_at or datetime.min, reverse=True)
return items[:limit]
def find_by_id(self, record_id: str) -> MockBillingRecord | None:
return self.records.get(record_id)
def mark_paid(self, record_id: str, payment_method: str, payment_id: str) -> bool:
self.mark_paid_count += 1
model = self.records.get(record_id)
if model is None or model.status == "paid":
return False
model.status = "paid"
model.payment_method = payment_method
model.payment_id = payment_id
model.paid_at = datetime.now(timezone.utc)
return True
def update_subscription_on_payment(self, user_id: str, plan: str, expires_at: datetime) -> None:
self.update_subscription_count += 1
self.updated_subscriptions[user_id] = {
"plan": plan,
"expires_at": expires_at,
"status": "active",
}
# ---------------------------------------------------------------------------
# 2. Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def mock_repo():
return MockBillingRepository()
def _make_client(mock_billing_repo: MockBillingRepository) -> TestClient:
"""创建带有 mock billing repository 的 TestClient。"""
test_app = FastAPI()
test_app.include_router(router, prefix="/subscription")
# Mock SessionLocal 和 BillingRepository
mock_session = MagicMock()
with patch(
"packages.adapters.sqlalchemy_impl.session.SessionLocal",
return_value=mock_session,
):
with patch(
"packages.adapters.sqlalchemy_impl.billing_repository.SQLAlchemyBillingRepository",
return_value=mock_billing_repo,
):
yield TestClient(test_app)
# ---------------------------------------------------------------------------
# 3. 支付成功回调测试
# ---------------------------------------------------------------------------
class TestPaymentCallbackSuccess:
"""支付成功回调测试。"""
@patch("packages.adapters.sqlalchemy_impl.billing_repository.SQLAlchemyBillingRepository")
@patch("packages.adapters.sqlalchemy_impl.session.SessionLocal")
def test_monthly_pro_payment_success(self, MockSession, MockRepo):
"""Pro 套餐月付支付成功。"""
mock_repo = MockBillingRepository()
MockRepo.return_value = mock_repo
MockSession.return_value = MagicMock()
test_app = FastAPI()
test_app.include_router(router, prefix="/subscription")
client = TestClient(test_app)
resp = client.post(
"/subscription/payment-callback",
params={
"user_id": "user-001",
"plan": "pro",
"billing_cycle": "monthly",
"amount": 299.0,
"payment_method": "alipay",
"payment_id": "pay_20240101_001",
},
)
assert resp.status_code == 200
data = resp.json()
assert data["success"] is True
assert "支付成功" in data["message"]
assert "record_id" in data
# 验证账单创建
assert mock_repo.created_count == 1
# 验证标记支付
assert mock_repo.mark_paid_count == 1
# 验证订阅更新
assert mock_repo.update_subscription_count == 1
assert "user-001" in mock_repo.updated_subscriptions
assert mock_repo.updated_subscriptions["user-001"]["plan"] == "pro"
@patch("packages.adapters.sqlalchemy_impl.billing_repository.SQLAlchemyBillingRepository")
@patch("packages.adapters.sqlalchemy_impl.session.SessionLocal")
def test_yearly_standard_payment_success(self, MockSession, MockRepo):
"""标准版年付支付成功。"""
mock_repo = MockBillingRepository()
MockRepo.return_value = mock_repo
MockSession.return_value = MagicMock()
test_app = FastAPI()
test_app.include_router(router, prefix="/subscription")
client = TestClient(test_app)
resp = client.post(
"/subscription/payment-callback",
params={
"user_id": "user-002",
"plan": "standard",
"billing_cycle": "yearly",
"amount": 999.0,
"payment_method": "wechat",
"payment_id": "wx_20240101_002",
},
)
assert resp.status_code == 200
data = resp.json()
assert data["success"] is True
assert mock_repo.updated_subscriptions["user-002"]["plan"] == "standard"
# 年付到期时间应为约 365 天后
expires_at = mock_repo.updated_subscriptions["user-002"]["expires_at"]
expected = datetime.now(timezone.utc) + timedelta(days=365)
assert abs((expires_at - expected).days) <= 1
@patch("packages.adapters.sqlalchemy_impl.billing_repository.SQLAlchemyBillingRepository")
@patch("packages.adapters.sqlalchemy_impl.session.SessionLocal")
def test_enterprise_payment_success(self, MockSession, MockRepo):
"""企业版支付成功。"""
mock_repo = MockBillingRepository()
MockRepo.return_value = mock_repo
MockSession.return_value = MagicMock()
test_app = FastAPI()
test_app.include_router(router, prefix="/subscription")
client = TestClient(test_app)
resp = client.post(
"/subscription/payment-callback",
params={
"user_id": "user-003",
"plan": "enterprise",
"billing_cycle": "monthly",
"amount": 999.0,
"payment_method": "bank_transfer",
"payment_id": "ent_20240101_003",
},
)
assert resp.status_code == 200
assert resp.json()["success"] is True
assert mock_repo.updated_subscriptions["user-003"]["plan"] == "enterprise"
@patch("packages.adapters.sqlalchemy_impl.billing_repository.SQLAlchemyBillingRepository")
@patch("packages.adapters.sqlalchemy_impl.session.SessionLocal")
def test_default_payment_params(self, MockSession, MockRepo):
"""使用默认 payment_method 和空 payment_id。"""
mock_repo = MockBillingRepository()
MockRepo.return_value = mock_repo
MockSession.return_value = MagicMock()
test_app = FastAPI()
test_app.include_router(router, prefix="/subscription")
client = TestClient(test_app)
resp = client.post(
"/subscription/payment-callback",
params={
"user_id": "user-004",
"plan": "standard",
"billing_cycle": "monthly",
"amount": 99.0,
},
)
assert resp.status_code == 200
assert resp.json()["success"] is True
# 默认 payment_method 应为 alipay
assert mock_repo.mark_paid_count == 1
# ---------------------------------------------------------------------------
# 4. 重复回调幂等性测试
# ---------------------------------------------------------------------------
class TestPaymentCallbackIdempotency:
"""支付回调幂等性测试。"""
@patch("packages.adapters.sqlalchemy_impl.billing_repository.SQLAlchemyBillingRepository")
@patch("packages.adapters.sqlalchemy_impl.session.SessionLocal")
def test_duplicate_callback_creates_new_record(self, MockSession, MockRepo):
"""重复回调(当前实现每次创建新账单,无幂等保护)。"""
mock_repo = MockBillingRepository()
MockRepo.return_value = mock_repo
MockSession.return_value = MagicMock()
test_app = FastAPI()
test_app.include_router(router, prefix="/subscription")
client = TestClient(test_app)
params = {
"user_id": "user-idem-1",
"plan": "pro",
"billing_cycle": "monthly",
"amount": 299.0,
"payment_id": "pay_dup_001",
}
# 第一次回调
resp1 = client.post("/subscription/payment-callback", params=params)
assert resp1.status_code == 200
# 第二次回调(当前实现会创建新账单,不做幂等)
resp2 = client.post("/subscription/payment-callback", params=params)
assert resp2.status_code == 200
# 当前实现每次都会创建新账单
assert mock_repo.created_count == 2
def test_mark_paid_is_idempotent(self):
"""mark_paid 方法对已支付账单返回 False(幂等)。"""
repo = MockBillingRepository()
repo.create(dict(
id="bill-001", user_id="u1",
plan_name="Pro 专业版", amount=299.0,
billing_cycle="monthly", status="pending",
))
# 第一次标记为已支付
result1 = repo.mark_paid("bill-001", "alipay", "pay-001")
assert result1 is True
assert repo.records["bill-001"].status == "paid"
# 第二次标记(幂等,应返回 False)
result2 = repo.mark_paid("bill-001", "alipay", "pay-001")
assert result2 is False
assert repo.records["bill-001"].status == "paid"
# ---------------------------------------------------------------------------
# 5. 参数校验测试
# ---------------------------------------------------------------------------
class TestPaymentCallbackValidation:
"""支付回调参数校验测试。"""
@patch("packages.adapters.sqlalchemy_impl.billing_repository.SQLAlchemyBillingRepository")
@patch("packages.adapters.sqlalchemy_impl.session.SessionLocal")
def test_missing_user_id_returns_422(self, MockSession, MockRepo):
"""缺少 user_id 参数返回 422。"""
MockRepo.return_value = MockBillingRepository()
MockSession.return_value = MagicMock()
test_app = FastAPI()
test_app.include_router(router, prefix="/subscription")
client = TestClient(test_app)
resp = client.post(
"/subscription/payment-callback",
params={"plan": "pro", "billing_cycle": "monthly", "amount": 299.0},
)
assert resp.status_code == 422
@patch("packages.adapters.sqlalchemy_impl.billing_repository.SQLAlchemyBillingRepository")
@patch("packages.adapters.sqlalchemy_impl.session.SessionLocal")
def test_missing_plan_returns_422(self, MockSession, MockRepo):
"""缺少 plan 参数返回 422。"""
MockRepo.return_value = MockBillingRepository()
MockSession.return_value = MagicMock()
test_app = FastAPI()
test_app.include_router(router, prefix="/subscription")
client = TestClient(test_app)
resp = client.post(
"/subscription/payment-callback",
params={"user_id": "u1", "billing_cycle": "monthly", "amount": 299.0},
)
assert resp.status_code == 422
@patch("packages.adapters.sqlalchemy_impl.billing_repository.SQLAlchemyBillingRepository")
@patch("packages.adapters.sqlalchemy_impl.session.SessionLocal")
def test_missing_amount_returns_422(self, MockSession, MockRepo):
"""缺少 amount 参数返回 422。"""
MockRepo.return_value = MockBillingRepository()
MockSession.return_value = MagicMock()
test_app = FastAPI()
test_app.include_router(router, prefix="/subscription")
client = TestClient(test_app)
resp = client.post(
"/subscription/payment-callback",
params={"user_id": "u1", "plan": "pro", "billing_cycle": "monthly"},
)
assert resp.status_code == 422
@patch("packages.adapters.sqlalchemy_impl.billing_repository.SQLAlchemyBillingRepository")
@patch("packages.adapters.sqlalchemy_impl.session.SessionLocal")
def test_negative_amount(self, MockSession, MockRepo):
"""负数金额(当前实现不校验,记录此行为)。"""
mock_repo = MockBillingRepository()
MockRepo.return_value = mock_repo
MockSession.return_value = MagicMock()
test_app = FastAPI()
test_app.include_router(router, prefix="/subscription")
client = TestClient(test_app)
resp = client.post(
"/subscription/payment-callback",
params={
"user_id": "u1", "plan": "pro", "billing_cycle": "monthly",
"amount": -100.0,
},
)
# 当前实现未校验金额正负
assert resp.status_code in (200, 400, 500)
# ---------------------------------------------------------------------------
# 6. 辅助函数测试
# ---------------------------------------------------------------------------
class TestHelperFunctions:
"""订阅辅助函数测试。"""
def test_get_plan_name_all_plans(self):
"""所有套餐名称映射正确。"""
assert _get_plan_name("free") == "体验版"
assert _get_plan_name("standard") == "标准版"
assert _get_plan_name("pro") == "专业版"
assert _get_plan_name("enterprise") == "企业版"
def test_get_plan_name_unknown(self):
"""未知套餐返回「未知套餐」。"""
assert _get_plan_name("unknown") == "未知套餐"
assert _get_plan_name("") == "未知套餐"
def test_get_plan_price_all_combinations(self):
"""所有套餐价格映射正确。"""
assert _get_plan_price("free", "monthly") == 0
assert _get_plan_price("free", "yearly") == 0
assert _get_plan_price("standard", "monthly") == 99
assert _get_plan_price("standard", "yearly") == 999
assert _get_plan_price("pro", "monthly") == 299
assert _get_plan_price("pro", "yearly") == 2999
assert _get_plan_price("enterprise", "monthly") == 999
assert _get_plan_price("enterprise", "yearly") == 9999
def test_get_plan_price_unknown(self):
"""未知组合返回 0。"""
assert _get_plan_price("unknown", "monthly") == 0
assert _get_plan_price("pro", "weekly") == 0
# ---------------------------------------------------------------------------
# 7. Mock Billing Repository 单元测试
# ---------------------------------------------------------------------------
class TestMockBillingRepository:
"""Billing Repository 行为单元测试。"""
def test_create_record(self):
"""创建账单记录。"""
repo = MockBillingRepository()
record = repo.create(dict(
id="bill-001", user_id="user-001",
plan_name="Pro 专业版", amount=299.0,
billing_cycle="monthly", status="pending",
))
assert record.id == "bill-001"
assert record.status == "pending"
assert repo.created_count == 1
def test_find_by_id(self):
"""按 ID 查询账单。"""
repo = MockBillingRepository()
repo.create(dict(
id="bill-001", user_id="user-1",
plan_name="Pro", amount=299, billing_cycle="monthly", status="pending",
))
found = repo.find_by_id("bill-001")
assert found is not None
assert found.id == "bill-001"
not_found = repo.find_by_id("nonexistent")
assert not_found is None
def test_find_by_user(self):
"""按用户查询账单。"""
repo = MockBillingRepository()
repo.create(dict(id="b1", user_id="u1", plan_name="Pro", amount=299, billing_cycle="monthly", status="pending"))
repo.create(dict(id="b2", user_id="u1", plan_name="Standard", amount=99, billing_cycle="monthly", status="pending"))
repo.create(dict(id="b3", user_id="u2", plan_name="Pro", amount=299, billing_cycle="monthly", status="pending"))
user1_records = repo.find_by_user("u1")
assert len(user1_records) == 2
user2_records = repo.find_by_user("u2")
assert len(user2_records) == 1
def test_mark_paid_transitions_status(self):
"""mark_paid 正确转换状态。"""
repo = MockBillingRepository()
repo.create(dict(
id="bill-001", user_id="u1",
plan_name="Pro", amount=299, billing_cycle="monthly", status="pending",
))
result = repo.mark_paid("bill-001", "alipay", "pay-001")
assert result is True
record = repo.find_by_id("bill-001")
assert record.status == "paid"
assert record.payment_method == "alipay"
assert record.payment_id == "pay-001"
assert record.paid_at is not None
def test_mark_paid_idempotent(self):
"""mark_paid 对已支付账单幂等。"""
repo = MockBillingRepository()
repo.create(dict(
id="bill-001", user_id="u1",
plan_name="Pro", amount=299, billing_cycle="monthly", status="pending",
))
repo.mark_paid("bill-001", "alipay", "pay-001")
paid_at_first = repo.find_by_id("bill-001").paid_at
result = repo.mark_paid("bill-001", "alipay", "pay-001")
assert result is False
# paid_at 不应更新
assert repo.find_by_id("bill-001").paid_at == paid_at_first
def test_mark_paid_nonexistent_returns_false(self):
"""标记不存在的账单返回 False。"""
repo = MockBillingRepository()
result = repo.mark_paid("nonexistent", "alipay", "pay-001")
assert result is False
def test_update_subscription_on_payment(self):
"""支付成功后更新订阅。"""
repo = MockBillingRepository()
expires = datetime.now(timezone.utc) + timedelta(days=30)
repo.update_subscription_on_payment("user-001", "pro", expires)
assert repo.update_subscription_count == 1
assert "user-001" in repo.updated_subscriptions
sub = repo.updated_subscriptions["user-001"]
assert sub["plan"] == "pro"
assert sub["status"] == "active"
assert sub["expires_at"] == expires
if __name__ == "__main__":
pytest.main([__file__, "-v"])
@@ -177,14 +177,12 @@ class TestGeneratedVideosAPIAvailability:
def test_generated_videos_routes_registered(self):
"""成片库路由已注册到 router。"""
from app.api.routes.generated_videos import router as gv_router
from apps.api.app.api.router import api_router
# 直接检查 generated_videos router 自身注册的路由
paths = [r.path for r in gv_router.routes if hasattr(r, "path")]
assert len(paths) > 0, "generated_videos router 没有注册任何路由"
# 验证关键端点存在
assert "" in paths, "列表端点不存在"
assert "/{video_id}" in paths, "详情端点不存在"
# 检查 router 包含 generated-videos 路径
routes = [r for r in api_router.routes if hasattr(r, "path")]
gv_routes = [r for r in routes if "generated-videos" in r.path]
assert len(gv_routes) > 0, "generated-videos 路由未注册"
def test_generated_videos_list_endpoint_exists(self):
"""GET /generated-videos 端点存在。"""
+75 -169
View File
@@ -4,17 +4,8 @@
而 worker_app.db 会在导入时调用 ensure_database_exists() 尝试连接 PostgreSQL。
因此必须在 @patch 装饰器解析模块路径之前,将 worker_app.db 预注入 sys.modules。
注意:production code 使用 VoiceCloneWorkflowService(非直接 CosyVoiceService),
Celery bind=True 任务的底层函数签名为 (self, profile_id)
CosyVoiceService 在 voice_clone.py 中被实例化传入 workflow,必须 mock 防止真实初始化。
跨环境兼容:
Python 3.13 + Celery 5.4.0 → import 返回 Celery Proxy
→ _get_current_object() 返回 Task 实例 → .run 是 bound methodself 已绑定)
→ 调用方式:task.run(profile_id)retry mock 在 task.run.retry
Python 3.10 + Celery 5.4.0 → import 返回原始函数(装饰器未生效)
→ 签名 (self, profile_id),需手动传 mock_self
→ 调用方式:func(mock_self, profile_id)retry mock 在 mock_self.retry
Celery 5.x 中 @task(bind=True) 装饰后,task.run 是绑定方法(self 已绑定),
直接调用 task(profile_id) 即可,不需要手动传 self。
"""
from __future__ import annotations
@@ -40,121 +31,83 @@ from celery.exceptions import Retry
from packages.application.cosyvoice_service import (
CosyVoiceError,
CosyVoiceService,
CosyVoiceTimeoutError,
)
from packages.application.voice_clone.use_cases import VoiceCloneNotFoundError
from packages.domain.voice_clone_profile import VoiceCloneProfile, VoiceCloneStatus
def _make_mock_profile(
def _make_profile(
*,
voice_id: str = "voice-xyz",
status: str = "ready",
) -> MagicMock:
"""创建测试用 mock profile。"""
profile = MagicMock()
profile.voice_id = voice_id
status: VoiceCloneStatus = VoiceCloneStatus.PROCESSING,
metadata: dict | None = None,
) -> VoiceCloneProfile:
"""创建测试用 VoiceCloneProfile。"""
if metadata is None:
metadata = {"cosyvoice_task_id": "task-abc"}
profile = VoiceCloneProfile.create(
user_id="user-123",
name="测试音色",
source_audio_url="https://example.com/audio.wav",
max_retries=3,
metadata=metadata,
)
profile.status = status
return profile
def _resolve_task(task_obj):
"""解析 Celery 任务对象,返回 (callable, mock_self_or_none)。
跨环境兼容 Celery Proxy / Task 实例 / 原始函数三种情况。
Returns:
tuple: (callable, mock_self)
- Proxy/Task: callable 是 bound method task.runmock_self=None
- 原始函数: callable 是原始函数,mock_self 需由调用方提供
"""
# Case 1: Celery Proxy → 提取 Task 实例的 .runbound method
if hasattr(task_obj, "_get_current_object"):
real_task = task_obj._get_current_object()
return real_task.run, None
# Case 2: Celery Task 实例(非 Proxy
if hasattr(task_obj, "run") and hasattr(task_obj, "retry"):
return task_obj.run, None
# Case 3: 原始函数(CI 环境中装饰器未生效)
return task_obj, MagicMock()
# ── 成功场景 ──────────────────────────────────────────────
class TestProcessVoiceCloneSuccess:
"""测试成功场景。"""
@patch("worker_app.tasks.voice_clone.SessionLocal")
@patch("worker_app.tasks.voice_clone.CosyVoiceService")
@patch("worker_app.tasks.voice_clone.VoiceCloneWorkflowService")
@patch("worker_app.tasks.voice_clone.SQLAlchemyVoiceCloneProfileRepository")
def test_process_voice_clone_success(
self,
mock_repo_cls: MagicMock,
mock_workflow_cls: MagicMock,
mock_cosy_cls: MagicMock,
mock_session_local: MagicMock,
) -> None:
@patch("worker_app.tasks.voice_clone.CosyVoiceService")
def test_process_voice_clone_success(self, mock_service_cls: MagicMock, mock_repo_cls: MagicMock) -> None:
"""克隆成功:轮询返回 voice_idprofile 标记为 ready。"""
mock_session = MagicMock()
mock_repo = MagicMock()
mock_workflow = MagicMock()
mock_service = MagicMock(spec=CosyVoiceService)
profile = _make_profile()
mock_repo.get.return_value = profile
mock_repo.update.side_effect = lambda p: p
mock_repo_cls.return_value = mock_repo
mock_workflow_cls.return_value = mock_workflow
mock_result = _make_mock_profile(voice_id="voice-xyz")
mock_workflow.poll_and_process_clone.return_value = mock_result
mock_service.poll_clone_task.return_value = {"voice_id": "voice-xyz"}
mock_service_cls.return_value = mock_service
mock_session_local.return_value = mock_session
_mock_db_module.SessionLocal.return_value = mock_session
from worker_app.tasks.voice_clone import process_voice_clone
func, mock_self = _resolve_task(process_voice_clone)
args = (mock_self, "profile-123") if mock_self else ("profile-123",)
result = func(*args)
# bind=True → run 是绑定方法,直接调用 task(profile_id)
result = process_voice_clone("profile-123")
assert result["ok"] is True
assert result["voice_id"] == "voice-xyz"
mock_workflow.poll_and_process_clone.assert_called_once_with(
"profile-123",
timeout=300,
)
mock_service.poll_clone_task.assert_called_once_with("task-abc", timeout=300)
mock_session.commit.assert_called_once()
mock_session.close.assert_called_once()
@patch("worker_app.tasks.voice_clone.SessionLocal")
@patch("worker_app.tasks.voice_clone.CosyVoiceService")
@patch("worker_app.tasks.voice_clone.VoiceCloneWorkflowService")
@patch("worker_app.tasks.voice_clone.SQLAlchemyVoiceCloneProfileRepository")
def test_process_voice_clone_profile_not_found(
self,
mock_repo_cls: MagicMock,
mock_workflow_cls: MagicMock,
mock_cosy_cls: MagicMock,
mock_session_local: MagicMock,
) -> None:
"""profile 不存在时 workflow 抛 VoiceCloneNotFoundError,返回 failed。"""
@patch("worker_app.tasks.voice_clone.CosyVoiceService")
def test_process_voice_clone_profile_not_found(self, mock_service_cls: MagicMock, mock_repo_cls: MagicMock) -> None:
"""profile 不存在时返回 failed。"""
mock_session = MagicMock()
mock_repo = MagicMock()
mock_workflow = MagicMock()
mock_repo.get.return_value = None
mock_repo_cls.return_value = mock_repo
mock_workflow_cls.return_value = mock_workflow
mock_workflow.poll_and_process_clone.side_effect = VoiceCloneNotFoundError("Voice clone nonexistent not found")
mock_session_local.return_value = mock_session
_mock_db_module.SessionLocal.return_value = mock_session
from worker_app.tasks.voice_clone import process_voice_clone
func, mock_self = _resolve_task(process_voice_clone)
args = (mock_self, "nonexistent") if mock_self else ("nonexistent",)
result = func(*args)
result = process_voice_clone("nonexistent")
assert result["ok"] is False
assert "not found" in result["error"].lower()
mock_session.rollback.assert_called_once()
mock_session.close.assert_called_once()
@@ -164,49 +117,29 @@ class TestProcessVoiceCloneSuccess:
class TestProcessVoiceCloneTimeout:
"""测试超时场景。"""
@patch("worker_app.tasks.voice_clone.SessionLocal")
@patch("worker_app.tasks.voice_clone.CosyVoiceService")
@patch("worker_app.tasks.voice_clone.VoiceCloneWorkflowService")
@patch("worker_app.tasks.voice_clone.SQLAlchemyVoiceCloneProfileRepository")
def test_process_voice_clone_timeout_retries(
self,
mock_repo_cls: MagicMock,
mock_workflow_cls: MagicMock,
mock_cosy_cls: MagicMock,
mock_session_local: MagicMock,
) -> None:
@patch("worker_app.tasks.voice_clone.CosyVoiceService")
def test_process_voice_clone_timeout_retries(self, mock_service_cls: MagicMock, mock_repo_cls: MagicMock) -> None:
"""超时时调用 self.retry() 进行重试,Retry 异常向上传播。"""
mock_session = MagicMock()
mock_repo = MagicMock()
mock_workflow = MagicMock()
mock_service = MagicMock(spec=CosyVoiceService)
profile = _make_profile()
mock_repo.get.return_value = profile
mock_repo_cls.return_value = mock_repo
mock_workflow_cls.return_value = mock_workflow
mock_workflow.poll_and_process_clone.side_effect = CosyVoiceTimeoutError("任务超时")
mock_service.poll_clone_task.side_effect = CosyVoiceTimeoutError("任务超时")
mock_service_cls.return_value = mock_service
mock_session_local.return_value = mock_session
_mock_db_module.SessionLocal.return_value = mock_session
from worker_app.tasks.voice_clone import process_voice_clone
func, mock_self = _resolve_task(process_voice_clone)
# 设置 retry mock:根据环境不同,retry 在不同对象上
if mock_self is None:
# Proxy/Task 环境:retry 在 Task 实例上(func 是 bound method task.run
real_task = process_voice_clone._get_current_object()
mock_retry = MagicMock()
mock_retry.side_effect = Retry("retrying")
with patch.object(real_task, "retry", mock_retry):
with pytest.raises(Retry):
func("profile-123")
mock_retry.assert_called_once()
else:
# 原始函数环境:retry 在 mock_self 上
mock_self.retry.side_effect = Retry("retrying")
# mock task.retry 使其抛出 Retry(模拟 Celery 行为)
with patch.object(process_voice_clone, "retry", side_effect=Retry("retrying")):
with pytest.raises(Retry):
func(mock_self, "profile-123")
mock_self.retry.assert_called_once()
process_voice_clone("profile-123")
mock_session.rollback.assert_called_once()
mock_session.close.assert_called_once()
@@ -218,104 +151,77 @@ class TestProcessVoiceCloneTimeout:
class TestProcessVoiceCloneFailure:
"""测试失败场景。"""
@patch("worker_app.tasks.voice_clone.SessionLocal")
@patch("worker_app.tasks.voice_clone.CosyVoiceService")
@patch("worker_app.tasks.voice_clone.VoiceCloneWorkflowService")
@patch("worker_app.tasks.voice_clone.SQLAlchemyVoiceCloneProfileRepository")
def test_process_voice_clone_cosyvoice_error(
self,
mock_repo_cls: MagicMock,
mock_workflow_cls: MagicMock,
mock_cosy_cls: MagicMock,
mock_session_local: MagicMock,
) -> None:
@patch("worker_app.tasks.voice_clone.CosyVoiceService")
def test_process_voice_clone_cosyvoice_error(self, mock_service_cls: MagicMock, mock_repo_cls: MagicMock) -> None:
"""CosyVoice 错误:profile 标记为 failed。"""
mock_session = MagicMock()
mock_repo = MagicMock()
mock_workflow = MagicMock()
mock_service = MagicMock(spec=CosyVoiceService)
profile = _make_profile()
mock_repo.get.return_value = profile
mock_repo.update.side_effect = lambda p: p
mock_repo_cls.return_value = mock_repo
mock_workflow_cls.return_value = mock_workflow
mock_workflow.poll_and_process_clone.side_effect = CosyVoiceError("克隆失败")
mock_service.poll_clone_task.side_effect = CosyVoiceError("克隆失败")
mock_service_cls.return_value = mock_service
mock_session_local.return_value = mock_session
_mock_db_module.SessionLocal.return_value = mock_session
from worker_app.tasks.voice_clone import process_voice_clone
func, mock_self = _resolve_task(process_voice_clone)
args = (mock_self, "profile-123") if mock_self else ("profile-123",)
result = func(*args)
result = process_voice_clone("profile-123")
assert result["ok"] is False
assert "克隆失败" in result["error"]
mock_session.rollback.assert_called_once()
mock_session.close.assert_called_once()
@patch("worker_app.tasks.voice_clone.SessionLocal")
@patch("worker_app.tasks.voice_clone.CosyVoiceService")
@patch("worker_app.tasks.voice_clone.VoiceCloneWorkflowService")
@patch("worker_app.tasks.voice_clone.SQLAlchemyVoiceCloneProfileRepository")
def test_process_voice_clone_unexpected_error(
self,
mock_repo_cls: MagicMock,
mock_workflow_cls: MagicMock,
mock_cosy_cls: MagicMock,
mock_session_local: MagicMock,
) -> None:
@patch("worker_app.tasks.voice_clone.CosyVoiceService")
def test_process_voice_clone_unexpected_error(self, mock_service_cls: MagicMock, mock_repo_cls: MagicMock) -> None:
"""意外异常:profile 标记为 failed。"""
mock_session = MagicMock()
mock_repo = MagicMock()
mock_workflow = MagicMock()
mock_service = MagicMock(spec=CosyVoiceService)
profile = _make_profile()
mock_repo.get.return_value = profile
mock_repo.update.side_effect = lambda p: p
mock_repo_cls.return_value = mock_repo
mock_workflow_cls.return_value = mock_workflow
mock_workflow.poll_and_process_clone.side_effect = RuntimeError("未知错误")
mock_service.poll_clone_task.side_effect = RuntimeError("未知错误")
mock_service_cls.return_value = mock_service
mock_session_local.return_value = mock_session
_mock_db_module.SessionLocal.return_value = mock_session
from worker_app.tasks.voice_clone import process_voice_clone
func, mock_self = _resolve_task(process_voice_clone)
args = (mock_self, "profile-123") if mock_self else ("profile-123",)
result = func(*args)
result = process_voice_clone("profile-123")
assert result["ok"] is False
assert "未知错误" in result["error"]
mock_session.rollback.assert_called_once()
mock_session.close.assert_called_once()
@patch("worker_app.tasks.voice_clone.SessionLocal")
@patch("worker_app.tasks.voice_clone.CosyVoiceService")
@patch("worker_app.tasks.voice_clone.VoiceCloneWorkflowService")
@patch("worker_app.tasks.voice_clone.SQLAlchemyVoiceCloneProfileRepository")
def test_process_voice_clone_no_task_id(
self,
mock_repo_cls: MagicMock,
mock_workflow_cls: MagicMock,
mock_cosy_cls: MagicMock,
mock_session_local: MagicMock,
) -> None:
@patch("worker_app.tasks.voice_clone.CosyVoiceService")
def test_process_voice_clone_no_task_id(self, mock_service_cls: MagicMock, mock_repo_cls: MagicMock) -> None:
"""metadata 中没有 cosyvoice_task_id 时返回 failed。"""
mock_session = MagicMock()
mock_repo = MagicMock()
mock_workflow = MagicMock()
# 显式传入空 dict,确保没有 cosyvoice_task_id
profile = _make_profile(metadata={})
mock_repo.get.return_value = profile
mock_repo.update.side_effect = lambda p: p
mock_repo_cls.return_value = mock_repo
mock_workflow_cls.return_value = mock_workflow
mock_workflow.poll_and_process_clone.side_effect = CosyVoiceError("missing task_id")
mock_session_local.return_value = mock_session
_mock_db_module.SessionLocal.return_value = mock_session
from worker_app.tasks.voice_clone import process_voice_clone
func, mock_self = _resolve_task(process_voice_clone)
args = (mock_self, "profile-123") if mock_self else ("profile-123",)
result = func(*args)
result = process_voice_clone("profile-123")
assert result["ok"] is False
assert "task_id" in result["error"]
mock_session.rollback.assert_called_once()
mock_session.close.assert_called_once()