feat(points): P2 step 1 - POINTS_ENABLED safety gate (default false) #1936

Merged
auto-approve-bot merged 5 commits from feat/points-gate-config into develop 2026-09-16 00:26:29 +08:00
5 changed files with 172 additions and 39 deletions
+7
View File
@@ -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
+5
View File
@@ -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。
+97 -39
View File
@@ -9,6 +9,7 @@ import asyncio
import functools
import inspect
import logging
import types
from collections.abc import Callable
from typing import Any
@@ -17,6 +18,77 @@ from fastapi import HTTPException
logger = logging.getLogger(__name__)
def _points_gate_enabled() -> bool:
"""读取配置开关:总开关关闭时装饰器完全放行(零开销/零副作用)。
放在模块顶层便于测试时 monkeypatch。
"""
try:
from app.config import settings as _settings
return bool(_settings.points_enabled)
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,
@@ -25,42 +97,30 @@ def points_gate(
) -> 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)):
...
当 POINTS_ENABLED=false(默认)时,装饰器完全透传原函数,零副作用。
开启后才会执行扣费逻辑:业务异常自动退费,积分不足返回 402。
"""
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
return _make_wrapper(func, is_async, scene_key, per_unit, unit_field, quantity_field)
return decorator
def _filter_kwargs(func: Callable, kwargs: dict) -> dict:
"""过滤掉目标函数签名不接受的 kwargs(避免 TypeError)。"""
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 _extract_kwargs(func: Callable, args: tuple, kwargs: dict) -> dict:
"""将位置参数映射到函数签名中的参数名,便于统一按 kwargs 提取。"""
sig = inspect.signature(func)
@@ -83,10 +143,9 @@ def _execute_with_gate(
"""积分扣费核心逻辑。"""
merged = _extract_kwargs(func, args, kwargs)
# 提取 current_user
current_user = merged.get("current_user")
# 提取 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
@@ -114,8 +173,8 @@ 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(func, args, _filter_kwargs(func, kwargs))
return func(*args, **_filter_kwargs(func, kwargs))
# ── 计算积分消耗 ──
if per_unit is not None:
@@ -139,12 +198,11 @@ 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(func, args, _filter_kwargs(func, kwargs))
return func(*args, **_filter_kwargs(func, kwargs))
# ── 扣减积分 ──
from packages.domain.points_service import PointsService
@@ -170,8 +228,8 @@ def _execute_with_gate(
# ── 执行业务函数,失败则退还积分 ──
try:
if is_async:
return _run_async(func, args, kwargs)
return func(*args, **kwargs)
return _run_async(func, args, _filter_kwargs(func, kwargs))
return func(*args, **_filter_kwargs(func, kwargs))
except Exception:
svc.refund_points(user.id, total_points, scene_key, db, ref_id=str(job_id))
raise
@@ -179,4 +237,4 @@ def _execute_with_gate(
def _run_async(func: Callable, args: tuple, kwargs: dict):
"""在 async wrapper 中 await 原始 async 函数。"""
return func(*args, **kwargs)
return func(*args, **_filter_kwargs(func, kwargs))
+55
View File
@@ -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
+8
View File
@@ -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