API文档维护Agent
312dad7497
六边形架构分层实现：
Domain 层:
- DuplicationRecord / DuplicateSegment 实体
- 状态机生命周期 (pending→processing→completed/failed)
Ports 层:
- DuplicationRecordRepository Protocol
Adapters 层:
- SQLAlchemyDuplicationRecordRepository 完整实现
- DuplicationRecordModel / DuplicationSegmentModel ORM 模型
Application 层:
- UploadForDuplicationUseCase (上传查重)
- ListDuplicationRecordsUseCase (记录列表)
- GetDuplicationDetailUseCase (详情查询)
- DeleteDuplicationRecordUseCase (删除)
- RetryDuplicationUseCase (重新查重)
API 层:
- POST /api/v1/duplication/upload (上传视频查重)
- GET /api/v1/duplication/records (记录列表)
- GET /api/v1/duplication/records/{id} (详情含片段)
- DELETE /api/v1/duplication/records/{id} (删除)
- POST /api/v1/duplication/records/{id}/retry (重新查重)
Alembic 迁移:
- 011_add_duplication_tables.py (duplication_records + duplication_segments)
API 契约与前端 feat/phase2-duplication-ui (PR#75) 完全对齐。
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
91 lines
3.1 KiB
Python
91 lines
3.1 KiB
Python
"""查重应用层用例。"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from dataclasses import dataclass
|
|
|
|
from packages.domain.duplication import DuplicationRecord
|
|
from packages.ports.duplication_repository import DuplicationRecordRepository
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class UploadForDuplicationCommand:
|
|
user_id: str
|
|
filename: str
|
|
file_size: int
|
|
storage_key: str
|
|
duration_seconds: float = 0.0
|
|
|
|
|
|
class UploadForDuplicationUseCase:
|
|
"""上传视频进行查重 — 创建查重记录并返回。"""
|
|
|
|
def __init__(self, duplication_repository: DuplicationRecordRepository):
|
|
self.duplication_repository = duplication_repository
|
|
|
|
def execute(self, command: UploadForDuplicationCommand) -> DuplicationRecord:
|
|
record = DuplicationRecord.create(
|
|
user_id=command.user_id,
|
|
filename=command.filename,
|
|
file_size=command.file_size,
|
|
storage_key=command.storage_key,
|
|
duration_seconds=command.duration_seconds,
|
|
)
|
|
record = self.duplication_repository.create(record)
|
|
logger.info("Duplication record created: %s for user %s", record.id, record.user_id)
|
|
return record
|
|
|
|
|
|
class ListDuplicationRecordsUseCase:
|
|
"""获取用户查重记录列表。"""
|
|
|
|
def __init__(self, duplication_repository: DuplicationRecordRepository):
|
|
self.duplication_repository = duplication_repository
|
|
|
|
def execute(self, user_id: str, *, offset: int = 0, limit: int = 50) -> list[DuplicationRecord]:
|
|
if not user_id.strip():
|
|
raise ValueError("user_id 不能为空")
|
|
return self.duplication_repository.list_by_user(user_id.strip(), offset=offset, limit=limit)
|
|
|
|
|
|
class GetDuplicationDetailUseCase:
|
|
"""获取查重记录详情（含重复片段）。"""
|
|
|
|
def __init__(self, duplication_repository: DuplicationRecordRepository):
|
|
self.duplication_repository = duplication_repository
|
|
|
|
def execute(self, record_id: str) -> DuplicationRecord | None:
|
|
return self.duplication_repository.get(record_id)
|
|
|
|
|
|
class DeleteDuplicationRecordUseCase:
|
|
"""删除查重记录。"""
|
|
|
|
def __init__(self, duplication_repository: DuplicationRecordRepository):
|
|
self.duplication_repository = duplication_repository
|
|
|
|
def execute(self, record_id: str) -> bool:
|
|
return self.duplication_repository.delete(record_id)
|
|
|
|
|
|
class RetryDuplicationUseCase:
|
|
"""重新提交查重 — 将记录状态重置为 pending。"""
|
|
|
|
def __init__(self, duplication_repository: DuplicationRecordRepository):
|
|
self.duplication_repository = duplication_repository
|
|
|
|
def execute(self, record_id: str) -> DuplicationRecord | None:
|
|
record = self.duplication_repository.get(record_id)
|
|
if record is None:
|
|
return None
|
|
record.status = "pending"
|
|
record.error_message = ""
|
|
record.duplicate_rate = None
|
|
record.duplicate_count = 0
|
|
record.segments = []
|
|
record = self.duplication_repository.update(record)
|
|
logger.info("Duplication record %s reset to pending for retry", record_id)
|
|
return record
|