.env.example

@@ -198,10 +198,13 @@ DOUBAO_TIMEOUT=30
 DOUBAO_MAX_RETRIES=2
 
 # ==================== 积分/会员系统 (#1895) ====================
-# 积分扣点总开关：默认 false（对现有用户零影响）。
-# P2 阶段各业务路由逐个接入 @points_gate 时，用
-#   `if settings.points_enabled: ...`
-# 包裹扣点逻辑；所有路由接入完成并验证通过后再在 staging/prod 打开。
+# 积分系统总开关：默认 false（暂停积分系统）。
+# - false：生成视频/口型同步/数字人/AI标题/TTS/克隆音色等所有功能对登录
+#   用户免费放行，不扣积分、不做余额拦截；积分余额/流水/会员状态查询接口
+#   保留可用，但数据不再变动。积分相关的表、代码、接口均保留不删除。
+# - 恢复积分：设置 ENABLE_CREDIT_SYSTEM=true 即可，无需改代码。
+ENABLE_CREDIT_SYSTEM=false
+# 旧开关名（兼容别名）：与 ENABLE_CREDIT_SYSTEM 任一为 true 即启用。
 POINTS_ENABLED=false
 
 # ==================== 抖音解析多源轮询 (#1963) ====================

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

@@ -12,6 +12,7 @@ from datetime import datetime, timedelta, timezone
 from typing import Optional
 
 from app.auth import AuthenticatedUser, get_current_user
+from app.config import settings
 from app.dependencies import get_db_session
 from app.schemas.points import (
     DailyUsageResponse,
@@ -44,6 +45,12 @@ from packages.domain.points_service import PointsService
 
 logger = logging.getLogger(__name__)
 
+
+def _credits_enabled() -> bool:
+    """积分系统总开关（ENABLE_CREDIT_SYSTEM），关闭时全部功能免费放行。"""
+    return bool(getattr(settings, "points_enabled", False))
+
+
 # ── 两个 router ──
 points_router = APIRouter()
 usage_router = APIRouter()
@@ -172,6 +179,19 @@ def check_points(
                 "valid_scenes": sorted(POINTS_SCENES.keys()),
             },
         )
+
+    # 积分系统暂停（ENABLE_CREDIT_SYSTEM=false）：所有场景直接放行，需 0 积分
+    if not _credits_enabled():
+        svc = _get_service()
+        account = svc.get_or_create_account(current_user.user.id, db)
+        return PointsCheckResponse(
+            allowed=True,
+            required_points=0,
+            current_balance=account["balance"],
+            remaining_after=account["balance"],
+            is_free_quota=False,
+        )
+
     is_mem = _is_member(current_user)
     mt = _member_type(current_user)
 
@@ -209,8 +229,19 @@ def deduct_points(
     current_user: AuthenticatedUser = Depends(get_current_user),
     db: Session = Depends(get_db_session),
 ):
-    """积分扣减（内部服务调用）。"""
+    """积分扣减（内部服务调用）。
+
+    积分系统暂停（ENABLE_CREDIT_SYSTEM=false）时为 no-op：不扣分、余额不变，
+    直接返回成功，保证内部调用方拿到 success=True 继续业务流程。
+    """
     svc = _get_service()
