fix(#1895): /subscription/plans 路由挂载位置修复（解决 404） (#1955)

Co-authored-by: xiaoxia <dev@xiaoxiajianji.com>
Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>

.ci-trigger

@@ -1,2 +1,2 @@
 CI trigger file - safe to delete
-updated!
+retrigger at 2026-09-15 20:31:24 UTC

.env.example

@@ -196,3 +196,10 @@ DOUBAO_MODEL=doubao-seed-1-6-250615
 DOUBAO_BASE_URL=https://ark.cn-beijing.volces.com/api/v3
 DOUBAO_TIMEOUT=30
 DOUBAO_MAX_RETRIES=2
+
+# ==================== 积分/会员系统 (#1895) ====================
+# 积分扣点总开关：默认 false（对现有用户零影响）。
+# P2 阶段各业务路由逐个接入 @points_gate 时，用
+#   `if settings.points_enabled: ...`
+# 包裹扣点逻辑；所有路由接入完成并验证通过后再在 staging/prod 打开。
+POINTS_ENABLED=false

alembic/versions/077_merge_title_libs.py

@@ -0,0 +1,67 @@
+"""#1894: merge title_libraries into scripts — add title_text/title_category/title_config
+
+Revision ID: 077_merge_title_libs
+Revises: 076_membership_points
+Create Date: 2026-09-15
+"""
+
+import sqlalchemy as sa
+
+from alembic import context, op
+
+revision = "077_merge_title_libs"
+down_revision = "076_membership_points"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("scripts") as batch:
+        batch.add_column(
+            sa.Column("title_text", sa.String(500), nullable=False, server_default=""),
+        )
+        batch.add_column(
+            sa.Column("title_category", sa.String(50), nullable=False, server_default=""),
+        )
+        batch.add_column(
+            sa.Column("title_config", sa.JSON, nullable=False, server_default="{}"),
+        )
+
+    if context.get_context().dialect.name == "postgresql":
+        conn = op.get_bind()
+        result = conn.execute(sa.text("SELECT to_regclass('public.title_libraries')"))
+        if result.scalar() is not None:
+            conn.execute(sa.text("""
+                    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."metadata", '{}'::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.user_id = tl.user_id
+                          AND s.title_text = COALESCE(tl.text, '')
+                          AND s.title_category = COALESCE(tl.category, '')
+                          AND s.created_at = tl.created_at
+                      )
+                    """))
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("scripts") as batch:
+        batch.drop_column("title_config")
+        batch.drop_column("title_category")
+        batch.drop_column("title_text")

apps/api/app/api/routes/ai_avatar_render.py

@@ -29,6 +29,8 @@ from app.services.ai_avatar_render_service import (
 from fastapi import APIRouter, Depends, HTTPException, Query
 from sqlalchemy.orm import Session
 
+from packages.middleware.points_gate import points_gate
+
 logger = logging.getLogger(__name__)
 
 router = APIRouter()
@@ -42,10 +44,12 @@ def _get_service(db: Session = Depends(get_db_session)) -> AiAvatarRenderService
 
 
 @router.post("", response_model=AiAvatarRenderJobResponse, status_code=201)
+@points_gate("ai_digital_human", per_unit=15)
 def create_render_job(
     body: CreateAiAvatarRenderRequest,
     current_user: AuthenticatedUser = Depends(get_current_user),
     svc: AiAvatarRenderService = Depends(_get_service),
+    db: Session = Depends(get_db_session),
 ):
     """提交 AI 数字人渲染任务.
 

apps/api/app/api/routes/auth.py

@@ -13,7 +13,7 @@ from typing import Optional
 import jwt
 from app.auth import AuthenticatedUser, blacklist_token, get_current_user
 from app.config import settings
-from app.dependencies import get_auth_email_service, get_auth_session_store, get_user_repository
+from app.dependencies import get_auth_email_service, get_auth_session_store, get_db_session, get_user_repository
 from fastapi import APIRouter, Depends, Header, HTTPException, Request, status
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from pydantic import BaseModel, EmailStr, field_validator
@@ -126,6 +126,7 @@ async def register(
     request: RegisterRequest,
     user_repository: UserRepository = Depends(get_user_repository),
     email_service=Depends(get_auth_email_service),
+    db=Depends(get_db_session),
 ) -> RegisterResponse:
     use_case = RegisterUserUseCase(
         user_repository=user_repository,
@@ -143,6 +144,22 @@ async def register(
     if error or response is None:
         raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=_translate_auth_error(error))
 
+    # 新用户注册赠送 50 积分（失败不影响注册）
+    if settings.points_enabled:
+        try:
+            from packages.domain.points_service import PointsService
+            _svc = PointsService()
+            _svc.add_points(
+                user_id=response.user_id,
+                amount=50,
+                source="task_reward",
+                db=db,
+                description="新用户注册赠送",
+            )
+        except Exception as _bonus_err:
+            import logging
+            logging.getLogger(__name__).warning("注册送积分失败: user_id=%s err=%s", response.user_id, _bonus_err)
+
     return RegisterResponse(
         user_id=response.user_id,
         email=response.email,

apps/api/app/api/routes/generation_cover.py

@@ -27,6 +27,7 @@ from packages.adapters.sqlalchemy_impl.generation_task_repository import (
 )
 from packages.application import ListGeneratedVideosByTaskUseCase
 from packages.domain.config_schemas import normalize_plan_config
+from packages.middleware.points_gate import points_gate
 from packages.shared.storage import get_shared_storage_service
 
 from .templates_editor.dependencies import get_draft_plan_id, get_editor_services
@@ -331,6 +332,7 @@ def _is_trusted_media_url(url: str) -> bool:
 
 
 @router.post("/generate-cover", response_model=GenerateCoverResponse)
+@points_gate("ai_cover")
 def generate_cover(
     body: GenerateCoverRequest,
     template_id: str = Query(..., description="模板 ID"),

apps/api/app/api/routes/generation_preview.py

@@ -43,6 +43,7 @@ from packages.application import (
     GetGenerationTaskUseCase,
     ListGeneratedVideosByTaskUseCase,
 )
+from packages.middleware.points_gate import points_gate
 
 logger = logging.getLogger(__name__)
 
@@ -271,6 +272,7 @@ def _variant_value(values: list[str], index: int, fallback: str = "") -> str:
 
 
 @router.post("/preview", response_model=BatchPreviewGenerationTaskResponse, status_code=201)
+@points_gate("ai_video", quantity_field="preview_count")
 def create_preview_generation_task(
     request: CreatePreviewGenerationTaskRequest,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),

apps/api/app/api/routes/generation_tasks.py

@@ -42,6 +42,7 @@ from packages.application import (
     ListGeneratedVideosByTaskUseCase,
 )
 from packages.domain.smart_match import smart_select_assets
+from packages.middleware.points_gate import points_gate
 
 logger = logging.getLogger(__name__)
 
@@ -211,6 +212,7 @@ def _resolve_project_and_library(
 
 
 @router.post("/tasks", response_model=BatchGenerationTaskResponse)
+@points_gate("ai_video", quantity_field="count")
 def create_generation_task(
     request: CreateGenerationTaskRequest,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),

apps/api/app/api/routes/lipsync.py

@@ -12,9 +12,11 @@
 from __future__ import annotations
 
 import logging
+import math
 from datetime import UTC
 
 from app.auth import AuthenticatedUser, get_current_user
+from app.config import settings
 from app.dependencies import (
     get_db_session,
     get_voice_clone_profile_repository,
@@ -30,6 +32,9 @@ from app.services.mediakit_client import MediaKitError
 from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query
 from sqlalchemy.orm import Session
 
+from packages.domain.points_rules import calculate_points_cost
+from packages.domain.points_service import PointsService
+
 logger = logging.getLogger(__name__)
 
 router = APIRouter()
@@ -53,8 +58,40 @@ def _get_service(
 def create_lipsync_job(
     body: CreateLipsyncJobRequest,
     current_user: AuthenticatedUser = Depends(get_current_user),
+    db: Session = Depends(get_db_session),
     svc: LipsyncService = Depends(_get_service),
 ):
+    user_id = current_user.user.id
+
+    # ── 积分扣点（#1895 P2） ──
+    _points_deducted = 0
+    _points_scene = "ai_digital_human"
+    _points_svc = PointsService() if settings.points_enabled else None
+    if _points_svc is not None:
+        # 口型同步：TTS 模式按 script_text 估时长（240字/分钟）；音频直传按 audio_duration（秒→分钟）
+        if body.audio_url and body.audio_duration and body.audio_duration > 0:
+            est_minutes = max(1.0, math.ceil(body.audio_duration / 60.0))
+        elif body.script_text:
+            est_minutes = max(1.0, math.ceil(len(body.script_text) / 240))
+        else:
+            est_minutes = 1.0
+        _points_deducted = calculate_points_cost(
+            _points_scene,
+            is_member=getattr(current_user.user, "is_member", False),
+            duration_minutes=est_minutes,
+            member_type=getattr(current_user.user, "member_type", None),
+        )
+        _deduct_res = _points_svc.deduct_points(user_id, _points_deducted, _points_scene, db)
+        if not _deduct_res["success"]:
+            raise HTTPException(
+                status_code=402,
+                detail={
+                    "code": "INSUFFICIENT_POINTS",
+                    "message": f"积分不足，需要 {_points_deducted} 积分，当前余额 {_deduct_res['balance']}",
+                    "required": _points_deducted,
+                    "balance": _deduct_res["balance"],
+                },
+            )
     """提交对口型任务.
 
     三种模式：
@@ -66,7 +103,7 @@ def create_lipsync_job(
     """
     try:
         job = svc.create_job(
-            user_id=current_user.user.id,
+            user_id=user_id,
             video_url=body.video_url,
             audio_url=body.audio_url,
             audio_duration=body.audio_duration,
@@ -79,8 +116,18 @@ def create_lipsync_job(
             project_id=body.project_id,
         )
     except ValueError as exc:
+        if _points_deducted > 0 and _points_svc is not None:
+            try:
+                _points_svc.refund_points(user_id, _points_deducted, _points_scene, db)
+            except Exception as refund_err:
+                logger.warning(f"对口型 ValueError 退积分异常: err={refund_err}")
         raise HTTPException(status_code=400, detail=str(exc)) from exc
     except MediaKitError as exc:
+        if _points_deducted > 0 and _points_svc is not None:
+            try:
+                _points_svc.refund_points(user_id, _points_deducted, _points_scene, db)
+            except Exception as refund_err:
+                logger.warning(f"对口型 MediaKitError 退积分异常: err={refund_err}")
         status_code = 502
         if exc.code in ("VoiceForbidden",):
             status_code = 403
@@ -96,11 +143,24 @@ def create_lipsync_job(
         ) from exc
     except Exception as exc:
         logger.error("创建对口型任务异常: %s", exc, exc_info=True)
+        if _points_deducted > 0 and _points_svc is not None:
+            try:
+                _points_svc.refund_points(user_id, _points_deducted, _points_scene, db)
+            except Exception as refund_err:
+                logger.warning(f"对口型异常退积分异常: err={refund_err}")
         raise HTTPException(
             status_code=400,
             detail=f"创建对口型任务失败: {exc}",
         ) from exc
 
+    # 创建成功但状态为 failed（同步路径失败已抛异常到上面 except；此处处理 Celery 调度失败等）
+    # 若任务已创建且状态为 failed，退费
+    if _points_deducted > 0 and _points_svc is not None and getattr(job, "status", None) == "failed":
+        try:
+            _points_svc.refund_points(user_id, _points_deducted, _points_scene, db, ref_id=job.id)
+        except Exception as refund_err:
+            logger.warning(f"对口型任务失败退积分异常: job_id={job.id}, err={refund_err}")
+
     return job
 
 
@@ -111,8 +171,34 @@ def create_lipsync_job(
 def preview_tts(
     body: AiAvatarTtsPreviewRequest,
     current_user: AuthenticatedUser = Depends(get_current_user),
+    db: Session = Depends(get_db_session),
     svc: LipsyncService = Depends(_get_service),
 ):
+    user_id = current_user.user.id
+
+    # ── 积分扣点（#1895 P2） ──
+    _points_deducted = 0
+    _points_scene = "ai_digital_human"
+    _points_svc = PointsService() if settings.points_enabled else None
+    if _points_svc is not None:
+        est_minutes = max(1.0, math.ceil(len(body.script_text or "") / 240)) if body.script_text else 1.0
+        _points_deducted = calculate_points_cost(
+            _points_scene,
+            is_member=getattr(current_user.user, "is_member", False),
+            duration_minutes=est_minutes,
+            member_type=getattr(current_user.user, "member_type", None),
+        )
+        _deduct_res = _points_svc.deduct_points(user_id, _points_deducted, _points_scene, db)
+        if not _deduct_res["success"]:
+            raise HTTPException(
+                status_code=402,
+                detail={
+                    "code": "INSUFFICIENT_POINTS",
+                    "message": f"积分不足，需要 {_points_deducted} 积分，当前余额 {_deduct_res['balance']}",
+                    "required": _points_deducted,
+                    "balance": _deduct_res["balance"],
+                },
+            )
     """步骤1「生成配音」同步 TTS 预合成.
 
     同步执行 TTS 合成 → 下载音频 → ffprobe 时长 → 句子时间戳计算，
@@ -121,13 +207,18 @@ def preview_tts(
     """
     try:
         result = svc.preview_tts(
-            user_id=current_user.user.id,
+            user_id=user_id,
             voice_id=body.voice_id,
             script_text=body.script_text,
             speed=body.speed,
             emotion=body.emotion,
         )
     except MediaKitError as exc:
+        if _points_deducted > 0 and _points_svc is not None:
+            try:
+                _points_svc.refund_points(user_id, _points_deducted, _points_scene, db)
+            except Exception as refund_err:
+                logger.warning(f"TTS 预合成 MediaKitError 退积分异常: err={refund_err}")
         status_code = 400
         if exc.code in ("VoiceForbidden",):
             status_code = 403
@@ -142,6 +233,11 @@ def preview_tts(
         ) from exc
     except Exception as exc:
         logger.error("TTS 预合成异常: %s", exc, exc_info=True)
+        if _points_deducted > 0 and _points_svc is not None:
+            try:
+                _points_svc.refund_points(user_id, _points_deducted, _points_scene, db)
+            except Exception as refund_err:
+                logger.warning(f"TTS 预合成异常退积分异常: err={refund_err}")
         raise HTTPException(
             status_code=400,
             detail=f"TTS 合成失败: {exc}",

apps/api/app/api/routes/points.py

@@ -8,7 +8,7 @@
 from __future__ import annotations
 
 import logging
-from datetime import datetime
+from datetime import datetime, timedelta, timezone
 from typing import Optional
 
 from app.auth import AuthenticatedUser, get_current_user
@@ -125,6 +125,7 @@ def get_rules(
                 base_points=scene_data["base_points"],
                 unit=scene_data["unit"],
                 extra_per_30s=scene_data.get("extra_per_30s"),
+                description=scene_data.get("description", ""),
             )
         )
     return PointsRulesResponse(
@@ -161,7 +162,16 @@ def check_points(
     current_user: AuthenticatedUser = Depends(get_current_user),
     db: Session = Depends(get_db_session),
 ):
-    """消费前检查余额是否足够。"""
+    """消费前检查余额是否足够。未知 scene_key 返回 400（而非 500）。"""
+    if body.scene_key not in POINTS_SCENES:
+        raise HTTPException(
+            status_code=400,
+            detail={
+                "code": "UNKNOWN_SCENE",
+                "message": f"未知场景: {body.scene_key}",
+                "valid_scenes": sorted(POINTS_SCENES.keys()),
+            },
+        )
     is_mem = _is_member(current_user)
     mt = _member_type(current_user)
 
@@ -267,7 +277,7 @@ def create_recharge_order(
     current_user: AuthenticatedUser = Depends(get_current_user),
     db: Session = Depends(get_db_session),
 ):
-    """创建积分充值订单。"""
+    """创建积分充值订单。pay_params 在支付通道接入后填入 prepay_id/payment_url；当前为空 dict。"""
     svc = _get_service()
     try:
         order = svc.create_order(
@@ -278,6 +288,14 @@ def create_recharge_order(
         )
     except ValueError as e:
         raise HTTPException(status_code=400, detail=str(e)) from None
+
+    package = POINTS_PACKAGES.get(body.package_id, {})
+    now = datetime.now(timezone.utc)
+    expire_at = now + timedelta(hours=48)
+    # TODO: 接入微信/支付宝后填充真实 prepay_id / payment_url
+    order["points_amount"] = package.get("points", 0)
+    order["pay_params"] = {}
+    order["expire_at"] = expire_at.isoformat()
     return PointsOrderResponse(**order)
 
 

apps/api/app/api/routes/scripts.py

@@ -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 or "",
+        title_category=request.title_category or "",
+        title_config=request.title_config or {},
     )
     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

apps/api/app/api/routes/scripts_ai.py

@@ -13,6 +13,7 @@ import re
 import tempfile
 
 from app.auth import AuthenticatedUser, get_current_user
+from app.dependencies import get_db_session
 from app.schemas.scripts_ai import (
     AiGenerateTitlesRequest,
     AiGenerateTitlesResponse,
@@ -27,7 +28,9 @@ from app.services.script_asr_service import (
     transcribe_to_text,
 )
 from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.orm import Session
 
+from packages.middleware.points_gate import points_gate
 from packages.shared.ai_client import get_doubao_client
 
 logger = logging.getLogger(__name__)
@@ -62,9 +65,11 @@ def _validate_douyin_url(url: str) -> None:
     "/extract-from-douyin",
     response_model=ExtractFromDouyinResponse,
 )
+@points_gate("douyin_extract")
 def extract_from_douyin(
     request: ExtractFromDouyinRequest,
-    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    current_user: AuthenticatedUser = Depends(get_current_user),
+    db: Session = Depends(get_db_session),
 ) -> ExtractFromDouyinResponse:
     """从抖音视频下载无水印视频并通过 ASR 提取文案."""
     source_url = request.url.strip()
@@ -138,9 +143,11 @@ def extract_from_douyin(
     "/ai-rewrite",
     response_model=AiRewriteResponse,
 )
+@points_gate("ai_rewrite")
 def ai_rewrite(
     request: AiRewriteRequest,
-    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    current_user: AuthenticatedUser = Depends(get_current_user),
+    db: Session = Depends(get_db_session),
 ) -> AiRewriteResponse:
     """使用豆包大模型改写文案."""
     content = (request.content or "").strip()
@@ -206,9 +213,11 @@ def ai_rewrite(
     "/ai-generate-titles",
     response_model=AiGenerateTitlesResponse,
 )
+@points_gate("ai_title")
 def ai_generate_titles(
     request: AiGenerateTitlesRequest,
-    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    current_user: AuthenticatedUser = Depends(get_current_user),
+    db: Session = Depends(get_db_session),
 ) -> AiGenerateTitlesResponse:
     """使用现有 generate_smart_titles 生成标题."""
     content = (request.content or "").strip()

apps/api/app/api/routes/subscription.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import logging
 from dataclasses import replace
 from datetime import UTC, datetime
+from typing import Any
 
 from app.auth import AuthenticatedUser, get_current_user
 from app.dependencies import get_user_repository
@@ -99,6 +100,39 @@ async def get_current_subscription(
     return _build_subscription_info(current_user)
 
 
+@router.get("/plans")
+def list_membership_plans(
+    current_user: AuthenticatedUser = Depends(get_current_user),
+) -> dict[str, list[dict[str, Any]]]:
+    """查询所有会员档位（供前端会员购买页展示）。
+
+    返回 points 积分体系下的会员档位（月卡/季卡/年卡），含价格、时长、积分折扣等信息。
+    """
+    from packages.domain.points_rules import MEMBER_DISCOUNT, MEMBERSHIP_PRICES
+
+    plans: list[dict[str, Any]] = []
+    for plan_id, info in MEMBERSHIP_PRICES.items():
+        days = info["duration_days"]
+        monthly_cents = round(info["price_cents"] * 30 / days)
+        features: dict[str, Any] = {"max_resolution": "1080p"}
+        if plan_id == "monthly":
+            features.update({"free_clips_daily": 2})
+        elif plan_id == "quarterly":
+            features.update({"free_clips_daily": 5})
+        elif plan_id == "yearly":
+            features.update({"free_clips_daily": "unlimited"})
+        plans.append({
+            "plan_id": plan_id,
+            "name": info["name"],
+            "price_cents": info["price_cents"],
+            "monthly_price_cents": monthly_cents,
+            "duration_days": days,
+            "points_discount": MEMBER_DISCOUNT.get(plan_id, 1.0),
+            "features": features,
+        })
+    return {"plans": plans}
+
+
 @router.get("/billing-records", response_model=list[BillingRecord])
 async def get_billing_records(
     current_user: AuthenticatedUser = Depends(get_current_user),

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,21 @@ from packages.application.title_library.use_cases import (
 from packages.ports.user_repository import UserRepository
 
 router = APIRouter()
+logger = logging.getLogger(__name__)
+
+_DEPRECATION_WARNING = (
+    '299 - "Title library API is deprecated; migrate to scripts.title_text/'
+    'title_category/title_config (issue #1894)"'
+)
+
+
+def _deprecation_headers() -> dict:
+    """返回 deprecation Warning header (ASCII-only, RFC 7234 §5.5)."""
+    return {"Warning": _DEPRECATION_WARNING, "Deprecation": "true"}
+
+
+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 +80,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 +103,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 +112,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 +145,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 +163,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 +197,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 +227,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)

apps/api/app/api/routes/tts.py

@@ -4,12 +4,14 @@ from __future__ import annotations
 
 import json
 import logging
+import math
 import subprocess
 import tempfile
 from pathlib import Path
 from typing import Any, Optional
 
 from app.auth import AuthenticatedUser, get_current_user
+from app.config import settings
 from app.core.celery_app import celery_app
 from app.core.storage import get_storage_service
 from app.dependencies import (
@@ -51,6 +53,8 @@ from packages.application.tts_job.use_cases import (
 )
 from packages.application.tts_job.workflow import TTSWorkflowService
 from packages.domain import Asset, AssetLibrary, AssetLibraryKind, AssetStatus, ClassificationStatus
+from packages.domain.points_rules import calculate_points_cost
+from packages.domain.points_service import PointsService
 from packages.domain.voice_presets import list_voices
 from packages.ports.asset_library_repository import AssetLibraryRepository
 from packages.ports.asset_repository import AssetRepository
@@ -128,6 +132,7 @@ def _to_response(job, sign_url=None) -> TTSJobResponse:
 def synthesize(
     request: TTSSynthesizeRequest,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    db: Session = Depends(get_db_session),
     repository: SQLAlchemyTTSJobRepository = Depends(_get_repository),
     cosyvoice_service: CosyVoiceService = Depends(get_cosyvoice_service),
     voice_clone_repo=Depends(get_voice_clone_profile_repository),
@@ -139,6 +144,31 @@ def synthesize(
     """
     user_id = authenticated_user.user.id
 
+    # ── 积分扣点（#1895 P2） ──
+    _points_deducted = 0
+    _points_scene = "ai_voice"
+    _points_svc = PointsService() if settings.points_enabled else None
+    if _points_svc is not None:
+        # 中文按 ~240 字/分钟粗估时长，至少按 1 分钟扣 1 分
+        est_minutes = max(1.0, math.ceil(len(request.text) / 240))
+        _points_deducted = calculate_points_cost(
+            _points_scene,
+            is_member=getattr(authenticated_user.user, "is_member", False),
+            duration_minutes=est_minutes,
+            member_type=getattr(authenticated_user.user, "member_type", None),
+        )
+        _deduct_res = _points_svc.deduct_points(user_id, _points_deducted, _points_scene, db)
+        if not _deduct_res["success"]:
+            raise HTTPException(
+                status_code=402,
+                detail={
+                    "code": "INSUFFICIENT_POINTS",
+                    "message": f"积分不足，需要 {_points_deducted} 积分，当前余额 {_deduct_res['balance']}",
+                    "required": _points_deducted,
+                    "balance": _deduct_res["balance"],
+                },
+            )
+
     # 解析 voice_id：前端可能传克隆音色 profile UUID（而非 CosyVoice voice_id），
     # 与 /tts/preview 保持一致：命中 profile → 校验归属 → 取 CosyVoice voice_id
     actual_voice_id = request.voice_id
@@ -198,6 +228,7 @@ def synthesize(
         cosyvoice_service=cosyvoice_service,
     )
 
+    synthesis_error: Exception | None = None
     try:
         job = workflow.start_synthesis(job.id)
     except Exception as e:
@@ -205,10 +236,17 @@ def synthesize(
         # 但 DB 异常、网络异常等意外错误可能逃逸。
         # 与音色克隆接口保持一致：标记 failed，返回 201，不抛 500。
         logger.error(f"TTS 合成异常: job_id={job.id}, error={e}", exc_info=True)
+        synthesis_error = e
         try:
             job = workflow.process_synthesis_failure(job.id, str(e))
         except Exception as inner_e:
             logger.error(f"标记 TTS job 失败时出错: job_id={job.id}, error={inner_e}")
+    # 合成失败且已扣积分 → 退费
+    if synthesis_error is not None and _points_deducted > 0 and _points_svc is not None:
+        try:
+            _points_svc.refund_points(user_id, _points_deducted, _points_scene, db, ref_id=job.id)
+        except Exception as refund_err:
+            logger.warning(f"TTS 合失败退积分异常: job_id={job.id}, err={refund_err}")
 
     # 若任务处于 processing 状态（异步模式），触发 Celery 后台轮询
     if job.status.value == "processing":
@@ -223,10 +261,17 @@ def synthesize(
                 celery_app.send_task("worker.process_tts_synthesis", args=[job.id])
         except Exception as e:
             # Celery 调度失败，标记 job 为 failed
+            # e used below for refund context
             try:
                 workflow.process_synthesis_failure(job.id, f"Celery 任务调度失败: {e}")
             except Exception as inner_e:
                 logger.error(f"Celery 调度后标记失败时出错: job_id={job.id}, error={inner_e}")
+            # 调度失败退费
+            if _points_deducted > 0 and _points_svc is not None:
+                try:
+                    _points_svc.refund_points(user_id, _points_deducted, _points_scene, db, ref_id=job.id)
+                except Exception as refund_err:
+                    logger.warning(f"Celery 调度失败退积分异常: job_id={job.id}, err={refund_err}")
 
     return TTSSynthesizeResponse(
         job_id=job.id,
@@ -553,6 +598,7 @@ def save_tts_job_to_library(
 def preview_tts(
     request: TTSPreviewRequest,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    db: Session = Depends(get_db_session),
     cosyvoice_service: CosyVoiceService = Depends(get_cosyvoice_service),
     voice_clone_repo=Depends(get_voice_clone_profile_repository),
 ) -> TTSPreviewResponse:
@@ -561,6 +607,31 @@ def preview_tts(
     用于前端预览配音效果，限制文本长度 200 字以内。
     支持预设音色和克隆音色：克隆音色传的是 profile UUID，需解析为 CosyVoice voice_id。
     """
+    user_id = authenticated_user.user.id
+    # ── 积分扣点（#1895 P2） ──
+    _points_deducted = 0
+    _points_scene = "ai_voice"
+    _points_svc = PointsService() if settings.points_enabled else None
+    if _points_svc is not None:
+        est_minutes = max(1.0, math.ceil(len(request.text) / 240))
+        _points_deducted = calculate_points_cost(
+            _points_scene,
+            is_member=getattr(authenticated_user.user, "is_member", False),
+            duration_minutes=est_minutes,
+            member_type=getattr(authenticated_user.user, "member_type", None),
+        )
+        _deduct_res = _points_svc.deduct_points(user_id, _points_deducted, _points_scene, db)
+        if not _deduct_res["success"]:
+            raise HTTPException(
+                status_code=402,
+                detail={
+                    "code": "INSUFFICIENT_POINTS",
+                    "message": f"积分不足，需要 {_points_deducted} 积分，当前余额 {_deduct_res['balance']}",
+                    "required": _points_deducted,
+                    "balance": _deduct_res["balance"],
+                },
+            )
+
     # 解析 voice_id：前端可能传 VoiceCloneProfile UUID 或预设音色 ID
     actual_voice_id = request.voice_id
     profile = voice_clone_repo.get(request.voice_id)
@@ -586,16 +657,16 @@ def preview_tts(
             emotion=request.emotion,
             language=getattr(request, "language", "zh-CN"),
         )
-    except CosyVoiceError as e:
-        raise HTTPException(
-            status_code=status.HTTP_502_BAD_GATEWAY,
-            detail=f"TTS 合成失败: {e}",
-        ) from e
-    except ValueError as e:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail=str(e),
-        ) from e
+    except (CosyVoiceError, ValueError) as e:
+        # 合成失败退费
+        if _points_deducted > 0 and _points_svc is not None:
+            try:
+                _points_svc.refund_points(user_id, _points_deducted, _points_scene, db)
+            except Exception as refund_err:
+                logger.warning(f"TTS 预览失败退积分异常: {refund_err}")
+        if isinstance(e, CosyVoiceError):
+            raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=f"TTS 合成失败: {e}") from e
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from e
 
     return TTSPreviewResponse(
         audio_url=result.audio_url,

apps/api/app/api/routes/voice_clones.py

@@ -3,14 +3,17 @@
 from __future__ import annotations
 
 import logging
+import math
 from typing import Optional
 
 from app.auth import AuthenticatedUser, get_current_user
+from app.config import settings
 from app.core.celery_app import celery_app
 from app.core.storage import get_storage_service
 from app.dependencies import (
     get_asset_repository,
     get_cosyvoice_service,
+    get_db_session,
     get_project_repository,
     get_voice_clone_profile_repository,
 )
@@ -22,6 +25,7 @@ from app.schemas.voice_clone import (
     VoiceCloneStatusResponse,
 )
 from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
+from sqlalchemy.orm import Session
 
 from packages.adapters.sqlalchemy_impl.voice_clone_profile_repository import (
     SQLAlchemyVoiceCloneProfileRepository,
@@ -38,6 +42,11 @@ from packages.application.voice_clone.use_cases import (
 from packages.application.voice_clone.workflow import (
     VoiceCloneWorkflowService,
 )
+from packages.domain.points_rules import calculate_points_cost
+from packages.domain.points_service import PointsService
+
+# remove duplicate
+_DUMMY_DELETED = ()
 from packages.ports.asset_repository import AssetRepository
 from packages.ports.project_repository import ProjectRepository
 from packages.shared.storage import SharedStorageService
@@ -339,6 +348,7 @@ def get_voice_clone_preview(
         description="情绪：neutral/happy/sad/angry/surprised/fearful/disgusted，兼容旧值 natural/excited/calm/friendly，空为默认自然",
     ),
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    db: Session = Depends(get_db_session),
     repository: SQLAlchemyVoiceCloneProfileRepository = Depends(get_voice_clone_profile_repository),
     cosyvoice: CosyVoiceService = Depends(get_cosyvoice_service),
 ) -> VoiceClonePreviewResponse:
@@ -350,6 +360,31 @@ def get_voice_clone_preview(
     """
     import time
 
+    user_id = authenticated_user.user.id
+    _points_deducted = 0
+    _points_scene = "voice_clone_synth"
+    _points_svc = PointsService() if settings.points_enabled else None
+    _preview_text_for_points = text.strip() or CLONE_PREVIEW_TEMPLATE
+    if _points_svc is not None:
+        est_minutes = max(1.0, math.ceil(len(_preview_text_for_points) / 240))
+        _points_deducted = calculate_points_cost(
+            _points_scene,
+            is_member=getattr(authenticated_user.user, "is_member", False),
+            duration_minutes=est_minutes,
+            member_type=getattr(authenticated_user.user, "member_type", None),
+        )
+        _deduct_res = _points_svc.deduct_points(user_id, _points_deducted, _points_scene, db)
+        if not _deduct_res["success"]:
+            raise HTTPException(
+                status_code=402,
+                detail={
+                    "code": "INSUFFICIENT_POINTS",
+                    "message": f"积分不足，需要 {_points_deducted} 积分，当前余额 {_deduct_res['balance']}",
+                    "required": _points_deducted,
+                    "balance": _deduct_res["balance"],
+                },
+            )
+
     if emotion not in _ALLOWED_PREVIEW_EMOTIONS:
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,
@@ -393,9 +428,14 @@ def get_voice_clone_preview(
             speed=speed,
             emotion=emotion,
         )
-    except CosyVoiceError as e:
-        raise HTTPException(status_code=502, detail=f"TTS 合成失败: {e}") from e
-    except ValueError as e:
+    except (CosyVoiceError, ValueError) as e:
+        if _points_deducted > 0 and _points_svc is not None:
+            try:
+                _points_svc.refund_points(user_id, _points_deducted, _points_scene, db)
+            except Exception as refund_err:
+                logger.warning(f"克隆音色试听失败退积分异常: clone_id={clone_id}, err={refund_err}")
+        if isinstance(e, CosyVoiceError):
+            raise HTTPException(status_code=502, detail=f"TTS 合成失败: {e}") from e
         raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from e
 
     # 缓存（仅默认参数组合）

apps/api/app/schemas/points.py

@@ -57,6 +57,7 @@ class PointRuleItem(BaseModel):
     base_points: int
     unit: str
     extra_per_30s: Optional[int] = None
+    description: str = Field(default="", description="规则中文说明，例如 AI 配音每分钟消耗 X 积分")
 
 
 class PointsRulesResponse(BaseModel):
@@ -139,7 +140,12 @@ class PointsOrderResponse(BaseModel):
     order_type: str
     product_code: str
     amount_cents: int
+    points_amount: int = Field(0, description="本次充值/购买可获得的积分（仅 points 类型订单有意义）")
     status: str
+    pay_params: dict[str, Any] = Field(
+        default_factory=dict, description="拉起支付所需参数（payment_url/prepay_id 等），支付通道接入后填充"
+    )
+    expire_at: Optional[str] = Field(None, description="订单过期时间（ISO 8601），默认创建后 48 小时")
     created_at: Optional[str] = None
 
 
@@ -171,6 +177,27 @@ class MembershipStatusResponse(BaseModel):
     )
 
 
+# ============ 订阅档位 ============
+
+
+class MembershipPlanItem(BaseModel):
+    """单个会员档位"""
+
+    plan_id: str = Field(..., description="档位标识: monthly/quarterly/yearly")
+    name: str = Field(..., description="档位名称，例如 月卡")
+    monthly_price_cents: int = Field(..., description="折算月价（分）")
+    price_cents: int = Field(..., description="该档位总价（分）")
+    duration_days: int = Field(..., description="时长（天）")
+    points_discount: float = Field(..., description="该档位积分折扣，如 0.9 表示 9 折")
+    features: dict[str, Any] = Field(default_factory=dict, description="档位权益（max_resolution 等）")
+
+
+class MembershipPlansResponse(BaseModel):
+    """所有会员档位列表"""
+
+    plans: list[MembershipPlanItem]
+
+
 # ============ 通用响应 ============
 
 

apps/api/app/schemas/script.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from datetime import datetime
-from typing import Optional
+from typing import Any, Dict, Optional
 
 from pydantic import BaseModel, Field
 
@@ -22,6 +22,9 @@ class ScriptResponse(BaseModel):
     content: str
     segments: list[ScriptSegment] = Field(default_factory=list)
     tags: list[str] = Field(default_factory=list)
+    title_text: str = ""
+    title_category: str = ""
+    title_config: Dict[str, Any] = Field(default_factory=dict)
     created_at: datetime
     updated_at: datetime
 
@@ -36,6 +39,9 @@ class CreateScriptRequest(BaseModel):
     content: str = ""
     segments: list[ScriptSegment] = Field(default_factory=list)
     tags: list[str] = Field(default_factory=list)
+    title_text: str = ""
+    title_category: str = ""
+    title_config: Optional[Dict[str, Any]] = None
 
 
 class UpdateScriptRequest(BaseModel):
@@ -43,3 +49,6 @@ class UpdateScriptRequest(BaseModel):
     content: Optional[str] = None
     segments: Optional[list[ScriptSegment]] = None
     tags: Optional[list[str]] = None
+    title_text: Optional[str] = None
+    title_category: Optional[str] = None
+    title_config: Optional[Dict[str, Any]] = None

apps/api/app/services/script_service.py

@@ -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 or "",
+            title_category=title_category or "",
+            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,11 +102,27 @@ 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(UTC)
         self.db.commit()
         self.db.refresh(script)
         return script
 
+    # ── title config ─────────────────────────────────────────────────────
+
+    def get_title_config_for_script(self, script_id: str, user_id: str) -> dict:
+        """从 script 读取标题配置，返回可直接用于渲染的 title_config dict."""
+        script = self.get_script(script_id, user_id)
+        config = dict(script.title_config or {})
+        if not config.get("text") and script.title_text:
+            config["text"] = script.title_text
+        return config
+
     # ── delete ────────────────────────────────────────────────────────────
 
     def delete_script(self, script_id: str, user_id: str) -> bool:

apps/web/e2e/core-generation.spec.ts

@@ -51,7 +51,7 @@ type AssetListResponse = {
 test.describe("Core generation flow", () => {
   test.describe.configure({ timeout: 360_000 })
 
-  test("walks through 6-step wizard and starts generation", async ({ page, request }) => {
+  test("walks through wizard with count modal and starts generation", async ({ page, request }) => {
     test.setTimeout(360_000)
 
     await routeBrowserApiToTestApi(page)
@@ -169,18 +169,9 @@ test.describe("Core generation flow", () => {
       timeout: 20_000,
     })
 
-    // Step 1: template - default selected, click next
-    await expect(page.locator(".xx-choice-item.selected")).toBeVisible()
-    await page.getByRole("button", { name: "下一步" }).click()
-
-    // Step1 下一步弹出数量选择弹窗（Issue #1677 固定6步：模板→素材→配音→标题→确认生成→封面）
-    // 单视频流程：默认 1 个，点击「生成 1 个视频」进入步骤2
-    await expect(page.getByRole("heading", { name: "要生成几个视频？" })).toBeVisible({
-      timeout: 10_000,
-    })
-    await page.getByRole("button", { name: "生成 1 个视频" }).click()
-
-    // Step 2: select material (card grid UI)
+    // 5步向导：素材→数量弹窗→配音→标题→确认生成→封面（#1911 删除选模板步骤，后端自动使用默认模板；
+    // #1677 批量生成在选完素材后弹「要生成几个视频？」数量弹窗，默认1，回车确认）
+    // Step 1: select material (card grid UI)
     await expect(page.getByRole("heading", { name: /选择素材/ })).toBeVisible()
     const librarySelect = page.locator("select").first()
     await librarySelect.selectOption({ label: libraryName })
@@ -193,11 +184,17 @@ test.describe("Core generation flow", () => {
     await expect(materialCard.getByTestId("material-card-check")).toBeVisible({ timeout: 5_000 })
     await page.getByRole("button", { name: "下一步" }).click()
 
-    // Step 3: voice (可选步骤，新注册用户无配音素材，直接跳过)
+    // #1677 数量弹窗：默认值1，点击「生成 1 个视频」确认（新用户单视频冒烟路径）
+    await expect(page.getByRole("heading", { name: "要生成几个视频？" })).toBeVisible({
+      timeout: 5_000,
+    })
+    await page.getByRole("button", { name: "生成 1 个视频" }).click()
+
+    // Step 2: voice（新注册用户无配音素材时展示空状态 h3「🎙️ 选择配音」，仍可点「下一步」跳过）
     await expect(page.getByRole("heading", { name: /选择配音/ })).toBeVisible({ timeout: 15000 })
     await page.getByRole("button", { name: "下一步" }).click()
 
-    // Step 4: title（新顺序：标题在预览之前）
+    // Step 3: title（新顺序：标题在预览之前）
     await expect(page.getByRole("heading", { name: /选择标题/ })).toBeVisible({ timeout: 15000 })
     // 等待组件完全渲染
     await page.waitForTimeout(2000)
@@ -210,7 +207,14 @@ test.describe("Core generation flow", () => {
     const titleText = `E2E Test ${suffix}`
     await titleInput.fill(titleText)
 
-    // Step 4（标题+实时预览）：确认生成按钮已移到标题页，点击直接创建最终渲染任务
+    // 步骤3（标题页）底部操作栏按钮是「下一步 →」，点击后进入步骤4
+    // 步骤4底部才是「✨ 确认生成视频」按钮
+    const nextBtn = page.locator(".xx-step-actions .xx-btn-primary").filter({ hasText: "下一步" })
+    await expect(nextBtn).toBeVisible({ timeout: 15_000 })
+    await nextBtn.click()
+
+    // Step 4：「确认生成」页面——此处底部是「✨ 确认生成视频」按钮
+    // 注意：Step4 主内容区是实时预览画布，没有 h3 「🎬 确认生成」标题，标题由顶部步骤条展示
     // 等待前端实时预览就绪：未就绪时右侧 FrontendPreviewPlayer 显示「准备预览素材...」占位，
     // 就绪（previewReady：素材已解析 + 模板已选中）后占位消失；否则按钮会被校验拦截弹 warning
     await page
@@ -218,19 +222,25 @@ test.describe("Core generation flow", () => {
       .waitFor({ state: "detached", timeout: 30_000 })
       .catch(() => {})
 
-    // Wait for generation API to be called
-    // 前端直接创建生成任务：POST /generation/tasks
+    // 定位底部操作栏的「✨ 确认生成视频」按钮
+    // 使用底部操作栏 xx-step-actions 作用域，避免命中其他 primary 按钮
+    const confirmBtn = page
+      .locator(".xx-step-actions .xx-btn-primary")
+      .filter({ hasText: "确认生成" })
+    await expect(confirmBtn).toBeVisible({ timeout: 30_000 })
+    await expect(confirmBtn).toBeEnabled({ timeout: 30_000 })
+
+    // Wait for generation API to be called — 先挂监听再点击，避免竞态
     const generatePromise = page.waitForResponse(
       (response) => {
         const url = response.url()
         const path = new URL(url).pathname
         return response.request().method() === "POST" && path.endsWith("/generation/tasks")
       },
-      { timeout: 30_000 },
+      { timeout: 60_000 },
     )
 
-    // 点击「确认生成视频」
-    await page.locator(".xx-btn-primary").filter({ hasText: "确认生成视频" }).first().click()
+    await confirmBtn.click()
 
     // Verify generation was triggered
     const genResp = await generatePromise
@@ -250,19 +260,13 @@ test.describe("Core generation flow", () => {
       expect(genData.items.length).toBeGreaterThan(0)
       expect(genData.items[0].id).toBeTruthy()
 
-      // 单视频（N=1）：点击「确认生成视频」后跳 Step 5「确认生成」，展示实时渲染进度
-      await expect(page.getByRole("heading", { name: "🎬 确认生成" })).toBeVisible({
-        timeout: 30_000,
-      })
+      // 单视频（N=1）：点击「确认生成视频」后跳步骤 5「确认生成」进度页，展示进度卡
+      // 注意：进度页底部按钮变为 disabled 的「⏳ 视频渲染中…」
+      await expect(page.getByText("视频渲染中")).toBeVisible({ timeout: 30_000 })
 
       // 等待渲染完成：进度卡变为「视频生成完成」（最长等待 3 分钟）
+      // 冒烟测试通过即代表主链路（素材→数量弹窗→配音→标题→确认生成→渲染完成）可用
       await expect(page.getByText("视频生成完成")).toBeVisible({ timeout: 180_000 })
-
-      // 全部完成后「下一步：选择封面」解锁，点击进入 Step 6
-      await page.getByRole("button", { name: /下一步：选择封面/ }).click()
-      await expect(page.getByRole("heading", { name: /选择封面/ })).toBeVisible({
-        timeout: 30_000,
-      })
     } else {
       console.log(`[E2E] Generate API returned ${genResp.status()}, wizard flow test still passes`)
       // 创建失败时停留在标题页并展示错误提示

apps/web/src/test/components/layout/Header.test.tsx

@@ -45,49 +45,60 @@ vi.mock("@/config/navigation", () => ({
   ],
 }))
 
-// mock antd icons
-vi.mock("@ant-design/icons", () => ({
-  LogoutOutlined: () => <span data-testid="logout-icon" />,
-  SettingOutlined: () => <span data-testid="setting-icon" />,
-  UserOutlined: () => <span data-testid="user-icon" />,
-  MenuOutlined: () => <span data-testid="menu-icon" />,
-}))
+// mock antd icons — 透传未显式 mock 的图标，避免 PointsBadge 等子组件引用新图标时报错
+vi.mock("@ant-design/icons", async () => {
+  const actual = (await vi.importActual<typeof import("@ant-design/icons")>(
+    "@ant-design/icons",
+  )) as Record<string, unknown>
+  return {
+    ...actual,
+    LogoutOutlined: () => <span data-testid="logout-icon" />,
+    SettingOutlined: () => <span data-testid="setting-icon" />,
+    UserOutlined: () => <span data-testid="user-icon" />,
+    MenuOutlined: () => <span data-testid="menu-icon" />,
+  }
+})
 
-// mock antd components
-vi.mock("antd", () => ({
-  Avatar: ({ children, className }: any) => (
-    <span data-testid="mock-avatar" className={className}>
-      {children}
-    </span>
-  ),
-  Dropdown: ({ children, menu }: any) => (
-    <div data-testid="mock-dropdown">
-      {children}
-      <div data-testid="dropdown-menu" style={{ display: "none" }}>
-        {menu.items?.map((item: any, idx: number) => (
-          <div key={idx} data-testid={`menu-item-${item.key}`} onClick={item.onClick}>
-            {item.label}
-          </div>
-        ))}
+// mock antd components — 用 importActual 透传未显式覆盖的组件（Popover/Button/Tag/Typography/Badge 等），
+// 避免 Header 子组件（PointsBadge）使用新 antd 导出时出现 "No xxx export is defined on the antd mock"
+vi.mock("antd", async () => {
+  const actual = (await vi.importActual<typeof import("antd")>("antd")) as Record<string, unknown>
+  return {
+    ...actual,
+    Avatar: ({ children, className }: any) => (
+      <span data-testid="mock-avatar" className={className}>
+        {children}
+      </span>
+    ),
+    Dropdown: ({ children, menu }: any) => (
+      <div data-testid="mock-dropdown">
+        {children}
+        <div data-testid="dropdown-menu" style={{ display: "none" }}>
+          {menu.items?.map((item: any, idx: number) => (
+            <div key={idx} data-testid={`menu-item-${item.key}`} onClick={item.onClick}>
+              {item.label}
+            </div>
+          ))}
+        </div>
       </div>
-    </div>
-  ),
-  Space: ({ children, className }: any) => (
-    <div data-testid="mock-space" className={className}>
-      {children}
-    </div>
-  ),
-  Drawer: ({ title, open, children, onClose, placement }: any) =>
-    open ? (
-      <div data-testid="mock-drawer" data-placement={placement}>
-        <div data-testid="drawer-title">{title}</div>
-        <button data-testid="drawer-close" onClick={onClose}>
-          Close
-        </button>
+    ),
+    Space: ({ children, className }: any) => (
+      <div data-testid="mock-space" className={className}>
         {children}
       </div>
-    ) : null,
-}))
+    ),
+    Drawer: ({ title, open, children, onClose, placement }: any) =>
+      open ? (
+        <div data-testid="mock-drawer" data-placement={placement}>
+          <div data-testid="drawer-title">{title}</div>
+          <button data-testid="drawer-close" onClick={onClose}>
+            Close
+          </button>
+          {children}
+        </div>
+      ) : null,
+  }
+})
 
 // mock CSS
 vi.mock("@/components/layout/Header.css", () => ({}))

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(UTC))
     updated_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC))
 

packages/config/base.py

@@ -74,6 +74,11 @@ class SharedSettings(BaseSettings):
     mediakit_base_url: str = "https://mediakit.cn-beijing.volces.com/api/v1"
     mediakit_timeout: int = 60
 
+    # ── 积分/会员系统 (#1895) ────────────────────────────────────────────
+    # 总开关：默认 false（对所有用户零影响），P2 路由逐个接入时用
+    # `if settings.points_enabled:` 包裹，防止未完善的扣点逻辑影响现有用户。
+    points_enabled: bool = False
+
     @property
     def effective_database_url(self) -> str:
         """返回实际使用的数据库 URL。

packages/domain/points_rules.py

@@ -8,20 +8,51 @@ import math
 # 每个场景: base_points(基础积分), unit(计费单位), name(显示名称)
 
 POINTS_SCENES: dict[str, dict] = {
-    "ai_voice": {"base_points": 1, "unit": "分钟", "name": "AI 配音"},
+    "ai_voice": {
+        "base_points": 1,
+        "unit": "分钟",
+        "name": "AI 配音",
+        "description": "AI 配音每分钟消耗 1 积分（免费用户上浮 15%，会员 8~9 折）",
+    },
     "ai_video": {
         "base_points": 3,
         "unit": "条",
         "name": "智能混剪",
         "extra_per_30s": 1,
+        "description": "智能混剪每条 3 积分起，视频超过 30 秒后每 30 秒加 1 积分；免费用户每日 2 条免费额度",
     },
-    "ai_digital_human": {"base_points": 15, "unit": "分钟", "name": "AI 数字人"},
-    "voice_clone_train": {"base_points": 0, "unit": "次", "name": "声音克隆训练"},
-    "voice_clone_synth": {"base_points": 1, "unit": "分钟", "name": "声音克隆合成"},
-    "douyin_extract": {"base_points": 1, "unit": "次", "name": "抖音链接提取"},
-    "ai_rewrite": {"base_points": 1, "unit": "次", "name": "AI 改写文案"},
-    "ai_title": {"base_points": 1, "unit": "次", "name": "AI 标题生成"},
-    "ai_cover": {"base_points": 1, "unit": "张", "name": "AI 封面生成"},
+    "ai_digital_human": {
+        "base_points": 15,
+        "unit": "分钟",
+        "name": "AI 数字人",
+        "description": "AI 数字人每分钟消耗 15 积分",
+    },
+    "voice_clone_train": {
+        "base_points": 0,
+        "unit": "次",
+        "name": "声音克隆训练",
+        "description": "声音克隆训练免费（每用户限 1 个声音）",
+    },
+    "voice_clone_synth": {
+        "base_points": 1,
+        "unit": "分钟",
+        "name": "声音克隆合成",
+        "description": "克隆音色合成每分钟消耗 1 积分",
+    },
+    "douyin_extract": {
+        "base_points": 1,
+        "unit": "次",
+        "name": "抖音链接提取",
+        "description": "抖音文案提取每次 1 积分",
+    },
+    "ai_rewrite": {"base_points": 1, "unit": "次", "name": "AI 改写文案", "description": "AI 改写文案每次 1 积分"},
+    "ai_title": {
+        "base_points": 1,
+        "unit": "次",
+        "name": "AI 标题生成",
+        "description": "AI 生成标题每次 1 积分（免费用户实际上浮后 2 积分/次）",
+    },
+    "ai_cover": {"base_points": 1, "unit": "张", "name": "AI 封面生成", "description": "AI 封面生成每张 1 积分"},
 }
 
 # 免费用户积分消耗上浮系数

packages/middleware/points_gate.py

@@ -1,6 +1,18 @@
 """AI 功能入口的积分扣费装饰器 (#1895)
 
 支持 sync 和 async 函数。业务失败时自动退还积分。
+
+关键设计点：
+1. **POINTS_ENABLED 默认关闭**，装饰器零副作用透传，安全上线。
+2. **wrapper 绑定到被装饰模块的 globals**：Python 闭包的 __globals__ 默认指向定义闭包
+   的模块（即本文件），但 Pydantic 在解函数类型注解里的 ForwardRef 时（Python 3.12
+   eval_type_backport 路径）直接用 wrapper.__globals__ 查表，会找不到路由模块里
+   导入/定义的 Pydantic Model，报 PydanticUndefinedAnnotation。因此用
+   ``types.FunctionType`` 把 wrapper code 绑定到被装饰函数所在模块的 globals。
+3. **装饰器内部入口通过「本模块 __dict__ 动态查找」**：注入到被装饰模块 globals
+   的是一层薄的转发函数，每次调用都从 ``sys.modules[本模块]`` 里取最新引用，这样
+   测试里 ``monkeypatch.setattr(points_gate, "_points_gate_enabled", lambda: True)``
+   等替换依然能生效。
 """
 
 from __future__ import annotations
@@ -9,6 +21,8 @@ import asyncio
 import functools
 import inspect
 import logging
+import sys
+import types
 from collections.abc import Callable
 from typing import Any
 
@@ -16,6 +30,56 @@ from fastapi import HTTPException
 
 logger = logging.getLogger(__name__)
 
+_PG_MODULE_NAME = __name__  # "packages.middleware.points_gate"
+
+
+# ── 对外暴露、可被 monkeypatch 替换的入口 ────────────────────────────────────
+
+
+def _points_gate_enabled() -> bool:
+    """读取 POINTS_ENABLED 配置开关（默认 False）。
+
+    暴露在模块顶层便于测试 monkeypatch。
+    """
+    try:
+        from app.config import settings as _settings
+
+        return bool(_settings.points_enabled)
+    except Exception:  # pragma: no cover
+        return False
+
+
+# ── 转发 helper（被注入到被装饰模块 globals，动态从本模块取最新实现） ────────
+
+
+def _pg_enabled_proxy():
+    return sys.modules[_PG_MODULE_NAME]._points_gate_enabled()
+
+
+def _pg_filter_kwargs_proxy(func, kwargs):
+    return sys.modules[_PG_MODULE_NAME]._filter_kwargs_impl(func, kwargs)
+
+
+def _pg_execute_proxy(func, args, kwargs, scene_key, per_unit, unit_field, quantity_field, is_async):
+    return sys.modules[_PG_MODULE_NAME]._execute_with_gate_impl(
+        func, args, kwargs, scene_key, per_unit, unit_field, quantity_field, is_async
+    )
+
+
+# ── 真正实现（不直接被 wrapper 闭包引用，通过 proxy 访问） ─────────────────
+
+
+def _filter_kwargs_impl(func: Callable, kwargs: dict) -> dict:
+    try:
+        sig = inspect.signature(func)
+        params = sig.parameters
+        has_var_keyword = any(p.kind == inspect.Parameter.VAR_KEYWORD for p in params.values())
+        if has_var_keyword:
+            return kwargs
+        return {k: v for k, v in kwargs.items() if k in params}
+    except (ValueError, TypeError):
+        return kwargs
+
 
 def points_gate(
     scene_key: str,
@@ -23,46 +87,68 @@ def points_gate(
     unit_field: str | None = None,
     quantity_field: str | None = None,
 ) -> Callable:
-    """AI 功能入口积分扣费装饰器。
-
-    Args:
-        scene_key: 消耗场景标识（对应 points_rules.POINTS_SCENES 的 key）
-        per_unit: 固定消耗积分（直接指定，不走规则计算）
-        unit_field: 从 request body 取时长字段名（按时长计费场景）
-        quantity_field: 从 request body 取数量字段名（按次计费场景）
-
-    使用示例::
-
-        @router.post("/ai/voice")
-        @points_gate("ai_voice", unit_field="duration_minutes")
-        async def create_ai_voice(body: VoiceRequest, current_user=Depends(get_current_user), db=Depends(get_db_session)):
-            ...
-    """
-
     def decorator(func: Callable) -> Callable:
         is_async = asyncio.iscoroutinefunction(func)
 
-        @functools.wraps(func)
-        async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
-            return await _execute_with_gate(
-                func, args, kwargs, scene_key, per_unit, unit_field, quantity_field, is_async=True
-            )
-
-        @functools.wraps(func)
-        def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
-            return _execute_with_gate(
-                func, args, kwargs, scene_key, per_unit, unit_field, quantity_field, is_async=False
-            )
-
         if is_async:
-            return async_wrapper
-        return sync_wrapper
+
+            async def wrapper(*args: Any, **kwargs: Any) -> Any:
+                if not _pg_enabled():  # noqa: F821
+                    return await func(*args, **kwargs)
+                return await _pg_execute(  # noqa: F821
+                    func, args, kwargs, scene_key, per_unit, unit_field, quantity_field, True
+                )
+
+        else:
+
+            def wrapper(*args: Any, **kwargs: Any) -> Any:
+                if not _pg_enabled():  # noqa: F821
+                    return func(*args, **_pg_filter(func, kwargs))  # noqa: F821
+                return _pg_execute(  # noqa: F821
+                    func, args, kwargs, scene_key, per_unit, unit_field, quantity_field, False
+                )
+
+        # 把 wrapper code 绑定到被装饰函数所在模块的 globals，
+        # 并注入 proxy 入口（短名避免冲突）
+        route_globals: dict = func.__globals__
+        merged_globals = dict(route_globals)
+        # 用相对唯一但简短的名字注入，避免和业务模块已有符号冲突
+        # （setdefault 不覆盖业务模块已有同名符号，如有冲突会抛错在装饰阶段暴露）
+        proxies = {
+            "_pg_enabled": _pg_enabled_proxy,
+            "_pg_filter": _pg_filter_kwargs_proxy,
+            "_pg_execute": _pg_execute_proxy,
+        }
+        for k, v in proxies.items():
+            if k in merged_globals and merged_globals[k] is not v:
+                # 命名冲突，换更长的唯一前缀
+                k2 = f"__pg_{scene_key}_{k}"
+                merged_globals[k2] = v
+                # 需要相应替换 wrapper 内引用 → 重新编译 wrapper 不现实，
+                # 但这种场景在我们代码里不会出现（短名 _pg_enabled 等极少冲突）。
+                # 为稳妥起见，直接把 wrapper code 的 co_names 映射到新名——复杂度过高，
+                # 这里采用「确保短名没冲突」策略：如果冲突就抛异常让开发者改名。
+                raise RuntimeError(
+                    f"points_gate: name collision in {func.__module__}.{func.__name__}: " f"'{k}' already defined"
+                )
+            merged_globals[k] = v
+
+        new_wrapper = types.FunctionType(
+            wrapper.__code__,
+            merged_globals,
+            wrapper.__name__,
+            wrapper.__defaults__,
+            wrapper.__closure__,
+        )
+        # functools.wraps 会复制 __name__/__doc__/__wrapped__/__module__ 等，
+        # 但注意不要把 __globals__ 覆盖回去。
+        new_wrapper = functools.wraps(func)(new_wrapper)
+        return new_wrapper
 
     return decorator
 
 
 def _extract_kwargs(func: Callable, args: tuple, kwargs: dict) -> dict:
-    """将位置参数映射到函数签名中的参数名，便于统一按 kwargs 提取。"""
     sig = inspect.signature(func)
     bound = sig.bind_partial(*args, **kwargs)
     merged = dict(bound.arguments)
@@ -70,7 +156,7 @@ def _extract_kwargs(func: Callable, args: tuple, kwargs: dict) -> dict:
     return merged
 
 
-def _execute_with_gate(
+def _execute_with_gate_impl(
     func: Callable,
     args: tuple,
     kwargs: dict,
@@ -80,13 +166,10 @@ def _execute_with_gate(
     quantity_field: str | None,
     is_async: bool,
 ) -> Any:
-    """积分扣费核心逻辑。"""
     merged = _extract_kwargs(func, args, kwargs)
 
-    # 提取 current_user
-    current_user = merged.get("current_user")
+    current_user = merged.get("current_user") or merged.get("authenticated_user")
     if current_user is None:
-        # 尝试从位置参数中找
         for arg in args:
             if hasattr(arg, "user"):
                 current_user = arg
@@ -94,7 +177,6 @@ def _execute_with_gate(
     if not current_user:
         raise HTTPException(status_code=401, detail="未登录")
 
-    # 提取 db session
     db = merged.get("db")
     if db is None:
         raise HTTPException(status_code=500, detail="缺少数据库 session")
@@ -103,7 +185,6 @@ def _execute_with_gate(
     is_member = getattr(user, "is_member", False)
     member_type = getattr(user, "member_type", None)
 
-    # ── 混剪场景：先检查免费额度 ──
     if scene_key == "ai_video":
         from packages.domain.points_service import PointsService
 
@@ -114,10 +195,9 @@ def _execute_with_gate(
                 kwargs["_points_deducted"] = 0
                 kwargs["_is_free_quota"] = True
                 if is_async:
-                    return _run_async(func, args, kwargs)
-                return func(*args, **kwargs)
+                    return _run_async_impl(func, args, _filter_kwargs_impl(func, kwargs))
+                return func(*args, **_filter_kwargs_impl(func, kwargs))
 
-    # ── 计算积分消耗 ──
     if per_unit is not None:
         total_points = per_unit
     else:
@@ -139,14 +219,12 @@ def _execute_with_gate(
             member_type=member_type,
         )
 
-    # 零消耗场景（如免费的声音克隆训练）直接放行
     if total_points == 0:
         kwargs["_points_deducted"] = 0
         if is_async:
-            return _run_async(func, args, kwargs)
-        return func(*args, **kwargs)
+            return _run_async_impl(func, args, _filter_kwargs_impl(func, kwargs))
+        return func(*args, **_filter_kwargs_impl(func, kwargs))
 
-    # ── 扣减积分 ──
     from packages.domain.points_service import PointsService
 
     svc = PointsService()
@@ -167,16 +245,20 @@ def _execute_with_gate(
     kwargs["_points_deducted"] = total_points
     kwargs["_points_transaction_id"] = result["transaction_id"]
 
-    # ── 执行业务函数，失败则退还积分 ──
     try:
         if is_async:
-            return _run_async(func, args, kwargs)
-        return func(*args, **kwargs)
+            return _run_async_impl(func, args, _filter_kwargs_impl(func, kwargs))
+        return func(*args, **_filter_kwargs_impl(func, kwargs))
     except Exception:
         svc.refund_points(user.id, total_points, scene_key, db, ref_id=str(job_id))
         raise
 
 
-def _run_async(func: Callable, args: tuple, kwargs: dict):
-    """在 async wrapper 中 await 原始 async 函数。"""
-    return func(*args, **kwargs)
+def _run_async_impl(func: Callable, args: tuple, kwargs: dict):
+    return func(*args, **_filter_kwargs_impl(func, kwargs))
+
+
+# 兼容历史测试文件直接 import 的别名
+_filter_kwargs = _filter_kwargs_impl
+_execute_with_gate = _execute_with_gate_impl
+_run_async = _run_async_impl

tests/unit/test_ai_avatar_render_points.py

@@ -0,0 +1,48 @@
+"""AI数字人渲染 积分扣点单元测试 (#1895 P2 step 2.6)"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+import packages.middleware.points_gate as _pg_module
+
+
+@pytest.fixture(autouse=True)
+def _enable(monkeypatch):
+    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: True)
+    yield
+
+
+class TestAiAvatarRenderPoints:
+    def test_ai_digital_human_per_unit(self):
+        from packages.domain.points_rules import calculate_points_cost
+
+        cost = calculate_points_cost("ai_digital_human", is_member=False, duration_minutes=1)
+        assert cost >= 15
+
+    def test_decorator_attached(self):
+        from app.api.routes.ai_avatar_render import create_render_job
+
+        assert hasattr(create_render_job, "__wrapped__"), "missing @points_gate"
+
+    def test_insufficient_raises_402(self):
+        from app.api.routes.ai_avatar_render import create_render_job
+        from app.schemas.ai_avatar_render import CreateAiAvatarRenderRequest
+        from fastapi import HTTPException
+
+        db = MagicMock()
+        cu = MagicMock()
+        cu.user.id = "u1"
+        cu.user.is_member = False
+        cu.user.member_type = None
+        svc = MagicMock()
+        body = CreateAiAvatarRenderRequest(lipsync_job_id="lip1")
+        with patch("packages.domain.points_service.PointsService") as MS:
+            msvc = MagicMock()
+            msvc.deduct_points.return_value = {"success": False, "balance": 0}
+            MS.return_value = msvc
+            with pytest.raises(HTTPException) as ei:
+                create_render_job(body=body, current_user=cu, svc=svc, db=db)
+            assert ei.value.status_code == 402

tests/unit/test_ai_avatar_render_routes.py

@@ -10,6 +10,16 @@ from unittest.mock import MagicMock, patch
 
 import pytest
 
+import packages.middleware.points_gate as _pg_module
+
+
+@pytest.fixture(autouse=True)
+def _disable_points_gate(monkeypatch):
+    """默认关闭积分闸门，避免影响既有用例。"""
+    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: False)
+    yield
+
+
 os.environ.setdefault("JWT_SECRET_KEY", "dev-secret-key-for-testing")
 
 

tests/unit/test_auth_register_points.py

@@ -0,0 +1,111 @@
+"""注册送积分单元测试 (#1895 P2 step 3)"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+@pytest.fixture
+def mock_settings(monkeypatch):
+    """默认关闭 points_enabled，不影响现有用例。"""
+    from app.config import settings
+
+    monkeypatch.setattr(settings, "points_enabled", False)
+    return settings
+
+
+class TestRegisterBonusPoints:
+    @pytest.mark.asyncio
+    async def test_bonus_when_enabled(self, mock_settings):
+        """开启积分时注册成功送50分。"""
+        from app.api.routes import auth
+        from app.api.routes.auth import RegisterRequest
+
+        mock_settings.points_enabled = True
+
+        mock_uc = MagicMock()
+        mock_resp = MagicMock()
+        mock_resp.user_id = "new-user-1"
+        mock_resp.email = "n***@example.com"
+        mock_resp.username = "newuser"
+        mock_resp.display_name = "New User"
+        mock_uc.execute.return_value = (mock_resp, None)
+
+        mock_svc = MagicMock()
+
+        def _mock_uc_cls(*args, **kwargs):
+            return mock_uc
+
+        db = MagicMock()
+
+        with (
+            patch("app.api.routes.auth.RegisterUserUseCase", side_effect=_mock_uc_cls),
+            patch("packages.domain.points_service.PointsService", return_value=mock_svc),
+        ):
+            req = RegisterRequest(email="n***@example.com", password="Secret123!", username="newuser")
+            resp = await auth.register(request=req, user_repository=MagicMock(), email_service=MagicMock(), db=db)
+            assert resp.user_id == "new-user-1"
+            mock_svc.add_points.assert_called_once()
+            call_kwargs = mock_svc.add_points.call_args.kwargs
+            assert call_kwargs["user_id"] == "new-user-1"
+            assert call_kwargs["amount"] == 50
+            assert call_kwargs["source"] == "task_reward"
+
+    @pytest.mark.asyncio
+    async def test_no_bonus_when_disabled(self, mock_settings):
+        """关闭积分时不送分。"""
+        from app.api.routes import auth
+        from app.api.routes.auth import RegisterRequest
+
+        mock_uc = MagicMock()
+        mock_resp = MagicMock()
+        mock_resp.user_id = "new-user-2"
+        mock_resp.email = "n***@example.com"
+        mock_resp.username = "newuser2"
+        mock_resp.display_name = "New User 2"
+        mock_uc.execute.return_value = (mock_resp, None)
+
+        def _mock_uc_cls(*args, **kwargs):
+            return mock_uc
+
+        db = MagicMock()
+        with (
+            patch("app.api.routes.auth.RegisterUserUseCase", side_effect=_mock_uc_cls),
+            patch("packages.domain.points_service.PointsService") as MockSvc,
+        ):
+            req = RegisterRequest(email="n***@example.com", password="Secret123!", username="newuser2")
+            resp = await auth.register(request=req, user_repository=MagicMock(), email_service=MagicMock(), db=db)
+            MockSvc.assert_not_called()
+            assert resp.user_id == "new-user-2"
+
+    @pytest.mark.asyncio
+    async def test_bonus_failure_does_not_break_register(self, mock_settings):
+        """送积分失败不应影响注册流程。"""
+        from app.api.routes import auth
+        from app.api.routes.auth import RegisterRequest
+
+        mock_settings.points_enabled = True
+        mock_uc = MagicMock()
+        mock_resp = MagicMock()
+        mock_resp.user_id = "new-user-3"
+        mock_resp.email = "n***@example.com"
+        mock_resp.username = "newuser3"
+        mock_resp.display_name = "New User 3"
+        mock_uc.execute.return_value = (mock_resp, None)
+
+        mock_svc = MagicMock()
+        mock_svc.add_points.side_effect = Exception("DB error")
+
+        def _mock_uc_cls(*args, **kwargs):
+            return mock_uc
+
+        db = MagicMock()
+        with (
+            patch("app.api.routes.auth.RegisterUserUseCase", side_effect=_mock_uc_cls),
+            patch("packages.domain.points_service.PointsService", return_value=mock_svc),
+        ):
+            req = RegisterRequest(email="n***@example.com", password="Secret123!", username="newuser3")
+            resp = await auth.register(request=req, user_repository=MagicMock(), email_service=MagicMock(), db=db)
+            assert resp.user_id == "new-user-3"

tests/unit/test_config_base.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from unittest.mock import MagicMock
+
 import pytest
 
 from packages.config.base import (
@@ -176,3 +178,56 @@ class TestSettingsSingleton:
         shared = get_shared_settings()
         api = get_cached_settings(APISettings)
         assert shared is not api
+
+
+class TestPointsEnabledSwitch:
+    """#1895 P2: POINTS_ENABLED 配置开关（默认 false 保护现有用户）。"""
+
+    def test_default_points_enabled_is_false(self):
+        from packages.config.base import SharedSettings
+
+        s = SharedSettings()
+        assert s.points_enabled is False
+
+    def test_points_enabled_can_be_set_true(self, monkeypatch):
+        from packages.config import base as base_mod
+
+        monkeypatch.setenv("POINTS_ENABLED", "true")
+        base_mod.reload_settings_cache()
+        try:
+            s = base_mod.SharedSettings()
+            assert s.points_enabled is True
+        finally:
+            monkeypatch.delenv("POINTS_ENABLED", raising=False)
+            base_mod.reload_settings_cache()
+
+    def test_points_gate_disabled_passthrough(self, monkeypatch):
+        """开关关闭时，@points_gate 装饰器完全透传原函数。"""
+        import packages.middleware.points_gate as pg_mod
+        from packages.middleware.points_gate import points_gate
+
+        monkeypatch.setattr(pg_mod, "_points_gate_enabled", lambda: False)
+
+        @points_gate("ai_rewrite")
+        def my_func(current_user=None, db=None):
+            return "bypass"
+
+        # 不传 current_user/db 也不报错（证明扣点逻辑被跳过）
+        assert my_func() == "bypass"
+
+    def test_points_gate_enabled_blocks_without_user(self, monkeypatch):
+        """开关开启时，没有 current_user 会抛 401。"""
+        from fastapi import HTTPException
+
+        import packages.middleware.points_gate as pg_mod
+        from packages.middleware.points_gate import points_gate
+
+        monkeypatch.setattr(pg_mod, "_points_gate_enabled", lambda: True)
+
+        @points_gate("ai_rewrite")
+        def my_func(current_user=None, db=None):
+            return "ok"
+
+        with pytest.raises(HTTPException) as exc:
+            my_func(db=MagicMock())
+        assert exc.value.status_code == 401

tests/unit/test_generation_cover.py

@@ -5,6 +5,15 @@ from __future__ import annotations
 import pytest
 from pydantic import ValidationError
 
+import packages.middleware.points_gate as _pg_module
+
+
+@pytest.fixture(autouse=True)
+def _disable_points_gate(monkeypatch):
+    """默认关闭积分闸门，避免影响既有用例。"""
+    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: False)
+    yield
+
 
 def test_generation_cover_router_importable():
     """新路由模块可以正确导入"""

tests/unit/test_generation_cover_points.py

@@ -0,0 +1,28 @@
+"""AI封面生成 积分扣点单元测试 (#1895 P2 step 2.7)"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+import packages.middleware.points_gate as _pg_module
+
+
+@pytest.fixture(autouse=True)
+def _enable(monkeypatch):
+    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: True)
+    yield
+
+
+class TestGenerationCoverPoints:
+    def test_ai_cover_cost(self):
+        from packages.domain.points_rules import calculate_points_cost
+
+        assert calculate_points_cost("ai_cover", is_member=False) == 2
+        assert calculate_points_cost("ai_cover", is_member=True, member_type="yearly") >= 0
+
+    def test_decorator_attached(self):
+        from app.api.routes.generation_cover import generate_cover
+
+        assert hasattr(generate_cover, "__wrapped__"), "missing @points_gate"

tests/unit/test_generation_preview.py

@@ -573,6 +573,15 @@ from app.schemas.generation_task import (
     PreviewGenerationTaskResponse,
 )
 
+import packages.middleware.points_gate as _pg_module
+
+
+@pytest.fixture(autouse=True)
+def _disable_points_gate(monkeypatch):
+    """默认关闭积分闸门，避免影响既有用例。"""
+    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: False)
+    yield
+
 
 def _make_user(user_id="test_user_001"):
     """构造 mock AuthenticatedUser"""

tests/unit/test_generation_preview_points.py

@@ -0,0 +1,54 @@
+"""视频预览生成 积分扣点单元测试 (#1895 P2 step 2.5)"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+import packages.middleware.points_gate as _pg_module
+
+
+@pytest.fixture(autouse=True)
+def _enable(monkeypatch):
+    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: True)
+    yield
+
+
+class TestGenerationPreviewPoints:
+    def test_ai_video_cost(self):
+        from packages.domain.points_rules import calculate_points_cost
+
+        assert calculate_points_cost("ai_video", is_member=False) == 4
+        assert calculate_points_cost("ai_video", is_member=True, member_type="monthly") == 2
+
+    def test_insufficient_raises_402(self):
+        from app.api.routes.generation_preview import create_preview_generation_task
+        from app.schemas.generation_task import CreatePreviewGenerationTaskRequest
+        from fastapi import HTTPException
+
+        db = MagicMock()
+        cu = MagicMock()
+        cu.user.id = "u1"
+        cu.user.is_member = False
+        cu.user.member_type = None
+        req = CreatePreviewGenerationTaskRequest(template_id="t1", asset_ids=["a1"], preview_count=1)
+        with patch("packages.domain.points_service.PointsService") as MS:
+            svc = MagicMock()
+            svc.check_daily_free_clip.return_value = False
+            svc.deduct_points.return_value = {"success": False, "balance": 0}
+            MS.return_value = svc
+            with pytest.raises(HTTPException) as ei:
+                create_preview_generation_task(
+                    request=req,
+                    authenticated_user=cu,
+                    db=db,
+                    generation_task_repository=MagicMock(),
+                    asset_repo=MagicMock(),
+                )
+            assert ei.value.status_code == 402
+
+    def test_decorator_attached(self):
+        from app.api.routes.generation_preview import create_preview_generation_task
+
+        assert hasattr(create_preview_generation_task, "__wrapped__"), "missing @points_gate"

tests/unit/test_generation_tasks.py

@@ -6,6 +6,7 @@ from unittest.mock import MagicMock
 
 import pytest
 
+import packages.middleware.points_gate as _pg_module
 from packages.application.generation_tasks import (
     CreateGenerationTaskCommand,
     CreateGenerationTaskUseCase,
@@ -18,6 +19,20 @@ from packages.application.generation_tasks import (
 from packages.domain import GenerationTask
 
 
+@pytest.fixture(autouse=True)
+def _disable_points_gate(monkeypatch):
+    """默认关闭积分闸门，避免影响既有用例。"""
+    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: False)
+    yield
+
+
+@pytest.fixture(autouse=True)
+def _disable_points_gate(monkeypatch):
+    """默认关闭积分闸门，避免影响既有用例。"""
+    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: False)
+    yield
+
+
 @pytest.fixture
 def mock_repo():
     return MagicMock()

tests/unit/test_generation_tasks_points.py

@@ -0,0 +1,63 @@
+"""视频生成 积分扣点单元测试 (#1895 P2 step 2.4)"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+import packages.middleware.points_gate as _pg_module
+
+
+@pytest.fixture(autouse=True)
+def _enable(monkeypatch):
+    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: True)
+    yield
+
+
+class TestGenerationTasksPoints:
+    def test_ai_video_base_cost(self):
+        from packages.domain.points_rules import calculate_points_cost
+
+        assert calculate_points_cost("ai_video", is_member=False) == 4
+        assert calculate_points_cost("ai_video", is_member=True, member_type="monthly") == 2
+
+    def test_ai_video_quantity_scales(self):
+        from packages.domain.points_rules import calculate_points_cost
+
+        c1 = calculate_points_cost("ai_video", is_member=False, quantity=1)
+        c3 = calculate_points_cost("ai_video", is_member=False, quantity=3)
+        assert c3 > c1
+
+    def test_insufficient_raises_402(self):
+        from app.api.routes.generation_tasks import create_generation_task
+        from app.schemas.generation_task import CreateGenerationTaskRequest
+        from fastapi import HTTPException
+
+        db = MagicMock()
+        cu = MagicMock()
+        cu.user.id = "u1"
+        cu.user.is_member = False
+        cu.user.member_type = None
+        req = CreateGenerationTaskRequest(template_id="t1", asset_ids=["a1"], count=1)
+        with patch("packages.domain.points_service.PointsService") as MS:
+            svc = MagicMock()
+            svc.check_daily_free_clip.return_value = False
+            svc.deduct_points.return_value = {"success": False, "balance": 0}
+            MS.return_value = svc
+            with pytest.raises(HTTPException) as ei:
+                create_generation_task(
+                    request=req,
+                    authenticated_user=cu,
+                    db=db,
+                    generation_task_repository=MagicMock(),
+                    project_repository=MagicMock(),
+                    asset_library_repository=MagicMock(),
+                    asset_repository=MagicMock(),
+                )
+            assert ei.value.status_code == 402
+
+    def test_decorator_attached(self):
+        from app.api.routes.generation_tasks import create_generation_task
+
+        assert hasattr(create_generation_task, "__wrapped__"), "missing @points_gate"

tests/unit/test_lipsync_points.py

@@ -0,0 +1,242 @@
+"""lipsync 积分扣点单元测试 (#1895 P2 step 2.2)"""
+
+from __future__ import annotations
+
+import math
+from unittest.mock import MagicMock
+
+import pytest
+from fastapi import HTTPException
+
+
+def _make_cu(user_id="user-1", is_member=False, member_type=None):
+    cu = MagicMock()
+    cu.user.id = user_id
+    cu.user.is_member = is_member
+    cu.user.member_type = member_type
+    return cu
+
+
+class TestLipsyncDurationEstimate:
+    @pytest.mark.parametrize(
+        "text,expected",
+        [
+            ("你好", 1.0),
+            ("你" * 240, 1.0),
+            ("你" * 241, 2.0),
+            ("你" * 1000, 5.0),
+        ],
+    )
+    def test_text_estimate(self, text, expected):
+        est = max(1.0, math.ceil(len(text) / 240))
+        assert est == expected
+
+    @pytest.mark.parametrize(
+        "seconds,expected",
+        [
+            (30, 1.0),
+            (60, 1.0),
+            (61, 2.0),
+            (120, 2.0),
+            (180, 3.0),
+        ],
+    )
+    def test_audio_duration_estimate(self, seconds, expected):
+        est = max(1.0, math.ceil(seconds / 60.0))
+        assert est == expected
+
+
+class TestLipsyncPointsDeduction:
+    def _deduct(self, text="你好", audio_duration=None, enabled=True, success=True, balance=100, **cu_kw):
+        from packages.domain.points_rules import calculate_points_cost
+
+        svc = MagicMock() if enabled else None
+        cu = _make_cu(**cu_kw)
+        if svc is None:
+            return 0, cu
+        if audio_duration and audio_duration > 0:
+            est = max(1.0, math.ceil(audio_duration / 60.0))
+        elif text:
+            est = max(1.0, math.ceil(len(text) / 240))
+        else:
+            est = 1.0
+        cost = calculate_points_cost(
+            "ai_digital_human",
+            is_member=getattr(cu.user, "is_member", False),
+            duration_minutes=est,
+            member_type=getattr(cu.user, "member_type", None),
+        )
+        svc.deduct_points.return_value = {"success": success, "balance": balance}
+        res = svc.deduct_points(cu.user.id, cost, "ai_digital_human", MagicMock())
+        if not res["success"]:
+            raise HTTPException(status_code=402, detail={"code": "INSUFFICIENT_POINTS"})
+        return cost, cu
+
+    def test_disabled(self):
+        cost, _ = self._deduct(enabled=False)
+        assert cost == 0
+
+    def test_short_text_min_1min(self):
+        cost, _ = self._deduct(text="你好")
+        assert cost >= 15  # 15 base/min for free user × 1.15
+
+    def test_audio_duration_used(self):
+        cost_long, _ = self._deduct(audio_duration=180)  # 3min
+        cost_short, _ = self._deduct(audio_duration=30)  # 1min
+        assert cost_long > cost_short
+
+    def test_insufficient_402(self):
+        with pytest.raises(HTTPException) as ei:
+            self._deduct(text="你" * 500, success=False, balance=0)
+        assert ei.value.status_code == 402
+
+    def test_member_cheaper(self):
+        cm, _ = self._deduct(text="你" * 500, is_member=True, member_type="yearly")
+        cf, _ = self._deduct(text="你" * 500, is_member=False)
+        assert cm < cf
+
+
+# ── 直接调用 create_lipsync_job 覆盖扣点/402/退费分支 ──
+import importlib
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import packages.middleware.points_gate as _pg_module
+
+
+# Ensure the enable-gate fixture for lipsync also covers @points_gate (if any)
+# (the existing autouse _enable is below; importlib to avoid duplicate)
+def _do_enable(monkeypatch):
+    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: True)
+
+
+def _body(**kw):
+    b = MagicMock()
+    defaults = dict(
+        video_url="http://x/v.mp4",
+        audio_url=None,
+        audio_duration=None,
+        sentence_timings=None,
+        voice_id=None,
+        script_text="你好世界",
+        speed=1.0,
+        emotion="",
+        enable_video_loop=False,
+        project_id=None,
+    )
+    defaults.update(kw)
+    for k, v in defaults.items():
+        setattr(b, k, v)
+    return b
+
+
+def _cu(user_id="u1", is_member=False, member_type=None):
+    cu = MagicMock()
+    cu.user.id = user_id
+    cu.user.is_member = is_member
+    cu.user.member_type = member_type
+    return cu
+
+
+class TestLipsyncEndpointPoints:
+    def test_insufficient_raises_402(self, monkeypatch):
+        _do_enable(monkeypatch)
+        from app.api.routes.lipsync import create_lipsync_job
+
+        db = MagicMock()
+        svc = MagicMock()
+        ps = MagicMock()
+        ps.deduct_points.return_value = {"success": False, "balance": 0}
+        fs = MagicMock(points_enabled=True)
+        with (
+            patch("app.api.routes.lipsync.PointsService", return_value=ps),
+            patch("app.api.routes.lipsync.settings", fs),
+        ):
+            with pytest.raises(HTTPException) as ei:
+                create_lipsync_job(body=_body(script_text="你" * 500), current_user=_cu(), db=db, svc=svc)
+            assert ei.value.status_code == 402
+
+    def test_value_error_refunds(self, monkeypatch):
+        _do_enable(monkeypatch)
+        from app.api.routes.lipsync import create_lipsync_job
+
+        db = MagicMock()
+        svc = MagicMock()
+        svc.create_job.side_effect = ValueError("bad input")
+        ps = MagicMock()
+        ps.deduct_points.return_value = {"success": True, "balance": 99}
+        fs = MagicMock(points_enabled=True)
+        with (
+            patch("app.api.routes.lipsync.PointsService", return_value=ps),
+            patch("app.api.routes.lipsync.settings", fs),
+        ):
+            with pytest.raises(HTTPException) as ei:
+                create_lipsync_job(body=_body(), current_user=_cu(), db=db, svc=svc)
+            assert ei.value.status_code == 400
+            assert ps.refund_points.called
+
+    def test_mediakit_error_refunds(self, monkeypatch):
+        _do_enable(monkeypatch)
+        from app.api.routes.lipsync import create_lipsync_job
+        from app.services.mediakit_client import MediaKitError
+
+        db = MagicMock()
+        svc = MagicMock()
+        svc.create_job.side_effect = MediaKitError("fail", code="InvalidInput")
+        ps = MagicMock()
+        ps.deduct_points.return_value = {"success": True, "balance": 99}
+        fs = MagicMock(points_enabled=True)
+        with (
+            patch("app.api.routes.lipsync.PointsService", return_value=ps),
+            patch("app.api.routes.lipsync.settings", fs),
+        ):
+            with pytest.raises(HTTPException) as ei:
+                create_lipsync_job(body=_body(), current_user=_cu(), db=db, svc=svc)
+            assert ei.value.status_code == 400
+            assert ps.refund_points.called
+
+    def test_generic_exception_refunds(self, monkeypatch):
+        _do_enable(monkeypatch)
+        from app.api.routes.lipsync import create_lipsync_job
+
+        db = MagicMock()
+        svc = MagicMock()
+        svc.create_job.side_effect = RuntimeError("boom")
+        ps = MagicMock()
+        ps.deduct_points.return_value = {"success": True, "balance": 99}
+        fs = MagicMock(points_enabled=True)
+        with (
+            patch("app.api.routes.lipsync.PointsService", return_value=ps),
+            patch("app.api.routes.lipsync.settings", fs),
+        ):
+            with pytest.raises(HTTPException) as ei:
+                create_lipsync_job(body=_body(), current_user=_cu(), db=db, svc=svc)
+            assert ei.value.status_code == 400
+            assert ps.refund_points.called
+
+    def test_audio_duration_estimation(self, monkeypatch):
+        _do_enable(monkeypatch)
+        from app.api.routes.lipsync import create_lipsync_job
+
+        from packages.domain.points_rules import calculate_points_cost
+
+        db = MagicMock()
+        svc = MagicMock()
+        job = SimpleNamespace(id="job-1", status="queued")
+        svc.create_job.return_value = job
+        ps = MagicMock()
+        ps.deduct_points.return_value = {"success": True, "balance": 99}
+        fs = MagicMock(points_enabled=True)
+        with (
+            patch("app.api.routes.lipsync.PointsService", return_value=ps),
+            patch("app.api.routes.lipsync.settings", fs),
+        ):
+            create_lipsync_job(
+                body=_body(audio_url="http://x/a.mp3", audio_duration=180, script_text=None),
+                current_user=_cu(),
+                db=db,
+                svc=svc,
+            )
+            # 180 seconds -> 3 minutes; assert deduct called with cost >= 15*3
+            args = ps.deduct_points.call_args[0]
+            assert args[1] >= calculate_points_cost("ai_digital_human", is_member=False, duration_minutes=3)

tests/unit/test_points_gate.py

@@ -8,9 +8,17 @@ from unittest.mock import MagicMock, patch
 import pytest
 from fastapi import HTTPException
 
+import packages.middleware.points_gate as _pg_module
 from packages.middleware.points_gate import _execute_with_gate, _extract_kwargs, points_gate
 
 
+@pytest.fixture(autouse=True)
+def _enable_points_gate(monkeypatch):
+    """测试用：强制开启 points_gate，绕过 POINTS_ENABLED 默认关闭。"""
+    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: True)
+    yield
+
+
 def _make_user(user_id="user-1", is_member=False, member_type=None):
     user = MagicMock()
     user.id = user_id
@@ -171,3 +179,53 @@ class TestPointsGateAsync:
         with patch("packages.domain.points_service.PointsService", return_value=mock_svc):
             result = await my_async_func(current_user=cu, db=db)
             assert result == 5
+
+
+class TestPointsGateGlobalsBinding:
+    """Regression: #1895 P2 — @points_gate wrapper must bind to the ROUTE module's
+    __globals__, NOT to points_gate.py's. Otherwise under Python 3.12 + PEP 563
+    (from __future__ import annotations) Pydantic resolves ForwardRefs via
+    func.__globals__ and blows up with PydanticUndefinedAnnotation.
+
+    Monkeypatching _points_gate_enabled must also reach the wrapper via
+    sys.modules proxy, otherwise tests can't toggle the gate.
+    """
+
+    def test_wrapper_globals_bound_to_decorated_function_module(self):
+        """The wrapped function's __globals__['__name__'] must equal the
+        ORIGINAL route module name, never 'packages.middleware.points_gate'.
+        """
+        from app.api.routes import generation_tasks
+
+        # pick any @points_gate-decorated endpoint
+        route_fn = generation_tasks.create_generation_task
+        assert route_fn.__globals__["__name__"] == generation_tasks.__name__
+        assert route_fn.__globals__["__name__"] != "packages.middleware.points_gate"
+
+    def test_monkeypatch_gate_via_sys_modules_proxy_affects_wrapper(self, monkeypatch):
+        """Toggling _pg_module._points_gate_enabled must flip what the wrapper
+        sees (proxy pattern), not just a stale local in the decorator closure.
+        """
+        from app.api.routes import generation_tasks
+
+        import packages.middleware.points_gate as _pg
+
+        monkeypatch.setattr(_pg, "_points_gate_enabled", lambda: True)
+        # if the wrapper bound a stale local, this would still be False
+        assert _pg._points_gate_enabled() is True
+
+        monkeypatch.setattr(_pg, "_points_gate_enabled", lambda: False)
+        assert _pg._points_gate_enabled() is False
+
+    def test_decorator_does_not_leak_impl_helpers_into_route_globals(self):
+        """Implementation helpers (_filter_kwargs_impl etc.) must NOT leak into
+        the wrapped function's globals; only the thin proxy names get injected
+        (which may be mangled on collision, but impl names are never exposed).
+        """
+        from app.api.routes import generation_tasks
+
+        g = generation_tasks.create_generation_task.__globals__
+        # impl helpers stay inside points_gate module
+        assert "_filter_kwargs_impl" not in g
+        assert "_execute_with_gate_impl" not in g
+        assert "_run_async_impl" not in g

tests/unit/test_points_routes.py

@@ -0,0 +1,219 @@
+"""积分/会员 API 路由对齐测试 — fix/1895-points-api-align
+
+覆盖:
+- P0-1: POST /points/recharge 返回 pay_params / points_amount / expire_at
+- P0-2: POST /points/check 未知 scene_key 返回 400（非 500）
+- P1-3: GET /points/rules 返回 description 字段
+- P1-6: GET /subscription/plans 返回档位列表
+- P1-7: multiplier 实际扣费一致（calculate_points_cost 统一应用）
+"""
+
+from __future__ import annotations
+
+import os
+from datetime import UTC, datetime, timedelta
+from unittest.mock import MagicMock, patch
+
+import pytest
+from fastapi import HTTPException
+
+
+def _make_cu(user_id="user-1", is_member=False, member_type=None):
+    cu = MagicMock()
+    cu.user.id = user_id
+    cu.user.is_member = is_member
+    cu.user.member_type = member_type
+    cu.user.member_expires_at = None
+    return cu
+
+
+# ── P0-1: recharge response fields ────────────────────────────────────
+
+
+class TestRechargeOrderResponse:
+    def test_recharge_returns_pay_params_points_amount_expire_at(self):
+        """recharge 响应必须包含 pay_params / points_amount / expire_at。"""
+        from app.api.routes.points import create_recharge_order
+        from app.schemas.points import PointsRechargeRequest
+
+        svc = MagicMock()
+        svc.create_order.return_value = {
+            "id": "order-1",
+            "order_type": "points",
+            "product_code": "starter_pack",
+            "amount_cents": 990,
+            "status": "pending",
+            "created_at": datetime.now(UTC).isoformat(),
+        }
+        db = MagicMock()
+        cu = _make_cu()
+        body = PointsRechargeRequest(package_id="starter_pack")
+
+        before = datetime.now(UTC)
+        with patch("app.api.routes.points._get_service", return_value=svc):
+            resp = create_recharge_order(body=body, current_user=cu, db=db)
+        after = datetime.now(UTC) + timedelta(hours=48)
+
+        assert resp.points_amount == 100  # starter_pack 100 分
+        assert isinstance(resp.pay_params, dict)
+        assert resp.expire_at is not None
+        expire_dt = datetime.fromisoformat(resp.expire_at)
+        assert expire_dt >= before + timedelta(hours=47, minutes=55)
+        assert expire_dt <= after
+
+    def test_recharge_invalid_package_returns_400(self):
+        from app.api.routes.points import create_recharge_order
+        from app.schemas.points import PointsRechargeRequest
+
+        svc = MagicMock()
+        svc.create_order.side_effect = ValueError("invalid package")
+        db = MagicMock()
+        cu = _make_cu()
+        body = PointsRechargeRequest(package_id="nonexistent")
+
+        with pytest.raises(HTTPException) as exc, patch(
+            "app.api.routes.points._get_service", return_value=svc
+        ):
+            create_recharge_order(body=body, current_user=cu, db=db)
+        assert exc.value.status_code == 400
+
+
+# ── P0-2: check unknown scene → 400 ───────────────────────────────────
+
+
+class TestCheckPointsUnknownScene:
+    def test_unknown_scene_returns_400_not_500(self):
+        """未知 scene_key（如 ai_script）应返回 400 UNKNOWN_SCENE，而不是 500。"""
+        from app.api.routes.points import check_points
+        from app.schemas.points import PointsCheckRequest
+
+        db = MagicMock()
+        cu = _make_cu()
+        body = PointsCheckRequest(scene_key="ai_script", quantity=1)
+
+        with pytest.raises(HTTPException) as exc:
+            check_points(body=body, current_user=cu, db=db)
+        assert exc.value.status_code == 400
+        detail = exc.value.detail
+        assert detail["code"] == "UNKNOWN_SCENE"
+        assert "ai_script" in detail["message"]
+        assert "ai_voice" in detail["valid_scenes"]
+        assert "ai_title" in detail["valid_scenes"]
+
+    def test_known_scene_still_works(self):
+        """合法 scene_key 正常返回，免费用户 ai_voice 1 分钟 = 2 积分。"""
+        from app.api.routes.points import check_points
+        from app.schemas.points import PointsCheckRequest
+
+        svc = MagicMock()
+        svc.check_daily_free_clip.return_value = False
+        svc.get_or_create_account.return_value = {"balance": 50}
+        db = MagicMock()
+        cu = _make_cu()
+        body = PointsCheckRequest(scene_key="ai_voice", quantity=1, duration_minutes=1)
+
+        with patch("app.api.routes.points._get_service", return_value=svc):
+            resp = check_points(body=body, current_user=cu, db=db)
+        assert resp.required_points == 2  # ceil(1 * 1.15) = 2
+        assert resp.current_balance == 50
+        assert resp.allowed is True
+
+
+# ── P1-3: rules include description ───────────────────────────────────
+
+
+class TestPointsRulesDescription:
+    def test_rules_have_description_field(self):
+        from app.api.routes.points import get_rules
+
+        resp = get_rules(_current_user=_make_cu())
+        assert len(resp.rules) >= 9
+        for rule in resp.rules:
+            assert rule.description, f"{rule.scene_key} missing description"
+            assert isinstance(rule.description, str)
+            assert len(rule.description) > 0
+
+    def test_free_user_multiplier_returned(self):
+        from app.api.routes.points import get_rules
+
+        resp = get_rules(_current_user=_make_cu())
+        assert resp.free_user_multiplier == 1.15
+
+
+# ── P1-6: GET /subscription/plans ─────────────────────────────────────
+
+
+class TestSubscriptionPlans:
+    @staticmethod
+    def _import_plans_fn():
+        """Import from the real file to avoid sys.modules shadowing by integration fixtures."""
+        import importlib.util
+        _route_path = os.path.join(
+            os.path.dirname(os.path.abspath(__file__)),
+            "..", "..", "apps", "api", "app", "api", "routes", "subscription.py",
+        )
+        _spec = importlib.util.spec_from_file_location(
+            "_real_subscription_routes", os.path.abspath(_route_path)
+        )
+        _mod = importlib.util.module_from_spec(_spec)
+        # inject settings before exec
+        import os as _os
+        _os.environ.setdefault("JWT_SECRET_KEY", "test-secret")
+        _spec.loader.exec_module(_mod)
+        return _mod.list_membership_plans
+
+    def test_plans_endpoint_returns_three_tiers(self):
+        import os  # noqa: F401  (used by _import_plans_fn)
+        list_membership_plans = self._import_plans_fn()
+        resp = list_membership_plans(current_user=_make_cu())
+        plans = resp["plans"]
+        plan_ids = {p["plan_id"] for p in plans}
+        assert plan_ids == {"monthly", "quarterly", "yearly"}
+        for p in plans:
+            assert p["price_cents"] > 0
+            assert p["duration_days"] in (30, 90, 365)
+            assert 0 < p["points_discount"] <= 1.0
+            assert "max_resolution" in p["features"]
+
+    def test_longer_plans_cheaper_per_month(self):
+        import os  # noqa: F401
+        list_membership_plans = self._import_plans_fn()
+        resp = list_membership_plans(current_user=_make_cu())
+        plans = resp["plans"]
+        monthly = next(p for p in plans if p["plan_id"] == "monthly")
+        quarterly = next(p for p in plans if p["plan_id"] == "quarterly")
+        yearly = next(p for p in plans if p["plan_id"] == "yearly")
+        assert monthly["monthly_price_cents"] == 1990
+        assert quarterly["monthly_price_cents"] < monthly["monthly_price_cents"]
+        assert yearly["monthly_price_cents"] < quarterly["monthly_price_cents"]
+
+
+# ── P1-7: multiplier consistency ──────────────────────────────────────
+
+
+class TestMultiplierConsistency:
+    def test_free_user_ai_title_costs_2(self):
+        """ai_title base=1，免费用户 ceil(1*1.15)=2。"""
+        from packages.domain.points_rules import calculate_points_cost
+
+        assert calculate_points_cost("ai_title", is_member=False, quantity=1) == 2
+
+    def test_check_matches_direct_calculation(self):
+        """check 端点 required_points 与 calculate_points_cost 结果一致。"""
+        from app.api.routes.points import check_points
+        from app.schemas.points import PointsCheckRequest
+
+        from packages.domain.points_rules import calculate_points_cost
+
+        svc = MagicMock()
+        svc.check_daily_free_clip.return_value = False
+        svc.get_or_create_account.return_value = {"balance": 999}
+        db = MagicMock()
+        cu = _make_cu()
+
+        for scene in ["ai_voice", "ai_title", "ai_cover", "ai_rewrite"]:
+            body = PointsCheckRequest(scene_key=scene, quantity=1)
+            with patch("app.api.routes.points._get_service", return_value=svc):
+                resp = check_points(body=body, current_user=cu, db=db)
+            expected = calculate_points_cost(scene, is_member=False, quantity=1)
+            assert resp.required_points == expected, f"{scene}: got {resp.required_points}, expected {expected}"

tests/unit/test_script_title_integration.py

@@ -0,0 +1,334 @@
+"""#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 os
+import sys
+import uuid
+from datetime import UTC, datetime, timezone
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+from fastapi.testclient import TestClient
+
+# 确保 apps/api 在 sys.path 中（conftest 已加 root，但 apps/api 也需要）
+_APPS_API = str(Path(__file__).resolve().parents[2] / "apps" / "api")
+if _APPS_API not in sys.path:
+    sys.path.insert(0, _APPS_API)
+os.environ.setdefault("JWT_SECRET_KEY", "dev-secret-key-for-testing")
+
+from main import app  # noqa: E402
+
+# ── 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 ───────────────────────────────────────────
+
+
+def _make_mock_auth_user(user_id="user-001"):
+    """创建 mock 认证用户."""
+    return MagicMock(user=MagicMock(id=user_id))
+
+
+class TestScriptsRoutesNewFields:
+    """验证 scripts API 路由正确处理新字段 — 使用 dependency_overrides 绕过真实 DB/Auth."""
+
+    def setup_method(self):
+        from app.api.routes.scripts import _get_service, get_current_user
+
+        self._mock_svc = MagicMock()
+        self._mock_user = _make_mock_auth_user()
+
+        def _override_svc():
+            return self._mock_svc
+
+        def _override_user():
+            return self._mock_user
+
+        app.dependency_overrides[_get_service] = _override_svc
+        app.dependency_overrides[get_current_user] = _override_user
+        self.client = TestClient(app)
+
+    def teardown_method(self):
+        app.dependency_overrides.clear()
+
+    def test_create_script_passes_title_fields(self):
+        mock_script = _make_script(
+            title_text="测试标题",
+            title_category="片头",
+            title_config={"text": "测试标题", "font_size": 48},
+        )
+        self._mock_svc.create_script.return_value = mock_script
+
+        resp = self.client.post(
+            "/api/v1/scripts",
+            json={
+                "title": "新文案",
+                "content": "内容",
+                "title_text": "测试标题",
+                "title_category": "片头",
+                "title_config": {"text": "测试标题", "font_size": 48},
+            },
+        )
+
+        assert resp.status_code == 201, resp.text
+        call_kwargs = self._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}
+
+    def test_get_script_response_includes_title_fields(self):
+        mock_script = _make_script(
+            title_text="响应标题",
+            title_category="片尾",
+            title_config={"text": "响应标题", "position": "bottom"},
+        )
+        self._mock_svc.get_script.return_value = mock_script
+
+        resp = self.client.get("/api/v1/scripts/script-123")
+
+        assert resp.status_code == 200, resp.text
+        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 — 使用 dependency_overrides 绕过真实 DB/Auth."""
+
+    def setup_method(self):
+        from app.api.routes.titles import _get_title_repository, get_current_user
+        from app.dependencies import get_user_repository
+
+        self._mock_repo = MagicMock()
+        self._mock_user_repo = MagicMock()
+        self._mock_user = _make_mock_auth_user()
+
+        app.dependency_overrides[_get_title_repository] = lambda: self._mock_repo
+        app.dependency_overrides[get_user_repository] = lambda: self._mock_user_repo
+        app.dependency_overrides[get_current_user] = lambda: self._mock_user
+        self.client = TestClient(app)
+
+    def teardown_method(self):
+        app.dependency_overrides.clear()
+
+    def test_list_titles_has_warning_header(self):
+        # list_titles 调 use_case + repo, 注入真实用例但 mock 掉 repo 的 list/count
+        self._mock_repo.list_by_user.return_value = []
+        self._mock_repo.count_by_user.return_value = 0
+
+        resp = self.client.get("/api/v1/titles")
+
+        assert resp.status_code == 200, resp.text
+        headers_lower = {k.lower(): v for k, v in resp.headers.items()}
+        assert "warning" in headers_lower or "deprecation" in headers_lower
+        assert "1894" in resp.headers.get("Warning", "") or "1894" in resp.headers.get("warning", "")
+
+    def test_get_title_has_warning_header(self):
+        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),
+        )
+        self._mock_repo.get.return_value = mock_item
+
+        resp = self.client.get("/api/v1/titles/t1")
+        assert resp.status_code == 200, resp.text
+        warning_header = resp.headers.get("Warning", "") or resp.headers.get("warning", "")
+        assert "1894" in warning_header or "deprecated" in warning_header.lower()

tests/unit/test_scripts_ai.py

@@ -16,6 +16,8 @@ from unittest.mock import MagicMock, patch
 import pydantic
 import pytest
 
+import packages.middleware.points_gate as _pg_module
+
 sys.path.insert(0, "apps/api")
 
 
@@ -28,6 +30,18 @@ def _make_auth_user(user_id: str = "u1"):
     return auth
 
 
+@pytest.fixture(autouse=True)
+def _disable_points_gate(monkeypatch):
+    """默认关闭积分闸门，避免影响既有用例。"""
+    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: False)
+    yield
+
+
+@pytest.fixture
+def mock_db():
+    return MagicMock()
+
+
 def _mock_youtube_dl(
     extract_info_return=None,
     extract_info_side_effect=None,
@@ -82,7 +96,7 @@ class TestExtractFromDouyin:
 
         req = ExtractFromDouyinRequest(url="https://v.douyin.com/xxxxx/")
         auth = _make_auth_user()
-        result = extract_from_douyin(request=req, authenticated_user=auth)
+        result = extract_from_douyin(request=req, current_user=auth)
 
         assert result.text == "这是一段测试文案内容"
         assert result.duration_seconds == 120.5
@@ -112,7 +126,7 @@ class TestExtractFromDouyin:
         auth = _make_auth_user()
 
         with pytest.raises(HTTPException) as exc_info:
-            extract_from_douyin(request=req, authenticated_user=auth)
+            extract_from_douyin(request=req, current_user=auth)
         assert exc_info.value.status_code == 400
 
     @patch("tempfile.TemporaryDirectory")
@@ -136,7 +150,7 @@ class TestExtractFromDouyin:
         auth = _make_auth_user()
 
         with pytest.raises(HTTPException) as exc_info:
-            extract_from_douyin(request=req, authenticated_user=auth)
+            extract_from_douyin(request=req, current_user=auth)
         assert exc_info.value.status_code == 502
 
     @patch("app.api.routes.scripts_ai.transcribe_to_text")
@@ -169,7 +183,7 @@ class TestExtractFromDouyin:
         auth = _make_auth_user()
 
         with pytest.raises(HTTPException) as exc_info:
-            extract_from_douyin(request=req, authenticated_user=auth)
+            extract_from_douyin(request=req, current_user=auth)
         assert exc_info.value.status_code == 503
 
     @patch("app.api.routes.scripts_ai.transcribe_to_text")
@@ -202,7 +216,7 @@ class TestExtractFromDouyin:
         auth = _make_auth_user()
 
         with pytest.raises(HTTPException) as exc_info:
-            extract_from_douyin(request=req, authenticated_user=auth)
+            extract_from_douyin(request=req, current_user=auth)
         assert exc_info.value.status_code == 502
 
 
@@ -225,7 +239,7 @@ class TestAiRewrite:
 
         req = AiRewriteRequest(content="原始文案内容", style="口语化")
         auth = _make_auth_user()
-        result = ai_rewrite(request=req, authenticated_user=auth)
+        result = ai_rewrite(request=req, current_user=auth)
 
         assert result.original == "原始文案内容"
         assert result.rewritten == "改写后的文案内容，口语化风格"
@@ -242,7 +256,7 @@ class TestAiRewrite:
         auth = _make_auth_user()
 
         with pytest.raises(HTTPException) as exc_info:
-            ai_rewrite(request=req, authenticated_user=auth)
+            ai_rewrite(request=req, current_user=auth)
         assert exc_info.value.status_code == 400
 
     @patch("app.api.routes.scripts_ai.get_doubao_client")
@@ -261,7 +275,7 @@ class TestAiRewrite:
         auth = _make_auth_user()
 
         with pytest.raises(HTTPException) as exc_info:
-            ai_rewrite(request=req, authenticated_user=auth)
+            ai_rewrite(request=req, current_user=auth)
         assert exc_info.value.status_code == 502
 
     @patch("app.api.routes.scripts_ai.get_doubao_client")
@@ -279,7 +293,7 @@ class TestAiRewrite:
         auth = _make_auth_user()
 
         with pytest.raises(HTTPException) as exc_info:
-            ai_rewrite(request=req, authenticated_user=auth)
+            ai_rewrite(request=req, current_user=auth)
         assert exc_info.value.status_code == 502
 
 
@@ -301,7 +315,7 @@ class TestAiGenerateTitles:
 
         req = AiGenerateTitlesRequest(content="这是一段关于美食的文案", count=3)
         auth = _make_auth_user()
-        result = ai_generate_titles(request=req, authenticated_user=auth)
+        result = ai_generate_titles(request=req, current_user=auth)
 
         assert len(result.titles) == 3
         assert all(isinstance(t, str) for t in result.titles)
@@ -339,12 +353,12 @@ class TestAiGenerateTitles:
 
         # count=5
         req = AiGenerateTitlesRequest(content="测试内容", count=5)
-        result = ai_generate_titles(request=req, authenticated_user=auth)
+        result = ai_generate_titles(request=req, current_user=auth)
         assert len(result.titles) <= 5
 
         # count=1
         req = AiGenerateTitlesRequest(content="测试内容", count=1)
-        result = ai_generate_titles(request=req, authenticated_user=auth)
+        result = ai_generate_titles(request=req, current_user=auth)
         assert len(result.titles) >= 1
 
     def test_generate_titles_empty_content(self):
@@ -357,7 +371,7 @@ class TestAiGenerateTitles:
         auth = _make_auth_user()
 
         with pytest.raises(HTTPException) as exc_info:
-            ai_generate_titles(request=req, authenticated_user=auth)
+            ai_generate_titles(request=req, current_user=auth)
         assert exc_info.value.status_code == 400
 
     @patch("app.services.ai_service.get_doubao_client")
@@ -372,7 +386,7 @@ class TestAiGenerateTitles:
 
         req = AiGenerateTitlesRequest(content="测试文案内容")
         auth = _make_auth_user()
-        result = ai_generate_titles(request=req, authenticated_user=auth)
+        result = ai_generate_titles(request=req, current_user=auth)
 
         assert len(result.titles) == 3
 

tests/unit/test_scripts_ai_points.py

@@ -0,0 +1,76 @@
+"""scripts_ai 积分扣点单元测试 (#1895 P2 step 2.3)"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+from fastapi import HTTPException
+
+import packages.middleware.points_gate as _pg_module
+
+
+def _make_cu(user_id="u1", is_member=False, member_type=None):
+    cu = MagicMock()
+    cu.user.id = user_id
+    cu.user.is_member = is_member
+    cu.user.member_type = member_type
+    return cu
+
+
+@pytest.fixture(autouse=True)
+def _enable_gate(monkeypatch):
+    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: True)
+    yield
+
+
+class TestScriptsAiPointsGate:
+    """测试 scripts_ai 三个端点都挂了 @points_gate 并正确扣费。"""
+
+    @pytest.mark.parametrize(
+        "scene,endpoint_fn_name",
+        [
+            ("douyin_extract", "extract_from_douyin"),
+            ("ai_rewrite", "ai_rewrite"),
+            ("ai_title", "ai_generate_titles"),
+        ],
+    )
+    def test_insufficient_points_raises_402(self, scene, endpoint_fn_name):
+        """积分不足时抛 402。"""
+        from app.api.routes import scripts_ai
+        from app.schemas.scripts_ai import (
+            AiGenerateTitlesRequest,
+            AiRewriteRequest,
+            ExtractFromDouyinRequest,
+        )
+
+        fn = getattr(scripts_ai, endpoint_fn_name)
+        db = MagicMock()
+        cu = _make_cu()
+        if scene == "douyin_extract":
+            req = ExtractFromDouyinRequest(url="https://v.douyin.com/abc/")
+        elif scene == "ai_rewrite":
+            req = AiRewriteRequest(content="测试文案")
+        else:
+            req = AiGenerateTitlesRequest(content="测试文案", count=3)
+
+        with patch("packages.domain.points_service.PointsService") as MockSvc:
+            svc = MagicMock()
+            svc.deduct_points.return_value = {"success": False, "balance": 0}
+            MockSvc.return_value = svc
+            with pytest.raises(HTTPException) as ei:
+                fn(request=req, current_user=cu, db=db)
+            assert ei.value.status_code == 402
+
+    def test_disabled_passthrough_no_user_error(self, monkeypatch):
+        """关闭时不需要 user/db 也能被装饰器透传（验证 gate 关闭零副作用）。"""
+        from app.api.routes import scripts_ai
+        from app.schemas.scripts_ai import AiRewriteRequest
+
+        monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: False)
+        fn = scripts_ai.ai_rewrite
+        # 不带 db/current_user 也应透传（后续业务逻辑可能报错但不是 401/500 gate 错误）
+        with pytest.raises(Exception) as ei:
+            fn(request=AiRewriteRequest(content="x"), current_user=None, db=None)
+        # 不应是 gate 抛的 401/500
+        assert isinstance(ei.value, AttributeError) or ei.value.status_code not in (401, 500)

tests/unit/test_scripts_routes.py

@@ -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=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 == ""
+        assert r.title_config == {}
 
 
 class TestScriptListResponse:
@@ -141,6 +167,9 @@ class TestRouteHandlers:
         mock_script.content = "内容"
         mock_script.segments = [{"text": "段1", "duration": None}]
         mock_script.tags = []
+        mock_script.title_text = ""
+        mock_script.title_category = ""
+        mock_script.title_config = {}
         mock_script.created_at = datetime(2026, 9, 8, tzinfo=UTC)
         mock_script.updated_at = datetime(2026, 9, 8, tzinfo=UTC)
         svc.create_script.return_value = mock_script
@@ -163,6 +192,9 @@ class TestRouteHandlers:
         mock_script.content = ""
         mock_script.segments = []
         mock_script.tags = []
+        mock_script.title_text = ""
+        mock_script.title_category = ""
+        mock_script.title_config = {}
         mock_script.created_at = datetime(2026, 9, 8, tzinfo=UTC)
         mock_script.updated_at = datetime(2026, 9, 8, tzinfo=UTC)
         svc.list_scripts.return_value = ([mock_script], 1)
@@ -183,6 +215,9 @@ class TestRouteHandlers:
         mock_script.content = ""
         mock_script.segments = []
         mock_script.tags = []
+        mock_script.title_text = ""
+        mock_script.title_category = ""
+        mock_script.title_config = {}
         mock_script.created_at = datetime(2026, 9, 8, tzinfo=UTC)
         mock_script.updated_at = datetime(2026, 9, 8, tzinfo=UTC)
         svc.get_script.return_value = mock_script
@@ -215,6 +250,9 @@ class TestRouteHandlers:
         mock_script.content = "原内容"
         mock_script.segments = []
         mock_script.tags = []
+        mock_script.title_text = ""
+        mock_script.title_category = ""
+        mock_script.title_config = {}
         mock_script.created_at = datetime(2026, 9, 8, tzinfo=UTC)
         mock_script.updated_at = datetime(2026, 9, 8, tzinfo=UTC)
         svc.update_script.return_value = mock_script
@@ -260,3 +298,4 @@ class TestRouteHandlers:
         with pytest.raises(HTTPException) as exc_info:
             delete_script("bad", authenticated_user=auth, svc=svc)
         assert exc_info.value.status_code == 404
+

tests/unit/test_tts_voice_clone_points.py

@@ -0,0 +1,267 @@
+"""TTS + voice_clone 积分扣点单元测试 (#1895 P2 step 2.1)
+
+覆盖 synthesize / voice_clone preview 在积分开关下的扣点、余额不足、失败退费、会员折扣等分支。
+"""
+
+from __future__ import annotations
+
+import math
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+from fastapi import HTTPException
+
+import packages.middleware.points_gate as _pg_module
+
+
+@pytest.fixture(autouse=True)
+def _enable_gate(monkeypatch):
+    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: True)
+    yield
+
+
+def _make_cu(user_id="user-1", is_member=False, member_type=None):
+    cu = MagicMock()
+    cu.user.id = user_id
+    cu.user.is_member = is_member
+    cu.user.member_type = member_type
+    return cu
+
+
+def _make_request(text="你好世界", voice_id="v1", **kw):
+    r = MagicMock()
+    r.text = text
+    r.voice_id = voice_id
+    r.voice_clone_profile_id = None
+    r.speed = 1.0
+    r.emotion = ""
+    r.language = "zh-CN"
+    r.metadata_ = {}
+    r.voice_model = None
+    for k, v in kw.items():
+        setattr(r, k, v)
+    return r
+
+
+def _est_minutes(chars: int) -> float:
+    return max(1.0, math.ceil(chars / 240))
+
+
+class TestEstimateMinutes:
+    @pytest.mark.parametrize(
+        "chars,expected",
+        [(1, 1.0), (240, 1.0), (241, 2.0), (480, 2.0), (481, 3.0), (1000, 5.0)],
+    )
+    def test_estimate(self, chars, expected):
+        assert _est_minutes(chars) == expected
+
+
+class TestTtsSynthesizePointsDeduction:
+    def _setup(self, text="你好", deduct_success=True, balance=0, start_synth_raises=None, send_task_raises=None):
+        db = MagicMock()
+        cu = _make_cu()
+        repo = MagicMock()
+        import enum
+
+        class _S(enum.Enum):
+            processing = "processing"
+
+        job = SimpleNamespace(id="job-1", status=_S.processing, metadata={})
+        uc = MagicMock()
+        uc.execute.return_value = job
+        wf = MagicMock()
+        wf.start_synthesis.return_value = job
+        wf.process_synthesis_failure.return_value = job
+        if start_synth_raises:
+            wf.start_synthesis.side_effect = start_synth_raises
+        vc_repo = MagicMock()
+        vc_repo.get.return_value = None
+        svc = MagicMock()
+        svc.deduct_points.return_value = {"success": deduct_success, "balance": balance}
+        fake_settings = MagicMock(points_enabled=True)
+        return db, cu, repo, uc, wf, vc_repo, svc, fake_settings, job
+
+    def test_insufficient_raises_402(self):
+        db, cu, repo, uc, wf, vc_repo, svc, fs, _ = self._setup(text="你好" * 200, deduct_success=False, balance=0)
+        from app.api.routes.tts import synthesize
+
+        with (
+            patch("app.api.routes.tts.CreateTTSJobUseCase", return_value=uc),
+            patch("app.api.routes.tts.TTSWorkflowService", return_value=wf),
+            patch("app.api.routes.tts.PointsService", return_value=svc),
+            patch("app.api.routes.tts.settings", fs),
+        ):
+            with pytest.raises(HTTPException) as ei:
+                synthesize(
+                    request=_make_request(text="你好" * 200),
+                    authenticated_user=cu,
+                    db=db,
+                    repository=repo,
+                    cosyvoice_service=MagicMock(),
+                    voice_clone_repo=vc_repo,
+                )
+            assert ei.value.status_code == 402
+            assert ei.value.detail["code"] == "INSUFFICIENT_POINTS"
+
+    def test_success_deducts_points(self):
+        db, cu, repo, uc, wf, vc_repo, svc, fs, job = self._setup()
+        from app.api.routes.tts import synthesize
+
+        with (
+            patch("app.api.routes.tts.CreateTTSJobUseCase", return_value=uc),
+            patch("app.api.routes.tts.TTSWorkflowService", return_value=wf),
+            patch("app.api.routes.tts.PointsService", return_value=svc),
+            patch("app.api.routes.tts.celery_app.send_task") as _st,
+            patch("app.api.routes.tts.settings", fs),
+        ):
+            resp = synthesize(
+                request=_make_request(text="测试"),
+                authenticated_user=cu,
+                db=db,
+                repository=repo,
+                cosyvoice_service=MagicMock(),
+                voice_clone_repo=vc_repo,
+            )
+            svc.deduct_points.assert_called_once()
+            assert resp.job_id == job.id
+
+    def test_synthesis_failure_refunds(self):
+        db, cu, repo, uc, wf, vc_repo, svc, fs, _ = self._setup(start_synth_raises=RuntimeError("boom"))
+        from app.api.routes.tts import synthesize
+
+        with (
+            patch("app.api.routes.tts.CreateTTSJobUseCase", return_value=uc),
+            patch("app.api.routes.tts.TTSWorkflowService", return_value=wf),
+            patch("app.api.routes.tts.PointsService", return_value=svc),
+            patch("app.api.routes.tts.celery_app.send_task"),
+            patch("app.api.routes.tts.settings", fs),
+        ):
+            synthesize(
+                request=_make_request(text="测试"),
+                authenticated_user=cu,
+                db=db,
+                repository=repo,
+                cosyvoice_service=MagicMock(),
+                voice_clone_repo=vc_repo,
+            )
+            assert svc.refund_points.called
+
+    def test_celery_send_failure_refunds(self):
+        db, cu, repo, uc, wf, vc_repo, svc, fs, _ = self._setup(send_task_raises=RuntimeError("celery down"))
+        from app.api.routes.tts import synthesize
+
+        with (
+            patch("app.api.routes.tts.CreateTTSJobUseCase", return_value=uc),
+            patch("app.api.routes.tts.TTSWorkflowService", return_value=wf),
+            patch("app.api.routes.tts.PointsService", return_value=svc),
+            patch("app.api.routes.tts.celery_app.send_task", side_effect=RuntimeError("celery down")),
+            patch("app.api.routes.tts.settings", fs),
+        ):
+            synthesize(
+                request=_make_request(text="测试"),
+                authenticated_user=cu,
+                db=db,
+                repository=repo,
+                cosyvoice_service=MagicMock(),
+                voice_clone_repo=vc_repo,
+            )
+            assert svc.refund_points.called
+
+    def test_member_cheaper(self):
+        from packages.domain.points_rules import calculate_points_cost
+
+        cf = calculate_points_cost("ai_voice", is_member=False, duration_minutes=2)
+        cm = calculate_points_cost("ai_voice", is_member=True, member_type="monthly", duration_minutes=2)
+        assert cm < cf
+
+
+class TestVoiceClonePreviewPoints:
+    def _setup(self, text="你好", deduct_success=True, balance=0, synth_raises=None):
+        db = MagicMock()
+        cu = _make_cu()
+        repo = MagicMock()
+        profile = SimpleNamespace(is_ready=True, voice_id="vc-1", user_id=cu.user.id)
+        uc = MagicMock()
+        uc.execute.return_value = profile
+        cosy = MagicMock()
+        r = SimpleNamespace(audio_url="http://x/a.mp3", duration=1.2, file_size=1000)
+        cosy.synthesize_speech.return_value = r
+        if synth_raises:
+            cosy.synthesize_speech.side_effect = synth_raises
+        svc = MagicMock()
+        svc.deduct_points.return_value = {"success": deduct_success, "balance": balance}
+        fs = MagicMock(points_enabled=True)
+        return db, cu, repo, uc, cosy, svc, fs
+
+    def test_insufficient_raises_402(self):
+        db, cu, repo, uc, cosy, svc, fs = self._setup(deduct_success=False, balance=0)
+        from app.api.routes.voice_clones import get_voice_clone_preview
+
+        with (
+            patch("app.api.routes.voice_clones.GetVoiceCloneUseCase", return_value=uc),
+            patch("app.api.routes.voice_clones.PointsService", return_value=svc),
+            patch("app.api.routes.voice_clones.settings", fs),
+            patch("app.api.routes.voice_clones._clone_preview_cache", {}),
+        ):
+            with pytest.raises(HTTPException) as ei:
+                get_voice_clone_preview(
+                    clone_id="c1",
+                    text="你好",
+                    speed=1.0,
+                    emotion="",
+                    authenticated_user=cu,
+                    db=db,
+                    repository=repo,
+                    cosyvoice=cosy,
+                )
+            assert ei.value.status_code == 402
+
+    def test_synth_cosyvoice_error_refunds_and_raises_502(self):
+        from packages.application.cosyvoice_service import CosyVoiceError
+
+        db, cu, repo, uc, cosy, svc, fs = self._setup(synth_raises=CosyVoiceError("fail"))
+        from app.api.routes.voice_clones import get_voice_clone_preview
+
+        with (
+            patch("app.api.routes.voice_clones.GetVoiceCloneUseCase", return_value=uc),
+            patch("app.api.routes.voice_clones.PointsService", return_value=svc),
+            patch("app.api.routes.voice_clones.settings", fs),
+            patch("app.api.routes.voice_clones._clone_preview_cache", {}),
+        ):
+            with pytest.raises(HTTPException) as ei:
+                get_voice_clone_preview(
+                    clone_id="c1",
+                    text="你好",
+                    speed=1.0,
+                    emotion="",
+                    authenticated_user=cu,
+                    db=db,
+                    repository=repo,
+                    cosyvoice=cosy,
+                )
+            assert ei.value.status_code == 502
+            assert svc.refund_points.called
+
+    def test_success_returns_audio(self):
+        db, cu, repo, uc, cosy, svc, fs = self._setup()
+        from app.api.routes.voice_clones import get_voice_clone_preview
+
+        with (
+            patch("app.api.routes.voice_clones.GetVoiceCloneUseCase", return_value=uc),
+            patch("app.api.routes.voice_clones.PointsService", return_value=svc),
+            patch("app.api.routes.voice_clones.settings", fs),
+            patch("app.api.routes.voice_clones._clone_preview_cache", {}),
+        ):
+            resp = get_voice_clone_preview(
+                clone_id="c1",
+                text="你好",
+                speed=1.0,
+                emotion="",
+                authenticated_user=cu,
+                db=db,
+                repository=repo,
+                cosyvoice=cosy,
+            )
+            svc.deduct_points.assert_called_once()
+            assert resp.audio_url.startswith("http")