diff --git a/apps/api/app/api/routes/generation_preview.py b/apps/api/app/api/routes/generation_preview.py index 2bf2cb11f..8c961e7c8 100755 --- a/apps/api/app/api/routes/generation_preview.py +++ b/apps/api/app/api/routes/generation_preview.py @@ -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) diff --git a/apps/api/app/api/routes/generation_tasks.py b/apps/api/app/api/routes/generation_tasks.py index ea25de342..d17eb0c2a 100755 --- a/apps/api/app/api/routes/generation_tasks.py +++ b/apps/api/app/api/routes/generation_tasks.py @@ -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) diff --git a/apps/api/app/api/routes/scripts.py b/apps/api/app/api/routes/scripts.py index 6ba326b6e..11f7276ee 100644 --- a/apps/api/app/api/routes/scripts.py +++ b/apps/api/app/api/routes/scripts.py @@ -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 diff --git a/apps/api/app/api/routes/titles.py b/apps/api/app/api/routes/titles.py index cd3cc313e..cf8d7b203 100755 --- a/apps/api/app/api/routes/titles.py +++ b/apps/api/app/api/routes/titles.py @@ -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) diff --git a/apps/api/app/schemas/generation_task.py b/apps/api/app/schemas/generation_task.py index 2b1557a9f..56e471749 100755 --- a/apps/api/app/schemas/generation_task.py +++ b/apps/api/app/schemas/generation_task.py @@ -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 作为标题配置", diff --git a/apps/api/app/schemas/script.py b/apps/api/app/schemas/script.py index fb06738c8..f4acb9ecb 100644 --- a/apps/api/app/schemas/script.py +++ b/apps/api/app/schemas/script.py @@ -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 diff --git a/apps/api/app/services/script_service.py b/apps/api/app/services/script_service.py index 113281d8d..8b069d57e 100644 --- a/apps/api/app/services/script_service.py +++ b/apps/api/app/services/script_service.py @@ -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 diff --git a/migrations/009_merge_title_library_into_scripts.sql b/migrations/009_merge_title_library_into_scripts.sql index 49e7aebdc..8c616b620 100644 --- a/migrations/009_merge_title_library_into_scripts.sql +++ b/migrations/009_merge_title_library_into_scripts.sql @@ -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; diff --git a/packages/adapters/sqlalchemy_impl/models.py b/packages/adapters/sqlalchemy_impl/models.py index 648aa9ca8..168e7b51b 100755 --- a/packages/adapters/sqlalchemy_impl/models.py +++ b/packages/adapters/sqlalchemy_impl/models.py @@ -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)) diff --git a/tests/unit/test_script_title_integration.py b/tests/unit/test_script_title_integration.py index dd4dfa2f6..63ed2707e 100644 --- a/tests/unit/test_script_title_integration.py +++ b/tests/unit/test_script_title_integration.py @@ -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