Files
xiaoxia-saas/packages/domain/asset_atom_clip.py
xiaoxia 7e88440ca9
CI/CD Pipeline / Check push changed paths (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (pull_request) Successful in 2s
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 3s
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (push) Successful in 2s
CI/CD Pipeline / Validate - Style (pull_request) Has been skipped
CI/CD Pipeline / Validate - Security (pull_request) Has been skipped
CI/CD Pipeline / Validate - Python (mypy + alembic) (pull_request) Has been skipped
CI/CD Pipeline / Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / Check push changed paths (push) Successful in 4s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
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
CI/CD Pipeline / Frontend Lint (push) Has been skipped
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 2m27s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 2m25s
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 / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
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 / Canary Release to Production (pull_request) Has been skipped
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m11s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 3m40s
AI Code Review / AI Code Review (pull_request) Successful in 6m41s
CI/CD Pipeline / Integration Tests (push) Failing after 7m5s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 7m20s
CI/CD Pipeline / CI Gate (pull_request) Successful in 2s
CI/CD Pipeline / Validate - Style (push) Successful in 8m25s
CI/CD Pipeline / Validate - Python (mypy + alembic) (push) Failing after 13m44s
CI/CD Pipeline / Unit Tests (push) Successful in 14m12s
CI/CD Pipeline / Validate - Security (push) Has been cancelled
CI/CD Pipeline / Retag skipped Staging API Image (push) Has been cancelled
CI/CD Pipeline / Retag skipped Staging Web Image (push) Has been cancelled
CI/CD Pipeline / Retag skipped Staging Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled
CI/CD Pipeline / Build Production API Image (push) Has been cancelled
CI/CD Pipeline / Build Production Web Image (push) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Production (push) Has been cancelled
CI/CD Pipeline / Production Browser E2E (push) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (push) Has been cancelled
CI/CD Pipeline / Canary Release to Production (push) Has been cancelled
CI/CD Pipeline / CI Gate (push) Has been cancelled
CI/CD Pipeline / Build Staging API Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Web Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Worker Image (push) Has been cancelled
feat: #1970 片段级 AI 标签 + 叙事加权匹配 + 冗余核查 (#1981)
feat: #1970 片段级 AI 标签 + 叙事加权匹配

- atom_clip_tagger.py: MediaKit 抽帧 + 豆包视觉 API 识别
- narrative_match.py: AI 标签加权匹配 (2.0 vs 1.0)
- Celery 链式触发 + 批量回填脚本
- migration 081 加 ai_tags 列
- 42 新测试,全量 15796 passed
2026-09-18 21:08:01 +08:00

83 lines
2.9 KiB
Python

"""素材原子片段(Atom Clip)领域实体 — #1970 智能剪辑流程重构。
原子片段是素材的逻辑切分单元,不物理切割视频文件。
每条记录指向某条素材的一段 [start_time, end_time] 区间。
"""
from __future__ import annotations
import uuid
from dataclasses import dataclass, field
from datetime import UTC, datetime
@dataclass
class AssetAtomClip:
"""素材原子片段。
Attributes:
id: 唯一标识。
asset_id: 所属素材 ID。
start_time: 片段起始时间(秒,浮点)。
end_time: 片段结束时间(秒,浮点)。
duration: 片段时长 = end_time - start_time(秒)。
clip_index: 在同一素材内的顺序编号(从 0 开始)。
tags: 继承自素材的标签,JSONB 存储,可为空列表。
scene_change_at: 片段尾部是否对齐了 scdet 镜头切换点(存储该切点的精确时间),
未对齐时为 None。
is_fallback: 是否为兜底逻辑在内存中生成的临时片段(不入库)。
created_at: 创建时间。
"""
id: str
asset_id: str
start_time: float
end_time: float
duration: float
clip_index: int
tags: list[str] = field(default_factory=list)
ai_tags: dict | None = None
scene_change_at: float | None = None
is_fallback: bool = False
created_at: datetime | None = None
def __post_init__(self):
if not self.id:
self.id = str(uuid.uuid4())
if self.duration <= 0:
self.duration = round(self.end_time - self.start_time, 3)
if self.duration < 0:
raise ValueError(f"duration must be >= 0, got start={self.start_time}, end={self.end_time}")
if self.start_time < 0:
raise ValueError(f"start_time must be >= 0, got {self.start_time}")
if self.end_time <= self.start_time:
raise ValueError(f"end_time must be > start_time, got start={self.start_time}, end={self.end_time}")
if self.clip_index < 0:
raise ValueError(f"clip_index must be >= 0, got {self.clip_index}")
if self.created_at is None:
self.created_at = datetime.now(UTC)
@classmethod
def create(
cls,
asset_id: str,
start_time: float,
end_time: float,
clip_index: int,
tags: list[str] | None = None,
scene_change_at: float | None = None,
is_fallback: bool = False,
) -> AssetAtomClip:
"""工厂方法:创建一个新的原子片段。"""
return cls(
id="", # __post_init__ 会自动生成
asset_id=asset_id,
start_time=round(start_time, 3),
end_time=round(end_time, 3),
duration=round(end_time - start_time, 3),
clip_index=clip_index,
tags=tags or [],
scene_change_at=scene_change_at,
is_fallback=is_fallback,
)