Compare commits
5 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| b64d384b91 | |||
| 6a4913a3b9 | |||
| bc85c79f39 | |||
| 08cad1f1ee | |||
| 29a127c7f1 |
@@ -355,7 +355,20 @@ def create_preview_generation_task(
|
||||
)
|
||||
|
||||
strategy_id = _resolve_strategy_id_from_template(request.template_id, db, user_id)
|
||||
base_title_config = request.title_config or {}
|
||||
|
||||
# #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)
|
||||
except Exception:
|
||||
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)
|
||||
|
||||
use_case = CreateGenerationTaskUseCase(generation_task_repository)
|
||||
|
||||
|
||||
@@ -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,7 +593,9 @@ def create_generation_task(
|
||||
# 变体级独立配置:titles[]/voice_library_ids[]/cover_urls[]
|
||||
# 长度1=所有变体共用,长度=count=每个变体独立,空数组=回退单值字段
|
||||
variant_title_text = _variant_value(request.titles, task_index, "")
|
||||
variant_title_config = dict(request.title_config or {})
|
||||
# #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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -43,6 +43,11 @@ class CreateGenerationTaskRequest(BaseModel):
|
||||
default_factory=list,
|
||||
description="POST /generation/variant-plans 返回的各变体 plan_id(长度须=count);为空则走服务端选片",
|
||||
)
|
||||
# ── 文案库关联(#1894 标题库整合到文案库)──
|
||||
script_id: str = Field(
|
||||
default="",
|
||||
description="关联的文案(script)ID;非空时从 script 读取 title_text/title_config 作为标题配置",
|
||||
)
|
||||
# ── 标题配置(结构化)──
|
||||
title_config: dict | None = Field(
|
||||
default=None,
|
||||
@@ -237,6 +242,11 @@ class CreatePreviewGenerationTaskRequest(BaseModel):
|
||||
default_factory=dict,
|
||||
description="标题配置(可选),渲染时烧录到预览视频中。支持字段: text/font/font_size/font_color/position/bold/stroke/shadow。N个变体时样式全局共用",
|
||||
)
|
||||
# ── #1894: 关联文案,从 script 读取标题配置 ──
|
||||
script_id: str = Field(
|
||||
default="",
|
||||
description="关联的文案(script)ID;非空时从 script 读取 title_text/title_config 作为标题配置",
|
||||
)
|
||||
# ── 多变体独立配置(preview_count > 1)──
|
||||
# 长度 1 = 所有变体共用;长度 = preview_count = 每个变体独立;空数组 = 回退单值字段
|
||||
titles: list[str] = Field(
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -0,0 +1,26 @@
|
||||
-- 009: 废弃标题库整合到文案库 (Issue #1894)
|
||||
-- 为 scripts 表新增标题相关字段,从 title_libraries 迁移数据
|
||||
-- 日期: 2026-09-15
|
||||
|
||||
-- 1. scripts 表新增字段
|
||||
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 '{}';
|
||||
|
||||
-- 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
|
||||
gen_random_uuid()::TEXT,
|
||||
tl.user_id,
|
||||
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;
|
||||
@@ -666,6 +666,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))
|
||||
|
||||
|
||||
@@ -0,0 +1,307 @@
|
||||
"""#1894 废弃标题库整合到文案库 — 集成测试.
|
||||
|
||||
覆盖:
|
||||
- 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.main import app
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
# ── 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)
|
||||
|
||||
|
||||
# ── 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 == ""
|
||||
|
||||
|
||||
# ── 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()
|
||||
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")
|
||||
|
||||
assert result["text"] == "从文案读取"
|
||||
assert result["font"] == "Arial"
|
||||
assert result["font_size"] == 36
|
||||
|
||||
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", "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")
|
||||
|
||||
|
||||
# ── 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}
|
||||
|
||||
|
||||
# ── TestScriptsRoutesNewFields ───────────────────────────────────────────
|
||||
|
||||
|
||||
class TestScriptsRoutesNewFields:
|
||||
"""验证 scripts API 路由正确处理新字段."""
|
||||
|
||||
@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
|
||||
|
||||
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
|
||||
@@ -70,12 +70,21 @@ class TestUpdateScriptRequest:
|
||||
assert r.content is None
|
||||
assert r.segments is None
|
||||
assert r.tags is None
|
||||
assert r.title_text is None
|
||||
assert r.title_category is None
|
||||
assert r.title_config is None
|
||||
|
||||
def test_partial_update(self):
|
||||
r = UpdateScriptRequest(title="新标题")
|
||||
assert r.title == "新标题"
|
||||
assert r.content is None
|
||||
|
||||
def test_partial_update_title_fields(self):
|
||||
r = UpdateScriptRequest(title_text="新标题文本", title_category="娱乐")
|
||||
assert r.title_text == "新标题文本"
|
||||
assert r.title_category == "娱乐"
|
||||
assert r.title is None
|
||||
|
||||
|
||||
class TestScriptResponse:
|
||||
def test_response_construction(self):
|
||||
@@ -87,11 +96,28 @@ class TestScriptResponse:
|
||||
content="内容",
|
||||
segments=[ScriptSegment(text="段1")],
|
||||
tags=["t1"],
|
||||
title_text="标题文案",
|
||||
title_category="科技",
|
||||
title_config={"font": "思源黑体", "size": 48},
|
||||
created_at=now,
|
||||
updated_at=now,
|
||||
)
|
||||
assert r.id == "s1"
|
||||
assert len(r.segments) == 1
|
||||
assert r.title_text == "标题文案"
|
||||
assert r.title_category == "科技"
|
||||
assert r.title_config["font"] == "思源黑体"
|
||||
|
||||
def test_response_defaults(self):
|
||||
"""新字段有默认值,不传也能构造."""
|
||||
now = datetime(2026, 9, 8, 12, 0, 0, tzinfo=timezone.utc)
|
||||
r = ScriptResponse(
|
||||
id="s1", user_id="u1", title="标题", content="",
|
||||
segments=[], tags=[], created_at=now, updated_at=now,
|
||||
)
|
||||
assert r.title_text == ""
|
||||
assert r.title_category == "default"
|
||||
assert r.title_config == {}
|
||||
|
||||
|
||||
class TestScriptListResponse:
|
||||
|
||||
Reference in New Issue
Block a user