Compare commits
10 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| d45f0d6bdb | |||
| bc9df316dc | |||
| 5ed2ee1192 | |||
| 061554af89 | |||
| 87cca302f4 | |||
| deb127ae08 | |||
| a9438ed996 | |||
| 72f47592a9 | |||
| b6c340a352 | |||
| 6e76b45d34 |
@@ -876,7 +876,7 @@ jobs:
|
||||
env:
|
||||
GITHUB_TOKEN: ${{ github.token }}
|
||||
run: |
|
||||
bash scripts/ci/ci_push_paths.sh
|
||||
curl -sfH "Authorization: token $GITHUB_TOKEN" -o /tmp/_ci_push_paths.sh "${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/raw/scripts/ci/ci_push_paths.sh?ref=${GITHUB_SHA}" && bash /tmp/_ci_push_paths.sh
|
||||
- name: Report CI trace
|
||||
if: always()
|
||||
shell: sh
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
"""migrate template_segments data to template_clip_configs
|
||||
|
||||
Revision ID: 060_migrate_segments
|
||||
Revises: 059_duplicate_rate
|
||||
Create Date: 2026-08-31
|
||||
"""
|
||||
|
||||
import sqlalchemy as sa
|
||||
|
||||
from alembic import op
|
||||
|
||||
revision = "060_migrate_segments"
|
||||
down_revision = "059_duplicate_rate"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
dialect = op.get_bind().dialect.name
|
||||
|
||||
if dialect == "postgresql":
|
||||
config_expr = (
|
||||
"CASE WHEN s.material_type IS NOT NULL AND s.material_type != '' "
|
||||
"THEN json_build_object('material_type', s.material_type)::jsonb "
|
||||
"ELSE '{}'::jsonb END"
|
||||
)
|
||||
empty_json = "'{}'::jsonb"
|
||||
else:
|
||||
config_expr = (
|
||||
"CASE WHEN s.material_type IS NOT NULL AND s.material_type != '' "
|
||||
"THEN JSON_OBJECT('material_type', s.material_type) "
|
||||
"ELSE '{}' END"
|
||||
)
|
||||
empty_json = "'{}'"
|
||||
|
||||
sql_str = (
|
||||
"INSERT INTO template_clip_configs "
|
||||
'(id, template_id, clip_type, "order", min_duration, max_duration, '
|
||||
"text_template, material_requirements, transition_effect, config, "
|
||||
"created_at, updated_at) "
|
||||
"SELECT "
|
||||
"s.id, s.template_id, 'main', s.segment_order, "
|
||||
"s.duration_min, s.duration_max, "
|
||||
"'', " + empty_json + ", "
|
||||
"'cut', " + config_expr + ", "
|
||||
"s.created_at, s.updated_at "
|
||||
"FROM template_segments s "
|
||||
"WHERE NOT EXISTS ("
|
||||
" SELECT 1 FROM template_clip_configs c "
|
||||
" WHERE c.template_id = s.template_id"
|
||||
")"
|
||||
)
|
||||
op.execute(sa.text(sql_str))
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
pass
|
||||
@@ -1,5 +1,6 @@
|
||||
# Build stage
|
||||
FROM git.xiaoxiajianji.com/xiaoxia/base/node:20 AS builder
|
||||
ARG SOURCE_HASH=""
|
||||
WORKDIR /app
|
||||
ARG VITE_API_URL=https://saas-api.xiaoxiajianji.com
|
||||
ENV VITE_API_URL=$VITE_API_URL
|
||||
@@ -18,8 +19,10 @@ COPY apps/web/ ./
|
||||
|
||||
# 构建:TS增量编译 + Vite构建,tsbuildinfo用cache mount持久化
|
||||
# node_modules直接使用镜像中已安装的(layer缓存保证完整性)
|
||||
# SOURCE_HASH 变化时强制重新执行(防止 buildkit 幽灵缓存命中)
|
||||
RUN --mount=type=cache,target=/app/apps/web/.tscache,sharing=locked \
|
||||
mkdir -p .tscache \
|
||||
&& echo "SOURCE_HASH=${SOURCE_HASH}" > .cache_bust \
|
||||
&& ./node_modules/.bin/tsc --incremental --tsBuildInfoFile .tscache/tsconfig.tsbuildinfo \
|
||||
&& ./node_modules/.bin/vite build
|
||||
|
||||
|
||||
@@ -1,4 +1,9 @@
|
||||
"""SQLAlchemy implementation of TemplateRepository."""
|
||||
"""SQLAlchemy implementation of TemplateRepository.
|
||||
|
||||
模板 segments 数据源已统一为 template_clip_configs 表。
|
||||
读取时优先 template_clip_configs,回退 template_segments(兼容历史数据)。
|
||||
写入全部走 template_clip_configs。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -10,6 +15,7 @@ from sqlalchemy.orm import Session
|
||||
from packages.adapters.sqlalchemy_impl.models import (
|
||||
EditPlanModel,
|
||||
TemplateCategoryModel,
|
||||
TemplateClipConfigModel,
|
||||
TemplateModel,
|
||||
TemplateSegmentModel,
|
||||
)
|
||||
@@ -47,27 +53,38 @@ class SQLAlchemyTemplateRepository:
|
||||
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}"%'))
|
||||
query = query.filter(TemplateModel.tags.like(f'"%{tag}"%'))
|
||||
models = query.order_by(TemplateModel.created_at.desc()).offset(skip).limit(limit).all()
|
||||
templates = [self._model_to_entity(m) for m in models]
|
||||
# 批量加载所有 segments,避免 N+1 查询
|
||||
# 批量加载 segments —— 优先 template_clip_configs
|
||||
if templates:
|
||||
template_ids = [t.id for t in templates]
|
||||
seg_models = (
|
||||
self.session.query(TemplateSegmentModel)
|
||||
.filter(TemplateSegmentModel.template_id.in_(template_ids))
|
||||
.order_by(TemplateSegmentModel.segment_order)
|
||||
clip_models = (
|
||||
self.session.query(TemplateClipConfigModel)
|
||||
.filter(TemplateClipConfigModel.template_id.in_(template_ids))
|
||||
.order_by(TemplateClipConfigModel.order)
|
||||
.all()
|
||||
)
|
||||
# 按 template_id 分组
|
||||
seg_map: dict[str, list] = {}
|
||||
for sm in seg_models:
|
||||
seg_map.setdefault(sm.template_id, []).append(
|
||||
self._segment_model_to_entity(sm),
|
||||
clip_map: dict[str, list] = {}
|
||||
for cm in clip_models:
|
||||
clip_map.setdefault(cm.template_id, []).append(
|
||||
self._clip_config_to_segment(cm),
|
||||
)
|
||||
# 对没有 clip_configs 的模板,回退读 template_segments
|
||||
missing_ids = [t.id for t in templates if t.id not in clip_map]
|
||||
if missing_ids:
|
||||
old_models = (
|
||||
self.session.query(TemplateSegmentModel)
|
||||
.filter(TemplateSegmentModel.template_id.in_(missing_ids))
|
||||
.order_by(TemplateSegmentModel.segment_order)
|
||||
.all()
|
||||
)
|
||||
for om in old_models:
|
||||
clip_map.setdefault(om.template_id, []).append(
|
||||
self._segment_model_to_entity(om),
|
||||
)
|
||||
for t in templates:
|
||||
t.segments = seg_map.get(t.id, [])
|
||||
t.segments = clip_map.get(t.id, [])
|
||||
return templates
|
||||
|
||||
def get(self, template_id: str, user_id: str) -> Optional[Template]:
|
||||
@@ -100,7 +117,6 @@ class SQLAlchemyTemplateRepository:
|
||||
is_active=template.is_active,
|
||||
)
|
||||
self.session.add(model)
|
||||
# flush 而非 commit,让 create + create_segments 在同一事务中提交
|
||||
self.session.flush()
|
||||
self.session.refresh(model)
|
||||
result = self._model_to_entity(model)
|
||||
@@ -145,11 +161,8 @@ class SQLAlchemyTemplateRepository:
|
||||
if model is None:
|
||||
return False
|
||||
model.is_active = False
|
||||
# 级联清理关联的 segments,避免孤儿数据
|
||||
self.session.query(TemplateSegmentModel).filter(
|
||||
TemplateSegmentModel.template_id == template_id,
|
||||
).delete(synchronize_session=False)
|
||||
self.session.commit()
|
||||
# 复用 delete_segments_by_template 清理两张表的关联数据
|
||||
self.delete_segments_by_template(template_id)
|
||||
return True
|
||||
|
||||
def count_by_user(
|
||||
@@ -172,7 +185,7 @@ class SQLAlchemyTemplateRepository:
|
||||
if keyword:
|
||||
query = query.filter(TemplateModel.name.like(f"%{keyword}%"))
|
||||
if tag:
|
||||
query = query.filter(TemplateModel.tags.like(f'%"{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:
|
||||
@@ -181,9 +194,8 @@ class SQLAlchemyTemplateRepository:
|
||||
if source is None:
|
||||
raise ValueError(f"Template {template_id} not found")
|
||||
|
||||
new_id = str(uuid.uuid4())
|
||||
new_template = Template(
|
||||
id=new_id,
|
||||
id=str(uuid.uuid4()),
|
||||
user_id=user_id,
|
||||
name=new_name,
|
||||
mode=source.mode,
|
||||
@@ -197,28 +209,22 @@ class SQLAlchemyTemplateRepository:
|
||||
)
|
||||
created = self.create(new_template)
|
||||
|
||||
# 复制 segments
|
||||
# 复用 create_segments 写入 template_clip_configs
|
||||
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(
|
||||
TemplateSegment(
|
||||
id=str(uuid.uuid4()),
|
||||
template_id=created.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.create_segments(new_segments)
|
||||
else:
|
||||
self.session.commit()
|
||||
|
||||
created.segments = new_segments
|
||||
@@ -227,34 +233,58 @@ class SQLAlchemyTemplateRepository:
|
||||
# ── Segments ──
|
||||
|
||||
def list_segments(self, template_id: str) -> List[TemplateSegment]:
|
||||
models = (
|
||||
"""优先从 template_clip_configs 读取,回退读 template_segments。"""
|
||||
clips = (
|
||||
self.session.query(TemplateClipConfigModel)
|
||||
.filter(TemplateClipConfigModel.template_id == template_id)
|
||||
.order_by(TemplateClipConfigModel.order)
|
||||
.all()
|
||||
)
|
||||
if clips:
|
||||
return [self._clip_config_to_segment(m) for m in clips]
|
||||
# 回退:旧表
|
||||
old = (
|
||||
self.session.query(TemplateSegmentModel)
|
||||
.filter(TemplateSegmentModel.template_id == template_id)
|
||||
.order_by(TemplateSegmentModel.segment_order)
|
||||
.all()
|
||||
)
|
||||
return [self._segment_model_to_entity(m) for m in models]
|
||||
return [self._segment_model_to_entity(m) for m in old]
|
||||
|
||||
def create_segments(self, segments: List[TemplateSegment]) -> List[TemplateSegment]:
|
||||
"""写入 template_clip_configs 表。material_type 存入 config JSON。"""
|
||||
for seg in segments:
|
||||
model = TemplateSegmentModel(
|
||||
config = {"material_type": seg.material_type} if seg.material_type else {}
|
||||
model = TemplateClipConfigModel(
|
||||
id=seg.id,
|
||||
template_id=seg.template_id,
|
||||
segment_order=seg.segment_order,
|
||||
duration_min=seg.duration_min,
|
||||
duration_max=seg.duration_max,
|
||||
material_type=seg.material_type,
|
||||
clip_type="main",
|
||||
order=seg.segment_order,
|
||||
min_duration=seg.duration_min,
|
||||
max_duration=seg.duration_max,
|
||||
text_template="",
|
||||
material_requirements={},
|
||||
transition_effect="cut",
|
||||
config=config,
|
||||
)
|
||||
self.session.add(model)
|
||||
self.session.commit()
|
||||
return segments
|
||||
|
||||
def delete_segments_by_template(self, template_id: str) -> int:
|
||||
count = (
|
||||
self.session.query(TemplateSegmentModel).filter(TemplateSegmentModel.template_id == template_id).delete()
|
||||
"""删除两张表中的 segments 数据,返回删除总数。"""
|
||||
c1 = (
|
||||
self.session.query(TemplateClipConfigModel)
|
||||
.filter(TemplateClipConfigModel.template_id == template_id)
|
||||
.delete(synchronize_session=False)
|
||||
)
|
||||
c2 = (
|
||||
self.session.query(TemplateSegmentModel)
|
||||
.filter(TemplateSegmentModel.template_id == template_id)
|
||||
.delete(synchronize_session=False)
|
||||
)
|
||||
self.session.commit()
|
||||
return count
|
||||
return c1 + c2
|
||||
|
||||
# ── Categories ──
|
||||
|
||||
@@ -366,6 +396,23 @@ class SQLAlchemyTemplateRepository:
|
||||
updated_at=model.updated_at,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _clip_config_to_segment(model: TemplateClipConfigModel) -> TemplateSegment:
|
||||
"""将 TemplateClipConfigModel 转换为 TemplateSegment 域实体。"""
|
||||
material_type = None
|
||||
if model.config and isinstance(model.config, dict):
|
||||
material_type = model.config.get("material_type")
|
||||
return TemplateSegment(
|
||||
id=model.id,
|
||||
template_id=model.template_id,
|
||||
segment_order=model.order,
|
||||
duration_min=model.min_duration,
|
||||
duration_max=model.max_duration,
|
||||
material_type=material_type,
|
||||
created_at=model.created_at,
|
||||
updated_at=model.updated_at,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _category_model_to_entity(model: TemplateCategoryModel) -> TemplateCategory:
|
||||
return TemplateCategory(
|
||||
|
||||
@@ -19,6 +19,18 @@ for arg in "$@"; do
|
||||
BUILD_ARGS="$BUILD_ARGS --build-arg $arg"
|
||||
done
|
||||
|
||||
# Web 镜像 cache bust:计算 apps/web/ 的 git tree hash
|
||||
# 当源码变化时 hash 变化,buildx 的 ARG 缓存键失效 → vite build 必定重新执行
|
||||
if [ "${DOCKERFILE##*/}" = "web.Dockerfile" ]; then
|
||||
SOURCE_HASH=$(git rev-parse HEAD:apps/web 2>/dev/null || echo "")
|
||||
if [ -n "$SOURCE_HASH" ]; then
|
||||
echo "Web cache bust: SOURCE_HASH=${SOURCE_HASH}"
|
||||
BUILD_ARGS="$BUILD_ARGS --build-arg SOURCE_HASH=${SOURCE_HASH}"
|
||||
else
|
||||
echo "⚠️ 无法计算 apps/web tree hash,跳过 cache bust"
|
||||
fi
|
||||
fi
|
||||
|
||||
BUILDER_NAME="ci-builder-persist"
|
||||
if ! docker buildx inspect "$BUILDER_NAME" > /dev/null 2>&1; then
|
||||
echo "持久 builder 不存在,创建中..."
|
||||
|
||||
@@ -31,6 +31,18 @@ for arg in "$@"; do
|
||||
BUILD_ARGS="$BUILD_ARGS --build-arg $arg"
|
||||
done
|
||||
|
||||
# Web 镜像 cache bust:计算 apps/web/ 的 git tree hash
|
||||
# 当源码变化时 hash 变化,buildx 的 ARG 缓存键失效 → vite build 必定重新执行
|
||||
if [ "${DOCKERFILE##*/}" = "web.Dockerfile" ]; then
|
||||
SOURCE_HASH=$(git rev-parse HEAD:apps/web 2>/dev/null || echo "")
|
||||
if [ -n "$SOURCE_HASH" ]; then
|
||||
echo "Web cache bust: SOURCE_HASH=${SOURCE_HASH}"
|
||||
BUILD_ARGS="$BUILD_ARGS --build-arg SOURCE_HASH=${SOURCE_HASH}"
|
||||
else
|
||||
echo "⚠️ 无法计算 apps/web tree hash,跳过 cache bust"
|
||||
fi
|
||||
fi
|
||||
|
||||
# 确保持久 builder 存在并使用(幂等)
|
||||
if ! docker buildx inspect "$BUILDER_NAME" > /dev/null 2>&1; then
|
||||
echo "持久 builder 不存在,创建中..."
|
||||
|
||||
@@ -8,7 +8,13 @@ echo "=== CI Validate: 安全扫描 ==="
|
||||
# --- 密钥检测 ---
|
||||
echo ""
|
||||
echo "=== [1/4] Secret detection (detect-secrets) ==="
|
||||
python3 -m pip install -q detect-secrets
|
||||
python3 -m pip install -q --no-cache-dir detect-secrets || {
|
||||
echo "⚠️ detect-secrets install failed, retrying without cache..."
|
||||
python3 -m pip install -q --no-cache-dir --no-binary :all: detect-secrets || {
|
||||
echo "❌ detect-secrets install failed after retry"
|
||||
exit 1
|
||||
}
|
||||
}
|
||||
detect-secrets --version
|
||||
|
||||
detect-secrets scan \
|
||||
@@ -72,17 +78,27 @@ fi
|
||||
# --- Pip-audit 依赖漏洞扫描(仅告警)---
|
||||
echo ""
|
||||
echo "=== [3/4] Python dependency vulnerability scan (pip-audit, advisory only) ==="
|
||||
python3 -m pip install -q pip-audit
|
||||
pip-audit --version
|
||||
EXIT_CODE=0
|
||||
for req_file in requirements.txt requirements-base.txt requirements-dev.txt; do
|
||||
if [ -f "$req_file" ]; then
|
||||
echo "--- Scanning $req_file ---"
|
||||
pip-audit -r "$req_file" --desc on 2>&1 | head -40 || EXIT_CODE=$?
|
||||
echo ""
|
||||
fi
|
||||
done
|
||||
echo "pip-audit scan completed (advisory mode - warnings only, not blocking CI)"
|
||||
python3 -m pip install -q --no-cache-dir pip-audit || {
|
||||
echo "⚠️ pip-audit install failed (cache issue?), retrying..."
|
||||
python3 -m pip install -q --no-cache-dir pip-audit || {
|
||||
echo "⚠️ pip-audit unavailable, skipping dependency vulnerability scan (advisory)"
|
||||
pip-audit --version 2>/dev/null || true
|
||||
}
|
||||
}
|
||||
if command -v pip-audit >/dev/null 2>&1 || python3 -m pip show pip-audit >/dev/null 2>&1; then
|
||||
pip-audit --version
|
||||
EXIT_CODE=0
|
||||
for req_file in requirements.txt requirements-base.txt requirements-dev.txt; do
|
||||
if [ -f "$req_file" ]; then
|
||||
echo "--- Scanning $req_file ---"
|
||||
pip-audit -r "$req_file" --desc on 2>&1 | head -40 || EXIT_CODE=$?
|
||||
echo ""
|
||||
fi
|
||||
done
|
||||
echo "pip-audit scan completed (advisory mode - warnings only, not blocking CI)"
|
||||
else
|
||||
echo "⚠️ pip-audit not available, skipping dependency vulnerability scan (advisory)"
|
||||
fi
|
||||
|
||||
# --- CI脚本语法校验 ---
|
||||
echo ""
|
||||
|
||||
@@ -24,7 +24,7 @@ echo "✅ Code formatting checks passed"
|
||||
echo ""
|
||||
echo "=== [3/3] Dead code detection (vulture, advisory only) ==="
|
||||
set +e
|
||||
python3 -m pip install -q vulture
|
||||
python3 -m pip install -q --no-cache-dir vulture || echo "⚠️ vulture install failed, skipping dead code detection"
|
||||
vulture --version
|
||||
echo "告警模式,不阻断CI。置信度>=90%建议尽快确认。"
|
||||
echo ""
|
||||
|
||||
@@ -0,0 +1,209 @@
|
||||
"""统一模板 segments 数据源单元测试。
|
||||
|
||||
验证 template_repository 从 template_clip_configs 读取 segments,
|
||||
写入走 template_clip_configs,回退兼容 template_segments。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from packages.adapters.sqlalchemy_impl.models import (
|
||||
Base,
|
||||
TemplateClipConfigModel,
|
||||
TemplateModel,
|
||||
TemplateSegmentModel,
|
||||
)
|
||||
from packages.adapters.sqlalchemy_impl.template_repository import (
|
||||
SQLAlchemyTemplateRepository,
|
||||
)
|
||||
from packages.domain.template import Template, TemplateSegment
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def session():
|
||||
engine = create_engine("sqlite:///:memory:")
|
||||
Base.metadata.create_all(engine)
|
||||
Session = sessionmaker(bind=engine)
|
||||
s = Session()
|
||||
try:
|
||||
yield s
|
||||
finally:
|
||||
s.close()
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def repo(session):
|
||||
return SQLAlchemyTemplateRepository(session)
|
||||
|
||||
|
||||
def _make_template(template_id=None, user_id="u1", name="测试模板", mode="one_take"):
|
||||
tid = template_id or str(uuid.uuid4())
|
||||
return Template(
|
||||
id=tid,
|
||||
user_id=user_id,
|
||||
name=name,
|
||||
mode=mode,
|
||||
category="",
|
||||
tags=[],
|
||||
estimated_duration=30.0,
|
||||
is_active=True,
|
||||
segments=[],
|
||||
)
|
||||
|
||||
|
||||
def _make_segment(template_id, order=1, material_type=None):
|
||||
return TemplateSegment(
|
||||
id=str(uuid.uuid4()),
|
||||
template_id=template_id,
|
||||
segment_order=order,
|
||||
duration_min=5.0,
|
||||
duration_max=10.0,
|
||||
material_type=material_type,
|
||||
)
|
||||
|
||||
|
||||
class TestCreateSegments:
|
||||
def test_writes_to_clip_configs(self, repo, session):
|
||||
tpl = _make_template()
|
||||
repo.create(tpl)
|
||||
seg = _make_segment(tpl.id, order=1)
|
||||
repo.create_segments([seg])
|
||||
clips = session.query(TemplateClipConfigModel).filter(TemplateClipConfigModel.template_id == tpl.id).all()
|
||||
assert len(clips) == 1
|
||||
assert clips[0].clip_type == "main"
|
||||
assert clips[0].order == 1
|
||||
assert clips[0].min_duration == 5.0
|
||||
|
||||
def test_material_type_stored_in_config(self, repo, session):
|
||||
tpl = _make_template()
|
||||
repo.create(tpl)
|
||||
seg = _make_segment(tpl.id, order=1, material_type="voiceover")
|
||||
repo.create_segments([seg])
|
||||
clip = session.query(TemplateClipConfigModel).filter(TemplateClipConfigModel.template_id == tpl.id).first()
|
||||
assert clip.config["material_type"] == "voiceover"
|
||||
|
||||
|
||||
class TestListSegments:
|
||||
def test_reads_from_clip_configs(self, repo, session):
|
||||
tpl = _make_template()
|
||||
repo.create(tpl)
|
||||
seg = _make_segment(tpl.id, order=1, material_type="voiceover")
|
||||
repo.create_segments([seg])
|
||||
result = repo.list_segments(tpl.id)
|
||||
assert len(result) == 1
|
||||
assert result[0].material_type == "voiceover"
|
||||
|
||||
def test_fallback_to_old_table(self, repo, session):
|
||||
tpl = _make_template()
|
||||
repo.create(tpl)
|
||||
old = TemplateSegmentModel(
|
||||
id=str(uuid.uuid4()),
|
||||
template_id=tpl.id,
|
||||
segment_order=1,
|
||||
duration_min=3.0,
|
||||
duration_max=8.0,
|
||||
material_type="场景",
|
||||
)
|
||||
session.add(old)
|
||||
session.commit()
|
||||
result = repo.list_segments(tpl.id)
|
||||
assert len(result) == 1
|
||||
assert result[0].material_type == "场景"
|
||||
|
||||
def test_clip_configs_takes_priority(self, repo, session):
|
||||
tpl = _make_template()
|
||||
repo.create(tpl)
|
||||
seg = _make_segment(tpl.id, order=1)
|
||||
repo.create_segments([seg])
|
||||
old = TemplateSegmentModel(
|
||||
id=str(uuid.uuid4()), template_id=tpl.id, segment_order=1, duration_min=1.0, duration_max=2.0
|
||||
)
|
||||
session.add(old)
|
||||
session.commit()
|
||||
result = repo.list_segments(tpl.id)
|
||||
assert len(result) == 1
|
||||
assert result[0].duration_min == 5.0
|
||||
|
||||
|
||||
class TestListByUser:
|
||||
def test_batch_loads_from_clip_configs(self, repo, session):
|
||||
tpl = _make_template()
|
||||
repo.create(tpl)
|
||||
seg = _make_segment(tpl.id, order=1, material_type="人物")
|
||||
repo.create_segments([seg])
|
||||
result = repo.list_by_user("u1")
|
||||
assert len(result) == 1
|
||||
assert len(result[0].segments) == 1
|
||||
assert result[0].segments[0].material_type == "人物"
|
||||
|
||||
def test_fallback_for_old_data(self, repo, session):
|
||||
tpl = _make_template()
|
||||
repo.create(tpl)
|
||||
old = TemplateSegmentModel(
|
||||
id=str(uuid.uuid4()), template_id=tpl.id, segment_order=1, duration_min=2.0, duration_max=6.0
|
||||
)
|
||||
session.add(old)
|
||||
session.commit()
|
||||
result = repo.list_by_user("u1")
|
||||
assert len(result) == 1
|
||||
assert len(result[0].segments) == 1
|
||||
assert result[0].segments[0].duration_min == 2.0
|
||||
|
||||
|
||||
class TestCopyTemplate:
|
||||
def test_copy_writes_to_clip_configs(self, repo, session):
|
||||
tpl = _make_template()
|
||||
repo.create(tpl)
|
||||
seg = _make_segment(tpl.id, order=1, material_type="voiceover")
|
||||
repo.create_segments([seg])
|
||||
copied = repo.copy_template(tpl.id, "u1", "副本模板")
|
||||
assert copied.id != tpl.id
|
||||
clips = session.query(TemplateClipConfigModel).filter(TemplateClipConfigModel.template_id == copied.id).all()
|
||||
assert len(clips) == 1
|
||||
assert clips[0].config["material_type"] == "voiceover"
|
||||
|
||||
def test_copy_empty_segments(self, repo, session):
|
||||
tpl = _make_template()
|
||||
repo.create(tpl)
|
||||
copied = repo.copy_template(tpl.id, "u1", "空副本")
|
||||
assert len(copied.segments) == 0
|
||||
|
||||
|
||||
class TestDelete:
|
||||
def test_delete_cleans_both_tables(self, repo, session):
|
||||
tpl = _make_template()
|
||||
repo.create(tpl)
|
||||
seg = _make_segment(tpl.id, order=1)
|
||||
repo.create_segments([seg])
|
||||
old = TemplateSegmentModel(
|
||||
id=str(uuid.uuid4()), template_id=tpl.id, segment_order=1, duration_min=1.0, duration_max=2.0
|
||||
)
|
||||
session.add(old)
|
||||
session.commit()
|
||||
repo.delete(tpl.id, "u1")
|
||||
c1 = session.query(TemplateClipConfigModel).filter(TemplateClipConfigModel.template_id == tpl.id).count()
|
||||
c2 = session.query(TemplateSegmentModel).filter(TemplateSegmentModel.template_id == tpl.id).count()
|
||||
assert c1 == 0
|
||||
assert c2 == 0
|
||||
|
||||
def test_delete_segments_by_template(self, repo, session):
|
||||
tpl = _make_template()
|
||||
repo.create(tpl)
|
||||
seg = _make_segment(tpl.id, order=1)
|
||||
repo.create_segments([seg])
|
||||
old = TemplateSegmentModel(
|
||||
id=str(uuid.uuid4()), template_id=tpl.id, segment_order=2, duration_min=1.0, duration_max=2.0
|
||||
)
|
||||
session.add(old)
|
||||
session.commit()
|
||||
count = repo.delete_segments_by_template(tpl.id)
|
||||
assert count == 2
|
||||
Reference in New Issue
Block a user