feat: 模板与剪辑计划后端能力补齐
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Failing after 18s
CI/CD Pipeline / Integration Tests (pull_request) Failing after 10s
CI/CD Pipeline / Unit Tests (pull_request) Failing after 51s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 36m12s
CI/CD Pipeline / Production Browser E2E (pull_request) Failing after 1551h25m55s
CI/CD Pipeline / Staging API Integration Tests (pull_request) Failing after 1551h25m57s
CI/CD Pipeline / Staging E2E Tests (pull_request) Failing after 1551h25m57s
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (pull_request) Failing after 1551h25m57s
CI/CD Pipeline / Deploy Production (pull_request) Failing after 1551h57m32s
CI/CD Pipeline / Build Production Runtime Images (pull_request) Failing after 1551h57m32s

- 模板复制接口(含所有片段配置)
- 模板列表多维度筛选(分类/标签/关键词/模式)
- 模板标签聚合接口(去重排序)
- 模板使用次数统计
- TemplateResponse 新增 usage_count 字段
- 新增 CountTemplatesUseCase / CopyTemplateUseCase / ListTagsUseCase / GetTemplateUsageUseCase
- 35个模板相关单元测试全部通过
This commit is contained in:
CI Bot
2026-07-13 21:57:40 +08:00
parent b562e152e0
commit 8a4afbb709
7 changed files with 566 additions and 21 deletions
+123 -13
View File
@@ -2,11 +2,14 @@
from __future__ import annotations
import uuid
from typing import List, Optional
from sqlalchemy import func
from sqlalchemy.orm import Session
from packages.adapters.sqlalchemy_impl.models import (
EditPlanModel,
TemplateCategoryModel,
TemplateModel,
TemplateSegmentModel,
@@ -28,13 +31,27 @@ class SQLAlchemyTemplateRepository:
*,
skip: int = 0,
limit: int = 50,
category: Optional[str] = None,
tag: Optional[str] = None,
keyword: Optional[str] = None,
mode: Optional[str] = None,
) -> List[Template]:
query = self.session.query(TemplateModel).filter(
TemplateModel.user_id == user_id,
TemplateModel.is_active.is_(True),
)
if category:
query = query.filter(TemplateModel.category == category)
if mode:
query = query.filter(TemplateModel.mode == mode)
if keyword:
like_pattern = f"%{keyword}%"
query = query.filter(TemplateModel.name.like(like_pattern))
if tag:
# JSON 数组包含指定标签(MySQL JSON_CONTAINS / SQLite json_each 兼容写法用 LIKE)
query = query.filter(TemplateModel.tags.like(f'%"{tag}"%'))
models = (
self.session.query(TemplateModel)
.filter(
TemplateModel.user_id == user_id,
TemplateModel.is_active.is_(True),
)
query
.order_by(TemplateModel.created_at.desc())
.offset(skip)
.limit(limit)
@@ -142,15 +159,77 @@ class SQLAlchemyTemplateRepository:
self.session.commit()
return True
def count_by_user(self, user_id: str) -> int:
return (
self.session.query(TemplateModel)
.filter(
TemplateModel.user_id == user_id,
TemplateModel.is_active.is_(True),
)
.count()
def count_by_user(
self,
user_id: str,
*,
category: Optional[str] = None,
tag: Optional[str] = None,
keyword: Optional[str] = None,
mode: Optional[str] = None,
) -> int:
query = self.session.query(TemplateModel).filter(
TemplateModel.user_id == user_id,
TemplateModel.is_active.is_(True),
)
if category:
query = query.filter(TemplateModel.category == category)
if mode:
query = query.filter(TemplateModel.mode == mode)
if keyword:
query = query.filter(TemplateModel.name.like(f"%{keyword}%"))
if tag:
query = query.filter(TemplateModel.tags.like(f'%"{tag}"%'))
return query.count()
def copy_template(self, template_id: str, user_id: str, new_name: str) -> Template:
"""复制模板(含所有 segments)。"""
source = self.get(template_id, user_id)
if source is None:
raise ValueError(f"Template {template_id} not found")
new_id = str(uuid.uuid4())
new_template = Template(
id=new_id,
user_id=user_id,
name=new_name,
mode=source.mode,
category=source.category,
tags=list(source.tags),
title_config=dict(source.title_config),
subtitle_config=dict(source.subtitle_config),
bgm_config=dict(source.bgm_config),
estimated_duration=source.estimated_duration,
is_active=True,
)
created = self.create(new_template)
# 复制 segments
new_segments: List[TemplateSegment] = []
for seg in source.segments:
new_seg = TemplateSegment(
id=str(uuid.uuid4()),
template_id=new_id,
segment_order=seg.segment_order,
duration_min=seg.duration_min,
duration_max=seg.duration_max,
material_type=seg.material_type,
)
new_segments.append(new_seg)
model = TemplateSegmentModel(
id=new_seg.id,
template_id=new_seg.template_id,
segment_order=new_seg.segment_order,
duration_min=new_seg.duration_min,
duration_max=new_seg.duration_max,
material_type=new_seg.material_type,
)
self.session.add(model)
if new_segments:
self.session.commit()
created.segments = new_segments
return created
# ── Segments ──
@@ -234,6 +313,37 @@ class SQLAlchemyTemplateRepository:
self.session.commit()
return True
# ── Tags ──
def list_tags(self, user_id: str) -> List[str]:
"""获取用户所有模板的标签(去重)。"""
models = (
self.session.query(TemplateModel)
.filter(
TemplateModel.user_id == user_id,
TemplateModel.is_active.is_(True),
TemplateModel.tags.isnot(None),
)
.all()
)
tags_set: set[str] = set()
for m in models:
if m.tags:
for t in m.tags:
if t:
tags_set.add(t)
return sorted(tags_set)
# ── Usage Stats ──
def get_usage_count(self, template_id: str) -> int:
"""获取模板被使用的次数(关联的剪辑计划数量)。"""
return (
self.session.query(EditPlanModel)
.filter(EditPlanModel.template_id == template_id)
.count()
)
# ── Mapping helpers ──
@staticmethod