feat(phase8-task202): TemplateClipConfig + EditPlanClip 数据模型 #143
@@ -0,0 +1,81 @@
|
||||
"""Phase 8: Create template_clip_configs and edit_plan_clips tables
|
||||
|
||||
Revision ID: 017
|
||||
Revises: 016
|
||||
Create Date: 2026-07-01
|
||||
|
||||
新增两张表：
|
||||
- template_clip_configs: 模板片段配置（定义模板中每个片段的规则）
|
||||
- edit_plan_clips: 剪辑计划片段（剪辑计划中的具体片段实例）
|
||||
"""
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
revision = "017"
|
||||
down_revision = "016"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
# template_clip_configs: 模板片段配置表
|
||||
op.create_table(
|
||||
"template_clip_configs",
|
||||
sa.Column("id", sa.String(32), primary_key=True),
|
||||
sa.Column("template_id", sa.String(32), nullable=False, index=True),
|
||||
sa.Column("clip_type", sa.String(20), nullable=False, index=True),
|
||||
sa.Column("order", sa.Integer, nullable=False),
|
||||
sa.Column("min_duration", sa.Float, nullable=False, server_default="0.0"),
|
||||
sa.Column("max_duration", sa.Float, nullable=False, server_default="0.0"),
|
||||
sa.Column("text_template", sa.Text, nullable=False, server_default=""),
|
||||
sa.Column("material_requirements", sa.JSON, nullable=False, server_default="{}"),
|
||||
sa.Column("transition_effect", sa.String(20), nullable=False, server_default="cut"),
|
||||
sa.Column("config", sa.JSON, nullable=False, server_default="{}"),
|
||||
sa.Column(
|
||||
"created_at",
|
||||
sa.DateTime,
|
||||
nullable=False,
|
||||
server_default=sa.func.now(),
|
||||
),
|
||||
sa.Column(
|
||||
"updated_at",
|
||||
sa.DateTime,
|
||||
nullable=False,
|
||||
server_default=sa.func.now(),
|
||||
),
|
||||
)
|
||||
|
||||
# edit_plan_clips: 剪辑计划片段表
|
||||
op.create_table(
|
||||
"edit_plan_clips",
|
||||
sa.Column("id", sa.String(32), primary_key=True),
|
||||
sa.Column("plan_id", sa.String(32), nullable=False, index=True),
|
||||
sa.Column("clip_type", sa.String(20), nullable=False, index=True),
|
||||
sa.Column("order", sa.Integer, nullable=False),
|
||||
sa.Column("template_clip_config_id", sa.String(32), nullable=False, server_default="", index=True),
|
||||
sa.Column("asset_id", sa.String(32), nullable=False, server_default="", index=True),
|
||||
sa.Column("text_content", sa.Text, nullable=False, server_default=""),
|
||||
sa.Column("start_time", sa.Float, nullable=False, server_default="0.0"),
|
||||
sa.Column("duration", sa.Float, nullable=False, server_default="0.0"),
|
||||
sa.Column("transition_effect", sa.String(20), nullable=False, server_default="cut"),
|
||||
sa.Column("status", sa.String(20), nullable=False, server_default="pending", index=True),
|
||||
sa.Column("config", sa.JSON, nullable=False, server_default="{}"),
|
||||
sa.Column(
|
||||
"created_at",
|
||||
sa.DateTime,
|
||||
nullable=False,
|
||||
server_default=sa.func.now(),
|
||||
),
|
||||
sa.Column(
|
||||
"updated_at",
|
||||
sa.DateTime,
|
||||
nullable=False,
|
||||
server_default=sa.func.now(),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_table("edit_plan_clips")
|
||||
op.drop_table("template_clip_configs")
|
||||
@@ -3,6 +3,7 @@
|
||||
from .asset_library_repository import SQLAlchemyAssetLibraryRepository
|
||||
from .asset_repository import SQLAlchemyAssetRepository
|
||||
from .classification_job_repository import SQLAlchemyClassificationJobRepository
|
||||
from .edit_plan_clip_repository import SQLAlchemyEditPlanClipRepository
|
||||
from .edit_plan_repository import SQLAlchemyEditPlanRepository
|
||||
from .edit_template_repository import SQLAlchemyEditTemplateRepository
|
||||
from .generated_video_repository import SQLAlchemyGeneratedVideoRepository
|
||||
@@ -16,18 +17,21 @@ from .session import (
|
||||
ensure_database_exists,
|
||||
initialize_database,
|
||||
)
|
||||
from .template_clip_config_repository import SQLAlchemyTemplateClipConfigRepository
|
||||
|
||||
__all__ = [
|
||||
"Base",
|
||||
"SQLAlchemyAssetLibraryRepository",
|
||||
"SQLAlchemyAssetRepository",
|
||||
"SQLAlchemyClassificationJobRepository",
|
||||
"SQLAlchemyEditPlanClipRepository",
|
||||
"SQLAlchemyEditPlanRepository",
|
||||
"SQLAlchemyEditTemplateRepository",
|
||||
"SQLAlchemyGeneratedVideoRepository",
|
||||
"SQLAlchemyGenerationTaskRepository",
|
||||
"SQLAlchemyIngestJobRepository",
|
||||
"SQLAlchemyProjectRepository",
|
||||
"SQLAlchemyTemplateClipConfigRepository",
|
||||
"build_engine",
|
||||
"build_session_factory",
|
||||
"ensure_database_exists",
|
||||
|
||||
@@ -0,0 +1,143 @@
|
||||
"""SQLAlchemy implementation of EditPlanClipRepository."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import List, Optional
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from packages.adapters.sqlalchemy_impl.models import EditPlanClipModel
|
||||
from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus
|
||||
|
||||
|
||||
class SQLAlchemyEditPlanClipRepository:
|
||||
"""SQLAlchemy 剪辑计划片段仓储"""
|
||||
|
||||
def __init__(self, session: Session) -> None:
|
||||
self.session = session
|
||||
|
||||
def list_by_plan(
|
||||
self,
|
||||
plan_id: str,
|
||||
*,
|
||||
status: Optional[EditPlanClipStatus] = None,
|
||||
skip: int = 0,
|
||||
limit: int = 100,
|
||||
) -> List[EditPlanClip]:
|
||||
"""按计划列出片段，按 order 排序"""
|
||||
query = self.session.query(EditPlanClipModel).filter(
|
||||
EditPlanClipModel.plan_id == plan_id,
|
||||
)
|
||||
if status:
|
||||
query = query.filter(EditPlanClipModel.status == status)
|
||||
query = query.order_by(EditPlanClipModel.order)
|
||||
models = query.offset(skip).limit(limit).all()
|
||||
return [self._model_to_entity(m) for m in models]
|
||||
|
||||
def get(self, clip_id: str) -> Optional[EditPlanClip]:
|
||||
"""根据 ID 获取片段"""
|
||||
model = (
|
||||
self.session.query(EditPlanClipModel)
|
||||
.filter(EditPlanClipModel.id == clip_id)
|
||||
.first()
|
||||
)
|
||||
if model is None:
|
||||
return None
|
||||
return self._model_to_entity(model)
|
||||
|
||||
def create(self, clip: EditPlanClip) -> EditPlanClip:
|
||||
"""创建片段"""
|
||||
model = EditPlanClipModel(
|
||||
id=clip.id,
|
||||
plan_id=clip.plan_id,
|
||||
clip_type=clip.clip_type,
|
||||
order=clip.order,
|
||||
template_clip_config_id=clip.template_clip_config_id,
|
||||
asset_id=clip.asset_id,
|
||||
text_content=clip.text_content,
|
||||
start_time=clip.start_time,
|
||||
duration=clip.duration,
|
||||
transition_effect=clip.transition_effect,
|
||||
status=clip.status,
|
||||
config=clip.config,
|
||||
)
|
||||
self.session.add(model)
|
||||
self.session.commit()
|
||||
self.session.refresh(model)
|
||||
return self._model_to_entity(model)
|
||||
|
||||
def update(self, clip: EditPlanClip) -> EditPlanClip:
|
||||
"""更新片段"""
|
||||
model = (
|
||||
self.session.query(EditPlanClipModel)
|
||||
.filter(EditPlanClipModel.id == clip.id)
|
||||
.first()
|
||||
)
|
||||
if model is None:
|
||||
raise ValueError(f"EditPlanClip {clip.id} not found")
|
||||
model.plan_id = clip.plan_id
|
||||
model.clip_type = clip.clip_type
|
||||
model.order = clip.order
|
||||
model.template_clip_config_id = clip.template_clip_config_id
|
||||
model.asset_id = clip.asset_id
|
||||
model.text_content = clip.text_content
|
||||
model.start_time = clip.start_time
|
||||
model.duration = clip.duration
|
||||
model.transition_effect = clip.transition_effect
|
||||
model.status = clip.status
|
||||
model.config = clip.config
|
||||
model.updated_at = clip.updated_at
|
||||
self.session.commit()
|
||||
self.session.refresh(model)
|
||||
return self._model_to_entity(model)
|
||||
|
||||
def delete(self, clip_id: str) -> bool:
|
||||
"""删除片段"""
|
||||
model = (
|
||||
self.session.query(EditPlanClipModel)
|
||||
.filter(EditPlanClipModel.id == clip_id)
|
||||
.first()
|
||||
)
|
||||
if model is None:
|
||||
return False
|
||||
self.session.delete(model)
|
||||
self.session.commit()
|
||||
return True
|
||||
|
||||
def delete_by_plan(self, plan_id: str) -> int:
|
||||
"""删除计划下所有片段，返回删除数量"""
|
||||
count = (
|
||||
self.session.query(EditPlanClipModel)
|
||||
.filter(EditPlanClipModel.plan_id == plan_id)
|
||||
.delete()
|
||||
)
|
||||
self.session.commit()
|
||||
return count
|
||||
|
||||
def count(self, *, plan_id: Optional[str] = None, status: Optional[EditPlanClipStatus] = None) -> int:
|
||||
"""统计片段数量"""
|
||||
query = self.session.query(EditPlanClipModel)
|
||||
if plan_id:
|
||||
query = query.filter(EditPlanClipModel.plan_id == plan_id)
|
||||
if status:
|
||||
query = query.filter(EditPlanClipModel.status == status)
|
||||
return query.count()
|
||||
|
||||
@staticmethod
|
||||
def _model_to_entity(model: EditPlanClipModel) -> EditPlanClip:
|
||||
return EditPlanClip(
|
||||
id=model.id,
|
||||
plan_id=model.plan_id,
|
||||
clip_type=model.clip_type,
|
||||
order=model.order,
|
||||
template_clip_config_id=model.template_clip_config_id or "",
|
||||
asset_id=model.asset_id or "",
|
||||
text_content=model.text_content or "",
|
||||
start_time=model.start_time or 0.0,
|
||||
duration=model.duration or 0.0,
|
||||
transition_effect=model.transition_effect or "cut",
|
||||
status=EditPlanClipStatus(model.status) if model.status else EditPlanClipStatus.PENDING,
|
||||
config=model.config or {},
|
||||
created_at=model.created_at,
|
||||
updated_at=model.updated_at,
|
||||
)
|
||||
@@ -123,6 +123,52 @@ class EditPlanModel(Base):
|
||||
updated_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
|
||||
|
||||
|
||||
class TemplateClipConfigModel(Base):
|
||||
"""Phase 8 模板片段配置 ORM 模型
|
||||
|
||||
定义模板中每个片段的规则：类型、时长范围、文案模板、素材要求、转场效果。
|
||||
"""
|
||||
|
||||
__tablename__ = "template_clip_configs"
|
||||
|
||||
id = Column(String(32), primary_key=True)
|
||||
template_id = Column(String(32), nullable=False, index=True)
|
||||
clip_type = Column(String(20), nullable=False, index=True)
|
||||
order = Column(Integer, nullable=False)
|
||||
min_duration = Column(Float, nullable=False, default=0.0)
|
||||
max_duration = Column(Float, nullable=False, default=0.0)
|
||||
text_template = Column(Text, nullable=False, default="")
|
||||
material_requirements = Column(JSON, nullable=False, default=dict)
|
||||
transition_effect = Column(String(20), nullable=False, default="cut")
|
||||
config = Column(JSON, nullable=False, default=dict)
|
||||
created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
|
||||
updated_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
|
||||
|
||||
|
||||
class EditPlanClipModel(Base):
|
||||
"""Phase 8 剪辑计划片段 ORM 模型
|
||||
|
||||
剪辑计划中的具体片段实例，关联 EditPlan 和可选的 TemplateClipConfig。
|
||||
"""
|
||||
|
||||
__tablename__ = "edit_plan_clips"
|
||||
|
||||
id = Column(String(32), primary_key=True)
|
||||
plan_id = Column(String(32), nullable=False, index=True)
|
||||
clip_type = Column(String(20), nullable=False, index=True)
|
||||
order = Column(Integer, nullable=False)
|
||||
template_clip_config_id = Column(String(32), nullable=False, default="", index=True)
|
||||
asset_id = Column(String(32), nullable=False, default="", index=True)
|
||||
text_content = Column(Text, nullable=False, default="")
|
||||
start_time = Column(Float, nullable=False, default=0.0)
|
||||
duration = Column(Float, nullable=False, default=0.0)
|
||||
transition_effect = Column(String(20), nullable=False, default="cut")
|
||||
status = Column(String(20), nullable=False, default="pending", index=True)
|
||||
config = Column(JSON, nullable=False, default=dict)
|
||||
created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
|
||||
updated_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
|
||||
|
||||
|
||||
class IngestJobModel(Base):
|
||||
__tablename__ = "ingest_jobs"
|
||||
|
||||
|
||||
@@ -0,0 +1,141 @@
|
||||
"""SQLAlchemy implementation of TemplateClipConfigRepository."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import List, Optional
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from packages.adapters.sqlalchemy_impl.models import TemplateClipConfigModel
|
||||
from packages.domain.template_clip_config import (
|
||||
ClipType,
|
||||
TemplateClipConfig,
|
||||
TransitionEffect,
|
||||
)
|
||||
|
||||
|
||||
class SQLAlchemyTemplateClipConfigRepository:
|
||||
"""SQLAlchemy 模板片段配置仓储"""
|
||||
|
||||
def __init__(self, session: Session) -> None:
|
||||
self.session = session
|
||||
|
||||
def list_by_template(
|
||||
self,
|
||||
template_id: str,
|
||||
*,
|
||||
clip_type: Optional[ClipType] = None,
|
||||
skip: int = 0,
|
||||
limit: int = 100,
|
||||
) -> List[TemplateClipConfig]:
|
||||
"""按模板列出片段配置，按 order 排序"""
|
||||
query = self.session.query(TemplateClipConfigModel).filter(
|
||||
TemplateClipConfigModel.template_id == template_id,
|
||||
)
|
||||
if clip_type:
|
||||
query = query.filter(TemplateClipConfigModel.clip_type == clip_type)
|
||||
query = query.order_by(TemplateClipConfigModel.order)
|
||||
models = query.offset(skip).limit(limit).all()
|
||||
return [self._model_to_entity(m) for m in models]
|
||||
|
||||
def get(self, config_id: str) -> Optional[TemplateClipConfig]:
|
||||
"""根据 ID 获取配置"""
|
||||
model = (
|
||||
self.session.query(TemplateClipConfigModel)
|
||||
.filter(TemplateClipConfigModel.id == config_id)
|
||||
.first()
|
||||
)
|
||||
if model is None:
|
||||
return None
|
||||
return self._model_to_entity(model)
|
||||
|
||||
def create(self, config: TemplateClipConfig) -> TemplateClipConfig:
|
||||
"""创建配置"""
|
||||
model = TemplateClipConfigModel(
|
||||
id=config.id,
|
||||
template_id=config.template_id,
|
||||
clip_type=config.clip_type,
|
||||
order=config.order,
|
||||
min_duration=config.min_duration,
|
||||
max_duration=config.max_duration,
|
||||
text_template=config.text_template,
|
||||
material_requirements=config.material_requirements,
|
||||
transition_effect=config.transition_effect,
|
||||
config=config.config,
|
||||
)
|
||||
self.session.add(model)
|
||||
self.session.commit()
|
||||
self.session.refresh(model)
|
||||
return self._model_to_entity(model)
|
||||
|
||||
def update(self, config: TemplateClipConfig) -> TemplateClipConfig:
|
||||
"""更新配置"""
|
||||
model = (
|
||||
self.session.query(TemplateClipConfigModel)
|
||||
.filter(TemplateClipConfigModel.id == config.id)
|
||||
.first()
|
||||
)
|
||||
if model is None:
|
||||
raise ValueError(f"TemplateClipConfig {config.id} not found")
|
||||
model.template_id = config.template_id
|
||||
model.clip_type = config.clip_type
|
||||
model.order = config.order
|
||||
model.min_duration = config.min_duration
|
||||
model.max_duration = config.max_duration
|
||||
model.text_template = config.text_template
|
||||
model.material_requirements = config.material_requirements
|
||||
model.transition_effect = config.transition_effect
|
||||
model.config = config.config
|
||||
model.updated_at = config.updated_at
|
||||
self.session.commit()
|
||||
self.session.refresh(model)
|
||||
return self._model_to_entity(model)
|
||||
|
||||
def delete(self, config_id: str) -> bool:
|
||||
"""删除配置"""
|
||||
model = (
|
||||
self.session.query(TemplateClipConfigModel)
|
||||
.filter(TemplateClipConfigModel.id == config_id)
|
||||
.first()
|
||||
)
|
||||
if model is None:
|
||||
return False
|
||||
self.session.delete(model)
|
||||
self.session.commit()
|
||||
return True
|
||||
|
||||
def delete_by_template(self, template_id: str) -> int:
|
||||
"""删除模板下所有片段配置，返回删除数量"""
|
||||
count = (
|
||||
self.session.query(TemplateClipConfigModel)
|
||||
.filter(TemplateClipConfigModel.template_id == template_id)
|
||||
.delete()
|
||||
)
|
||||
self.session.commit()
|
||||
return count
|
||||
|
||||
def count(self, *, template_id: Optional[str] = None) -> int:
|
||||
"""统计配置数量"""
|
||||
query = self.session.query(TemplateClipConfigModel)
|
||||
if template_id:
|
||||
query = query.filter(TemplateClipConfigModel.template_id == template_id)
|
||||
return query.count()
|
||||
|
||||
@staticmethod
|
||||
def _model_to_entity(model: TemplateClipConfigModel) -> TemplateClipConfig:
|
||||
return TemplateClipConfig(
|
||||
id=model.id,
|
||||
template_id=model.template_id,
|
||||
clip_type=ClipType(model.clip_type),
|
||||
order=model.order,
|
||||
min_duration=model.min_duration or 0.0,
|
||||
max_duration=model.max_duration or 0.0,
|
||||
text_template=model.text_template or "",
|
||||
material_requirements=model.material_requirements or {},
|
||||
transition_effect=TransitionEffect(model.transition_effect)
|
||||
if model.transition_effect
|
||||
else TransitionEffect.CUT,
|
||||
config=model.config or {},
|
||||
created_at=model.created_at,
|
||||
updated_at=model.updated_at,
|
||||
)
|
||||
@@ -6,9 +6,10 @@ from .classification import (
|
||||
ClassificationJobStatus,
|
||||
)
|
||||
from .duplication import DuplicateSegment, DuplicationRecord
|
||||
from .editing_mode import EditingMode
|
||||
from .edit_plan import EditPlan, EditPlanStatus
|
||||
from .edit_plan_clip import EditPlanClip, EditPlanClipStatus
|
||||
from .edit_template import EditTemplate, EditTemplateStatus
|
||||
from .editing_mode import EditingMode
|
||||
from .entities import (
|
||||
Asset,
|
||||
AssetLibrary,
|
||||
@@ -22,6 +23,7 @@ from .entities import (
|
||||
)
|
||||
from .generated_video import GeneratedVideo
|
||||
from .generation_task import GenerationTask, GenerationTaskStatus
|
||||
from .template_clip_config import ClipType, TemplateClipConfig, TransitionEffect
|
||||
from .title_library import TitleLibraryItem
|
||||
from .voice_library import VoiceLibraryItem
|
||||
|
||||
@@ -34,19 +36,24 @@ __all__ = [
|
||||
"ClassificationJob",
|
||||
"ClassificationJobStatus",
|
||||
"ClassificationStatus",
|
||||
"ClipType",
|
||||
"DuplicateSegment",
|
||||
"DuplicationRecord",
|
||||
"EditingMode",
|
||||
"EditPlan",
|
||||
"EditPlanClip",
|
||||
"EditPlanClipStatus",
|
||||
"EditPlanStatus",
|
||||
"EditTemplate",
|
||||
"EditTemplateStatus",
|
||||
"EditingMode",
|
||||
"GeneratedVideo",
|
||||
"GenerationTask",
|
||||
"GenerationTaskStatus",
|
||||
"IngestJob",
|
||||
"IngestJobStatus",
|
||||
"Project",
|
||||
"TemplateClipConfig",
|
||||
"TransitionEffect",
|
||||
"User",
|
||||
"TitleLibraryItem",
|
||||
"VoiceLibraryItem",
|
||||
|
||||
@@ -0,0 +1,142 @@
|
||||
"""EditPlanClip domain entity for Phase 8 模板编排引擎.
|
||||
|
||||
剪辑计划中的具体片段实例，关联 EditPlan 和 TemplateClipConfig，
|
||||
包含实际素材、实际文案、排序和时长等信息。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime, timezone
|
||||
|
||||
if sys.version_info >= (3, 11):
|
||||
from enum import StrEnum
|
||||
else:
|
||||
from enum import Enum
|
||||
|
||||
class StrEnum(str, Enum):
|
||||
pass
|
||||
|
||||
|
||||
from typing import Any
|
||||
from uuid import uuid4
|
||||
|
||||
|
||||
class EditPlanClipStatus(StrEnum):
|
||||
"""片段状态"""
|
||||
|
||||
PENDING = "pending" # 待处理（素材未就绪）
|
||||
READY = "ready" # 就绪（素材已就绪，可渲染）
|
||||
RENDERED = "rendered" # 已渲染
|
||||
FAILED = "failed" # 渲染失败
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class EditPlanClip:
|
||||
"""剪辑计划片段
|
||||
|
||||
表示 EditPlan 中的一个具体片段实例，包含实际素材、文案和渲染状态。
|
||||
可选地关联 TemplateClipConfig 以继承模板规则。
|
||||
"""
|
||||
|
||||
id: str
|
||||
plan_id: str
|
||||
clip_type: str
|
||||
order: int
|
||||
template_clip_config_id: str = ""
|
||||
asset_id: str = ""
|
||||
text_content: str = ""
|
||||
start_time: float = 0.0
|
||||
duration: float = 0.0
|
||||
transition_effect: str = "cut"
|
||||
status: EditPlanClipStatus = EditPlanClipStatus.PENDING
|
||||
config: dict[str, Any] = field(default_factory=dict)
|
||||
created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
|
||||
updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
|
||||
|
||||
@classmethod
|
||||
def create(
|
||||
cls,
|
||||
plan_id: str,
|
||||
clip_type: str,
|
||||
order: int,
|
||||
*,
|
||||
template_clip_config_id: str = "",
|
||||
asset_id: str = "",
|
||||
text_content: str = "",
|
||||
start_time: float = 0.0,
|
||||
duration: float = 0.0,
|
||||
transition_effect: str = "cut",
|
||||
config: dict[str, Any] | None = None,
|
||||
) -> EditPlanClip:
|
||||
"""创建剪辑计划片段"""
|
||||
if not plan_id.strip():
|
||||
raise ValueError("plan_id 不能为空")
|
||||
if not clip_type.strip():
|
||||
raise ValueError("clip_type 不能为空")
|
||||
if start_time < 0:
|
||||
raise ValueError("start_time 不能为负数")
|
||||
if duration < 0:
|
||||
raise ValueError("duration 不能为负数")
|
||||
|
||||
return cls(
|
||||
id=uuid4().hex,
|
||||
plan_id=plan_id.strip(),
|
||||
clip_type=clip_type.strip(),
|
||||
order=order,
|
||||
template_clip_config_id=template_clip_config_id.strip()
|
||||
if template_clip_config_id
|
||||
else "",
|
||||
asset_id=asset_id.strip() if asset_id else "",
|
||||
text_content=text_content.strip(),
|
||||
start_time=start_time,
|
||||
duration=duration,
|
||||
transition_effect=transition_effect.strip() or "cut",
|
||||
status=EditPlanClipStatus.PENDING,
|
||||
config=config or {},
|
||||
)
|
||||
|
||||
def assign_asset(self, asset_id: str) -> None:
|
||||
"""分配素材"""
|
||||
if not asset_id.strip():
|
||||
raise ValueError("asset_id 不能为空")
|
||||
self.asset_id = asset_id.strip()
|
||||
self.updated_at = datetime.now(timezone.utc)
|
||||
|
||||
def mark_ready(self) -> None:
|
||||
"""标记为就绪"""
|
||||
if self.status != EditPlanClipStatus.PENDING:
|
||||
raise ValueError(
|
||||
f"只有 pending 状态的片段可以标记就绪，当前状态: {self.status}"
|
||||
)
|
||||
self.status = EditPlanClipStatus.READY
|
||||
self.updated_at = datetime.now(timezone.utc)
|
||||
|
||||
def mark_rendered(self) -> None:
|
||||
"""标记为已渲染"""
|
||||
if self.status != EditPlanClipStatus.READY:
|
||||
raise ValueError(
|
||||
f"只有 ready 状态的片段可以标记已渲染，当前状态: {self.status}"
|
||||
)
|
||||
self.status = EditPlanClipStatus.RENDERED
|
||||
self.updated_at = datetime.now(timezone.utc)
|
||||
|
||||
def mark_failed(self) -> None:
|
||||
"""标记为失败"""
|
||||
if self.status != EditPlanClipStatus.READY:
|
||||
raise ValueError(
|
||||
f"只有 ready 状态的片段可以标记失败，当前状态: {self.status}"
|
||||
)
|
||||
self.status = EditPlanClipStatus.FAILED
|
||||
self.updated_at = datetime.now(timezone.utc)
|
||||
|
||||
@property
|
||||
def end_time(self) -> float:
|
||||
"""片段结束时间"""
|
||||
return self.start_time + self.duration
|
||||
|
||||
@property
|
||||
def has_asset(self) -> bool:
|
||||
"""是否已分配素材"""
|
||||
return bool(self.asset_id)
|
||||
@@ -0,0 +1,115 @@
|
||||
"""TemplateClipConfig domain entity for Phase 8 模板编排引擎.
|
||||
|
||||
定义模板中每个片段的配置规则：片段类型、时长范围、文案模板、素材要求、转场效果等。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime, timezone
|
||||
|
||||
if sys.version_info >= (3, 11):
|
||||
from enum import StrEnum
|
||||
else:
|
||||
from enum import Enum
|
||||
|
||||
class StrEnum(str, Enum):
|
||||
pass
|
||||
|
||||
|
||||
from typing import Any
|
||||
from uuid import uuid4
|
||||
|
||||
|
||||
class ClipType(StrEnum):
|
||||
"""片段类型"""
|
||||
|
||||
INTRO = "intro" # 片头
|
||||
MAIN = "main" # 主体片段
|
||||
TRANSITION = "transition" # 转场片段
|
||||
OUTRO = "outro" # 片尾
|
||||
TITLE = "title" # 标题卡
|
||||
SUBTITLE = "subtitle" # 字幕卡
|
||||
|
||||
|
||||
class TransitionEffect(StrEnum):
|
||||
"""转场效果"""
|
||||
|
||||
CUT = "cut" # 硬切
|
||||
FADE = "fade" # 淡入淡出
|
||||
SLIDE_LEFT = "slide_left" # 左滑
|
||||
SLIDE_RIGHT = "slide_right" # 右滑
|
||||
DISSOLVE = "dissolve" # 溶解
|
||||
WIPE = "wipe" # 擦除
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class TemplateClipConfig:
|
||||
"""模板片段配置
|
||||
|
||||
定义模板中单个片段的规则，包括类型、时长范围、文案模板、素材要求和转场效果。
|
||||
属于 EditTemplate 的子实体。
|
||||
"""
|
||||
|
||||
id: str
|
||||
template_id: str
|
||||
clip_type: ClipType
|
||||
order: int
|
||||
min_duration: float = 0.0
|
||||
max_duration: float = 0.0
|
||||
text_template: str = ""
|
||||
material_requirements: dict[str, Any] = field(default_factory=dict)
|
||||
transition_effect: TransitionEffect = TransitionEffect.CUT
|
||||
config: dict[str, Any] = field(default_factory=dict)
|
||||
created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
|
||||
updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
|
||||
|
||||
@classmethod
|
||||
def create(
|
||||
cls,
|
||||
template_id: str,
|
||||
clip_type: ClipType | str,
|
||||
order: int,
|
||||
*,
|
||||
min_duration: float = 0.0,
|
||||
max_duration: float = 0.0,
|
||||
text_template: str = "",
|
||||
material_requirements: dict[str, Any] | None = None,
|
||||
transition_effect: TransitionEffect | str = TransitionEffect.CUT,
|
||||
config: dict[str, Any] | None = None,
|
||||
) -> TemplateClipConfig:
|
||||
"""创建模板片段配置"""
|
||||
if not template_id.strip():
|
||||
raise ValueError("template_id 不能为空")
|
||||
if min_duration < 0:
|
||||
raise ValueError("min_duration 不能为负数")
|
||||
if max_duration < 0:
|
||||
raise ValueError("max_duration 不能为负数")
|
||||
if max_duration > 0 and min_duration > max_duration:
|
||||
raise ValueError("min_duration 不能大于 max_duration")
|
||||
|
||||
return cls(
|
||||
id=uuid4().hex,
|
||||
template_id=template_id.strip(),
|
||||
clip_type=ClipType(clip_type),
|
||||
order=order,
|
||||
min_duration=min_duration,
|
||||
max_duration=max_duration,
|
||||
text_template=text_template.strip(),
|
||||
material_requirements=material_requirements or {},
|
||||
transition_effect=TransitionEffect(transition_effect),
|
||||
config=config or {},
|
||||
)
|
||||
|
||||
@property
|
||||
def has_duration_range(self) -> bool:
|
||||
"""是否设置了时长范围"""
|
||||
return self.min_duration > 0 or self.max_duration > 0
|
||||
|
||||
@property
|
||||
def default_duration(self) -> float:
|
||||
"""默认时长（取中间值）"""
|
||||
if self.min_duration > 0 and self.max_duration > 0:
|
||||
return (self.min_duration + self.max_duration) / 2
|
||||
return self.max_duration or self.min_duration or 0.0
|
||||
@@ -0,0 +1,536 @@
|
||||
"""Unit tests for Phase 8 任务 2.02: TemplateClipConfig + EditPlanClip 数据模型."""
|
||||
|
||||
import pytest
|
||||
|
||||
from packages.domain.template_clip_config import (
|
||||
ClipType,
|
||||
TemplateClipConfig,
|
||||
TransitionEffect,
|
||||
)
|
||||
from packages.domain.edit_plan_clip import (
|
||||
EditPlanClip,
|
||||
EditPlanClipStatus,
|
||||
)
|
||||
|
||||
|
||||
# ── TemplateClipConfig 领域实体测试 ─────────────────────────────────────────
|
||||
|
||||
|
||||
class TestTemplateClipConfig:
|
||||
"""TemplateClipConfig 领域实体测试"""
|
||||
|
||||
def test_create_basic(self):
|
||||
"""基本创建"""
|
||||
config = TemplateClipConfig.create(
|
||||
template_id="tpl_001",
|
||||
clip_type=ClipType.INTRO,
|
||||
order=0,
|
||||
)
|
||||
assert config.template_id == "tpl_001"
|
||||
assert config.clip_type == ClipType.INTRO
|
||||
assert config.order == 0
|
||||
assert config.min_duration == 0.0
|
||||
assert config.max_duration == 0.0
|
||||
assert config.text_template == ""
|
||||
assert config.material_requirements == {}
|
||||
assert config.transition_effect == TransitionEffect.CUT
|
||||
assert config.config == {}
|
||||
assert config.id # 自动生成 ID
|
||||
|
||||
def test_create_with_all_fields(self):
|
||||
"""完整字段创建"""
|
||||
config = TemplateClipConfig.create(
|
||||
template_id="tpl_001",
|
||||
clip_type=ClipType.MAIN,
|
||||
order=1,
|
||||
min_duration=3.0,
|
||||
max_duration=10.0,
|
||||
text_template="欢迎使用{product_name}",
|
||||
material_requirements={"type": "video", "min_resolution": "1080p"},
|
||||
transition_effect=TransitionEffect.FADE,
|
||||
config={"speed": 1.0},
|
||||
)
|
||||
assert config.min_duration == 3.0
|
||||
assert config.max_duration == 10.0
|
||||
assert config.text_template == "欢迎使用{product_name}"
|
||||
assert config.material_requirements == {"type": "video", "min_resolution": "1080p"}
|
||||
assert config.transition_effect == TransitionEffect.FADE
|
||||
assert config.config == {"speed": 1.0}
|
||||
|
||||
def test_create_with_string_enum_values(self):
|
||||
"""字符串枚举值创建"""
|
||||
config = TemplateClipConfig.create(
|
||||
template_id="tpl_001",
|
||||
clip_type="outro",
|
||||
order=2,
|
||||
transition_effect="slide_left",
|
||||
)
|
||||
assert config.clip_type == ClipType.OUTRO
|
||||
assert config.transition_effect == TransitionEffect.SLIDE_LEFT
|
||||
|
||||
def test_create_empty_template_id_raises(self):
|
||||
"""空 template_id 报错"""
|
||||
with pytest.raises(ValueError, match="template_id 不能为空"):
|
||||
TemplateClipConfig.create(template_id="", clip_type=ClipType.INTRO, order=0)
|
||||
|
||||
def test_create_whitespace_template_id_raises(self):
|
||||
"""空白 template_id 报错"""
|
||||
with pytest.raises(ValueError, match="template_id 不能为空"):
|
||||
TemplateClipConfig.create(template_id=" ", clip_type=ClipType.INTRO, order=0)
|
||||
|
||||
def test_create_negative_min_duration_raises(self):
|
||||
"""负数 min_duration 报错"""
|
||||
with pytest.raises(ValueError, match="min_duration 不能为负数"):
|
||||
TemplateClipConfig.create(
|
||||
template_id="tpl_001", clip_type=ClipType.INTRO, order=0, min_duration=-1.0
|
||||
)
|
||||
|
||||
def test_create_negative_max_duration_raises(self):
|
||||
"""负数 max_duration 报错"""
|
||||
with pytest.raises(ValueError, match="max_duration 不能为负数"):
|
||||
TemplateClipConfig.create(
|
||||
template_id="tpl_001", clip_type=ClipType.INTRO, order=0, max_duration=-1.0
|
||||
)
|
||||
|
||||
def test_create_min_greater_than_max_raises(self):
|
||||
"""min_duration > max_duration 报错"""
|
||||
with pytest.raises(ValueError, match="min_duration 不能大于 max_duration"):
|
||||
TemplateClipConfig.create(
|
||||
template_id="tpl_001",
|
||||
clip_type=ClipType.INTRO,
|
||||
order=0,
|
||||
min_duration=10.0,
|
||||
max_duration=5.0,
|
||||
)
|
||||
|
||||
def test_has_duration_range(self):
|
||||
"""has_duration_range 属性"""
|
||||
config_no_range = TemplateClipConfig.create(
|
||||
template_id="tpl_001", clip_type=ClipType.INTRO, order=0
|
||||
)
|
||||
assert config_no_range.has_duration_range is False
|
||||
|
||||
config_with_range = TemplateClipConfig.create(
|
||||
template_id="tpl_001", clip_type=ClipType.INTRO, order=0, min_duration=3.0, max_duration=10.0
|
||||
)
|
||||
assert config_with_range.has_duration_range is True
|
||||
|
||||
def test_default_duration(self):
|
||||
"""default_duration 属性"""
|
||||
# 无时长范围
|
||||
config_no_range = TemplateClipConfig.create(
|
||||
template_id="tpl_001", clip_type=ClipType.INTRO, order=0
|
||||
)
|
||||
assert config_no_range.default_duration == 0.0
|
||||
|
||||
# 只有 min
|
||||
config_min_only = TemplateClipConfig.create(
|
||||
template_id="tpl_001", clip_type=ClipType.INTRO, order=0, min_duration=5.0
|
||||
)
|
||||
assert config_min_only.default_duration == 5.0
|
||||
|
||||
# 只有 max
|
||||
config_max_only = TemplateClipConfig.create(
|
||||
template_id="tpl_001", clip_type=ClipType.INTRO, order=0, max_duration=10.0
|
||||
)
|
||||
assert config_max_only.default_duration == 10.0
|
||||
|
||||
# 两者都有 → 平均值
|
||||
config_both = TemplateClipConfig.create(
|
||||
template_id="tpl_001", clip_type=ClipType.INTRO, order=0, min_duration=4.0, max_duration=10.0
|
||||
)
|
||||
assert config_both.default_duration == 7.0
|
||||
|
||||
def test_clip_type_enum_values(self):
|
||||
"""ClipType 枚举值"""
|
||||
assert ClipType.INTRO == "intro"
|
||||
assert ClipType.MAIN == "main"
|
||||
assert ClipType.TRANSITION == "transition"
|
||||
assert ClipType.OUTRO == "outro"
|
||||
assert ClipType.TITLE == "title"
|
||||
assert ClipType.SUBTITLE == "subtitle"
|
||||
|
||||
def test_transition_effect_enum_values(self):
|
||||
"""TransitionEffect 枚举值"""
|
||||
assert TransitionEffect.CUT == "cut"
|
||||
assert TransitionEffect.FADE == "fade"
|
||||
assert TransitionEffect.SLIDE_LEFT == "slide_left"
|
||||
assert TransitionEffect.SLIDE_RIGHT == "slide_right"
|
||||
assert TransitionEffect.DISSOLVE == "dissolve"
|
||||
assert TransitionEffect.WIPE == "wipe"
|
||||
|
||||
def test_timestamps_auto_set(self):
|
||||
"""创建时自动设置时间戳"""
|
||||
config = TemplateClipConfig.create(
|
||||
template_id="tpl_001", clip_type=ClipType.INTRO, order=0
|
||||
)
|
||||
assert config.created_at is not None
|
||||
assert config.updated_at is not None
|
||||
|
||||
|
||||
# ── EditPlanClip 领域实体测试 ───────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestEditPlanClip:
|
||||
"""EditPlanClip 领域实体测试"""
|
||||
|
||||
def test_create_basic(self):
|
||||
"""基本创建"""
|
||||
clip = EditPlanClip.create(plan_id="plan_001", clip_type="main", order=0)
|
||||
assert clip.plan_id == "plan_001"
|
||||
assert clip.clip_type == "main"
|
||||
assert clip.order == 0
|
||||
assert clip.status == EditPlanClipStatus.PENDING
|
||||
assert clip.template_clip_config_id == ""
|
||||
assert clip.asset_id == ""
|
||||
assert clip.text_content == ""
|
||||
assert clip.start_time == 0.0
|
||||
assert clip.duration == 0.0
|
||||
assert clip.transition_effect == "cut"
|
||||
assert clip.config == {}
|
||||
assert clip.id
|
||||
|
||||
def test_create_with_all_fields(self):
|
||||
"""完整字段创建"""
|
||||
clip = EditPlanClip.create(
|
||||
plan_id="plan_001",
|
||||
clip_type="intro",
|
||||
order=0,
|
||||
template_clip_config_id="cfg_001",
|
||||
asset_id="asset_001",
|
||||
text_content="欢迎",
|
||||
start_time=0.0,
|
||||
duration=5.0,
|
||||
transition_effect="fade",
|
||||
config={"zoom": 1.2},
|
||||
)
|
||||
assert clip.template_clip_config_id == "cfg_001"
|
||||
assert clip.asset_id == "asset_001"
|
||||
assert clip.text_content == "欢迎"
|
||||
assert clip.duration == 5.0
|
||||
assert clip.transition_effect == "fade"
|
||||
assert clip.config == {"zoom": 1.2}
|
||||
|
||||
def test_create_empty_plan_id_raises(self):
|
||||
"""空 plan_id 报错"""
|
||||
with pytest.raises(ValueError, match="plan_id 不能为空"):
|
||||
EditPlanClip.create(plan_id="", clip_type="main", order=0)
|
||||
|
||||
def test_create_empty_clip_type_raises(self):
|
||||
"""空 clip_type 报错"""
|
||||
with pytest.raises(ValueError, match="clip_type 不能为空"):
|
||||
EditPlanClip.create(plan_id="plan_001", clip_type="", order=0)
|
||||
|
||||
def test_create_negative_start_time_raises(self):
|
||||
"""负数 start_time 报错"""
|
||||
with pytest.raises(ValueError, match="start_time 不能为负数"):
|
||||
EditPlanClip.create(plan_id="plan_001", clip_type="main", order=0, start_time=-1.0)
|
||||
|
||||
def test_create_negative_duration_raises(self):
|
||||
"""负数 duration 报错"""
|
||||
with pytest.raises(ValueError, match="duration 不能为负数"):
|
||||
EditPlanClip.create(plan_id="plan_001", clip_type="main", order=0, duration=-1.0)
|
||||
|
||||
def test_assign_asset(self):
|
||||
"""分配素材"""
|
||||
clip = EditPlanClip.create(plan_id="plan_001", clip_type="main", order=0)
|
||||
assert clip.has_asset is False
|
||||
clip.assign_asset("asset_001")
|
||||
assert clip.asset_id == "asset_001"
|
||||
assert clip.has_asset is True
|
||||
|
||||
def test_assign_asset_empty_raises(self):
|
||||
"""分配空素材报错"""
|
||||
clip = EditPlanClip.create(plan_id="plan_001", clip_type="main", order=0)
|
||||
with pytest.raises(ValueError, match="asset_id 不能为空"):
|
||||
clip.assign_asset("")
|
||||
|
||||
def test_status_transitions(self):
|
||||
"""状态流转: pending → ready → rendered"""
|
||||
clip = EditPlanClip.create(plan_id="plan_001", clip_type="main", order=0)
|
||||
assert clip.status == EditPlanClipStatus.PENDING
|
||||
|
||||
clip.mark_ready()
|
||||
assert clip.status == EditPlanClipStatus.READY
|
||||
|
||||
clip.mark_rendered()
|
||||
assert clip.status == EditPlanClipStatus.RENDERED
|
||||
|
||||
def test_status_transition_pending_to_failed(self):
|
||||
"""状态流转: pending → ready → failed"""
|
||||
clip = EditPlanClip.create(plan_id="plan_001", clip_type="main", order=0)
|
||||
clip.mark_ready()
|
||||
clip.mark_failed()
|
||||
assert clip.status == EditPlanClipStatus.FAILED
|
||||
|
||||
def test_mark_ready_from_non_pending_raises(self):
|
||||
"""非 pending 状态标记 ready 报错"""
|
||||
clip = EditPlanClip.create(plan_id="plan_001", clip_type="main", order=0)
|
||||
clip.mark_ready()
|
||||
with pytest.raises(ValueError, match="只有 pending 状态"):
|
||||
clip.mark_ready()
|
||||
|
||||
def test_mark_rendered_from_non_ready_raises(self):
|
||||
"""非 ready 状态标记 rendered 报错"""
|
||||
clip = EditPlanClip.create(plan_id="plan_001", clip_type="main", order=0)
|
||||
with pytest.raises(ValueError, match="只有 ready 状态"):
|
||||
clip.mark_rendered()
|
||||
|
||||
def test_mark_failed_from_non_ready_raises(self):
|
||||
"""非 ready 状态标记 failed 报错"""
|
||||
clip = EditPlanClip.create(plan_id="plan_001", clip_type="main", order=0)
|
||||
with pytest.raises(ValueError, match="只有 ready 状态"):
|
||||
clip.mark_failed()
|
||||
|
||||
def test_end_time_property(self):
|
||||
"""end_time 属性"""
|
||||
clip = EditPlanClip.create(
|
||||
plan_id="plan_001", clip_type="main", order=0, start_time=5.0, duration=10.0
|
||||
)
|
||||
assert clip.end_time == 15.0
|
||||
|
||||
def test_has_asset_property(self):
|
||||
"""has_asset 属性"""
|
||||
clip_no_asset = EditPlanClip.create(plan_id="plan_001", clip_type="main", order=0)
|
||||
assert clip_no_asset.has_asset is False
|
||||
|
||||
clip_with_asset = EditPlanClip.create(
|
||||
plan_id="plan_001", clip_type="main", order=0, asset_id="asset_001"
|
||||
)
|
||||
assert clip_with_asset.has_asset is True
|
||||
|
||||
def test_edit_plan_clip_status_enum(self):
|
||||
"""EditPlanClipStatus 枚举值"""
|
||||
assert EditPlanClipStatus.PENDING == "pending"
|
||||
assert EditPlanClipStatus.READY == "ready"
|
||||
assert EditPlanClipStatus.RENDERED == "rendered"
|
||||
assert EditPlanClipStatus.FAILED == "failed"
|
||||
|
||||
|
||||
# ── Repository 集成测试（使用 SQLite 内存数据库）────────────────────────────
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from packages.adapters.sqlalchemy_impl.models import Base, TemplateClipConfigModel, EditPlanClipModel
|
||||
from packages.adapters.sqlalchemy_impl.template_clip_config_repository import (
|
||||
SQLAlchemyTemplateClipConfigRepository,
|
||||
)
|
||||
from packages.adapters.sqlalchemy_impl.edit_plan_clip_repository import (
|
||||
SQLAlchemyEditPlanClipRepository,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def db_session():
|
||||
"""创建内存数据库 session"""
|
||||
engine = create_engine("sqlite:///:memory:")
|
||||
Base.metadata.create_all(engine)
|
||||
Session = sessionmaker(bind=engine)
|
||||
session = Session()
|
||||
yield session
|
||||
session.close()
|
||||
|
||||
|
||||
class TestTemplateClipConfigRepository:
|
||||
"""TemplateClipConfig 仓储测试"""
|
||||
|
||||
def test_create_and_get(self, db_session):
|
||||
"""创建并获取"""
|
||||
repo = SQLAlchemyTemplateClipConfigRepository(db_session)
|
||||
config = TemplateClipConfig.create(
|
||||
template_id="tpl_001", clip_type=ClipType.INTRO, order=0, min_duration=2.0
|
||||
)
|
||||
created = repo.create(config)
|
||||
assert created.id == config.id
|
||||
|
||||
fetched = repo.get(config.id)
|
||||
assert fetched is not None
|
||||
assert fetched.template_id == "tpl_001"
|
||||
assert fetched.clip_type == ClipType.INTRO
|
||||
assert fetched.min_duration == 2.0
|
||||
|
||||
def test_list_by_template(self, db_session):
|
||||
"""按模板列出"""
|
||||
repo = SQLAlchemyTemplateClipConfigRepository(db_session)
|
||||
for i in range(3):
|
||||
repo.create(
|
||||
TemplateClipConfig.create(
|
||||
template_id="tpl_001", clip_type=ClipType.MAIN, order=i
|
||||
)
|
||||
)
|
||||
repo.create(
|
||||
TemplateClipConfig.create(template_id="tpl_002", clip_type=ClipType.INTRO, order=0)
|
||||
)
|
||||
|
||||
results = repo.list_by_template("tpl_001")
|
||||
assert len(results) == 3
|
||||
assert all(r.template_id == "tpl_001" for r in results)
|
||||
# 按 order 排序
|
||||
assert results[0].order == 0
|
||||
assert results[1].order == 1
|
||||
assert results[2].order == 2
|
||||
|
||||
def test_list_by_template_with_clip_type_filter(self, db_session):
|
||||
"""按模板+类型过滤"""
|
||||
repo = SQLAlchemyTemplateClipConfigRepository(db_session)
|
||||
repo.create(TemplateClipConfig.create(template_id="tpl_001", clip_type=ClipType.INTRO, order=0))
|
||||
repo.create(TemplateClipConfig.create(template_id="tpl_001", clip_type=ClipType.MAIN, order=1))
|
||||
repo.create(TemplateClipConfig.create(template_id="tpl_001", clip_type=ClipType.MAIN, order=2))
|
||||
|
||||
results = repo.list_by_template("tpl_001", clip_type=ClipType.MAIN)
|
||||
assert len(results) == 2
|
||||
assert all(r.clip_type == ClipType.MAIN for r in results)
|
||||
|
||||
def test_update(self, db_session):
|
||||
"""更新"""
|
||||
repo = SQLAlchemyTemplateClipConfigRepository(db_session)
|
||||
config = TemplateClipConfig.create(template_id="tpl_001", clip_type=ClipType.INTRO, order=0)
|
||||
repo.create(config)
|
||||
|
||||
config.min_duration = 5.0
|
||||
config.max_duration = 15.0
|
||||
config.text_template = "Hello {name}"
|
||||
updated = repo.update(config)
|
||||
assert updated.min_duration == 5.0
|
||||
assert updated.max_duration == 15.0
|
||||
assert updated.text_template == "Hello {name}"
|
||||
|
||||
def test_delete(self, db_session):
|
||||
"""删除"""
|
||||
repo = SQLAlchemyTemplateClipConfigRepository(db_session)
|
||||
config = TemplateClipConfig.create(template_id="tpl_001", clip_type=ClipType.INTRO, order=0)
|
||||
repo.create(config)
|
||||
assert repo.delete(config.id) is True
|
||||
assert repo.get(config.id) is None
|
||||
assert repo.delete("nonexistent") is False
|
||||
|
||||
def test_delete_by_template(self, db_session):
|
||||
"""按模板批量删除"""
|
||||
repo = SQLAlchemyTemplateClipConfigRepository(db_session)
|
||||
for i in range(3):
|
||||
repo.create(
|
||||
TemplateClipConfig.create(template_id="tpl_001", clip_type=ClipType.MAIN, order=i)
|
||||
)
|
||||
deleted = repo.delete_by_template("tpl_001")
|
||||
assert deleted == 3
|
||||
assert repo.count(template_id="tpl_001") == 0
|
||||
|
||||
def test_count(self, db_session):
|
||||
"""统计"""
|
||||
repo = SQLAlchemyTemplateClipConfigRepository(db_session)
|
||||
assert repo.count() == 0
|
||||
repo.create(TemplateClipConfig.create(template_id="tpl_001", clip_type=ClipType.INTRO, order=0))
|
||||
repo.create(TemplateClipConfig.create(template_id="tpl_001", clip_type=ClipType.MAIN, order=1))
|
||||
assert repo.count() == 2
|
||||
assert repo.count(template_id="tpl_001") == 2
|
||||
assert repo.count(template_id="tpl_999") == 0
|
||||
|
||||
|
||||
class TestEditPlanClipRepository:
|
||||
"""EditPlanClip 仓储测试"""
|
||||
|
||||
def test_create_and_get(self, db_session):
|
||||
"""创建并获取"""
|
||||
repo = SQLAlchemyEditPlanClipRepository(db_session)
|
||||
clip = EditPlanClip.create(
|
||||
plan_id="plan_001", clip_type="main", order=0, duration=5.0
|
||||
)
|
||||
created = repo.create(clip)
|
||||
assert created.id == clip.id
|
||||
|
||||
fetched = repo.get(clip.id)
|
||||
assert fetched is not None
|
||||
assert fetched.plan_id == "plan_001"
|
||||
assert fetched.clip_type == "main"
|
||||
assert fetched.duration == 5.0
|
||||
assert fetched.status == EditPlanClipStatus.PENDING
|
||||
|
||||
def test_list_by_plan(self, db_session):
|
||||
"""按计划列出"""
|
||||
repo = SQLAlchemyEditPlanClipRepository(db_session)
|
||||
for i in range(3):
|
||||
repo.create(
|
||||
EditPlanClip.create(plan_id="plan_001", clip_type="main", order=i)
|
||||
)
|
||||
repo.create(
|
||||
EditPlanClip.create(plan_id="plan_002", clip_type="intro", order=0)
|
||||
)
|
||||
|
||||
results = repo.list_by_plan("plan_001")
|
||||
assert len(results) == 3
|
||||
assert all(r.plan_id == "plan_001" for r in results)
|
||||
assert results[0].order == 0
|
||||
assert results[1].order == 1
|
||||
assert results[2].order == 2
|
||||
|
||||
def test_list_by_plan_with_status_filter(self, db_session):
|
||||
"""按计划+状态过滤"""
|
||||
repo = SQLAlchemyEditPlanClipRepository(db_session)
|
||||
clip1 = EditPlanClip.create(plan_id="plan_001", clip_type="main", order=0)
|
||||
clip2 = EditPlanClip.create(plan_id="plan_001", clip_type="main", order=1)
|
||||
repo.create(clip1)
|
||||
repo.create(clip2)
|
||||
|
||||
clip1.mark_ready()
|
||||
repo.update(clip1)
|
||||
|
||||
results = repo.list_by_plan("plan_001", status=EditPlanClipStatus.READY)
|
||||
assert len(results) == 1
|
||||
assert results[0].status == EditPlanClipStatus.READY
|
||||
|
||||
def test_update(self, db_session):
|
||||
"""更新"""
|
||||
repo = SQLAlchemyEditPlanClipRepository(db_session)
|
||||
clip = EditPlanClip.create(plan_id="plan_001", clip_type="main", order=0)
|
||||
repo.create(clip)
|
||||
|
||||
clip.assign_asset("asset_001")
|
||||
clip.duration = 10.0
|
||||
updated = repo.update(clip)
|
||||
assert updated.asset_id == "asset_001"
|
||||
assert updated.duration == 10.0
|
||||
|
||||
def test_delete(self, db_session):
|
||||
"""删除"""
|
||||
repo = SQLAlchemyEditPlanClipRepository(db_session)
|
||||
clip = EditPlanClip.create(plan_id="plan_001", clip_type="main", order=0)
|
||||
repo.create(clip)
|
||||
assert repo.delete(clip.id) is True
|
||||
assert repo.get(clip.id) is None
|
||||
assert repo.delete("nonexistent") is False
|
||||
|
||||
def test_delete_by_plan(self, db_session):
|
||||
"""按计划批量删除"""
|
||||
repo = SQLAlchemyEditPlanClipRepository(db_session)
|
||||
for i in range(3):
|
||||
repo.create(
|
||||
EditPlanClip.create(plan_id="plan_001", clip_type="main", order=i)
|
||||
)
|
||||
deleted = repo.delete_by_plan("plan_001")
|
||||
assert deleted == 3
|
||||
assert repo.count(plan_id="plan_001") == 0
|
||||
|
||||
def test_count(self, db_session):
|
||||
"""统计"""
|
||||
repo = SQLAlchemyEditPlanClipRepository(db_session)
|
||||
assert repo.count() == 0
|
||||
repo.create(EditPlanClip.create(plan_id="plan_001", clip_type="main", order=0))
|
||||
repo.create(EditPlanClip.create(plan_id="plan_001", clip_type="main", order=1))
|
||||
assert repo.count() == 2
|
||||
assert repo.count(plan_id="plan_001") == 2
|
||||
assert repo.count(plan_id="plan_999") == 0
|
||||
|
||||
def test_count_with_status(self, db_session):
|
||||
"""按状态统计"""
|
||||
repo = SQLAlchemyEditPlanClipRepository(db_session)
|
||||
clip1 = EditPlanClip.create(plan_id="plan_001", clip_type="main", order=0)
|
||||
clip2 = EditPlanClip.create(plan_id="plan_001", clip_type="main", order=1)
|
||||
repo.create(clip1)
|
||||
repo.create(clip2)
|
||||
|
||||
clip1.mark_ready()
|
||||
repo.update(clip1)
|
||||
|
||||
assert repo.count(status=EditPlanClipStatus.PENDING) == 1
|
||||
assert repo.count(status=EditPlanClipStatus.READY) == 1
|
||||
Reference in New Issue
Block a user