xiaoxia-saas/packages/application/catalog/admin_catalog.py

"""读取管理后台配置的会员套餐 / 积分充值包（共享库真实数据）。

替代旧的硬编码 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)