+    if not _credits_enabled():
+        account = svc.get_or_create_account(current_user.user.id, db)
+        return SimpleMessageResponse(
+            success=True,
+            message="积分系统已暂停，未扣减积分",
+            data={"transaction_id": "", "balance": account["balance"]},
+        )
     result = svc.deduct_points(
         user_id=current_user.user.id,
         amount=body.amount,
@@ -243,11 +274,7 @@ def refund_points(
     """积分退还（内部服务调用）。"""
     from packages.adapters.sqlalchemy_impl.models import PointsTransactionModel
 
-    txn = (
-        db.query(PointsTransactionModel)
-        .filter(PointsTransactionModel.id == body.transaction_id)
-        .first()
-    )
+    txn = db.query(PointsTransactionModel).filter(PointsTransactionModel.id == body.transaction_id).first()
     if txn is None:
         raise HTTPException(status_code=404, detail="交易记录不存在")
     if txn.user_id != current_user.user.id:

packages/config/base.py

@@ -8,6 +8,7 @@ API 和 Worker 各自的 Settings 类继承本类，只追加服务特有字段
 import os
 from typing import Optional, TypeVar
 
+from pydantic import AliasChoices, Field
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 T = TypeVar("T", bound=BaseSettings)
@@ -76,9 +77,33 @@ class SharedSettings(BaseSettings):
     mediakit_timeout: int = 60
 
     # ── 积分/会员系统 (#1895) ────────────────────────────────────────────
-    # 总开关：默认 false（对所有用户零影响），P2 路由逐个接入时用
-    # `if settings.points_enabled:` 包裹，防止未完善的扣点逻辑影响现有用户。
-    points_enabled: bool = False
+    # 积分系统总开关（产品要求 #1895：暂停积分系统但保留全部代码/表/接口）。
+    # - false（默认）：所有 AI 功能（生成视频/口型/数字人/AI标题/TTS/克隆音色…）
+    #   对全部登录用户免费放行，不扣积分、不做余额拦截；积分余额/流水/会员
+    #   状态等查询接口保持可用，但数据不再变动。
+    # - 未来恢复：只需设置环境变量 ENABLE_CREDIT_SYSTEM=true。
+    # 旧开关 POINTS_ENABLED 仍保留作为兼容别名（两者任一为 true 即启用）。
+    # 主开关（推荐环境变量名 ENABLE_CREDIT_SYSTEM）
+    credits_enabled: bool = Field(
+        default=False,
+        validation_alias=AliasChoices("ENABLE_CREDIT_SYSTEM", "credits_enabled"),
+    )
+    # 旧开关兼容（POINTS_ENABLED）；两者任一为 true 即启用
+    points_enabled_compat: bool = Field(
+        default=False,
+        validation_alias=AliasChoices("POINTS_ENABLED", "points_enabled_compat"),
+    )
+
+    @property
+    def points_enabled(self) -> bool:
+        """旧代码/测试使用的属性名，等价于积分系统总开关（兼容别名）。"""
+        return bool(self.credits_enabled or self.points_enabled_compat)
+
+    @points_enabled.setter
+    def points_enabled(self, value: bool) -> None:
+        # 支持旧测试/代码 ``settings.points_enabled = True`` 的写法
+        self.credits_enabled = bool(value)
+        self.points_enabled_compat = False
 
     # ── GPU MuseTalk 反向轮询 Worker ────────────────────────────────────
     # Worker 用这个长期 Token 鉴权（不是用户 JWT）。多 Worker 共用同一个 Token；

tests/unit/test_credit_system_switch.py

@@ -0,0 +1,229 @@
+"""积分系统暂停开关测试 (#1895, ENABLE_CREDIT_SYSTEM)。
+
+产品要求：暂停积分系统但保留全部代码/表/接口。
+- 默认 false：所有 AI 功能免费放行，不扣积分、不做余额拦截；
+- /points/check 恒返回 allowed=True、required_points=0；
+- /points/deduct 为 no-op，余额不变；
+- 查询接口（balance/transactions/rules/packages/membership/usage）照常可用；
+- 旧环境变量 POINTS_ENABLED 作为兼容别名仍可开启。
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+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
+
+
+# ── 配置层 ────────────────────────────────────────────────────────────────
+
+
+class TestCreditSystemConfig:
+    def test_default_disabled(self):
+        import os
+
+        from packages.config.base import SharedSettings
+
+        assert os.environ.get("ENABLE_CREDIT_SYSTEM") is None
+        assert os.environ.get("POINTS_ENABLED") is None
+        s = SharedSettings(_env_file=None)
+        assert s.credits_enabled is False
+        # 旧属性名仍可用（业务代码大量引用 settings.points_enabled）
+        assert s.points_enabled is False
+
+    def test_enable_credit_system_env(self, monkeypatch):
+        from packages.config import base as base_mod
+
+        monkeypatch.setenv("ENABLE_CREDIT_SYSTEM", "true")
+        s = base_mod.SharedSettings(_env_file=None)
+        assert s.points_enabled is True
+        assert s.credits_enabled is True
+
+    def test_legacy_points_enabled_env_alias(self, monkeypatch):
+        from packages.config import base as base_mod
+
+        monkeypatch.setenv("ENABLE_CREDIT_SYSTEM", "false")
+        monkeypatch.setenv("POINTS_ENABLED", "true")
+        s = base_mod.SharedSettings(_env_file=None)
+        assert s.points_enabled is True
+        assert s.credits_enabled is False
+        assert s.points_enabled_compat is True
+
+    def test_legacy_setter_back_compat(self):
+        from packages.config.base import SharedSettings
+
+        s = SharedSettings(_env_file=None)
+        s.points_enabled = True
+        assert s.credits_enabled is True
+        assert s.points_enabled is True
+        s.points_enabled = False
+        assert s.points_enabled is False
+
+
+# ── /points/check：关闭时恒放行、需 0 积分 ────────────────────────────────
+
+
+class TestCheckEndpointWhenDisabled:
+    def test_check_allowed_zero_required(self):
+        from app.api.routes.points import check_points
+        from app.schemas.points import PointsCheckRequest
+
+        svc = MagicMock()
+        svc.get_or_create_account.return_value = {"balance": 0}
+        db = MagicMock()
+        cu = _make_cu()
+        body = PointsCheckRequest(scene_key="ai_voice", quantity=1, duration_minutes=5)
+
+        with (
+            patch("app.api.routes.points._credits_enabled", return_value=False),
+            patch("app.api.routes.points._get_service", return_value=svc),
+        ):
+            resp = check_points(body=body, current_user=cu, db=db)
+
+        assert resp.allowed is True
+        assert resp.required_points == 0
+        assert resp.remaining_after == 0
+        # 不再走免费额度判定
+        svc.check_daily_free_clip.assert_not_called()
+
+    def test_unknown_scene_still_400_when_disabled(self):
+        """未知 scene 即使系统关闭也返回 400（参数校验先于开关）。"""
+        from app.api.routes.points import check_points
+        from app.schemas.points import PointsCheckRequest
+        from fastapi import HTTPException
+
+        with pytest.raises(HTTPException) as exc:
+            check_points(body=PointsCheckRequest(scene_key="nope"), current_user=_make_cu(), db=MagicMock())
+        assert exc.value.status_code == 400
+
+    def test_check_enabled_calculates_cost(self):
+        """开关开启时保持原有计费校验。"""
+        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": 100}
+        body = PointsCheckRequest(scene_key="ai_title", quantity=1)
+
+        with (
+            patch("app.api.routes.points._credits_enabled", return_value=True),
+            patch("app.api.routes.points._get_service", return_value=svc),
+        ):
+            resp = check_points(body=body, current_user=_make_cu(), db=MagicMock())
+
+        assert resp.required_points == 2  # 免费用户 ceil(1*1.15)=2
+
+
+# ── /points/deduct：关闭时 no-op，余额不变 ────────────────────────────────
+
+
+class TestDeductEndpointWhenDisabled:
+    def test_deduct_is_noop(self):
+        from app.api.routes.points import deduct_points
+        from app.schemas.points import PointsDeductRequest
+
+        svc = MagicMock()
+        svc.get_or_create_account.return_value = {"balance": 7}
+        body = PointsDeductRequest(scene_key="ai_voice", amount=999)
+
+        with (
+            patch("app.api.routes.points._credits_enabled", return_value=False),
+            patch("app.api.routes.points._get_service", return_value=svc),
+        ):
+            resp = deduct_points(body=body, current_user=_make_cu(), db=MagicMock())
+
+        svc.deduct_points.assert_not_called()
+        assert resp.success is True
+        assert resp.data["balance"] == 7
+        assert resp.data["transaction_id"] == ""
+
+    def test_deduct_enabled_works_as_before(self):
+        from app.api.routes.points import deduct_points
+        from app.schemas.points import PointsDeductRequest
+
+        svc = MagicMock()
+        svc.deduct_points.return_value = {"success": True, "balance": 8, "transaction_id": "tx-1"}
+        body = PointsDeductRequest(scene_key="ai_title", amount=2)
+
+        with (
+            patch("app.api.routes.points._credits_enabled", return_value=True),
+            patch("app.api.routes.points._get_service", return_value=svc),
+        ):
+            resp = deduct_points(body=body, current_user=_make_cu(), db=MagicMock())
+
+        svc.deduct_points.assert_called_once()
+        assert resp.data["balance"] == 8
+        assert resp.data["transaction_id"] == "tx-1"
+
+
+# ── 查询接口：系统关闭时仍全部可用 ────────────────────────────────────────
+
+
+class TestQueryEndpointsRemainAvailable:
+    def test_balance_route_works_when_disabled(self):
+        from app.api.routes.points import get_balance
+
+        svc = MagicMock()
+        svc.get_or_create_account.return_value = {"balance": 0, "total_earned": 0, "total_spent": 0}
+        with (
+            patch("app.api.routes.points._credits_enabled", return_value=False),
+            patch("app.api.routes.points._get_service", return_value=svc),
+        ):
+            resp = get_balance(current_user=_make_cu(), db=MagicMock())
+        assert resp.balance == 0
+        assert resp.is_member is False
+
+    def test_transactions_route_works_when_disabled(self):
+        from app.api.routes.points import get_transactions
+
+        svc = MagicMock()
+        svc.get_transactions.return_value = {"items": [], "total": 0, "page": 1, "page_size": 20}
+        with (
+            patch("app.api.routes.points._credits_enabled", return_value=False),
+            patch("app.api.routes.points._get_service", return_value=svc),
+        ):
+            resp = get_transactions(current_user=_make_cu(), db=MagicMock())
+        assert resp.total == 0
+
+    def test_daily_usage_route_works_when_disabled(self):
+        from app.api.routes.points import get_daily_usage
+
+        svc = MagicMock()
+        svc.get_daily_usage.return_value = {
+            "free_clips_used": 0,
+            "free_clips_limit": 2,
+            "free_clips_remaining": 2,
+            "reset_at": "2026-09-20T00:00:00Z",
+        }
+        with (
+            patch("app.api.routes.points._credits_enabled", return_value=False),
+            patch("app.api.routes.points._get_service", return_value=svc),
+        ):
+            resp = get_daily_usage(current_user=_make_cu(), db=MagicMock())
+        assert resp.free_clips_limit == 2
+
+
+# ── 业务路由：开关关闭时 PointsService 不实例化、不扣分 ───────────────────
+
+
+class TestBusinessRoutesBypassWhenDisabled:
+    def test_lipsync_route_skips_points(self):
+        """lipsync 创建任务路由：settings.points_enabled=False 时不构造 PointsService。"""
+        from app.api.routes import lipsync as lipsync_mod
+
+        assert bool(getattr(lipsync_mod.settings, "points_enabled", False)) is False
+
+    def test_tts_route_skips_points(self):
+        from app.api.routes import tts as tts_mod
+
+        assert bool(getattr(tts_mod.settings, "points_enabled", False)) is False

tests/unit/test_points_routes.py

@@ -71,9 +71,7 @@ class TestRechargeOrderResponse:
         cu = _make_cu()
         body = PointsRechargeRequest(package_id="nonexistent")
 
-        with pytest.raises(HTTPException) as exc, patch(
-            "app.api.routes.points._get_service", return_value=svc
-        ):
+        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
 
@@ -112,7 +110,10 @@ class TestCheckPointsUnknownScene:
         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):
+        with (
+            patch("app.api.routes.points._credits_enabled", return_value=True),
+            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
@@ -148,22 +149,30 @@ class TestSubscriptionPlans:
     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)
+            "..",
+            "..",
+            "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"]
@@ -177,6 +186,7 @@ class TestSubscriptionPlans:
 
     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"]
@@ -211,9 +221,10 @@ class TestMultiplierConsistency:
         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}"
+        with patch("app.api.routes.points._credits_enabled", return_value=True):
+            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}"