video-gen/video-gen-api/app/api/admin/credit_management.py

from __future__ import annotations

import json
from datetime import timedelta

from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.dependencies import get_admin_user, get_db
from app.enums.credit_balance import CreditBalanceSourceType
from app.models.credit.balance import UserCreditBalance
from app.models.credit.product import CreditProduct
from app.models.user import User
from app.schemas.credit_balance import AdminCreditDeductRequest, AdminCreditGrantRequest
from app.schemas.credit_product import CreditProductCreate, CreditProductRenewalUpdate, CreditProductUpdate
from app.services.credit.ledger_service import deduct_credits, grant_credits
from app.services.credit.product_service import product_to_dict
from app.services.credit.query_service import (
    apply_balance_status_filter,
    effective_balance_status,
    get_balance_summary,
)
from app.services.credit.time_policy import add_natural_months, last_usable_at
from app.services.credit.utils import utc_now
from app.services.notification import create_notification
from app.services.operation_log import log_operation
from app.services.operation_log_service import log_operation_event
from app.utils.id_gen import generate_id

router = APIRouter(prefix="/admin/credit-management", tags=["admin-credit-management"])


def _apply_product_payload(product: CreditProduct, payload: dict) -> None:
    mapping = {"features": "features_json"}
    for key, value in payload.items():
        setattr(product, mapping.get(key, key), value)
    if product.product_type == "credit_addon" and product.validity_months is None:
        # 兼容旧管理端未提交有效期的请求，新建增值包仍默认1个月；
        # 更新时未提交该字段则保留原值。
        product.validity_months = 1
    if product.product_type == "subscription":
        product.price = product.regular_price or 0
        product.grant_credits = None
        product.validity_months = None
    else:
        product.renewal_enabled = False
        product.tier_code = None
        product.tier_rank = None
        product.billing_cycle = None
        product.monthly_grant_credits = None
        product.first_purchase_price = None
        product.regular_price = None
        product.activity_price = None
        product.activity_start_at = None
        product.activity_end_at = None


def _validate_product_entity(product: CreditProduct) -> None:
    if product.product_type == "subscription":
        required = {
            "套餐等级编码": product.tier_code,
            "套餐等级顺序": product.tier_rank,
            "订阅周期": product.billing_cycle,
            "每月积分": product.monthly_grant_credits,
            "首充价格": product.first_purchase_price,
            "原价": product.regular_price,
        }
        missing = [label for label, value in required.items() if value is None]
        if missing:
            raise HTTPException(status_code=400, detail=f"订阅套餐缺少字段：{'、'.join(missing)}")
        if product.activity_price is None:
            if product.activity_start_at is not None or product.activity_end_at is not None:
                raise HTTPException(status_code=400, detail="未配置活动价时不能单独配置活动周期")
        elif product.activity_start_at is None or product.activity_end_at is None:
            raise HTTPException(status_code=400, detail="配置活动价时必须同时配置活动开始和结束时间")
        elif product.activity_end_at <= product.activity_start_at:
            raise HTTPException(status_code=400, detail="活动结束时间必须晚于开始时间")
    elif product.product_type == "credit_addon":
        if product.grant_credits is None or product.price is None:
            raise HTTPException(status_code=400, detail="积分增值包必须配置价格和积分数量")
        if product.validity_months is None or not 1 <= int(product.validity_months) <= 36:
            raise HTTPException(status_code=400, detail="积分增值包有效期必须为1-36个月")
    else:
        raise HTTPException(status_code=400, detail="不支持的积分商品类型")


