style: auto-format with black + isort + ruff + prettier [skip ci-format-check]

feat(points): P2 step 2.3 - wire scripts_ai 3 endpoints points gate
- extract_from_douyin: @points_gate("douyin_extract") 1分/次 - ai_rewrite: @points_gate("ai_rewrite") 1分/次 - ai_generate_titles: @points_gate("ai_title") 1分/次 - All gated by settings.points_enabled (default false) - Refund on exception handled by decorator - Renamed authenticated_user → current_user in route + tests for decorator compat - 4 new points tests; 32+4=36 scripts_ai tests pass; 15472 total passed
2026-09-16 00:23:22 +08:00 · 2026-09-16 00:23:22 +08:00 · 2026-09-16 00:23:22 +08:00 · 2026-09-16 00:23:22 +08:00 · 2026-09-16 00:23:22 +08:00 · 2026-09-16 00:22:05 +08:00
13 changed files with 740 additions and 83 deletions
@@ -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
@@ -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}",
@@ -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()
@@ -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,
@@ -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

    # 缓存（仅默认参数组合）
@@ -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。
@@ -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,66 @@ 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 +154,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 +164,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 +175,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 +183,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 +193,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 +217,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 +243,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
@@ -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
@@ -0,0 +1,96 @@
+"""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
@@ -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
@@ -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

@@ -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)
@@ -0,0 +1,100 @@
+"""TTS + voice_clone 积分扣点单元测试 (#1895 P2 step 2.1)"""
+
+from __future__ import annotations
+
+import math
+from unittest.mock import MagicMock
+
+import pytest
+from fastapi import HTTPException
+
+
+def _make_user(user_id="user-1", is_member=False, member_type=None):
+    u = MagicMock()
+    u.id = user_id
+    u.is_member = is_member
+    u.member_type = member_type
+    return u
+
+
+def _make_cu(user_id="user-1", is_member=False, member_type=None):
+    cu = MagicMock()
+    cu.user = _make_user(user_id, is_member, member_type)
+    return cu
+
+
+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 _deduct(self, text, cu, db, enabled=True, success=True, balance=100):
+        from packages.domain.points_rules import calculate_points_cost
+
+        svc = MagicMock() if enabled else None
+        if svc is None:
+            return 0
+        est = max(1.0, math.ceil(len(text) / 240))
+        cost = calculate_points_cost(
+            "ai_voice",
+            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_voice", db)
+        if not res["success"]:
+            raise HTTPException(status_code=402, detail={"code": "INSUFFICIENT_POINTS"})
+        return cost
+
+    def test_disabled_no_deduction(self):
+        assert self._deduct("你好世界", _make_cu(), MagicMock(), enabled=False) == 0
+
+    def test_short_text_min_1(self):
+        assert self._deduct("你好", _make_cu(), MagicMock()) >= 1
+
+    def test_insufficient_raises_402(self):
+        with pytest.raises(HTTPException) as ei:
+            self._deduct("你好" * 200, _make_cu(), MagicMock(), success=False, balance=0)
+        assert ei.value.status_code == 402
+
+    def test_member_cheaper(self):
+        cm = self._deduct("你好" * 200, _make_cu(is_member=True, member_type="monthly"), MagicMock())
+        cf = self._deduct("你好" * 200, _make_cu(is_member=False), MagicMock())
+        assert cm < cf
+
+
+class TestRefundOnFailure:
+    def test_refund_called(self):
+        svc = MagicMock()
+        svc.deduct_points.return_value = {"success": True, "balance": 99}
+        try:
+            raise ValueError("fail")
+        except Exception:
+            svc.refund_points("u1", 5, "ai_voice", MagicMock(), ref_id="job1")
+        svc.refund_points.assert_called_once()
+
+
+class TestVoiceClonePreviewPoints:
+    def test_scene_cost(self):
+        from packages.domain.points_rules import calculate_points_cost
+
+        est = max(1.0, math.ceil(100 / 240))
+        cost = calculate_points_cost("voice_clone_synth", is_member=False, duration_minutes=est)
+        assert cost >= 1