import json
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any

from fastapi import HTTPException
from sqlalchemy import func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.cart_item import CartItem
from app.models.coupon_activity_claim import CouponActivityClaim, CouponActivityClaimStatus
from app.models.coupon_mall_claim import CouponMallClaim, CouponMallClaimStatus
from app.models.coupon_mall_item import CouponMallItem, CouponMallItemStatus
from app.models.order import Order, OrderStatus
from app.models.product import Product
from app.models.product_group import ProductGroup
from app.models.promo_code import (
    PromoCode,
    PromoCodeClaimMethod,
    PromoCodeDiscountLevel,
    PromoCodeScope,
    PromoCodeType,
)
from app.models.promo_code_usage import PromoCodeUsage
from app.models.user import User
from app.models.user_agent import AgentStatus, UserAgent


@dataclass(frozen=True)
class CouponApplication:
    promo_code: PromoCode
    code: str
    order_amount: int
    eligible_amount: int
    discount_amount: int
    type: str
    value: int
    description: str | None
    discount_level: str
    target_cart_item_id: int | None = None
    target_product_id: int | None = None
    item_discounts: dict[int, int] = field(default_factory=dict)


def normalize_coupon_code(code: str | None) -> str | None:
    text = (code or "").strip().upper()
    return text or None


def _enum_value(value: Any) -> str:
    return value.value if hasattr(value, "value") else str(value)


def _as_aware(value: datetime | None) -> datetime | None:
    if value is None:
        return None
    if value.tzinfo is None:
        return value.replace(tzinfo=timezone.utc)
    return value


def _parse_list(raw: str | None) -> set[str]:
    if not raw:
        return set()
    try:
        data = json.loads(raw)
        if isinstance(data, list):
            return {str(item).strip().lower() for item in data if str(item).strip()}
    except Exception:
        pass
    return {part.strip().lower() for part in raw.split(",") if part.strip()}


def _parse_int_set(raw: str | None) -> set[int]:
    values = set()
    for item in _parse_list(raw):
        try:
            values.add(int(item))
        except (TypeError, ValueError):
            continue
    return values


