fix(points): make @points_gate wrapper globals + sys.modules proxy for Py3.12/PEP563 compat

- rebind wrapper code object to decorated route's __globals__ via types.FunctionType
  so PEP563 ForwardRefs (ExtractFromDouyinRequest etc.) resolve on Py3.12
- use thin sys.modules proxies for internal helpers so tests can monkeypatch
  _points_gate_enabled / _execute_with_gate and see the change at call time
- preserve backward-compat aliases (_filter_kwargs, _execute_with_gate, _run_async)
  for existing unit tests

.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

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/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,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
+            async def wrapper(*args: Any, **kwargs: Any) -> Any:
-        return sync_wrapper
+                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") or merged.get("authenticated_user")
-    current_user = merged.get("current_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 _run_async_impl(func, args, _filter_kwargs_impl(func, kwargs))
-                return func(*args, **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 _run_async_impl(func, args, _filter_kwargs_impl(func, kwargs))
-        return func(*args, **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 _run_async_impl(func, args, _filter_kwargs_impl(func, kwargs))
-        return func(*args, **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):
+def _run_async_impl(func: Callable, args: tuple, kwargs: dict):
-    """在 async wrapper 中 await 原始 async 函数。"""
+    return func(*args, **_filter_kwargs_impl(func, kwargs))
-    return func(*args, **kwargs)
+
+
+# 兼容历史测试文件直接 import 的别名
+_filter_kwargs = _filter_kwargs_impl
+_execute_with_gate = _execute_with_gate_impl
+_run_async = _run_async_impl

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_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