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

@@ -145,19 +145,22 @@ def get_rules(
 def get_packages(
     current_user: AuthenticatedUser = Depends(get_current_user),
 ):
-    """查询可购买的积分包列表。"""
-    packages = []
-    for code, pkg in POINTS_PACKAGES.items():
-        unit_price = f"¥{pkg['price_cents'] / 100 / pkg['points']:.3f}/积分"
-        packages.append(
-            PointsPackageItem(
-                code=code,
-                name=pkg["name"],
-                points=pkg["points"],
-                price_cents=pkg["price_cents"],
-                unit_price=unit_price,
-            )
+    """查询可购买的积分包列表（读管理后台 credit_packages 表真实数据）。
+
+    仅返回 is_active=true；后台改价/启停后最多 30 秒生效。
+    """
+    from packages.application.catalog.admin_catalog import get_points_packages
+
+    packages = [
+        PointsPackageItem(
+            code=row["code"],
+            name=row["name"],
+            points=row["points"],
+            price_cents=row["price_cents"],
+            unit_price=row["unit_price"],
         )
+        for row in get_points_packages()
+    ]
     mt = _member_type(current_user)
     discount = MEMBER_DISCOUNT.get(mt) if mt else None
     return PointsPackagesResponse(packages=packages, user_discount=discount)

apps/api/app/api/routes/subscription.py

@@ -86,33 +86,13 @@ async def get_current_subscription(
 def list_membership_plans(
     current_user: AuthenticatedUser = Depends(get_current_user),
 ) -> dict[str, list[dict[str, Any]]]:
-    """查询所有会员档位（供前端会员购买页展示）。
+    """查询可购买的会员套餐（读管理后台 plans 表真实数据）。
 
-    返回 points 积分体系下的会员档位（月卡/季卡/年卡），含价格、时长、积分折扣等信息。
+    仅返回 is_enabled=true 的套餐；后台启停/改价后最多 30 秒生效。
     """
-    from packages.domain.points_rules import MEMBER_DISCOUNT, MEMBERSHIP_PRICES
+    from packages.application.catalog.admin_catalog import get_membership_plans
 
-    plans: list[dict[str, Any]] = []
-    for plan_id, info in MEMBERSHIP_PRICES.items():
-        days = info["duration_days"]
-        monthly_cents = round(info["price_cents"] * 30 / days)
-        features: dict[str, Any] = {"max_resolution": "1080p"}
-        if plan_id == MembershipType.MONTHLY:
-            features.update({"free_clips_daily": 2})
-        elif plan_id == MembershipType.QUARTERLY:
-            features.update({"free_clips_daily": 5})
-        elif plan_id == MembershipType.YEARLY:
-            features.update({"free_clips_daily": "unlimited"})
-        plans.append({
-            "plan_id": plan_id,
-            "name": info["name"],
-            "price_cents": info["price_cents"],
-            "monthly_price_cents": monthly_cents,
-            "duration_days": days,
-            "points_discount": MEMBER_DISCOUNT.get(plan_id, 1.0),
-            "features": features,
-        })
-    return {"plans": plans}
+    return {"plans": get_membership_plans()}
 
 
 @router.get("/billing-records", response_model=list[BillingRecord])

packages/application/catalog/__init__.py

@@ -0,0 +1 @@
+"""应用层：对外展示目录（套餐/积分包）。"""

packages/application/catalog/admin_catalog.py

@@ -0,0 +1,152 @@
+"""读取管理后台配置的会员套餐 / 积分充值包（共享库真实数据）。
+
+替代旧的硬编码 MEMBERSHIP_PRICES / POINTS_PACKAGES。
+短 TTL 缓存（30 秒），后台改价/启停后用户端最多 30 秒可见。
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+from typing import Any
+
+_CACHE_TTL = 30.0
+_lock = threading.Lock()
+_cache: dict[str, tuple[float, Any]] = {}
+
+_QUOTA_LABELS = {
+    "4k": "4K 超清分辨率",
+    "batch_render": "批量渲染",
+    "priority_queue": "优先处理队列",
+    "ai_matting": "AI 智能抠像",
+    "remove_watermark": "去水印",
+}
+
+
+def _cached(key: str, loader):
+    now = time.time()
+    hit = _cache.get(key)
+    if hit and now - hit[0] < _CACHE_TTL:
+        return hit[1]
+    with _lock:
+        hit = _cache.get(key)
+        if hit and time.time() - hit[0] < _CACHE_TTL:
+            return hit[1]
+        value = loader()
+        _cache[key] = (time.time(), value)
+        return value
+
+
+def _quota_features(quotas: dict[str, Any] | None) -> dict[str, Any]:
+    quotas = quotas or {}
+    features: dict[str, Any] = {}
+    for k, v in quotas.items():
+        if k == "credits_per_month":
+            features["credits_per_month"] = v
+        elif k in _QUOTA_LABELS:
+            features[_QUOTA_LABELS[k]] = v
+        else:
+            features[k] = v
+    return features
+
+
+def get_membership_plans() -> list[dict[str, Any]]:
+    """读取 is_enabled=true 的套餐，按年/月周期展开为用户端档位。"""
+
+    def _load() -> list[dict[str, Any]]:
+        from sqlalchemy import text
+
+        from packages.adapters.sqlalchemy_impl.session import SessionLocal
+
+        if SessionLocal is None:
+            return []
+
+        session = SessionLocal()
+        try:
+            rows = session.execute(text("""
+                    SELECT plan_key, name, description, monthly_price, yearly_price,
+                           quotas, display_order
+                    FROM plans
+                    WHERE is_enabled = TRUE
+                    ORDER BY display_order NULLS LAST, created_at
+                    """)).fetchall()
+        finally:
+            session.close()
+
+        plans: list[dict[str, Any]] = []
+        for r in rows:
+            base_features = _quota_features(r.quotas if isinstance(r.quotas, dict) else None)
+            if r.yearly_price and float(r.yearly_price) > 0:
+                plans.append(
+                    {
+                        "plan_id": r.plan_key,
+                        "billing_cycle": "yearly",
+                        "name": r.name,
+                        "description": r.description,
+                        "price_cents": int(round(float(r.yearly_price) * 100)),
+                        "monthly_price_cents": int(round(float(r.yearly_price) * 100 / 12)),
+                        "duration_days": 365,
+                        "features": dict(base_features),
+                    }
+                )
+            if r.monthly_price and float(r.monthly_price) > 0:
+                plans.append(
+                    {
+                        "plan_id": r.plan_key,
+                        "billing_cycle": "monthly",
+                        "name": r.name,
+                        "description": r.description,
+                        "price_cents": int(round(float(r.monthly_price) * 100)),
+                        "monthly_price_cents": int(round(float(r.monthly_price) * 100)),
+                        "duration_days": 30,
+                        "features": dict(base_features),
+                    }
+                )
+        return plans
+
+    return _cached("membership_plans", _load)
+
+
+def get_points_packages() -> list[dict[str, Any]]:
+    """读取 is_active=true 的积分充值包。"""
+
+    def _load() -> list[dict[str, Any]]:
+        from sqlalchemy import text
+
+        from packages.adapters.sqlalchemy_impl.session import SessionLocal
+
+        if SessionLocal is None:
+            return []
+
+        session = SessionLocal()
+        try:
+            rows = session.execute(text("""
+                    SELECT package_key, name, price, credits, bonus_credits,
+                           is_recommended, description, sort_order
+                    FROM credit_packages
+                    WHERE is_active = TRUE
+                    ORDER BY sort_order NULLS LAST, price
+                    """)).fetchall()
+        finally:
+            session.close()
+
+        packages: list[dict[str, Any]] = []
+        for r in rows:
+            total_points = int(r.credits or 0) + int(r.bonus_credits or 0)
+            price_cents = int(round(float(r.price) * 100))
+            unit = (price_cents / 100 / total_points) if total_points else 0
+            packages.append(
+                {
+                    "code": r.package_key,
+                    "name": r.name,
+                    "points": total_points,
+                    "bonus_credits": int(r.bonus_credits or 0),
+                    "price_cents": price_cents,
+                    "unit_price": f"¥{unit:.3f}/积分",
+                    "is_recommended": bool(r.is_recommended),
+                    "description": r.description,
+                }
+            )
+        return packages
+
+    return _cached("points_packages", _load)

tests/unit/test_catalog.py

@@ -0,0 +1,198 @@
+"""catalog 应用服务单测：会员套餐 / 积分包从共享库读取与字段映射。"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def _clear_cache():
+    from packages.application.catalog import admin_catalog
+
+    admin_catalog._cache.clear()
+    yield
+    admin_catalog._cache.clear()
+
+
+def _row(**kw):
+    row = MagicMock()
+    for k, v in kw.items():
+        setattr(row, k, v)
+    return row
+
+
+class TestMembershipPlans:
+    def test_yearly_plan_mapping(self):
+        from packages.application.catalog import admin_catalog
+
+        row = _row(
+            plan_key="premium_yearly",
+            name="高级会员年卡",
+            description="年度订阅",
+            monthly_price=0,
+            yearly_price=399,
+            quotas={"4k": True, "batch_render": True, "credits_per_month": 500},
+            display_order=1,
+        )
+        session = MagicMock()
+        session.execute.return_value.fetchall.return_value = [row]
+        sl = MagicMock(return_value=session)
+
+        with patch("packages.adapters.sqlalchemy_impl.session.SessionLocal", sl, create=True):
+            plans = admin_catalog.get_membership_plans()
+
+        assert len(plans) == 1
+        p = plans[0]
+        assert p["plan_id"] == "premium_yearly"
+        assert p["billing_cycle"] == "yearly"
+        assert p["price_cents"] == 39900
+        assert p["monthly_price_cents"] == 3325
+        assert p["duration_days"] == 365
+        assert p["features"]["4K 超清分辨率"] is True
+        assert p["features"]["credits_per_month"] == 500
+        session.close.assert_called_once()
+
+    def test_monthly_plan_mapping(self):
+        from packages.application.catalog import admin_catalog
+
+        row = _row(
+            plan_key="premium_monthly",
+            name="高级会员月卡",
+            description=None,
+            monthly_price=39,
+            yearly_price=0,
+            quotas=None,
+            display_order=2,
+        )
+        session = MagicMock()
+        session.execute.return_value.fetchall.return_value = [row]
+        sl = MagicMock(return_value=session)
+
+        with patch("packages.adapters.sqlalchemy_impl.session.SessionLocal", sl, create=True):
+            plans = admin_catalog.get_membership_plans()
+
+        assert len(plans) == 1
+        p = plans[0]
+        assert p["billing_cycle"] == "monthly"
+        assert p["price_cents"] == 3900
+        assert p["monthly_price_cents"] == 3900
+        assert p["duration_days"] == 30
+        assert p["features"] == {}
+
+    def test_both_cycles_expanded(self):
+        from packages.application.catalog import admin_catalog
+
+        row = _row(
+            plan_key="premium",
+            name="高级会员",
+            description=None,
+            monthly_price=39,
+            yearly_price=399,
+            quotas={},
+            display_order=1,
+        )
+        session = MagicMock()
+        session.execute.return_value.fetchall.return_value = [row]
+        sl = MagicMock(return_value=session)
+
+        with patch("packages.adapters.sqlalchemy_impl.session.SessionLocal", sl, create=True):
+            plans = admin_catalog.get_membership_plans()
+
+        cycles = {p["billing_cycle"] for p in plans}
+        assert cycles == {"yearly", "monthly"}
+
+    def test_no_session_returns_empty(self):
+        from packages.application.catalog import admin_catalog
+
+        with patch("packages.adapters.sqlalchemy_impl.session.SessionLocal", None, create=True):
+            assert admin_catalog.get_membership_plans() == []
+
+
+class TestPointsPackages:
+    def test_package_mapping_with_bonus(self):
+        from packages.application.catalog import admin_catalog
+
+        row = _row(
+            package_key="pkg_100",
+            name="100元充值包",
+            price=100,
+            credits=1000,
+            bonus_credits=100,
+            is_recommended=True,
+            description="推荐",
+            sort_order=4,
+        )
+        session = MagicMock()
+        session.execute.return_value.fetchall.return_value = [row]
+        sl = MagicMock(return_value=session)
+
+        with patch("packages.adapters.sqlalchemy_impl.session.SessionLocal", sl, create=True):
+            packages = admin_catalog.get_points_packages()
+
+        assert len(packages) == 1
+        pkg = packages[0]
+        assert pkg["code"] == "pkg_100"
+        assert pkg["points"] == 1100
+        assert pkg["price_cents"] == 10000
+        assert pkg["is_recommended"] is True
+        assert pkg["unit_price"] == "¥0.091/积分"
+
+    def test_zero_credits_unit_price_safe(self):
+        from packages.application.catalog import admin_catalog
+
+        row = _row(
+            package_key="pkg_0",
+            name="空包",
+            price=0,
+            credits=0,
+            bonus_credits=0,
+            is_recommended=False,
+            description=None,
+            sort_order=0,
+        )
+        session = MagicMock()
+        session.execute.return_value.fetchall.return_value = [row]
+        sl = MagicMock(return_value=session)
+
+        with patch("packages.adapters.sqlalchemy_impl.session.SessionLocal", sl, create=True):
+            packages = admin_catalog.get_points_packages()
+
+        assert packages[0]["points"] == 0
+        assert packages[0]["price_cents"] == 0
+        assert packages[0]["unit_price"] == "¥0.000/积分"
+
+    def test_no_session_returns_empty(self):
+        from packages.application.catalog import admin_catalog
+
+        with patch("packages.adapters.sqlalchemy_impl.session.SessionLocal", None, create=True):
+            assert admin_catalog.get_points_packages() == []
+
+
+class TestPackagesRoute:
+    def test_get_packages_route_returns_items(self):
+        from app.api.routes.points import get_packages
+
+        cu = MagicMock()
+        cu.user.member_type = None
+        rows = [
+            {
+                "code": "pkg_10",
+                "name": "10元充值包",
+                "points": 100,
+                "price_cents": 1000,
+                "unit_price": "¥0.100/积分",
+            }
+        ]
+        with patch(
+            "packages.application.catalog.admin_catalog.get_points_packages",
+            return_value=rows,
+        ):
+            resp = get_packages(current_user=cu)
+
+        assert len(resp.packages) == 1
+        item = resp.packages[0]
+        assert item.code == "pkg_10"
+        assert item.points == 100
+        assert item.price_cents == 1000

tests/unit/test_points_routes.py

@@ -173,33 +173,46 @@ class TestSubscriptionPlans:
         _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)
-
+    def test_plans_endpoint_reads_admin_table(self):
+        """/subscription/plans 改读管理后台 plans 表：返回 catalog 服务提供的真实档位。"""
         list_membership_plans = self._import_plans_fn()
-        resp = list_membership_plans(current_user=_make_cu())
+        real_plan = {
+            "plan_id": "premium_yearly",
+            "billing_cycle": "yearly",
+            "name": "高级会员年卡",
+            "description": "高级会员年度订阅，享受全部功能",
+            "price_cents": 39900,
+            "monthly_price_cents": 3325,
+            "duration_days": 365,
+            "features": {
+                "4K 超清分辨率": True,
+                "批量渲染": True,
+                "优先处理队列": True,
+                "credits_per_month": 500,
+            },
+        }
+        with patch(
+            "packages.application.catalog.admin_catalog.get_membership_plans",
+            return_value=[real_plan],
+        ):
+            resp = list_membership_plans(current_user=_make_cu())
         plans = resp["plans"]
-        plan_ids = {p["plan_id"] for p in plans}
-        assert plan_ids == {"monthly", "quarterly", "yearly"}
-        for p in plans:
-            assert p["price_cents"] > 0
-            assert p["duration_days"] in (30, 90, 365)
-            assert 0 < p["points_discount"] <= 1.0
-            assert "max_resolution" in p["features"]
-
-    def test_longer_plans_cheaper_per_month(self):
-        import os  # noqa: F401
+        assert len(plans) == 1
+        p0 = plans[0]
+        assert p0["plan_id"] == "premium_yearly"
+        assert p0["price_cents"] == 39900
+        assert p0["duration_days"] == 365
+        assert p0["features"]["4K 超清分辨率"] is True
 
+    def test_plans_endpoint_empty_when_all_disabled(self):
+        """后台停用全部套餐时，用户端返回空列表。"""
         list_membership_plans = self._import_plans_fn()
-        resp = list_membership_plans(current_user=_make_cu())
-        plans = resp["plans"]
-        monthly = next(p for p in plans if p["plan_id"] == "monthly")
-        quarterly = next(p for p in plans if p["plan_id"] == "quarterly")
-        yearly = next(p for p in plans if p["plan_id"] == "yearly")
-        assert monthly["monthly_price_cents"] == 1990
-        assert quarterly["monthly_price_cents"] < monthly["monthly_price_cents"]
-        assert yearly["monthly_price_cents"] < quarterly["monthly_price_cents"]
-
+        with patch(
+            "packages.application.catalog.admin_catalog.get_membership_plans",
+            return_value=[],
+        ):
+            resp = list_membership_plans(current_user=_make_cu())
+        assert resp["plans"] == []
 
 # ── P1-7: multiplier consistency ──────────────────────────────────────