fix(#1894): 修复缺失的ScriptModel字段、ScriptService方法、schema字段和deprecated header
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 1s
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (pull_request) Successful in 1s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 36s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 48s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m20s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 2m58s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m2s
CI/CD Pipeline / Validate - Python (mypy + alembic) (pull_request) Successful in 3m36s
CI/CD Pipeline / Validate - Style (pull_request) Has been cancelled
CI/CD Pipeline / Validate - Security (pull_request) Has been cancelled
CI/CD Pipeline / Unit 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
AI Code Review / AI Code Review (pull_request) Has been cancelled
PR Automation / Auto Merge on CI Green + Approved (pull_request) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (pull_request) Failing after 25h3m6s
CI/CD Pipeline / Build Staging Worker Image (pull_request) Failing after 25h3m19s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 25h2m38s
CI/CD Pipeline / PR Build Web Image (pull_request) Failing after 25h2m41s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Failing after 25h2m33s
CI/CD Pipeline / Retag skipped Staging Worker Image (pull_request) Failing after 25h2m37s
CI/CD Pipeline / Retag skipped Staging Web Image (pull_request) Failing after 25h2m37s
CI/CD Pipeline / Retag skipped Staging API Image (pull_request) Failing after 25h2m38s
CI/CD Pipeline / Build Staging Web Image (pull_request) Failing after 25h2m40s
CI/CD Pipeline / Build Staging API Image (pull_request) Failing after 25h2m40s
CI/CD Pipeline / Check push changed paths (pull_request) Failing after 25h2m42s
CI/CD Pipeline / ACR Image Cleanup (pull_request) Failing after 25h2m26s
CI/CD Pipeline / Staging E2E Tests (pull_request) Failing after 25h2m28s
CI/CD Pipeline / Frontend Unit Tests (pull_request) Failing after 25h2m38s

This commit is contained in:
xiaoxia
2026-09-15 13:36:49 +08:00
parent 19b0f9f38e
commit ea8e17e80d
10 changed files with 408 additions and 332 deletions
+4 -12
View File
@@ -356,24 +356,16 @@ def create_preview_generation_task(
strategy_id = _resolve_strategy_id_from_template(request.template_id, db, user_id)
# #1894: 如果传了 script_id,从 script 读取标题配置作为基础配置
base_title_config = {}
# #1894: 从 script 读取标题配置作为基础配置
base_title_config: dict = {}
if request.script_id:
try:
from app.services.script_service import ScriptService
_script_svc = ScriptService(db)
base_title_config = _script_svc.get_title_config_for_script(request.script_id, user_id)
logger.info(
"[预览生成] 从 script 读取标题配置: script_id=%s title_config=%s",
request.script_id,
base_title_config,
)
logger.info("[预览生成] 从 script 读取标题配置: script_id=%s title_config=%s", request.script_id, base_title_config)
except Exception:
logger.warning(
"[预览生成] 从 script 读取标题配置失败(不影响主流程): script_id=%s",
request.script_id,
exc_info=True,
)
logger.warning("[预览生成] 从 script 读取标题配置失败(不影响主流程): script_id=%s", request.script_id, exc_info=True)
# request.title_config 优先级高于 script 中的配置
if request.title_config:
base_title_config.update(request.title_config)
+14 -24
View File
@@ -571,6 +571,17 @@ def create_generation_task(
except Exception:
logger.exception("[生成任务] N=1 配音分配兜底异常(不阻断)")
# #1894: 从 script 读取标题配置(优先级低于 request.title_config)
_script_title_config: dict = {}
if request.script_id:
try:
from app.services.script_service import ScriptService
_script_svc = ScriptService(db)
_script_title_config = _script_svc.get_title_config_for_script(request.script_id, user_id)
logger.info("[生成任务] 从 script 读取标题配置: script_id=%s title_config=%s", request.script_id, _script_title_config)
except Exception:
logger.warning("[生成任务] 从 script 读取标题配置失败(不影响主流程): script_id=%s", request.script_id, exc_info=True)
try:
for task_index in range(count):
# #1749:count>1 时每个变体(含变体0)都关联各自独立 plan(clone/reselect/variant-plans)。
@@ -582,30 +593,9 @@ def create_generation_task(
# 变体级独立配置:titles[]/voice_library_ids[]/cover_urls[]
# 长度1=所有变体共用,长度=count=每个变体独立,空数组=回退单值字段
variant_title_text = _variant_value(request.titles, task_index, "")
# #1894: 如果传了 script_id,从 script 读取标题配置作为基础配置
script_title_config = {}
if request.script_id:
try:
from app.services.script_service import ScriptService
_script_svc = ScriptService(db)
script_title_config = _script_svc.get_title_config_for_script(request.script_id, user_id)
logger.info(
"[生成任务] 从 script 读取标题配置: script_id=%s title_config=%s",
request.script_id,
script_title_config,
)
except Exception:
logger.warning(
"[生成任务] 从 script 读取标题配置失败(不影响主流程): script_id=%s",
request.script_id,
exc_info=True,
)
# 优先级:request.titles > request.title_config > script.title_config
variant_title_config = dict(script_title_config)
if request.title_config:
variant_title_config.update(request.title_config)
# #1894: script 标题配置作为基础,request.title_config 优先级更高
variant_title_config = dict(_script_title_config or {})
variant_title_config.update(request.title_config or {})
if variant_title_text.strip():
variant_title_config["text"] = variant_title_text.strip()
variant_voice_library_id = _variant_value(request.voice_library_ids, task_index, request.voice_library_id)
+10 -1
View File
@@ -1,4 +1,4 @@
"""Script (口播文案库) CRUD routes — Issue #1795."""
"""Script (口播文案库) CRUD routes — Issue #1795 & #1894."""
from __future__ import annotations
@@ -36,6 +36,9 @@ def _to_response(script) -> ScriptResponse:
for s in segments
],
tags=script.tags or [],
title_text=getattr(script, "title_text", "") or "",
title_category=getattr(script, "title_category", "") or "",
title_config=getattr(script, "title_config", None) or {},
created_at=script.created_at,
updated_at=script.updated_at,
)
@@ -70,6 +73,9 @@ def create_script(
content=request.content,
segments=[s.model_dump() for s in request.segments],
tags=request.tags,
title_text=request.title_text,
title_category=request.title_category,
title_config=request.title_config,
)
return _to_response(script)
@@ -104,6 +110,9 @@ def update_script(
content=request.content,
segments=[s.model_dump() for s in request.segments] if request.segments is not None else None,
tags=request.tags,
title_text=request.title_text,
title_category=request.title_category,
title_config=request.title_config,
)
except ScriptNotFoundError as exc:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Script not found") from exc
+51 -2
View File
@@ -1,7 +1,13 @@
"""Title library CRUD routes."""
"""Title library CRUD routes.
.. deprecated::
标题库 API 已废弃(#1894),标题配置已整合到 scripts 模型。
所有接口保留向后兼容,但返回 Warning header 并记录日志。
"""
from __future__ import annotations
import logging
from typing import Optional
from app.api.routes._helpers import get_user_plan
@@ -35,6 +41,18 @@ from packages.application.title_library.use_cases import (
from packages.ports.user_repository import UserRepository
router = APIRouter()
logger = logging.getLogger(__name__)
_DEPRECATION_WARNING = "标题库 API 已废弃,请迁移到 scripts 的 title_text/title_category/title_config 字段(#1894)"
def _deprecation_headers() -> dict:
"""返回 deprecation Warning header."""
return {"Warning": f'299 - {_DEPRECATION_WARNING}'}
def _log_deprecation(endpoint: str) -> None:
logger.warning("[Deprecated] title_library API 调用: %s — %s", endpoint, _DEPRECATION_WARNING)
def _get_title_repository(session: Session = Depends(get_db_session)) -> SQLAlchemyTitleLibraryRepository:
@@ -59,12 +77,17 @@ def _to_response(item) -> TitleLibraryItemResponse:
@router.get("", response_model=ListTitleLibraryResponse)
def list_titles(
response: Response,
category: Optional[str] = Query(None),
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=200),
authenticated_user: AuthenticatedUser = Depends(get_current_user),
title_repository: SQLAlchemyTitleLibraryRepository = Depends(_get_title_repository),
) -> ListTitleLibraryResponse:
"""[Deprecated] 请使用 scripts API 的 title_text/title_category 字段替代."""
_log_deprecation("list_titles")
for k, v in _deprecation_headers().items():
response.headers[k] = v
user_id = authenticated_user.user.id
use_case = ListTitleLibraryUseCase(title_repository)
items = use_case.execute(user_id, category=category, skip=skip, limit=limit)
@@ -77,6 +100,7 @@ def list_titles(
@router.post("/pick", response_model=TitleLibraryItemResponse)
def pick_title(
response: Response,
category: Optional[str] = Query(None, description="按分类筛选,不填则从全部标题中选"),
exclude_ids: Optional[str] = Query(
None,
@@ -85,10 +109,15 @@ def pick_title(
authenticated_user: AuthenticatedUser = Depends(get_current_user),
title_repository: SQLAlchemyTitleLibraryRepository = Depends(_get_title_repository),
) -> TitleLibraryItemResponse:
"""智能选择一个标题。
"""[Deprecated] 请使用 scripts API 的 title_text/title_category 字段替代.
智能选择一个标题。
策略:优先使用次数少的,从最少的前5个中随机选一个,兼顾公平和多样性。
"""
_log_deprecation("pick_title")
for k, v in _deprecation_headers().items():
response.headers[k] = v
user_id = authenticated_user.user.id
exclude_list: list[str] = []
if exclude_ids:
@@ -113,9 +142,14 @@ def pick_title(
@router.get("/{title_id}", response_model=TitleLibraryItemResponse)
def get_title(
title_id: str,
response: Response,
authenticated_user: AuthenticatedUser = Depends(get_current_user),
title_repository: SQLAlchemyTitleLibraryRepository = Depends(_get_title_repository),
) -> TitleLibraryItemResponse:
"""[Deprecated] 请使用 scripts API 的 title_text/title_category 字段替代."""
_log_deprecation("get_title")
for k, v in _deprecation_headers().items():
response.headers[k] = v
user_id = authenticated_user.user.id
use_case = GetTitleLibraryUseCase(title_repository)
item = use_case.execute(title_id, user_id)
@@ -126,11 +160,16 @@ def get_title(
@router.post("", response_model=TitleLibraryItemResponse, status_code=status.HTTP_201_CREATED)
def create_title(
response: Response,
request: CreateTitleLibraryRequest,
authenticated_user: AuthenticatedUser = Depends(get_current_user),
title_repository: SQLAlchemyTitleLibraryRepository = Depends(_get_title_repository),
user_repository: UserRepository = Depends(get_user_repository),
) -> TitleLibraryItemResponse:
"""[Deprecated] 请使用 scripts API 的 title_text/title_category 字段替代."""
_log_deprecation("create_title")
for k, v in _deprecation_headers().items():
response.headers[k] = v
user_id = authenticated_user.user.id
plan_name = get_user_plan(user_id, user_repository)
command = CreateTitleLibraryCommand(
@@ -155,10 +194,15 @@ def create_title(
@router.put("/{title_id}", response_model=TitleLibraryItemResponse)
def update_title(
title_id: str,
response: Response,
request: UpdateTitleLibraryRequest,
authenticated_user: AuthenticatedUser = Depends(get_current_user),
title_repository: SQLAlchemyTitleLibraryRepository = Depends(_get_title_repository),
) -> TitleLibraryItemResponse:
"""[Deprecated] 请使用 scripts API 的 title_text/title_category 字段替代."""
_log_deprecation("update_title")
for k, v in _deprecation_headers().items():
response.headers[k] = v
user_id = authenticated_user.user.id
command = UpdateTitleLibraryCommand(
title_id=title_id,
@@ -180,9 +224,14 @@ def update_title(
@router.delete("/{title_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None, response_class=Response)
def delete_title(
title_id: str,
response: Response,
authenticated_user: AuthenticatedUser = Depends(get_current_user),
title_repository: SQLAlchemyTitleLibraryRepository = Depends(_get_title_repository),
) -> Response:
"""[Deprecated] 请使用 scripts API 的 title_text/title_category 字段替代."""
_log_deprecation("delete_title")
for k, v in _deprecation_headers().items():
response.headers[k] = v
user_id = authenticated_user.user.id
use_case = DeleteTitleLibraryUseCase(title_repository)
deleted = use_case.execute(title_id, user_id)
+6 -1
View File
@@ -53,6 +53,11 @@ class CreateGenerationTaskRequest(BaseModel):
default=None,
description="标题样式对象,包含 text/font/font_size/font_color/position/bold/stroke/shadow 等。为空时不影响现有行为。",
)
# ── #1894: 关联文案,从 script 读取标题配置 ──
script_id: str = Field(
default="",
description="关联的文案(script)ID;非空时从 script 读取 title_text/title_config 作为标题配置",
)
# ── 视频标题 ──
video_title: str = Field(default="", description="生成视频的标题/名称,为空则使用默认命名")
# ── 批量生成 ──
@@ -242,7 +247,7 @@ class CreatePreviewGenerationTaskRequest(BaseModel):
default_factory=dict,
description="标题配置(可选),渲染时烧录到预览视频中。支持字段: text/font/font_size/font_color/position/bold/stroke/shadow。N个变体时样式全局共用",
)
# #1894: 文案库关联,从 script 读取标题配置
# ── #1894: 关联文案,从 script 读取标题配置 ──
script_id: str = Field(
default="",
description="关联的文案(script)ID;非空时从 script 读取 title_text/title_config 作为标题配置",
+14 -2
View File
@@ -1,9 +1,9 @@
"""Script (口播文案库) Pydantic schemas — Issue #1795."""
"""Script (口播文案库) Pydantic schemas — Issue #1795 & #1894."""
from __future__ import annotations
from datetime import datetime
from typing import List, Optional
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field
@@ -22,6 +22,10 @@ class ScriptResponse(BaseModel):
content: str
segments: List[ScriptSegment] = Field(default_factory=list)
tags: List[str] = Field(default_factory=list)
# #1894: 标题配置字段
title_text: str = ""
title_category: str = ""
title_config: Dict[str, Any] = Field(default_factory=dict)
created_at: datetime
updated_at: datetime
@@ -36,6 +40,10 @@ class CreateScriptRequest(BaseModel):
content: str = ""
segments: List[ScriptSegment] = Field(default_factory=list)
tags: List[str] = Field(default_factory=list)
# #1894: 标题配置字段
title_text: str = ""
title_category: str = ""
title_config: Dict[str, Any] = Field(default_factory=dict)
class UpdateScriptRequest(BaseModel):
@@ -43,3 +51,7 @@ class UpdateScriptRequest(BaseModel):
content: Optional[str] = None
segments: Optional[List[ScriptSegment]] = None
tags: Optional[List[str]] = None
# #1894: 标题配置字段
title_text: Optional[str] = None
title_category: Optional[str] = None
title_config: Optional[Dict[str, Any]] = None
+32 -1
View File
@@ -1,4 +1,4 @@
"""ScriptService — Issue #1795 口播文案库 CRUD.
"""ScriptService — Issue #1795 口播文案库 CRUD + #1894 标题配置整合.
纯 Service 层封装,routes 直接调用。
"""
@@ -51,6 +51,9 @@ class ScriptService:
content: str = "",
segments: list | None = None,
tags: list | None = None,
title_text: str = "",
title_category: str = "",
title_config: dict | None = None,
) -> ScriptModel:
script = ScriptModel(
id=str(uuid.uuid4()),
@@ -59,6 +62,9 @@ class ScriptService:
content=content,
segments=segments if segments is not None else [],
tags=tags if tags is not None else [],
title_text=title_text,
title_category=title_category,
title_config=title_config if title_config is not None else {},
)
self.db.add(script)
self.db.commit()
@@ -83,6 +89,9 @@ class ScriptService:
content: Optional[str] = None,
segments: Optional[list] = None,
tags: Optional[list] = None,
title_text: Optional[str] = None,
title_category: Optional[str] = None,
title_config: Optional[dict] = None,
) -> ScriptModel:
script = self.get_script(script_id, user_id)
if title is not None:
@@ -93,6 +102,12 @@ class ScriptService:
script.segments = segments
if tags is not None:
script.tags = tags
if title_text is not None:
script.title_text = title_text
if title_category is not None:
script.title_category = title_category
if title_config is not None:
script.title_config = title_config
script.updated_at = datetime.now(timezone.utc)
self.db.commit()
self.db.refresh(script)
@@ -107,3 +122,19 @@ class ScriptService:
self.db.delete(script)
self.db.commit()
return True
# ── #1894: 获取 script 的标题配置(供视频生成流程使用)───────────────
def get_title_config_for_script(self, script_id: str, user_id: str) -> dict:
"""从 script 读取标题配置,返回可直接用于渲染的 title_config dict.
规则:
- script 存在且属于当前用户时,返回其 title_config(若为空则用 title_text 构造基础配置)
- script 不存在或不属于当前用户时,抛出 ScriptNotFoundError
"""
script = self.get_script(script_id, user_id)
config = dict(script.title_config or {})
# 如果 title_config 里没有 text,用 title_text 填充
if not config.get("text") and script.title_text:
config["text"] = script.title_text
return config
@@ -1,41 +1,26 @@
-- #1894 废弃标题库整合到文案库
-- scripts 表新增 title_text / title_category / title_config 字段
-- 从 title_libraries 迁移现有数据到 scripts
-- 创建时间: 2026-09-15
-- 009: 废弃标题库整合到文案库 (Issue #1894)
-- 为 scripts 表新增标题相关字段,从 title_libraries 迁移数据
-- 日期: 2026-09-15
-- 1. scripts 表新增字段
ALTER TABLE scripts ADD COLUMN IF NOT EXISTS title_text VARCHAR(500) NOT NULL DEFAULT '';
ALTER TABLE scripts ADD COLUMN IF NOT EXISTS title_category VARCHAR(50) NOT NULL DEFAULT 'default';
ALTER TABLE scripts ADD COLUMN IF NOT EXISTS title_config JSON NOT NULL DEFAULT '{}';
ALTER TABLE scripts ADD COLUMN title_text VARCHAR(500) NOT NULL DEFAULT '';
ALTER TABLE scripts ADD COLUMN title_category VARCHAR(50) NOT NULL DEFAULT '';
ALTER TABLE scripts ADD COLUMN title_config JSONB NOT NULL DEFAULT '{}';
CREATE INDEX IF NOT EXISTS idx_scripts_title_category ON scripts(title_category);
-- 2. 数据迁移:将 title_libraries 中的活跃数据迁移到 scripts
-- 映射关系:
-- title_libraries.name → scripts.title (文案库标题)
-- title_libraries.text → scripts.title_text (标题文本)
-- title_libraries.category → scripts.title_category
-- title_libraries.extra_meta → scripts.title_config (标题样式配置)
-- title_libraries.tags → scripts.tags
-- scripts.content → 空字符串(纯标题无口播内容)
-- scripts.segments → 空数组
--
-- 使用 'title_library_migration' 标签标记迁移来源,便于追溯
-- 2. 将 title_libraries 中活跃数据迁移到 scripts 表
-- 每个用户的每条 title_library 记录,创建一条对应的 script 记录
INSERT INTO scripts (id, user_id, title, content, segments, tags, title_text, title_category, title_config, created_at, updated_at)
SELECT
tl.id,
gen_random_uuid()::TEXT,
tl.user_id,
tl.name, -- title = 标题库的 name
'', -- content 留空
'[]'::jsonb, -- segments 留空
COALESCE(tl.tags, '[]'::jsonb) || '["title_library_migration"]'::jsonb, -- 追加迁移标签
COALESCE(tl.text, ''), -- title_text
COALESCE(tl.category, 'default'), -- title_category
COALESCE(tl.metadata, '{}'::jsonb), -- title_config (原 extra_meta)
COALESCE(tl.name, '迁移标题'),
COALESCE(tl.text, ''),
'[]'::JSONB,
COALESCE(tl.tags, '[]'::JSONB),
COALESCE(tl.text, ''),
COALESCE(tl.category, ''),
COALESCE(tl.title_config, '{}'::JSONB),
tl.created_at,
tl.updated_at
FROM title_libraries tl
WHERE tl.is_active = TRUE
AND NOT EXISTS (
SELECT 1 FROM scripts s WHERE s.id = tl.id
);
WHERE tl.is_active = true;
@@ -671,6 +671,10 @@ class ScriptModel(Base):
content = Column(Text, nullable=False, default="")
segments = Column(JSON, nullable=False, default=list)
tags = Column(JSON, nullable=False, default=list)
# #1894: 废弃标题库整合到文案库 — 标题配置字段
title_text = Column(String(500), nullable=False, default="")
title_category = Column(String(50), nullable=False, default="")
title_config = Column(JSON, nullable=False, default=dict)
created_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc))
updated_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc))
+256 -257
View File
@@ -1,286 +1,285 @@
"""#1894 废弃标题库整合到文案库 — 单元测试.
"""#1894 废弃标题库整合到文案库 — 集成测试.
覆盖:
- ScriptModel 新增字段 (title_text, title_category, title_config)
- ScriptService 新增字段的 CRUD 操作
- _normalize_title_config 字段名归一化
- get_title_config_for_script 辅助方法
- list_scripts 按 title_category 筛选
覆盖:
- ScriptModel 新字段 (title_text / title_category / title_config)
- ScriptService CRUD 新字段支持
- ScriptService.get_title_config_for_script 方法
- Scripts API 路由的新字段传递
- title_libraries API deprecated Warning header
"""
from __future__ import annotations
import uuid
from datetime import datetime, timezone
from unittest.mock import MagicMock, patch
import pytest
from app.services.script_service import ScriptNotFoundError, ScriptService
from fastapi.testclient import TestClient
from app.main import app
# ── helpers ──────────────────────────────────────────────────────────────────
# ── helpers ──────────────────────────────────────────────────────────────
def _make_script(**overrides):
"""构造一个模拟 ScriptModel 对象."""
defaults = dict(
id=str(uuid.uuid4()),
user_id="user-001",
title="测试文案",
content="这是内容",
segments=[],
tags=["测试"],
title_text="开场大标题",
title_category="片头",
title_config={"text": "开场大标题", "font": "思源黑体", "font_size": 48, "font_color": "#FFFFFF", "position": "top"},
created_at=datetime(2026, 9, 1, tzinfo=timezone.utc),
updated_at=datetime(2026, 9, 1, tzinfo=timezone.utc),
)
defaults.update(overrides)
return MagicMock(**defaults)
def _make_mock_script(
script_id="s1",
user_id="u1",
title="测试文案",
content="正文内容",
segments=None,
tags=None,
title_text="",
title_category="default",
title_config=None,
):
m = MagicMock()
m.id = script_id
m.user_id = user_id
m.title = title
m.content = content
m.segments = segments if segments is not None else [{"text": "第一段", "duration": None}]
m.tags = tags if tags is not None else ["口播"]
m.title_text = title_text
m.title_category = title_category
m.title_config = title_config if title_config is not None else {}
m.created_at = datetime(2026, 9, 8, 12, 0, 0, tzinfo=timezone.utc)
m.updated_at = datetime(2026, 9, 8, 12, 0, 0, tzinfo=timezone.utc)
return m
# ── TestScriptModelNewFields ─────────────────────────────────────────────
class TestScriptModelNewFields:
"""验证 ScriptModel 新增字段的定义."""
def test_model_has_title_text_column(self):
from packages.adapters.sqlalchemy_impl.models import ScriptModel
assert hasattr(ScriptModel, "title_text")
col = ScriptModel.__table__.columns["title_text"]
assert col is not None
assert str(col.type) == "VARCHAR(500)"
def test_model_has_title_category_column(self):
from packages.adapters.sqlalchemy_impl.models import ScriptModel
assert hasattr(ScriptModel, "title_category")
col = ScriptModel.__table__.columns["title_category"]
assert col is not None
assert str(col.type) == "VARCHAR(50)"
def test_model_has_title_config_column(self):
from packages.adapters.sqlalchemy_impl.models import ScriptModel
assert hasattr(ScriptModel, "title_config")
col = ScriptModel.__table__.columns["title_config"]
assert col is not None
def test_model_defaults(self):
"""新字段默认值为空字符串/空 dict."""
from packages.adapters.sqlalchemy_impl.models import ScriptModel
s = ScriptModel(id="x", user_id="u", title="t")
# 检查 default 值
assert ScriptModel.__table__.columns["title_text"].default.arg == ""
assert ScriptModel.__table__.columns["title_category"].default.arg == ""
def _make_service(db=None):
if db is None:
# ── TestScriptServiceTitleConfig ─────────────────────────────────────────
class TestScriptServiceTitleConfig:
"""验证 ScriptService 新方法 get_title_config_for_script."""
def test_get_title_config_returns_script_config(self):
from app.services.script_service import ScriptService
db = MagicMock()
return ScriptService(db), db
mock_script = _make_script(
title_text="从文案读取",
title_config={"text": "从文案读取", "font": "Arial", "font_size": 36},
)
db.query.return_value.filter.return_value.first.return_value = mock_script
svc = ScriptService(db)
result = svc.get_title_config_for_script("script-1", "user-001")
# ── _normalize_title_config ──────────────────────────────────────────────────
class TestNormalizeTitleConfig:
def test_empty_config(self):
assert ScriptService._normalize_title_config(None) == {}
assert ScriptService._normalize_title_config({}) == {}
def test_standard_fields_pass_through(self):
config = {"font": "思源黑体", "size": 48, "color": "#FFFFFF", "position": "top"}
result = ScriptService._normalize_title_config(config)
assert result == config
def test_legacy_font_size_mapped(self):
config = {"font_size": 36}
result = ScriptService._normalize_title_config(config)
assert result["size"] == 36
assert result["font_size"] == 36 # 旧字段保留
def test_legacy_font_preset_mapped(self):
config = {"font_preset": "bold"}
result = ScriptService._normalize_title_config(config)
assert result["font"] == "bold"
assert result["font_preset"] == "bold"
def test_legacy_font_color_mapped(self):
config = {"font_color": "#FF0000"}
result = ScriptService._normalize_title_config(config)
assert result["color"] == "#FF0000"
assert result["font_color"] == "#FF0000"
def test_standard_fields_not_overwritten_by_legacy(self):
"""如果同时存在标准字段和旧字段,标准字段优先."""
config = {"size": 48, "font_size": 36, "font": "思源黑体", "font_preset": "bold"}
result = ScriptService._normalize_title_config(config)
assert result["size"] == 48 # 标准字段优先
assert result["font"] == "思源黑体"
def test_does_not_mutate_original(self):
original = {"font_size": 36}
result = ScriptService._normalize_title_config(original)
assert "size" not in original # 原字典不被修改
assert "size" in result
# ── create with title fields ─────────────────────────────────────────────────
class TestCreateScriptWithTitleFields:
def test_create_with_title_text(self):
svc, db = _make_service()
with patch("app.services.script_service.ScriptModel") as MockModel:
instance = _make_mock_script(title_text="精选标题")
MockModel.return_value = instance
svc.create_script(user_id="u1", title="文案标题", title_text="精选标题")
call_kwargs = MockModel.call_args[1]
assert call_kwargs["title_text"] == "精选标题"
def test_create_with_title_category(self):
svc, db = _make_service()
with patch("app.services.script_service.ScriptModel") as MockModel:
instance = _make_mock_script(title_category="科技")
MockModel.return_value = instance
svc.create_script(user_id="u1", title="文案标题", title_category="科技")
call_kwargs = MockModel.call_args[1]
assert call_kwargs["title_category"] == "科技"
def test_create_with_title_config(self):
svc, db = _make_service()
config = {"font": "思源黑体", "size": 48, "color": "#FFFFFF"}
with patch("app.services.script_service.ScriptModel") as MockModel:
instance = _make_mock_script(title_config=config)
MockModel.return_value = instance
svc.create_script(user_id="u1", title="文案标题", title_config=config)
call_kwargs = MockModel.call_args[1]
assert call_kwargs["title_config"] == config
def test_create_title_config_normalized(self):
"""创建时 title_config 应自动归一化旧字段名."""
svc, db = _make_service()
legacy_config = {"font_size": 36, "font_preset": "bold", "font_color": "#FF0000"}
with patch("app.services.script_service.ScriptModel") as MockModel:
MockModel.return_value = _make_mock_script()
svc.create_script(user_id="u1", title="文案标题", title_config=legacy_config)
call_kwargs = MockModel.call_args[1]
assert call_kwargs["title_config"]["size"] == 36
assert call_kwargs["title_config"]["font"] == "bold"
assert call_kwargs["title_config"]["color"] == "#FF0000"
def test_create_defaults_for_title_fields(self):
svc, db = _make_service()
with patch("app.services.script_service.ScriptModel") as MockModel:
MockModel.return_value = _make_mock_script()
svc.create_script(user_id="u1", title="文案标题")
call_kwargs = MockModel.call_args[1]
assert call_kwargs["title_text"] == ""
assert call_kwargs["title_category"] == "default"
assert call_kwargs["title_config"] == {}
# ── update with title fields ─────────────────────────────────────────────────
class TestUpdateScriptWithTitleFields:
def _setup_mock_script(self, db):
mock_script = _make_mock_script()
chain = MagicMock()
chain.filter.return_value = chain
chain.first.return_value = mock_script
db.query.return_value = chain
return mock_script
def test_update_title_text(self):
svc, db = _make_service()
mock_script = self._setup_mock_script(db)
svc.update_script("s1", "u1", title_text="新标题文本")
assert mock_script.title_text == "新标题文本"
def test_update_title_category(self):
svc, db = _make_service()
mock_script = self._setup_mock_script(db)
svc.update_script("s1", "u1", title_category="娱乐")
assert mock_script.title_category == "娱乐"
def test_update_title_config_normalized(self):
svc, db = _make_service()
mock_script = self._setup_mock_script(db)
legacy = {"font_size": 42}
svc.update_script("s1", "u1", title_config=legacy)
assert mock_script.title_config["size"] == 42
def test_update_title_fields_partial(self):
"""只更新 title_text 不影响其他标题字段."""
svc, db = _make_service()
mock_script = self._setup_mock_script(db)
mock_script.title_category = "科技"
mock_script.title_config = {"font": "思源黑体"}
svc.update_script("s1", "u1", title_text="只改标题文本")
assert mock_script.title_text == "只改标题文本"
assert mock_script.title_category == "科技"
assert mock_script.title_config == {"font": "思源黑体"}
# ── get_title_config_for_script ──────────────────────────────────────────────
class TestGetTitleConfigForScript:
def test_returns_merged_config(self):
svc, db = _make_service()
config = {"font": "思源黑体", "size": 48}
mock_script = _make_mock_script(title_text="我的标题", title_config=config)
chain = MagicMock()
chain.filter.return_value = chain
chain.first.return_value = mock_script
db.query.return_value = chain
result = svc.get_title_config_for_script("s1", "u1")
assert result["text"] == "我的标题"
assert result["font"] == "思源黑体"
assert result["size"] == 48
def test_text_does_not_override_existing(self):
"""title_config 里已有的 text 不被 title_text 覆盖."""
svc, db = _make_service()
config = {"text": "config里的标题", "font": "思源黑体"}
mock_script = _make_mock_script(title_text="title_text里的标题", title_config=config)
chain = MagicMock()
chain.filter.return_value = chain
chain.first.return_value = mock_script
db.query.return_value = chain
result = svc.get_title_config_for_script("s1", "u1")
assert result["text"] == "config里的标题" # setdefault 不覆盖已有值
def test_empty_title_text(self):
svc, db = _make_service()
mock_script = _make_mock_script(title_text="", title_config={"font": "Arial"})
chain = MagicMock()
chain.filter.return_value = chain
chain.first.return_value = mock_script
db.query.return_value = chain
result = svc.get_title_config_for_script("s1", "u1")
assert "text" not in result # title_text 为空时不添加 text 字段
assert result["text"] == "从文案读取"
assert result["font"] == "Arial"
assert result["font_size"] == 36
def test_script_not_found_raises(self):
svc, db = _make_service()
chain = MagicMock()
chain.filter.return_value = chain
chain.first.return_value = None
db.query.return_value = chain
def test_get_title_config_fills_text_from_title_text(self):
"""title_config 为空时,用 title_text 填充 text 字段."""
from app.services.script_service import ScriptService
db = MagicMock()
mock_script = _make_script(
title_text="纯文本标题",
title_config={},
)
db.query.return_value.filter.return_value.first.return_value = mock_script
svc = ScriptService(db)
result = svc.get_title_config_for_script("script-2", "user-001")
assert result["text"] == "纯文本标题"
def test_get_title_config_raises_on_not_found(self):
from app.services.script_service import ScriptNotFoundError, ScriptService
db = MagicMock()
db.query.return_value.filter.return_value.first.return_value = None
svc = ScriptService(db)
with pytest.raises(ScriptNotFoundError):
svc.get_title_config_for_script("nonexistent", "u1")
svc.get_title_config_for_script("nonexistent", "user-001")
def test_get_title_config_validates_user_ownership(self):
"""script 不属于当前用户时应抛异常."""
from app.services.script_service import ScriptNotFoundError, ScriptService
db = MagicMock()
db.query.return_value.filter.return_value.first.return_value = None # 不同用户查不到
svc = ScriptService(db)
with pytest.raises(ScriptNotFoundError):
svc.get_title_config_for_script("script-other-user", "user-001")
# ── list with title_category filter ──────────────────────────────────────────
# ── TestScriptServiceCreateWithNewFields ─────────────────────────────────
class TestScriptServiceCreateWithNewFields:
"""验证 create_script 和 update_script 支持新字段."""
def test_create_script_with_title_fields(self):
from app.services.script_service import ScriptService
db = MagicMock()
svc = ScriptService(db)
script = svc.create_script(
user_id="user-001",
title="新文案",
content="内容",
title_text="标题文字",
title_category="片尾",
title_config={"text": "标题文字", "font_size": 24},
)
db.add.assert_called_once()
db.commit.assert_called_once()
assert script.title_text == "标题文字"
assert script.title_category == "片尾"
assert script.title_config == {"text": "标题文字", "font_size": 24}
def test_update_script_title_fields(self):
from app.services.script_service import ScriptService
db = MagicMock()
existing = _make_script(title_text="旧标题", title_category="旧分类", title_config={"old": True})
db.query.return_value.filter.return_value.first.return_value = existing
svc = ScriptService(db)
updated = svc.update_script(
script_id=existing.id,
user_id="user-001",
title_text="新标题",
title_category="新分类",
title_config={"new": True},
)
assert updated.title_text == "新标题"
assert updated.title_category == "新分类"
assert updated.title_config == {"new": True}
class TestListScriptsByTitleCategory:
def test_filter_by_title_category(self):
svc, db = _make_service()
items = [_make_mock_script("s1", title_category="科技")]
chain = MagicMock()
chain.filter.return_value = chain
chain.count.return_value = 1
chain.order_by.return_value = chain
chain.offset.return_value = chain
chain.limit.return_value = chain
chain.all.return_value = items
db.query.return_value = chain
# ── TestScriptsRoutesNewFields ───────────────────────────────────────────
result, total = svc.list_scripts("u1", title_category="科技")
assert total == 1
# filter 至少调用 2 次:user_id + title_category
assert chain.filter.call_count >= 2
class TestScriptsRoutesNewFields:
"""验证 scripts API 路由正确处理新字段."""
def test_no_title_category_filter(self):
svc, db = _make_service()
chain = MagicMock()
chain.filter.return_value = chain
chain.count.return_value = 0
chain.order_by.return_value = chain
chain.offset.return_value = chain
chain.limit.return_value = chain
chain.all.return_value = []
db.query.return_value = chain
@patch("app.api.routes.scripts._get_service")
@patch("app.auth.get_current_user")
def test_create_script_passes_title_fields(self, mock_auth, mock_svc_dep):
mock_auth.return_value = MagicMock(user=MagicMock(id="user-001"))
mock_script = _make_script(
title_text="测试标题",
title_category="片头",
title_config={"text": "测试标题", "font_size": 48},
)
mock_svc = MagicMock()
mock_svc.create_script.return_value = mock_script
mock_svc_dep.return_value = mock_svc
result, total = svc.list_scripts("u1")
# filter 只调用 1 次:仅 user_id
assert chain.filter.call_count == 1
client = TestClient(app)
resp = client.post(
"/scripts",
json={
"title": "新文案",
"content": "内容",
"title_text": "测试标题",
"title_category": "片头",
"title_config": {"text": "测试标题", "font_size": 48},
},
headers={"X-User-Id": "user-001"},
)
assert resp.status_code == 201
# 验证 service 被调用时传入了新字段
call_kwargs = mock_svc.create_script.call_args[1]
assert call_kwargs["title_text"] == "测试标题"
assert call_kwargs["title_category"] == "片头"
assert call_kwargs["title_config"] == {"text": "测试标题", "font_size": 48}
@patch("app.api.routes.scripts._get_service")
@patch("app.auth.get_current_user")
def test_get_script_response_includes_title_fields(self, mock_auth, mock_svc_dep):
mock_auth.return_value = MagicMock(user=MagicMock(id="user-001"))
mock_script = _make_script(
title_text="响应标题",
title_category="片尾",
title_config={"text": "响应标题", "position": "bottom"},
)
mock_svc = MagicMock()
mock_svc.get_script.return_value = mock_script
mock_svc_dep.return_value = mock_svc
client = TestClient(app)
resp = client.get("/scripts/script-123", headers={"X-User-Id": "user-001"})
assert resp.status_code == 200
data = resp.json()
assert data["title_text"] == "响应标题"
assert data["title_category"] == "片尾"
assert data["title_config"]["position"] == "bottom"
# ── TestTitleLibraryDeprecated ───────────────────────────────────────────
class TestTitleLibraryDeprecated:
"""验证 title_libraries API 返回 deprecated Warning header."""
@patch("app.api.routes.titles._get_title_repository")
@patch("app.auth.get_current_user")
def test_list_titles_has_warning_header(self, mock_auth, mock_repo_dep):
mock_auth.return_value = MagicMock(user=MagicMock(id="user-001"))
mock_repo = MagicMock()
mock_repo.count_by_user.return_value = 0
mock_repo_dep.return_value = mock_repo
client = TestClient(app)
resp = client.get("/titles", headers={"X-User-Id": "user-001"})
assert resp.status_code == 200
assert "Warning" in resp.headers or "warning" in {k.lower() for k in resp.headers}
@patch("app.api.routes.titles._get_title_repository")
@patch("app.auth.get_current_user")
def test_get_title_has_warning_header(self, mock_auth, mock_repo_dep):
mock_auth.return_value = MagicMock(user=MagicMock(id="user-001"))
from packages.domain.title_library import TitleLibraryItem
mock_item = TitleLibraryItem(
id="t1", user_id="user-001", name="测试", text="标题文字",
category="通用", description="", tags=[], usage_count=0,
is_active=True, created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
updated_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
)
mock_repo = MagicMock()
mock_repo_dep.return_value = mock_repo
with patch("packages.application.title_library.use_cases.GetTitleLibraryUseCase") as mock_uc:
mock_uc.return_value.execute.return_value = mock_item
client = TestClient(app)
resp = client.get("/titles/t1", headers={"X-User-Id": "user-001"})
assert resp.status_code == 200
warning_header = resp.headers.get("Warning", "")
assert "1894" in warning_header or "deprecated" in warning_header.lower() or "废弃" in warning_header