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

- 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

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

@@ -17,8 +17,6 @@ from datetime import UTC
 
 from app.auth import AuthenticatedUser, get_current_user
 from app.config import settings
-from packages.domain.points_rules import calculate_points_cost
-from packages.domain.points_service import PointsService
 from app.dependencies import (
     get_db_session,
     get_voice_clone_profile_repository,
@@ -34,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()

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/tts.py

@@ -41,8 +41,6 @@ from sqlalchemy.orm import Session
 from packages.adapters.sqlalchemy_impl.tts_job_repository import (
     SQLAlchemyTTSJobRepository,
 )
-from packages.domain.points_rules import calculate_points_cost
-from packages.domain.points_service import PointsService
 from packages.application.cosyvoice_service import CosyVoiceError, CosyVoiceService
 from packages.application.tts_job.streaming_service import TTSStreamingService
 from packages.application.tts_job.use_cases import (
@@ -55,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

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

@@ -25,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,
@@ -43,7 +44,6 @@ from packages.application.voice_clone.workflow import (
 )
 from packages.domain.points_rules import calculate_points_cost
 from packages.domain.points_service import PointsService
-from sqlalchemy.orm import Session
 
 # remove duplicate
 _DUMMY_DELETED = ()

packages/middleware/points_gate.py

@@ -9,7 +9,6 @@ import asyncio
 import functools
 import inspect
 import logging
-import types
 from collections.abc import Callable
 from typing import Any
 
@@ -27,65 +26,10 @@ def _points_gate_enabled() -> bool:
         from app.config import settings as _settings
 
         return bool(_settings.points_enabled)
-    except Exception:  # pragma: no cover
+    except Exception:  # pragma: no cover - 配置加载失败默认关闭
         return False
 
 
-def _make_wrapper(
-    func: Callable,
-    is_async: bool,
-    scene_key: str,
-    per_unit,
-    unit_field,
-    quantity_field,
-) -> Callable:
-    """在被装饰函数所在模块的 globals 下创建 wrapper。
-
-    关键点：Python 闭包默认在「定义闭包的模块」globals 下查找自由变量。若直接
-    在 points_gate 模块内定义 wrapper，wrapper.__globals__ 将指向本模块，导致
-    FastAPI/Pydantic 在解析函数类型注解（ForwardRef）时找不到路由模块中
-    导入的 Pydantic Model，出现 PydanticUndefinedAnnotation。
-
-    这里通过 types.FunctionType 把 wrapper 的 code 绑定到「被装饰函数所在模块
-    的 globals（补充装饰器内部符号）」，使 wrapper 的 ForwardRef 解析行为
-    与原路由函数一致。
-    """
-    if is_async:
-        async def wrapper(*args: Any, **kwargs: Any) -> Any:
-            if not _points_gate_enabled():
-                return await func(*args, **kwargs)
-            return await _execute_with_gate(
-                func, args, kwargs, scene_key, per_unit, unit_field, quantity_field, is_async=True
-            )
-    else:
-        def wrapper(*args: Any, **kwargs: Any) -> Any:
-            if not _points_gate_enabled():
-                return func(*args, **_filter_kwargs(func, kwargs))
-            return _execute_with_gate(
-                func, args, kwargs, scene_key, per_unit, unit_field, quantity_field, is_async=False
-            )
-
-    # 把闭包所需的内部符号注入到被装饰模块的 globals，避免闭包找不到名字
-    merged_globals: dict = dict(func.__globals__)
-    for _k, _v in (
-        ("_points_gate_enabled", _points_gate_enabled),
-        ("_execute_with_gate", _execute_with_gate),
-        ("_run_async", _run_async),
-        ("_filter_kwargs", _filter_kwargs),
-    ):
-        merged_globals.setdefault(_k, _v)
-
-    new_wrapper = types.FunctionType(
-        wrapper.__code__,
-        merged_globals,
-        wrapper.__name__,
-        wrapper.__defaults__,
-        wrapper.__closure__,
-    )
-    new_wrapper = functools.wraps(func)(new_wrapper)
-    return new_wrapper
-
-
 def points_gate(
     scene_key: str,
     per_unit: int | None = None,
@@ -96,17 +40,51 @@ def points_gate(
 
     当 POINTS_ENABLED=false（默认）时，装饰器完全透传原函数，零副作用。
     开启后才会执行扣费逻辑：业务异常自动退费，积分不足返回 402。
+
+    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)
-        return _make_wrapper(func, is_async, scene_key, per_unit, unit_field, quantity_field)
+
+        @functools.wraps(func)
+        async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
+            if not _points_gate_enabled():
+                return await func(*args, **kwargs)
+            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:
+            if not _points_gate_enabled():
+                return func(*args, **_filter_kwargs(func, kwargs))
+            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
 
     return decorator
 
 
 def _filter_kwargs(func: Callable, kwargs: dict) -> dict:
     """过滤掉目标函数签名不接受的 kwargs（避免 TypeError）。"""
+    import inspect
+
     try:
         sig = inspect.signature(func)
         params = sig.parameters
@@ -143,6 +121,7 @@ def _execute_with_gate(
     # 提取 current_user（兼容 authenticated_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
@@ -195,6 +174,7 @@ def _execute_with_gate(
             member_type=member_type,
         )
 
+    # 零消耗场景（如免费的声音克隆训练）直接放行
     if total_points == 0:
         kwargs["_points_deducted"] = 0
         if is_async:

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)