xiaoxia-saas/packages/middleware/points_gate.py

"""PointsGate — AI 路由积分扣费中间件（#1895 P4）。

使用方式::

    from packages.middleware.points_gate import points_deduction

    @router.post("/generate")
    def generate_title(
        request: TitleRequest,
        current_user: User = Depends(get_current_user),
        points_svc: PointsService = Depends(get_points_service),
    ):
        with points_deduction(points_svc, current_user, "ai_title", description="AI 标题生成"):
            result = do_generate(...)
        return result

- 默认受 ``settings.POINTS_ENABLED`` 开关控制，关闭时不扣费，直接放行。
- ``ai_video`` 场景会优先走每日免费混剪额度（免费用户每日 2 条），会员直接走扣费。
- contextmanager 内业务异常会自动 refund；未抛异常视为成功，积分正常扣除。
"""

from __future__ import annotations

import logging
from contextlib import contextmanager
from datetime import datetime, timezone
from typing import Any, Iterator

from fastapi import HTTPException, status

logger = logging.getLogger(__name__)


def _is_active_member(user: Any) -> bool:
    """判断用户是否为「在有效期内」的付费会员."""
    if not user:
        return False
    if not bool(getattr(user, "is_member", False)):
        return False
    expires_at = getattr(user, "member_expires_at", None)
    if expires_at is None:
        return True
    now = datetime.now(timezone.utc)
    if getattr(expires_at, "tzinfo", None) is None:
        now = now.replace(tzinfo=None)
    return expires_at > now


def _insufficient_points(amount: int, balance: int, scene_name: str = "") -> HTTPException:
    return HTTPException(
        status_code=status.HTTP_402_PAYMENT_REQUIRED,
        detail={
            "code": "INSUFFICIENT_POINTS",
            "message": f"积分不足，需要 {amount} 积分，当前余额 {balance}",
            "required_points": amount,
            "current_balance": balance,
            "scene": scene_name,
        },
    )


@contextmanager
def points_deduction(
    points_svc: Any,
    user: Any,
    scene_key: str,
    *,
    duration_minutes: float = 1.0,
    extra_segments: int = 0,
    description: str = "",
    enabled: bool | None = None,
) -> Iterator[str | None]:
    """积分扣费 contextmanager：业务成功 → 确认扣费；业务抛异常 → 自动退款.

    :param points_svc: PointsService 实例
    :param user: User entity（需带 id/is_member/member_expires_at）
    :param scene_key: POINTS_RULES 中的场景 key
    :param duration_minutes: 时长（分钟），按场景语义解释
    :param extra_segments: 额外片段数（ai_video 预留）
    :param description: 流水描述
    :param enabled: 显式开关；None 时读取 settings.POINTS_ENABLED
    :yields: transaction_id 或 None（免费/未启用场景）
    """
    if enabled is None:
        from app.config import settings

        enabled = bool(getattr(settings, "POINTS_ENABLED", False))
    if not enabled:
        yield None
        return

    from packages.domain.points import POINTS_RULES, calc_points

    if scene_key not in POINTS_RULES:
        logger.debug("points_gate: 未配置场景 scene=%s，放行", scene_key)
        yield None
        return

    user_id = getattr(user, "id", None)
    if not user_id:
        yield None
        return

    is_member = _is_active_member(user)
    scene_name = POINTS_RULES[scene_key].get("name", scene_key)
    desc = description or scene_name

    if scene_key == "ai_video" and not is_member:
        try:
            if points_svc.check_and_incr_daily_free_clips(user_id):
                logger.info("points_gate: 免费混剪额度占用 user=%s", user_id)
                yield None
                return
        except Exception:
            logger.warning("points_gate: daily_free_clips 检查失败，降级走扣费 user=%s", user_id, exc_info=True)

    amount = calc_points(
        scene_key,
        is_member,
        duration_minutes=duration_minutes,
        extra_segments=extra_segments,
    )
    if amount <= 0:
        logger.debug("points_gate: 免费场景 scene=%s，放行", scene_key)
        yield None
        return

    result = points_svc.check_and_deduct(
        user_id=user_id,
        scene_key=scene_key,
        duration_minutes=duration_minutes,
        extra_segments=extra_segments,
        description=desc,
        is_member=is_member,
    )
    if not result.success:
        logger.info(
            "points_gate: 扣费失败 user=%s scene=%s reason=%s need=%d bal=%d",
            user_id,
            scene_key,
            result.reason,
            amount,
            result.balance,
        )
        raise _insufficient_points(amount, result.balance, scene_name)

    tx_id = result.transaction_id
    logger.info(
        "points_gate: 扣费成功 user=%s scene=%s amount=%d tx=%s",
        user_id,
        scene_key,
        amount,
        tx_id,
    )
    try:
        yield tx_id
    except Exception:
        if tx_id:
            try:
                points_svc.refund(user_id, tx_id, reason=f"{scene_key} 业务失败: {desc}")
                logger.info("points_gate: 业务异常已退款 user=%s tx=%s scene=%s", user_id, tx_id, scene_key)
            except Exception:
                logger.exception("points_gate: 退款失败 user=%s tx=%s", user_id, tx_id)
        raise