Files
xiaoxia-saas/packages/domain/generated_video.py
T
xiaoxia a40de0338e
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (pull_request) Successful in 15s
CI/CD Pipeline / Check push changed paths (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 2m54s
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
AI Code Review / AI Code Review (pull_request) Failing after 4m9s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 3m6s
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 3m39s
CI/CD Pipeline / Retag skipped Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 3m29s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 2m46s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 7m14s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 3m33s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 6m42s
CI/CD Pipeline / Validate - Code Quality (pull_request) Has been cancelled
CI/CD Pipeline / Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled
feat: 成片 duplicate_rate 百分比 + 素材高频使用自动排除
任务一:生成视频补 duplicate_rate 百分比
- GeneratedVideoModel 加 duplicate_rate 列 (Float, nullable)
- GeneratedVideo 领域实体 + repository 映射同步更新
- Alembic migration 059
- VideoDeduplicator.compute_duplicate_rate:遍历项目内已有视频,
  MD5精确=100%,pHash用 (1-avg_distance/64)*100,取最高值
- dedup_helpers 在指纹计算后调用并存入 duplicate_rate
- VideoItemResponse schema 加 duplicate_rate 字段
- _to_video_response 透传 duplicate_rate

任务二:素材级自动排除高频使用素材
- asset_segment_tracker.get_asset_recent_use_counts:
  统计每个素材在最近 N 个不同 plan_id 中的使用次数
- smart_match_assets 路由加高频排除层(阈值 3 次 / 最近 5 个视频)
- 排除后不够 limit 时自动放宽,宁多不少
- 异常时降级跳过排除,不影响正常选素材

测试:12 个新用例覆盖 duplicate_rate 计算 + API 响应 + 高频统计
2026-08-31 15:14:34 +08:00

73 lines
2.2 KiB
Python
Executable File

from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any
from uuid import uuid4
@dataclass(slots=True)
class GeneratedVideo:
id: str
project_id: str
generation_task_id: str
name: str
file_url: str
file_size: int
duration: float
width: int
height: int
fps: float
user_id: str = ""
thumbnail_url: str | None = None
status: str = "completed"
review_status: str = "pending_review"
generation_params: dict[str, Any] = field(default_factory=dict)
video_fingerprint: dict[str, Any] | None = None
is_duplicate: bool = False
duplicate_of: str | None = None
duplicate_rate: float | None = None
generated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
@classmethod
def create(
cls,
project_id: str,
generation_task_id: str,
name: str,
file_url: str,
*,
user_id: str = "",
file_size: int = 0,
duration: float = 0.0,
width: int = 0,
height: int = 0,
fps: float = 0.0,
thumbnail_url: str | None = None,
generation_params: dict[str, Any] | None = None,
) -> "GeneratedVideo":
if not project_id.strip():
raise ValueError("project_id cannot be empty")
if not generation_task_id.strip():
raise ValueError("generation_task_id cannot be empty")
if not name.strip():
raise ValueError("name cannot be empty")
if not file_url.strip():
raise ValueError("file_url cannot be empty")
return cls(
id=uuid4().hex,
project_id=project_id.strip(),
user_id=user_id.strip(),
generation_task_id=generation_task_id.strip(),
name=name.strip(),
file_url=file_url.strip(),
file_size=file_size,
duration=duration,
width=width,
height=height,
fps=fps,
thumbnail_url=thumbnail_url,
generation_params=generation_params or {},
)