@router.get("/products")
async def list_products(
    product_type: str | None = Query(default=None),
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    stmt = select(CreditProduct)
    if product_type:
        stmt = stmt.where(CreditProduct.product_type == product_type)
    result = await db.execute(stmt.order_by(CreditProduct.product_type, CreditProduct.sort_order, CreditProduct.id))
    return [product_to_dict(item) for item in result.scalars().all()]


@router.post("/products")
async def create_product(
    data: CreditProductCreate,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    exists = await db.execute(select(CreditProduct.id).where(CreditProduct.product_code == data.product_code).limit(1))
    if exists.scalar_one_or_none():
        raise HTTPException(status_code=409, detail="商品编码已存在")
    product = CreditProduct(id=generate_id())
    _apply_product_payload(product, data.model_dump())
    _validate_product_entity(product)
    db.add(product)
    await db.flush()
    snapshot = product_to_dict(product)
    await log_operation(db, admin.id, admin.username, f"创建积分商品 {product.name}", "POST", "/admin/credit-management/products", detail=json.dumps(snapshot, ensure_ascii=False, default=str))
    log_operation_event(domain="credit_product", module="credit", event_type="CREDIT_PRODUCT_CREATED", user_id=admin.id, detail=snapshot)
    # 商品保存后前端会立即使用返回值刷新列表。这里显式提交，避免依赖
    # yield 依赖退出阶段提交时出现紧随其后的 GET 读到旧状态。
    await db.commit()
    return snapshot


@router.put("/products/{product_id}")
async def update_product(
    product_id: str,
    data: CreditProductUpdate,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(select(CreditProduct).where(CreditProduct.id == product_id).limit(1).with_for_update())
    product = result.scalar_one_or_none()
    if not product:
        raise HTTPException(status_code=404, detail="商品不存在")
    before = product_to_dict(product)
    payload = data.model_dump(exclude_unset=True)
    new_code = payload.get("product_code")
    if new_code and new_code != product.product_code:
        duplicate = await db.execute(
            select(CreditProduct.id).where(
                CreditProduct.product_code == new_code, CreditProduct.id != product.id
            ).limit(1)
        )
        if duplicate.scalar_one_or_none():
            raise HTTPException(status_code=409, detail="商品编码已存在")
    _apply_product_payload(product, payload)
    _validate_product_entity(product)
    await db.flush()
    after = product_to_dict(product)
    await log_operation(db, admin.id, admin.username, f"更新积分商品 {product.name}", "PUT", f"/admin/credit-management/products/{product_id}", detail=json.dumps({"before": before, "after": after}, ensure_ascii=False, default=str))
    log_operation_event(domain="credit_product", module="credit", event_type="CREDIT_PRODUCT_UPDATED", user_id=admin.id, detail={"product_id": product_id})
    await db.commit()
    refreshed = await db.execute(select(CreditProduct).where(CreditProduct.id == product_id).limit(1))
    persisted = refreshed.scalar_one_or_none()
    if persisted is None:
        raise HTTPException(status_code=404, detail="商品不存在")
    return product_to_dict(persisted)


@router.put("/products/{product_id}/renewal")
async def update_product_renewal(
    product_id: str,
    data: CreditProductRenewalUpdate,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(CreditProduct)
        .where(CreditProduct.id == product_id)
        .limit(1)
        .with_for_update()
    )
    product = result.scalar_one_or_none()
    if not product:
        raise HTTPException(status_code=404, detail="商品不存在")
    if product.product_type != "subscription":
        raise HTTPException(status_code=400, detail="积分增值包不支持续费开关")

    before = bool(product.renewal_enabled)
    product.renewal_enabled = bool(data.renewal_enabled)
    await db.flush()
    await log_operation(
        db,
        admin.id,
        admin.username,
        f"{'开启' if product.renewal_enabled else '关闭'}积分商品续费 {product.name}",
        "PUT",
        f"/admin/credit-management/products/{product_id}/renewal",
        detail=json.dumps(
            {"before": before, "after": bool(product.renewal_enabled)},
            ensure_ascii=False,
        ),
    )
    await db.commit()

    # 提交后重新查询，返回数据库真实持久化结果，避免前端误用事务内快照。
    refreshed = await db.execute(
        select(CreditProduct).where(CreditProduct.id == product_id).limit(1)
    )
    persisted = refreshed.scalar_one_or_none()
    if persisted is None:
        raise HTTPException(status_code=404, detail="商品不存在")
    if bool(persisted.renewal_enabled) != bool(data.renewal_enabled):
        raise HTTPException(status_code=500, detail="续费状态保存后校验失败")
    return product_to_dict(persisted)


@router.delete("/products/{product_id}")
async def disable_product(
    product_id: str,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(select(CreditProduct).where(CreditProduct.id == product_id).limit(1).with_for_update())
    product = result.scalar_one_or_none()
    if not product:
        raise HTTPException(status_code=404, detail="商品不存在")
    product.is_active = False
    await db.flush()
    await log_operation(db, admin.id, admin.username, f"下架积分商品 {product.name}", "DELETE", f"/admin/credit-management/products/{product_id}")
    await db.commit()
    return {"ok": True}


@router.get("/users/{user_id}/summary")
async def get_user_credit_summary(
    user_id: str,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    return (await get_balance_summary(db, user_id)).to_dict()


@router.get("/users/{user_id}/balances")
async def list_user_credit_balances(
    user_id: str,
    status: str | None = Query(default=None),
    page: int = Query(1, ge=1),
    page_size: int = Query(50, ge=1, le=200),
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    checked_at = utc_now()
    stmt = select(UserCreditBalance).where(UserCreditBalance.user_id == user_id)
    stmt = apply_balance_status_filter(stmt, status, request_time=checked_at)
    result = await db.execute(stmt.order_by(UserCreditBalance.expires_at.asc(), UserCreditBalance.id.asc()).offset((page - 1) * page_size).limit(page_size))
    return [
        {
            "id": item.id,
            "credit_level": item.credit_level,
            "source_type": item.source_type,
            "source_id": item.source_id,
            "grant_amount": float(item.grant_amount),
            "unspent_amount": float(item.unspent_amount),
            "consumed_amount": float(item.consumed_amount),
            "expired_amount": float(item.expired_amount),
            "revoked_amount": float(item.revoked_amount),
            "valid_from": item.valid_from,
            "expires_at": item.expires_at,
            "last_usable_at": last_usable_at(item.expires_at),
            "status": effective_balance_status(item, request_time=checked_at),
        }
        for item in result.scalars().all()
    ]


@router.post("/users/{user_id}/grant")
async def admin_grant_credit(
    user_id: str,
    data: AdminCreditGrantRequest,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    starts_at = data.valid_from or utc_now()
    ends_at = starts_at + timedelta(days=data.validity_value) if data.validity_unit == "day" else add_natural_months(starts_at, data.validity_value)
    result = await grant_credits(
        db,
        user_id=user_id,
        amount=data.amount,
        description=data.description,
        source_type=CreditBalanceSourceType.ADMIN_GRANT.value,
        source_id=admin.id,
        valid_from=starts_at,
        expires_at=ends_at,
        credit_level=data.credit_level,
        related_id=admin.id,
        biz_key=f"admin-grant:{admin.id}:{generate_id()}",
    )
    await create_notification(
        db, user_id, "积分变动通知",
        f"您的积分已增加{data.amount}积分。原因：{data.description}", "credit",
    )
    await log_operation(db, admin.id, admin.username, f"给用户 {user_id} 增加积分 {data.amount}", "POST", f"/admin/credit-management/users/{user_id}/grant")
    return {"ok": True, "credits": result.balance_after}


@router.post("/users/{user_id}/deduct")
async def admin_deduct_credit(
    user_id: str,
    data: AdminCreditDeductRequest,
    admin: User = Depends(get_admin_user),
    db: AsyncSession = Depends(get_db),
):
    try:
        result = await deduct_credits(
            db,
            user_id=user_id,
            amount=data.amount,
            description=data.description,
            related_id=admin.id,
            biz_key=f"admin-deduct:{admin.id}:{generate_id()}",
        )
    except Exception as exc:
        if exc.__class__.__name__ == "InsufficientCreditsError":
            raise HTTPException(status_code=400, detail="用户有效积分不足") from exc
        raise
    await create_notification(
        db, user_id, "积分变动通知",
        f"您的积分已扣除{data.amount}积分。原因：{data.description}", "credit",
    )
    await log_operation(db, admin.id, admin.username, f"扣除用户 {user_id} 积分 {data.amount}", "POST", f"/admin/credit-management/users/{user_id}/deduct")
    return {"ok": True, "credits": result.balance_after}