def _allocate_amounts(amounts: list[int], target_amount: int, total_amount: int) -> list[int]:
    if not amounts:
        return []
    if target_amount <= 0 or total_amount <= 0:
        return [0 for _ in amounts]
    allocated = [(max(0, amount) * target_amount) // total_amount for amount in amounts]
    remainder = target_amount - sum(allocated)
    for idx in range(min(remainder, len(allocated))):
        allocated[idx] += 1
    return allocated


async def _load_coupon(db: AsyncSession, code: str, *, for_update: bool = False) -> PromoCode:
    stmt = select(PromoCode).where(func.upper(PromoCode.code) == code)
    if for_update:
        stmt = stmt.with_for_update()
    result = await db.execute(stmt)
    coupon = result.scalar_one_or_none()
    if not coupon:
        raise HTTPException(status_code=404, detail="优惠券不存在")
    return coupon


async def _user_used_count(db: AsyncSession, promo_code_id: int, user_id: int) -> int:
    result = await db.execute(
        select(func.count(PromoCodeUsage.id)).where(
            PromoCodeUsage.promo_code_id == promo_code_id,
            PromoCodeUsage.user_id == user_id,
            PromoCodeUsage.status == "used",
        )
    )
    return int(result.scalar_one() or 0)


async def _has_prior_paid_order(db: AsyncSession, user_id: int) -> bool:
    result = await db.execute(
        select(func.count(Order.id)).where(
            Order.user_id == user_id,
            Order.status.in_([OrderStatus.PAID.value, OrderStatus.REFUNDED.value]),
        )
    )
    return int(result.scalar_one() or 0) > 0


async def _validate_mall_claim(db: AsyncSession, coupon: PromoCode, user_id: int) -> CouponMallClaim | None:
    claim_method = _enum_value(getattr(coupon, "claim_method", PromoCodeClaimMethod.CODE.value))
    if claim_method != PromoCodeClaimMethod.MALL.value:
        return None

    claim = await db.scalar(
        select(CouponMallClaim).where(
            CouponMallClaim.promo_code_id == coupon.id,
            CouponMallClaim.user_id == user_id,
            CouponMallClaim.status == CouponMallClaimStatus.CLAIMED,
        )
    )
    if not claim:
        raise HTTPException(status_code=400, detail="该优惠券需先在积分商城领取")
    return claim


async def _validate_user_can_use_coupons(db: AsyncSession, user_id: int) -> None:
    """激活代理身份的用户不能参与普通优惠券抵扣。"""
    is_agent = await db.scalar(
        select(UserAgent.id).where(
            UserAgent.user_id == user_id,
            UserAgent.status.in_([AgentStatus.ACTIVE, AgentStatus.OBSERVING]),
        )
    )
    if is_agent:
        raise HTTPException(status_code=400, detail="代理身份用户不可使用优惠券")


async def _validate_agent_coupon_owner(db: AsyncSession, coupon: PromoCode, user_id: int) -> None:
    if getattr(coupon, "issuer_type", None) != "agent":
        return

    issuer_user_id = getattr(coupon, "issuer_user_id", None)
    if not issuer_user_id:
        raise HTTPException(status_code=400, detail="代理优惠券归属异常，暂不可使用")
    if int(issuer_user_id) == int(user_id):
        raise HTTPException(status_code=400, detail="不能使用自己创建的代理优惠券")

    active_agent = await db.scalar(
        select(UserAgent.id).where(
            UserAgent.user_id == int(issuer_user_id),
            UserAgent.status.in_([AgentStatus.ACTIVE, AgentStatus.OBSERVING]),
        )
    )
    if not active_agent:
        raise HTTPException(status_code=400, detail="该代理优惠券所属代理未激活，暂不可使用")

    customer = await db.get(User, user_id)
    if customer and customer.bound_agent_id and int(customer.bound_agent_id) != int(issuer_user_id):
        raise HTTPException(status_code=400, detail="该优惠券不属于您绑定的代理")


async def _category_ids_for_products(db: AsyncSession, products: dict[int, Product]) -> dict[int, int | None]:
    group_ids = {int(product.group_id) for product in products.values() if product.group_id}
    if not group_ids:
        return {}
    result = await db.execute(select(ProductGroup.id, ProductGroup.category_id).where(ProductGroup.id.in_(group_ids)))
    return {
        int(group_id): int(category_id) if category_id is not None else None
        for group_id, category_id in result.all()
    }


async def _matched_item_ids(
    db: AsyncSession,
    coupon: PromoCode,
    cart_items: list[CartItem],
    products: dict[int, Product],
) -> set[int]:
    cycles = _parse_list(coupon.applicable_cycles)
    scope = _enum_value(coupon.scope)
    scope_ids = _parse_int_set(coupon.scope_ids)
    group_categories: dict[int, int | None] = {}
    if scope == PromoCodeScope.CATEGORY.value:
        group_categories = await _category_ids_for_products(db, products)

    matched: set[int] = set()
    for item in cart_items:
        product = products.get(item.product_id)
        if not product:
            continue
        if cycles and str(item.cycle or "").lower() not in cycles:
            continue

        item_matches = scope == PromoCodeScope.ALL.value
        if scope == PromoCodeScope.PRODUCT.value:
            item_matches = int(product.id) in scope_ids
        elif scope == PromoCodeScope.GROUP.value:
            item_matches = bool(product.group_id and int(product.group_id) in scope_ids)
        elif scope == PromoCodeScope.CATEGORY.value:
            category_id = group_categories.get(int(product.group_id)) if product.group_id else None
            item_matches = bool(category_id and int(category_id) in scope_ids)

        if item_matches:
            matched.add(int(item.id))

    return matched


def _discount_amount(coupon: PromoCode, eligible_amount: int) -> int:
    coupon_type = _enum_value(coupon.type)
    value = int(coupon.value or 0)
    if coupon_type == PromoCodeType.PERCENT.value:
        percent = max(0, min(100, value))
        discount = eligible_amount * percent // 100
    elif coupon_type == PromoCodeType.FIXED.value:
        discount = min(eligible_amount, max(0, value))
    elif coupon_type == PromoCodeType.THRESHOLD_FIXED.value:
        threshold = int(coupon.min_order_amount or 0)
        if threshold > 0 and eligible_amount < threshold:
            discount = 0
        else:
            discount = min(eligible_amount, max(0, value))
    elif coupon_type == PromoCodeType.OVERRIDE.value:
        discount = max(0, eligible_amount - max(0, value))
    else:
        raise HTTPException(status_code=400, detail="优惠券类型不支持")

    max_discount_amount = int(getattr(coupon, "max_discount_amount", 0) or 0)
    if max_discount_amount > 0:
        discount = min(discount, max_discount_amount)
    return discount


def _order_item_discounts(item_totals: dict[int, int], discount: int) -> dict[int, int]:
    ordered = [(int(item_id), max(0, int(amount))) for item_id, amount in item_totals.items() if amount > 0]
    total = sum(amount for _, amount in ordered)
    allocated = _allocate_amounts([amount for _, amount in ordered], discount, total)
    return {
        item_id: amount
        for (item_id, _), amount in zip(ordered, allocated)
        if amount > 0
    }


def _coupon_summary(coupon: PromoCode, *, order_amount: int, eligible_amount: int = 0) -> dict[str, Any]:
    return {
        "code": coupon.code,
        "description": coupon.description,
        "type": _enum_value(coupon.type),
        "value": int(coupon.value or 0),
        "discount_level": _enum_value(getattr(coupon, "discount_level", PromoCodeDiscountLevel.ORDER.value)),
        "target_cart_item_id": None,
        "target_product_id": None,
        "order_amount": order_amount,
        "eligible_amount": eligible_amount,
        "discount_amount": 0,
        "item_discounts": {},
        "max_discount_amount": int(getattr(coupon, "max_discount_amount", 0) or 0),
        "min_quantity": int(getattr(coupon, "min_quantity", 0) or 0),
    }


async def _validate_advanced_rules(
    db: AsyncSession,
    coupon: PromoCode,
    *,
    user_id: int,
    cart_items: list[CartItem],
    products: dict[int, Product],
) -> None:
    allowed_user_ids = _parse_int_set(getattr(coupon, "allowed_user_ids", None))
    if allowed_user_ids and int(user_id) not in allowed_user_ids:
        raise HTTPException(status_code=400, detail="该优惠券仅限指定用户使用")

    min_quantity = int(getattr(coupon, "min_quantity", 0) or 0)
    total_quantity = sum(max(0, int(item.quantity or 0)) for item in cart_items)
    if min_quantity > 0 and total_quantity < min_quantity:
        raise HTTPException(status_code=400, detail=f"订单商品数量需满 {min_quantity} 件才可使用该优惠券")

    required_product_ids = _parse_int_set(getattr(coupon, "required_product_ids", None))
    if required_product_ids:
        product_ids = {int(item.product_id) for item in cart_items}
        if not product_ids.intersection(required_product_ids):
            raise HTTPException(status_code=400, detail="订单需包含指定产品才可使用该优惠券")

    required_group_ids = _parse_int_set(getattr(coupon, "required_group_ids", None))
    if required_group_ids:
        group_ids = {
            int(product.group_id)
            for product in products.values()
            if product.group_id is not None
        }
        if not group_ids.intersection(required_group_ids):
            raise HTTPException(status_code=400, detail="订单需包含指定产品组才可使用该优惠券")

    required_category_ids = _parse_int_set(getattr(coupon, "required_category_ids", None))
    if required_category_ids:
        group_categories = await _category_ids_for_products(db, products)
        category_ids = {
            int(category_id)
            for category_id in group_categories.values()
            if category_id is not None
        }
        if not category_ids.intersection(required_category_ids):
            raise HTTPException(status_code=400, detail="订单需包含指定分类产品才可使用该优惠券")


async def get_product_coupon_target_selection(
    db: AsyncSession,
    *,
    user_id: int,
    cart_items: list[CartItem],
    products: dict[int, Product],
    item_totals: dict[int, int],
    code: str | None,
) -> dict[str, Any] | None:
    normalized = normalize_coupon_code(code)
    if not normalized:
        return None

    coupon = await _load_coupon(db, normalized, for_update=False)
    if not coupon.is_enabled:
        raise HTTPException(status_code=400, detail="优惠券已停用")

    now = datetime.now(timezone.utc)
    starts_at = _as_aware(coupon.starts_at)
    expires_at = _as_aware(coupon.expires_at)
    if starts_at and starts_at > now:
        raise HTTPException(status_code=400, detail="优惠券尚未开始")
    if expires_at and expires_at < now:
        raise HTTPException(status_code=400, detail="优惠券已过期")

    await _validate_user_can_use_coupons(db, user_id)

    if coupon.max_uses > 0 and coupon.used_count >= coupon.max_uses:
        raise HTTPException(status_code=400, detail="优惠券已被领用完")
    if coupon.max_uses_per_user > 0:
        used_count = await _user_used_count(db, coupon.id, user_id)
        if used_count >= coupon.max_uses_per_user:
            raise HTTPException(status_code=400, detail="该账号已达到优惠券使用次数上限")
    if coupon.new_users_only and await _has_prior_paid_order(db, user_id):
        raise HTTPException(status_code=400, detail="该优惠券仅限新用户首单使用")
    await _validate_agent_coupon_owner(db, coupon, user_id)
    await _validate_mall_claim(db, coupon, user_id)

    await _validate_advanced_rules(
        db,
        coupon,
        user_id=user_id,
        cart_items=cart_items,
        products=products,
    )

    order_amount = sum(max(0, int(value)) for value in item_totals.values())
    if coupon.min_order_amount > 0 and order_amount < coupon.min_order_amount:
        raise HTTPException(
            status_code=400,
            detail=f"订单金额需满 ¥{coupon.min_order_amount / 100:.2f} 才可使用该优惠券",
        )

    discount_level = _enum_value(getattr(coupon, "discount_level", PromoCodeDiscountLevel.ORDER.value))
    if discount_level != PromoCodeDiscountLevel.PRODUCT.value:
        return None

    matched_ids = await _matched_item_ids(db, coupon, cart_items, products)
    if not matched_ids:
        raise HTTPException(status_code=400, detail="当前购物车中没有可使用该优惠券的商品")

    options: list[dict[str, Any]] = []
    for item in cart_items:
        cart_item_id = int(item.id)
        if cart_item_id not in matched_ids:
            continue
        item_total = max(0, int(item_totals.get(cart_item_id, 0)))
        discount = min(item_total, _discount_amount(coupon, item_total))
        if discount <= 0:
            continue
        product = products.get(item.product_id)
        options.append({
            "cart_item_id": cart_item_id,
            "product_id": int(item.product_id),
            "product_name": product.name if product else f"商品 {item.product_id}",
            "sku": product.sku if product else None,
            "cycle": item.cycle,
            "quantity": int(item.quantity or 0),
            "item_total": item_total,
            "discount_amount": discount,
            "final_amount": max(0, item_total - discount),
        })

    if not options:
        raise HTTPException(status_code=400, detail="当前购物车中没有可使用该优惠券的商品")

    return {
        "requires_target_selection": True,
        "coupon": _coupon_summary(
            coupon,
            order_amount=order_amount,
            eligible_amount=sum(option["item_total"] for option in options),
        ),
        "target_options": options,
    }


async def claim_activity_coupon(
    db: AsyncSession,
    *,
    promo_code_id: int,
    user_id: int,
    activity_id: int | None = None,
) -> CouponActivityClaim:
    """用户从活动页面领取优惠券。幂等：同一用户对同一张券已有 claimed 记录则直接返回。

    限制逻辑：
    - 券必须启用且未过期
    - 按券自身的 max_uses_per_user 控制每用户可领次数（0=不限）
    - 全局 max_uses 达到上限时拒绝
    - 不需要积分，不绑定 mall_item
    """
    coupon = await db.get(PromoCode, promo_code_id)
    if not coupon or not coupon.is_enabled:
        raise HTTPException(status_code=404, detail="优惠券不存在或已停用")

    now = datetime.now(timezone.utc)
    starts_at = _as_aware(coupon.starts_at)
    expires_at = _as_aware(coupon.expires_at)
    if starts_at and starts_at > now:
        raise HTTPException(status_code=400, detail="该活动券尚未开始")
    if expires_at and expires_at < now:
        raise HTTPException(status_code=400, detail="该活动券已过期")

    if coupon.max_uses > 0 and coupon.used_count >= coupon.max_uses:
        raise HTTPException(status_code=400, detail="该优惠券已被领完")

    # 已有 claimed 记录则幂等返回
    existing = await db.scalar(
        select(CouponActivityClaim).where(
            CouponActivityClaim.promo_code_id == promo_code_id,
            CouponActivityClaim.user_id == user_id,
            CouponActivityClaim.status == CouponActivityClaimStatus.CLAIMED,
        )
    )
    if existing:
        return existing

    # 按 max_uses_per_user 控制（已领取次数 = claimed + used）
    if coupon.max_uses_per_user > 0:
        claim_count_result = await db.execute(
            select(func.count(CouponActivityClaim.id)).where(
                CouponActivityClaim.promo_code_id == promo_code_id,
                CouponActivityClaim.user_id == user_id,
            )
        )
        claim_count = int(claim_count_result.scalar_one() or 0)
        if claim_count >= coupon.max_uses_per_user:
            raise HTTPException(status_code=400, detail="您已达到该优惠券的领取次数上限")

    claim = CouponActivityClaim(
        promo_code_id=promo_code_id,
        user_id=user_id,
        activity_id=activity_id,
        status=CouponActivityClaimStatus.CLAIMED,
    )
    db.add(claim)
    await db.flush()
    return claim


async def list_available_coupons(db: AsyncSession, user_id: int) -> list[PromoCode]:
    """返回当前用户可用、未过期、未超限的优惠券列表（不含 code 解析）。

    注意：激活代理身份的用户实际结算时会被拒绝，但列表仍返回，方便前端提示原因。
    """
    now = datetime.now(timezone.utc)

    # 从积分商城领取的券
    mall_stmt = (
        select(PromoCode)
        .join(CouponMallClaim, CouponMallClaim.promo_code_id == PromoCode.id)
        .join(CouponMallItem, CouponMallItem.promo_code_id == PromoCode.id)
        .where(
            CouponMallClaim.user_id == user_id,
            CouponMallClaim.status == CouponMallClaimStatus.CLAIMED,
            CouponMallItem.status == CouponMallItemStatus.ACTIVE,
            PromoCode.is_enabled.is_(True),
            or_(PromoCode.expires_at.is_(None), PromoCode.expires_at > now),
        )
    )

    # 用户自己创建/发放的 code 券（不包括代理券，代理券由绑定关系使用）
    own_stmt = (
        select(PromoCode)
        .where(
            PromoCode.issuer_user_id == user_id,
            PromoCode.claim_method == PromoCodeClaimMethod.CODE.value,
            PromoCode.is_enabled.is_(True),
            or_(PromoCode.expires_at.is_(None), PromoCode.expires_at > now),
        )
    )

    # 从活动页面领取的券
    activity_stmt = (
        select(PromoCode)
        .join(CouponActivityClaim, CouponActivityClaim.promo_code_id == PromoCode.id)
        .where(
            CouponActivityClaim.user_id == user_id,
            CouponActivityClaim.status == CouponActivityClaimStatus.CLAIMED,
            PromoCode.is_enabled.is_(True),
            or_(PromoCode.expires_at.is_(None), PromoCode.expires_at > now),
        )
    )

    result = await db.execute(mall_stmt)
    mall_coupons = result.scalars().all()
    result = await db.execute(own_stmt)
    own_coupons = result.scalars().all()
    result = await db.execute(activity_stmt)
    activity_coupons = result.scalars().all()

    seen: set[int] = set()
    coupons: list[PromoCode] = []
    for coupon in list(mall_coupons) + list(own_coupons) + list(activity_coupons):
        if coupon.id in seen:
            continue
        seen.add(coupon.id)

        if coupon.max_uses > 0 and coupon.used_count >= coupon.max_uses:
            continue
        if coupon.max_uses_per_user > 0:
            used_count = await _user_used_count(db, coupon.id, user_id)
            if used_count >= coupon.max_uses_per_user:
                continue
        if coupon.new_users_only and await _has_prior_paid_order(db, user_id):
            continue
        if getattr(coupon, "issuer_type", None) == "agent" and int(coupon.issuer_user_id) == user_id:
            continue
        coupons.append(coupon)

    return coupons


async def calculate_cart_coupon(
    db: AsyncSession,
    *,
    user_id: int,
    cart_items: list[CartItem],
    products: dict[int, Product],
    item_totals: dict[int, int],
    code: str | None,
    target_cart_item_id: int | None = None,
    for_update: bool = False,
) -> CouponApplication | None:
    normalized = normalize_coupon_code(code)
    if not normalized:
        return None

    coupon = await _load_coupon(db, normalized, for_update=for_update)
    if not coupon.is_enabled:
        raise HTTPException(status_code=400, detail="优惠券已停用")

    now = datetime.now(timezone.utc)
    starts_at = _as_aware(coupon.starts_at)
    expires_at = _as_aware(coupon.expires_at)
    if starts_at and starts_at > now:
        raise HTTPException(status_code=400, detail="优惠券尚未开始")
    if expires_at and expires_at < now:
        raise HTTPException(status_code=400, detail="优惠券已过期")

    await _validate_user_can_use_coupons(db, user_id)

    if coupon.max_uses > 0 and coupon.used_count >= coupon.max_uses:
        raise HTTPException(status_code=400, detail="优惠券已被领用完")
    if coupon.max_uses_per_user > 0:
        used_count = await _user_used_count(db, coupon.id, user_id)
        if used_count >= coupon.max_uses_per_user:
            raise HTTPException(status_code=400, detail="该账号已达到优惠券使用次数上限")
    if coupon.new_users_only and await _has_prior_paid_order(db, user_id):
        raise HTTPException(status_code=400, detail="该优惠券仅限新用户首单使用")
    await _validate_agent_coupon_owner(db, coupon, user_id)
    mall_claim = await _validate_mall_claim(db, coupon, user_id)

    await _validate_advanced_rules(
        db,
        coupon,
        user_id=user_id,
        cart_items=cart_items,
        products=products,
    )

    order_amount = sum(max(0, int(value)) for value in item_totals.values())
    if coupon.min_order_amount > 0 and order_amount < coupon.min_order_amount:
        raise HTTPException(
            status_code=400,
            detail=f"订单金额需满 ¥{coupon.min_order_amount / 100:.2f} 才可使用该优惠券",
        )

    matched_ids = await _matched_item_ids(db, coupon, cart_items, products)
    if not matched_ids:
        raise HTTPException(status_code=400, detail="当前购物车中没有可使用该优惠券的商品")

    discount_level = _enum_value(getattr(coupon, "discount_level", PromoCodeDiscountLevel.ORDER.value))
    target_product_id: int | None = None
    if discount_level == PromoCodeDiscountLevel.PRODUCT.value:
        if target_cart_item_id is None:
            raise HTTPException(status_code=400, detail="请先选择要使用产品级优惠券的商品")
        target_cart_item_id = int(target_cart_item_id)
        if target_cart_item_id not in item_totals:
            raise HTTPException(status_code=400, detail="所选商品不在当前购物车中")
        if target_cart_item_id not in matched_ids:
            raise HTTPException(status_code=400, detail="所选商品不符合该优惠券的使用范围")
        eligible_amount = max(0, int(item_totals.get(target_cart_item_id, 0)))
        target_item = next((item for item in cart_items if int(item.id) == target_cart_item_id), None)
        target_product_id = int(target_item.product_id) if target_item else None
        discount = min(eligible_amount, _discount_amount(coupon, eligible_amount))
        item_discounts = {target_cart_item_id: discount} if discount > 0 else {}
    elif discount_level == PromoCodeDiscountLevel.ORDER.value:
        eligible_amount = order_amount
        discount = min(eligible_amount, _discount_amount(coupon, eligible_amount))
        item_discounts = _order_item_discounts(item_totals, discount)
    else:
        raise HTTPException(status_code=400, detail="优惠券层级不支持")

    if discount <= 0:
        raise HTTPException(status_code=400, detail="该优惠券未产生可用优惠")

    return CouponApplication(
        promo_code=coupon,
        code=coupon.code,
        order_amount=order_amount,
        eligible_amount=eligible_amount,
        discount_amount=discount,
        type=_enum_value(coupon.type),
        value=int(coupon.value or 0),
        description=coupon.description,
        discount_level=discount_level,
        target_cart_item_id=target_cart_item_id if discount_level == PromoCodeDiscountLevel.PRODUCT.value else None,
        target_product_id=target_product_id,
        item_discounts=item_discounts,
    )


async def record_coupon_usage(
    db: AsyncSession,
    *,
    application: CouponApplication,
    user_id: int,
    order_id: int,
) -> PromoCodeUsage:
    usage = PromoCodeUsage(
        promo_code_id=application.promo_code.id,
        code_snapshot=application.code,
        user_id=user_id,
        order_id=order_id,
        order_amount=application.order_amount,
        eligible_amount=application.eligible_amount,
        discount_amount=application.discount_amount,
        status="used",
        used_at=datetime.now(timezone.utc),
        payload_snapshot={
            "type": application.type,
            "value": application.value,
            "description": application.description,
            "issuer_type": getattr(application.promo_code, "issuer_type", None),
            "issuer_user_id": getattr(application.promo_code, "issuer_user_id", None),
            "discount_level": application.discount_level,
            "target_cart_item_id": application.target_cart_item_id,
            "target_product_id": application.target_product_id,
            "item_discounts": application.item_discounts,
            "max_discount_amount": int(getattr(application.promo_code, "max_discount_amount", 0) or 0),
            "min_quantity": int(getattr(application.promo_code, "min_quantity", 0) or 0),
            "required_product_ids": list(_parse_int_set(getattr(application.promo_code, "required_product_ids", None))),
            "required_group_ids": list(_parse_int_set(getattr(application.promo_code, "required_group_ids", None))),
            "required_category_ids": list(_parse_int_set(getattr(application.promo_code, "required_category_ids", None))),
            "allowed_user_ids": list(_parse_int_set(getattr(application.promo_code, "allowed_user_ids", None))),
        },
    )
    application.promo_code.used_count = int(application.promo_code.used_count or 0) + 1
    db.add(usage)

    # 商城领取的券标记为已使用
    claim_method = _enum_value(getattr(application.promo_code, "claim_method", PromoCodeClaimMethod.CODE.value))
    if claim_method == PromoCodeClaimMethod.MALL.value:
        claim = await db.scalar(
            select(CouponMallClaim).where(
                CouponMallClaim.promo_code_id == application.promo_code.id,
                CouponMallClaim.user_id == user_id,
                CouponMallClaim.status == CouponMallClaimStatus.CLAIMED,
            )
        )
        if claim:
            claim.status = CouponMallClaimStatus.USED
            claim.used_at = datetime.now(timezone.utc)
            claim.order_id = order_id

    # 活动领取的券标记为已使用（取最早一条 claimed 记录）
    activity_claim = await db.scalar(
        select(CouponActivityClaim).where(
            CouponActivityClaim.promo_code_id == application.promo_code.id,
            CouponActivityClaim.user_id == user_id,
            CouponActivityClaim.status == CouponActivityClaimStatus.CLAIMED,
        ).order_by(CouponActivityClaim.id.asc())
    )
    if activity_claim:
        activity_claim.status = CouponActivityClaimStatus.USED
        activity_claim.used_at = datetime.now(timezone.utc)
        activity_claim.order_id = order_id

    if getattr(application.promo_code, "issuer_type", None) == "agent" and getattr(application.promo_code, "issuer_user_id", None):
        customer = await db.get(User, user_id)
        if customer and not customer.bound_agent_id:
            from app.services import agent_service

            await agent_service.bind_customer_to_agent(
                db,
                agent_user_id=int(application.promo_code.issuer_user_id),
                customer_user_id=user_id,
                bind_source="agent_coupon",
            )

    await db.flush()
    return usage


def coupon_application_to_dict(application: CouponApplication, *, final_amount: int | None = None) -> dict[str, Any]:
    data: dict[str, Any] = {
        "description": application.description,
        "issuer_type": getattr(application.promo_code, "issuer_type", None),
        "issuer_user_id": getattr(application.promo_code, "issuer_user_id", None),
        "type": application.type,
        "value": application.value,
        "discount_level": application.discount_level,
        "target_cart_item_id": application.target_cart_item_id,
        "target_product_id": application.target_product_id,
        "order_amount": application.order_amount,
        "eligible_amount": application.eligible_amount,
        "discount_amount": application.discount_amount,
        "item_discounts": application.item_discounts,
        "max_discount_amount": int(getattr(application.promo_code, "max_discount_amount", 0) or 0),
        "min_quantity": int(getattr(application.promo_code, "min_quantity", 0) or 0),
    }
    if final_amount is not None:
        data["final_amount"] = final_amount
    return data
