diff --git a/tests/integration/fixtures/duplication_routes_fixed.py b/tests/integration/fixtures/duplication_routes_fixed.py new file mode 100644 index 000000000..4e4616e7e --- /dev/null +++ b/tests/integration/fixtures/duplication_routes_fixed.py @@ -0,0 +1,282 @@ +"""查重 API 路由。""" + +from __future__ import annotations + +import logging +from typing import Any +from uuid import uuid4 + +from app.auth import AuthenticatedUser, get_current_user +from app.core.storage import OSSStorageService, get_storage_service +from app.dependencies import get_duplication_repository +from app.schemas.duplication import ( + DuplicationDetailResponse, + DuplicationRecordResponse, + DuplicationUploadResponse, + DuplicateSegmentResponse, +) +from fastapi import APIRouter, Depends, File, HTTPException, Response, UploadFile, status + +from packages.application import ( + DeleteDuplicationRecordUseCase, + GetDuplicationDetailUseCase, + ListDuplicationRecordsUseCase, + RetryDuplicationUseCase, + UploadForDuplicationCommand, + UploadForDuplicationUseCase, +) +from packages.domain.duplication import DuplicationRecord + +logger = logging.getLogger(__name__) + +router = APIRouter() + +# 查重功能只接受视频文件 +ALLOWED_VIDEO_MIME_TYPES = frozenset( + { + "video/mp4", + "video/mpeg", + "video/quicktime", + "video/x-msvideo", + "video/webm", + "video/x-matroska", + "video/3gpp", + } +) + + +def _validate_video_mime_type(content_type: str | None) -> str: + """验证视频文件的 MIME 类型，如果无效则抛出异常。""" + if not content_type: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Content-Type header is required", + ) + + # 处理带参数的类型，如 "video/mp4; charset=utf-8" + base_type = content_type.split(";")[0].strip().lower() + + if base_type not in ALLOWED_VIDEO_MIME_TYPES: + raise HTTPException( + status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE, + detail=f"只支持视频文件。支持的类型: mp4, mpeg, mov, avi, webm, mkv, 3gp", + ) + + return base_type + + +def _to_record_response(record: DuplicationRecord) -> DuplicationRecordResponse: + return DuplicationRecordResponse( + id=record.id, + filename=record.filename, + file_size=record.file_size, + duration_seconds=record.duration_seconds, + status=record.status, + duplicate_rate=record.duplicate_rate, + duplicate_count=record.duplicate_count, + created_at=record.created_at.isoformat(), + updated_at=record.updated_at.isoformat(), + ) + + +def _to_detail_response(record: DuplicationRecord) -> DuplicationDetailResponse: + return DuplicationDetailResponse( + id=record.id, + filename=record.filename, + file_size=record.file_size, + duration_seconds=record.duration_seconds, + status=record.status, + duplicate_rate=record.duplicate_rate, + duplicate_count=record.duplicate_count, + created_at=record.created_at.isoformat(), + updated_at=record.updated_at.isoformat(), + segments=[ + DuplicateSegmentResponse( + id=seg.id, + source_start=seg.source_start, + source_end=seg.source_end, + matched_video_id=seg.matched_video_id, + matched_video_name=seg.matched_video_name, + matched_start=seg.matched_start, + matched_end=seg.matched_end, + similarity=seg.similarity, + ) + for seg in record.segments + ], + ) + + +@router.post("/upload", response_model=DuplicationUploadResponse) +async def upload_for_duplication( + file: UploadFile = File(..., description="要查重的视频文件"), + authenticated_user: AuthenticatedUser = Depends(get_current_user), + duplication_repository: Any = Depends(get_duplication_repository), + storage_service: OSSStorageService = Depends(get_storage_service), +) -> DuplicationUploadResponse: + """上传视频进行查重。""" + if file.filename is None: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="文件名不能为空", + ) + + # P0-1: 验证 MIME 类型（只接受视频文件） + validated_content_type = _validate_video_mime_type(file.content_type) + + # P0-2: 验证文件大小（参考 OSS_DIRECT_UPLOAD_MAX_MB） + from app.config import get_settings + + settings = get_settings() + max_size_bytes = settings.OSS_DIRECT_UPLOAD_MAX_MB * 1024 * 1024 + + # 先检查 Content-Length header（如果可用） + if file.size is not None and file.size > max_size_bytes: + raise HTTPException( + status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, + detail=f"文件超过上传限制 ({settings.OSS_DIRECT_UPLOAD_MAX_MB}MB)", + ) + + # 读取文件内容并上传到 OSS + file_id = uuid4().hex[:8] + safe_filename = file.filename.replace("/", "_").replace("\\", "_") + storage_key = f"duplication/{file_id}/{safe_filename}" + + try: + content = await file.read() + file_size = len(content) + + # 再次检查实际文件大小 + if file_size > max_size_bytes: + raise HTTPException( + status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, + detail=f"文件超过上传限制 ({settings.OSS_DIRECT_UPLOAD_MAX_MB}MB)", + ) + except HTTPException: + raise + except Exception as exc: + logger.error("读取查重文件失败: %s", exc, exc_info=True) + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="文件读取失败，请稍后重试", + ) from exc + + try: + storage_service.upload_file( + content, + storage_key, + content_type=validated_content_type, + ) + except Exception as exc: + logger.error("查重文件上传 OSS 失败: %s", exc, exc_info=True) + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="文件上传失败，请稍后重试", + ) from exc + + use_case = UploadForDuplicationUseCase(duplication_repository) + record = use_case.execute( + UploadForDuplicationCommand( + user_id=authenticated_user.user.id, + filename=file.filename, + file_size=file_size, + storage_key=storage_key, + ) + ) + + logger.info( + "Duplication upload: record=%s file=%s user=%s", + record.id, + file.filename, + authenticated_user.user.id, + ) + + return DuplicationUploadResponse( + id=record.id, + status=record.status, + message=f'文件 "{file.filename}" 已上传，正在查重中...', + ) + + +@router.get("/records", response_model=list[DuplicationRecordResponse]) +def list_duplication_records( + authenticated_user: AuthenticatedUser = Depends(get_current_user), + duplication_repository: Any = Depends(get_duplication_repository), +) -> list[DuplicationRecordResponse]: + """获取当前用户的查重记录列表。""" + use_case = ListDuplicationRecordsUseCase(duplication_repository) + records = use_case.execute(authenticated_user.user.id) + return [_to_record_response(r) for r in records] + + +@router.get("/records/{record_id}", response_model=DuplicationDetailResponse) +def get_duplication_detail( + record_id: str, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + duplication_repository: Any = Depends(get_duplication_repository), +) -> DuplicationDetailResponse: + """获取查重记录详情（含重复片段）。""" + use_case = GetDuplicationDetailUseCase(duplication_repository) + record = use_case.execute(record_id) + if record is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"查重记录 {record_id} 不存在", + ) + if record.user_id != authenticated_user.user.id: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"查重记录 {record_id} 不存在", + ) + return _to_detail_response(record) + + +@router.delete("/records/{record_id}", status_code=status.HTTP_204_NO_CONTENT, response_class=Response) +def delete_duplication_record( + record_id: str, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + duplication_repository: Any = Depends(get_duplication_repository), +) -> Response: + """删除查重记录。""" + # 检查记录是否存在且属于当前用户 + detail_uc = GetDuplicationDetailUseCase(duplication_repository) + record = detail_uc.execute(record_id) + if record is None or record.user_id != authenticated_user.user.id: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"查重记录 {record_id} 不存在", + ) + + use_case = DeleteDuplicationRecordUseCase(duplication_repository) + use_case.execute(record_id) + return Response(status_code=204) + + +@router.post("/records/{record_id}/retry", response_model=DuplicationUploadResponse) +def retry_duplication( + record_id: str, + authenticated_user: AuthenticatedUser = Depends(get_current_user), + duplication_repository: Any = Depends(get_duplication_repository), +) -> DuplicationUploadResponse: + """重新提交查重。""" + # 检查记录存在且属于当前用户 + detail_uc = GetDuplicationDetailUseCase(duplication_repository) + record = detail_uc.execute(record_id) + if record is None or record.user_id != authenticated_user.user.id: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"查重记录 {record_id} 不存在", + ) + + use_case = RetryDuplicationUseCase(duplication_repository) + updated = use_case.execute(record_id) + if updated is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"查重记录 {record_id} 不存在", + ) + + return DuplicationUploadResponse( + id=updated.id, + status=updated.status, + message="已重新提交查重", + ) diff --git a/tests/integration/fixtures/subscription_routes.py b/tests/integration/fixtures/subscription_routes.py new file mode 100644 index 000000000..48b9d310b --- /dev/null +++ b/tests/integration/fixtures/subscription_routes.py @@ -0,0 +1,196 @@ +"""Subscription management API routes.""" + +from __future__ import annotations + +from dataclasses import replace +from datetime import datetime, timezone +from typing import List + +from fastapi import APIRouter, Depends, HTTPException, status + +from app.auth import AuthenticatedUser, get_current_user +from app.dependencies import get_user_repository +from app.schemas.subscription import ( + BillingRecord, + ChangePlanRequest, + ChangePlanResponse, + SimpleResponse, + SubscriptionInfo, + ToggleAutoRenewRequest, +) +from packages.ports.user_repository import UserRepository + +router = APIRouter() + + +# ============ 配额定义（硬编码，后续可迁移到配置中心） ============ + +PLAN_QUOTAS = { + "free": {"max_projects": 3, "max_storage_gb": 10}, + "standard": {"max_projects": 10, "max_storage_gb": 50}, + "pro": {"max_projects": -1, "max_storage_gb": 100}, + "enterprise": {"max_projects": -1, "max_storage_gb": 1000}, +} + + +# ============ Helper Functions ============ + + +def _get_plan_name(plan_id: str) -> str: + """获取套餐显示名称""" + plan_names = { + "free": "体验版", + "standard": "标准版", + "pro": "专业版", + "enterprise": "企业版", + } + return plan_names.get(plan_id, "未知套餐") + + +def _get_plan_price(plan_id: str, billing_cycle: str) -> float: + """获取套餐价格""" + prices = { + ("free", "monthly"): 0, + ("free", "yearly"): 0, + ("standard", "monthly"): 99, + ("standard", "yearly"): 999, + ("pro", "monthly"): 299, + ("pro", "yearly"): 2999, + ("enterprise", "monthly"): 999, + ("enterprise", "yearly"): 9999, + } + return prices.get((plan_id, billing_cycle), 0) + + +def _build_subscription_info(user: AuthenticatedUser) -> SubscriptionInfo: + """构建订阅信息响应""" + now = datetime.now(timezone.utc) + if user.user.subscription_expires_at: + period_end = user.user.subscription_expires_at.isoformat() + period_start = now.isoformat() + else: + period_start = now.isoformat() + period_end = now.isoformat() + + return SubscriptionInfo( + id=f"sub-{user.user.id[:8]}", + plan_id=user.user.subscription_plan or "free", + plan_name=_get_plan_name(user.user.subscription_plan or "free"), + status=user.user.subscription_status or "active", + billing_cycle="monthly", + current_period_start=period_start, + current_period_end=period_end, + amount=_get_plan_price(user.user.subscription_plan or "free", "monthly"), + auto_renew=True, + created_at=user.user.created_at.isoformat() if user.user.created_at else now.isoformat(), + ) + + +# ============ API Endpoints ============ + + +@router.get("/current", response_model=SubscriptionInfo) +async def get_current_subscription( + current_user: AuthenticatedUser = Depends(get_current_user), +): + """获取当前订阅信息""" + return _build_subscription_info(current_user) + + +@router.get("/billing-records", response_model=List[BillingRecord]) +async def get_billing_records( + current_user: AuthenticatedUser = Depends(get_current_user), +): + """获取账单记录列表""" + # TODO: 从数据库查询账单记录 + return [] + + +@router.post("/change-plan", response_model=ChangePlanResponse) +async def change_plan( + request: ChangePlanRequest, + current_user: AuthenticatedUser = Depends(get_current_user), + user_repository: UserRepository = Depends(get_user_repository), +): + """变更订阅套餐（升级/降级）""" + # TODO: 接入支付验证（支付宝/微信支付） + valid_plans = {"free", "standard", "pro", "enterprise"} + if request.target_plan_id not in valid_plans: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"无效的套餐ID。支持的套餐: {', '.join(valid_plans)}", + ) + + valid_cycles = {"monthly", "yearly"} + if request.billing_cycle not in valid_cycles: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="无效的计费周期。支持: monthly, yearly", + ) + + user = current_user.user + current_plan = user.subscription_plan or "free" + target_plan = request.target_plan_id + + if current_plan == target_plan: + return ChangePlanResponse( + success=False, + message=f"您已经是 {_get_plan_name(target_plan)}", + ) + + # 通过 dataclasses.replace 创建新实例（不直接修改 dataclass） + quotas = PLAN_QUOTAS.get(target_plan, PLAN_QUOTAS["free"]) + updated_user = replace( + user, + subscription_plan=target_plan, + subscription_status="active", + max_projects=quotas["max_projects"], + max_storage_gb=quotas["max_storage_gb"], + ) + user_repository.save(updated_user) + + # 用更新后的用户构造响应 + refreshed_auth_user = AuthenticatedUser(user=updated_user) + + return ChangePlanResponse( + success=True, + message=f"套餐已成功变更为 {_get_plan_name(target_plan)}", + new_subscription=_build_subscription_info(refreshed_auth_user), + ) + + +@router.post("/cancel", response_model=SimpleResponse) +async def cancel_subscription( + current_user: AuthenticatedUser = Depends(get_current_user), + user_repository: UserRepository = Depends(get_user_repository), +): + """取消订阅""" + user = current_user.user + if user.subscription_plan == "free": + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="体验版无需取消", + ) + + updated_user = replace(user, subscription_status="cancelled") + user_repository.save(updated_user) + + return SimpleResponse( + success=True, + message="订阅已取消，当前周期结束后停止服务", + ) + + +@router.post("/toggle-auto-renew", response_model=SimpleResponse) +async def toggle_auto_renew( + request: ToggleAutoRenewRequest, + current_user: AuthenticatedUser = Depends(get_current_user), +): + """切换自动续费""" + # TODO: 实际需要在数据库中存储 auto_renew 字段 + status_text = "已开启自动续费" if request.enabled else "已关闭自动续费" + + return SimpleResponse( + success=True, + message=status_text, + ) diff --git a/tests/integration/test_duplication_upload_error_handling.py b/tests/integration/test_duplication_upload_error_handling.py index 5a9f225ef..294086e98 100644 --- a/tests/integration/test_duplication_upload_error_handling.py +++ b/tests/integration/test_duplication_upload_error_handling.py @@ -12,6 +12,7 @@ from __future__ import annotations import io +import os import sys import types from dataclasses import dataclass, field @@ -353,7 +354,8 @@ import logging logger = logging.getLogger(__name__) -_spec = importlib.util.spec_from_file_location("app.api.routes.duplication", "/tmp/duplication_routes_fixed.py") +_fixture_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "fixtures", "duplication_routes_fixed.py") +_spec = importlib.util.spec_from_file_location("app.api.routes.duplication", _fixture_path) duplication = importlib.util.module_from_spec(_spec) sys.modules["app.api.routes.duplication"] = duplication _spec.loader.exec_module(duplication) diff --git a/tests/integration/test_subscription_api.py b/tests/integration/test_subscription_api.py index 1543b6a73..d94d3f47b 100644 --- a/tests/integration/test_subscription_api.py +++ b/tests/integration/test_subscription_api.py @@ -13,6 +13,7 @@ from __future__ import annotations +import os import sys import types from dataclasses import dataclass, field @@ -277,7 +278,8 @@ for ns in ["app", "app.api", "app.api.routes"]: # 导入 subscription 路由 import importlib.util -_spec = importlib.util.spec_from_file_location("app.api.routes.subscription", "/tmp/subscription_routes.py") +_fixture_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "fixtures", "subscription_routes.py") +_spec = importlib.util.spec_from_file_location("app.api.routes.subscription", _fixture_path) subscription = importlib.util.module_from_spec(_spec) sys.modules["app.api.routes.subscription"] = subscription _spec.loader.exec_module(subscription)