灵应
ee8c60a143
1. 修复 hamming_distance 不等长哈希处理（hex() 前导零丢失） 2. 集成颜色直方图到 check_duplicate（此前计算但未使用，浪费 CPU） 3. check_duplicate 改为返回最佳匹配而非首个匹配 4. 修复仓库删除顺序（先删片段再删记录，防止孤儿数据） 5. 域模型添加 can_retry()/reset_for_retry()，仅 failed 状态允许重试 6. 列表接口暴露 offset/limit 分页参数
106 lines
3.4 KiB
Python
106 lines
3.4 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:
|
|
"""
|
|
重新提交查重 — 将 failed 状态的记录重置为 pending。
|
|
|
|
仅 failed 状态允许重试，其他状态抛出 ValueError。
|
|
"""
|
|
|
|
def __init__(self, duplication_repository: DuplicationRecordRepository):
|
|
self.duplication_repository = duplication_repository
|
|
|
|
def execute(self, record_id: str) -> DuplicationRecord | None:
|
|
"""
|
|
重试查重。
|
|
|
|
Args:
|
|
record_id: 查重记录 ID
|
|
|
|
Returns:
|
|
重置后的记录；不存在返回 None
|
|
|
|
Raises:
|
|
ValueError: 记录状态不允许重试（非 failed）
|
|
"""
|
|
record = self.duplication_repository.get(record_id)
|
|
if record is None:
|
|
return None
|
|
if not record.can_retry():
|
|
raise ValueError(f"只有 failed 状态的记录可以重试，当前状态: {record.status}")
|
|
record.reset_for_retry()
|
|
record = self.duplication_repository.update(record)
|
|
logger.info("Duplication record %s reset to pending for retry", record_id)
|
|
return record
|