from datetime import datetime, timezone

from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.member_level import MemberLevel
from app.models.point_log import PointChangeType, PointLog
from app.models.point_reward import PointReward, PointRewardStatus
from app.models.user_membership import UserMembership


# 积分兑换比例：1000积分 = 1元 = 100分
POINTS_TO_CENTS_RATE = 1000  # 1000积分 = 100分（1元）
POINTS_REWARD_RATE = 100  # 1元消费 = 100积分


async def get_or_create_membership(db: AsyncSession, user_id: int) -> UserMembership:
    """获取或创建用户会员信息"""
    result = await db.execute(select(UserMembership).where(UserMembership.user_id == user_id))
    membership = result.scalar_one_or_none()
    if membership is None:
        # 查找默认等级（level=0）
        result = await db.execute(
            select(MemberLevel).where(MemberLevel.level == 0, MemberLevel.enabled == True)
        )
        default_level = result.scalar_one_or_none()

        membership = UserMembership(
            user_id=user_id,
            member_level_id=default_level.id if default_level else None,
            total_spent=0,
            total_points=0,
            available_points=0,
        )
        db.add(membership)
        await db.flush()
        await db.refresh(membership)
    return membership


async def get_member_levels(db: AsyncSession) -> list[MemberLevel]:
    """获取所有启用的会员等级，按 level 排序"""
    result = await db.execute(
        select(MemberLevel).where(MemberLevel.enabled == True).order_by(MemberLevel.level)
    )
    return list(result.scalars().all())


async def add_points(
    db: AsyncSession,
    user_id: int,
    delta: int,
    change_type: PointChangeType,
    order_id: int | None = None,
    ref_no: str | None = None,
    remark: str | None = None,
    operator_id: int | None = None,
    allow_negative: bool = False,
) -> PointLog:
    """增加/扣除用户积分，并写流水

    Args:
        delta: 正数为增加，负数为扣除
    """
    membership = await get_or_create_membership(db, user_id)

    # 检查积分是否足够（扣除时）
    if delta < 0 and not allow_negative and membership.available_points + delta < 0:
        raise ValueError("积分不足")

    new_balance = membership.available_points + delta

    # 更新会员积分
    membership.available_points = new_balance
    if delta > 0:
        membership.total_points += delta

    # 写流水
    log = PointLog(
        user_id=user_id,
        change_type=change_type,
        delta=delta,
        balance_after=new_balance,
        order_id=order_id,
        ref_no=ref_no,
        remark=remark,
        operator_id=operator_id,
    )
    db.add(log)
    await db.flush()
    await db.refresh(log)
    return log


async def calculate_self_purchase_reward(
    db: AsyncSession,
    user_id: int,
    amount_cents: int,
) -> tuple[int, int, int]:
    """Return (points, cashback_rate, reward_cents) for self-purchase reward."""
    membership = await get_or_create_membership(db, user_id)

    # 自购返积分按固定返现比例计算，不叠加积分获取倍率。
    cashback_rate = 0
    if membership.member_level_id:
        result = await db.execute(
            select(MemberLevel).where(MemberLevel.id == membership.member_level_id)
        )
        level = result.scalar_one_or_none()
        if level:
            cashback_rate = level.points_cashback_rate

    if cashback_rate <= 0 or amount_cents <= 0:
        return 0, cashback_rate, 0

    reward_cents = int(amount_cents * cashback_rate / 10000)
    final_points = reward_cents * 10

    return final_points, cashback_rate, reward_cents


async def create_self_purchase_point_reward(
    db: AsyncSession,
    *,
    user_id: int,
    order_id: int,
    ref_no: str,
    amount_cents: int,
    source_type: str,
    instance_id: int | None = None,
    status: str = PointRewardStatus.PENDING.value,
    remark: str | None = None,
    payload_snapshot: dict | None = None,
) -> PointReward | None:
    """Create a claimable self-purchase reward record without adding usable points."""
    existing = await db.execute(
        select(PointReward).where(
            PointReward.order_id == order_id,
            PointReward.instance_id.is_(None) if instance_id is None else PointReward.instance_id == instance_id,
            PointReward.source_type == source_type,
        )
    )
    reward = existing.scalar_one_or_none()
    if reward:
        return reward

    points, cashback_rate, reward_cents = await calculate_self_purchase_reward(db, user_id, amount_cents)
    if points <= 0:
        return None

    reward = PointReward(
        user_id=user_id,
        order_id=order_id,
        instance_id=instance_id,
        source_type=source_type,
        status=status,
        amount_cents=amount_cents,
        reward_rate=cashback_rate,
        points=points,
        ref_no=ref_no,
        remark=remark or f"自购积分返现，比例{cashback_rate / 100:.1f}%，可领取{points}积分",
        payload_snapshot={
            "reward_cents": reward_cents,
            **(payload_snapshot or {}),
        },
    )
    db.add(reward)
    await db.flush()
    return reward


async def create_self_purchase_rewards_for_instances(
    db: AsyncSession,
    *,
    user_id: int,
    order_id: int,
    ref_no: str,
    instance_amounts: list[tuple[int, int]],
    source_type: str,
    status: str = PointRewardStatus.PENDING.value,
) -> list[PointReward]:
    rewards: list[PointReward] = []
    for instance_id, amount_cents in instance_amounts:
        reward = await create_self_purchase_point_reward(
            db,
            user_id=user_id,
            order_id=order_id,
            ref_no=ref_no,
            amount_cents=amount_cents,
            source_type=source_type,
            instance_id=instance_id,
            status=status,
            payload_snapshot={"instance_id": instance_id},
        )
        if reward:
            rewards.append(reward)
    return rewards


async def reward_points_for_order(
    db: AsyncSession,
    user_id: int,
    order_amount_cents: int,
    order_id: int,
    order_no: str,
) -> PointLog | None:
    """Legacy immediate self-purchase point reward. New flows should use point_rewards."""
    final_points, cashback_rate, _reward_cents = await calculate_self_purchase_reward(db, user_id, order_amount_cents)
    if final_points <= 0:
        return None

    return await add_points(
        db=db,
        user_id=user_id,
        delta=final_points,
        change_type=PointChangeType.ORDER_REWARD,
        order_id=order_id,
        ref_no=order_no,
        remark=f"订单自购积分返现，比例{cashback_rate / 100:.1f}%，奖励{final_points}积分",
    )


def point_reward_to_dict(reward: PointReward) -> dict:
    status_label = {
        PointRewardStatus.PENDING.value: "待开通",
        PointRewardStatus.CLAIMABLE.value: "可领取",
        PointRewardStatus.CLAIMED.value: "已领取",
        PointRewardStatus.CANCELLED.value: "已取消",
        PointRewardStatus.REVERSED.value: "已扣回",
    }.get(reward.status, reward.status)
    return {
        "id": reward.id,
        "order_id": reward.order_id,
        "instance_id": reward.instance_id,
        "source_type": reward.source_type,
        "status": reward.status,
        "status_label": status_label,
        "amount_cents": reward.amount_cents,
        "reward_rate": reward.reward_rate,
        "points": reward.points,
        "ref_no": reward.ref_no,
        "remark": reward.remark,
        "created_at": reward.created_at.isoformat() if reward.created_at else None,
        "claimed_at": reward.claimed_at.isoformat() if reward.claimed_at else None,
    }


async def get_unclaimed_instance_point_rewards(
    db: AsyncSession,
    *,
    user_id: int,
    instance_id: int,
) -> list[PointReward]:
    result = await db.execute(
        select(PointReward)
        .where(
            PointReward.user_id == user_id,
            PointReward.instance_id == instance_id,
            PointReward.status.in_([
                PointRewardStatus.PENDING.value,
                PointRewardStatus.CLAIMABLE.value,
            ]),
        )
        .order_by(PointReward.id.asc())
    )
    return list(result.scalars().all())


async def mark_instance_point_rewards_claimable(
    db: AsyncSession,
    *,
    instance_id: int,
    order_id: int | None = None,
) -> int:
    stmt = select(PointReward).where(
        PointReward.instance_id == instance_id,
        PointReward.status == PointRewardStatus.PENDING.value,
    )
    if order_id is not None:
        stmt = stmt.where(PointReward.order_id == order_id)
    result = await db.execute(stmt)
    rewards = list(result.scalars().all())
    for reward in rewards:
        reward.status = PointRewardStatus.CLAIMABLE.value
    if rewards:
        await db.flush()
    return len(rewards)


async def claim_instance_point_rewards(
    db: AsyncSession,
    *,
    user_id: int,
    instance_id: int,
) -> dict:
    result = await db.execute(
        select(PointReward)
        .where(
            PointReward.user_id == user_id,
            PointReward.instance_id == instance_id,
            PointReward.status.in_([
                PointRewardStatus.PENDING.value,
                PointRewardStatus.CLAIMABLE.value,
            ]),
        )
        .order_by(PointReward.id.asc())
        .with_for_update()
    )
    rewards = list(result.scalars().all())
    if not rewards:
        return {"claimed_points": 0, "claimed_count": 0, "pending_count": 0}

    pending_count = sum(1 for reward in rewards if reward.status == PointRewardStatus.PENDING.value)
    claimable = [reward for reward in rewards if reward.status == PointRewardStatus.CLAIMABLE.value]
    if not claimable:
        return {"claimed_points": 0, "claimed_count": 0, "pending_count": pending_count}

    claimed_points = 0
    now = datetime.now(timezone.utc)
    for reward in claimable:
        log = await add_points(
            db=db,
            user_id=user_id,
            delta=reward.points,
            change_type=PointChangeType.ORDER_REWARD,
            order_id=reward.order_id,
            ref_no=reward.ref_no,
            remark=f"领取自购返积分：{reward.points}积分",
        )
        reward.status = PointRewardStatus.CLAIMED.value
        reward.point_log_id = log.id
        reward.claimed_at = now
        claimed_points += reward.points

        from app.models.order import Order

        order = await db.get(Order, reward.order_id)
        if order:
            order.points_claimed = True
            order.points_claimed_at = order.points_claimed_at or now
            order.points_claimed_amount = (order.points_claimed_amount or 0) + reward.points

    await db.flush()
    return {
        "claimed_points": claimed_points,
        "claimed_count": len(claimable),
        "pending_count": pending_count,
    }


async def reverse_order_point_rewards(
    db: AsyncSession,
    *,
    order_id: int,
    reason: str,
) -> dict:
    """Cancel unclaimed rewards and claw back claimed self-purchase rewards for an order."""
    result = await db.execute(
        select(PointReward)
        .where(
            PointReward.order_id == order_id,
            PointReward.status.in_([
                PointRewardStatus.PENDING.value,
                PointRewardStatus.CLAIMABLE.value,
                PointRewardStatus.CLAIMED.value,
            ]),
        )
        .with_for_update()
    )
    rewards = list(result.scalars().all())
    now = datetime.now(timezone.utc)
    cancelled = 0
    reversed_points = 0
    for reward in rewards:
        if reward.status in (PointRewardStatus.PENDING.value, PointRewardStatus.CLAIMABLE.value):
            reward.status = PointRewardStatus.CANCELLED.value
            reward.cancelled_at = now
            reward.remark = reason
            cancelled += 1
            continue

        if reward.status == PointRewardStatus.CLAIMED.value:
            log = await add_points(
                db=db,
                user_id=reward.user_id,
                delta=-reward.points,
                change_type=PointChangeType.ORDER_REFUND,
                order_id=reward.order_id,
                ref_no=reward.ref_no,
                remark=f"自购返积分扣回：{reason[:120]}",
                allow_negative=True,
            )
            reward.status = PointRewardStatus.REVERSED.value
            reward.reverse_point_log_id = log.id
            reward.reversed_at = now
            reward.remark = reason
            reversed_points += reward.points

    bonus_reverse = await reverse_order_immediate_point_logs(
        db,
        order_id=order_id,
        reason=reason,
    )

    if rewards:
        await db.flush()
    return {
        "cancelled_count": cancelled,
        "reversed_points": reversed_points,
        **bonus_reverse,
    }


async def cancel_unclaimed_order_point_rewards(
    db: AsyncSession,
    *,
    order_id: int,
    reason: str,
) -> dict:
    """Cancel unclaimed self-purchase rewards without clawing back claimed points."""
    result = await db.execute(
        select(PointReward)
        .where(
            PointReward.order_id == order_id,
            PointReward.status.in_([
                PointRewardStatus.PENDING.value,
                PointRewardStatus.CLAIMABLE.value,
            ]),
        )
        .with_for_update()
    )
    rewards = list(result.scalars().all())
    now = datetime.now(timezone.utc)
    for reward in rewards:
        reward.status = PointRewardStatus.CANCELLED.value
        reward.cancelled_at = now
        reward.remark = reason

    if rewards:
        await db.flush()
    return {"cancelled_count": len(rewards)}


async def refund_used_points_for_order(
    db: AsyncSession,
    *,
    user_id: int,
    order_id: int,
    order_no: str,
    ratio_bps: int,
    reason: str,
) -> PointLog | None:
    """Return points used as order discount according to refund ratio."""
    ratio_bps = max(0, min(10000, ratio_bps))
    if ratio_bps <= 0:
        return None

    existing = await db.execute(
        select(PointLog).where(
            PointLog.user_id == user_id,
            PointLog.order_id == order_id,
            PointLog.change_type == PointChangeType.POINTS_REFUND,
        )
    )
    if existing.scalar_one_or_none():
        return None

    result = await db.execute(
        select(PointLog).where(
            PointLog.user_id == user_id,
            PointLog.order_id == order_id,
            PointLog.change_type == PointChangeType.POINTS_USE,
        )
    )
    used_points = sum(max(0, -int(log.delta or 0)) for log in result.scalars().all())
    refund_points = (used_points * ratio_bps) // 10000
    if refund_points <= 0:
        return None

    return await add_points(
        db=db,
        user_id=user_id,
        delta=refund_points,
        change_type=PointChangeType.POINTS_REFUND,
        order_id=order_id,
        ref_no=f"{order_no}:points_refund",
        remark=f"订单退订返还积分抵扣：{reason[:120]}",
    )


async def reverse_order_immediate_point_logs(
    db: AsyncSession,
    *,
    order_id: int,
    reason: str,
) -> dict:
    """Reverse point logs that were already credited directly for an order."""
    from app.models.agent_settlement import AgentSettlement

    rewarded_log_ids = await db.execute(
        select(PointReward.point_log_id).where(
            PointReward.order_id == order_id,
            PointReward.point_log_id.is_not(None),
        )
    )
    claimed_reward_log_ids = {int(log_id) for log_id in rewarded_log_ids.scalars().all() if log_id}
    settlement_log_ids_res = await db.execute(
        select(AgentSettlement.point_log_id).where(
            AgentSettlement.order_id == order_id,
            AgentSettlement.point_log_id.is_not(None),
        )
    )
    settlement_log_ids = {int(log_id) for log_id in settlement_log_ids_res.scalars().all() if log_id}

    result = await db.execute(
        select(PointLog).where(
            PointLog.order_id == order_id,
            PointLog.delta > 0,
            PointLog.change_type.in_([
                PointChangeType.ORDER_REWARD.value,
                PointChangeType.REFERRAL_REWARD.value,
                PointChangeType.AGENT_PROFIT.value,
            ]),
        )
    )
    logs = list(result.scalars().all())
    reversed_points = 0
    reversed_count = 0
    for log in logs:
        if log.id in claimed_reward_log_ids or log.id in settlement_log_ids:
            continue
        reverse_ref = f"{log.ref_no or order_id}:reverse:{log.id}"
        existing = await db.execute(
            select(PointLog).where(
                PointLog.ref_no == reverse_ref,
                PointLog.change_type == PointChangeType.ORDER_REFUND,
            )
        )
        if existing.scalar_one_or_none():
            continue
        await add_points(
            db=db,
            user_id=log.user_id,
            delta=-log.delta,
            change_type=PointChangeType.ORDER_REFUND,
            order_id=order_id,
            ref_no=reverse_ref,
            remark=f"订单奖励积分扣回：{reason[:120]}",
            allow_negative=True,
        )
        reversed_points += log.delta
        reversed_count += 1

    return {
        "reversed_bonus_points": reversed_points,
        "reversed_bonus_count": reversed_count,
    }


async def reverse_instance_point_rewards(
    db: AsyncSession,
    *,
    instance_id: int,
    reason: str,
) -> dict:
    """Cancel or claw back rewards tied to a single server instance."""
    result = await db.execute(
        select(PointReward)
        .where(
            PointReward.instance_id == instance_id,
            PointReward.status.in_([
                PointRewardStatus.PENDING.value,
                PointRewardStatus.CLAIMABLE.value,
                PointRewardStatus.CLAIMED.value,
            ]),
        )
        .with_for_update()
    )
    rewards = list(result.scalars().all())
    now = datetime.now(timezone.utc)
    cancelled = 0
    reversed_points = 0
    for reward in rewards:
        if reward.status in (PointRewardStatus.PENDING.value, PointRewardStatus.CLAIMABLE.value):
            reward.status = PointRewardStatus.CANCELLED.value
            reward.cancelled_at = now
            reward.remark = reason
            cancelled += 1
            continue

        if reward.status == PointRewardStatus.CLAIMED.value:
            log = await add_points(
                db=db,
                user_id=reward.user_id,
                delta=-reward.points,
                change_type=PointChangeType.ORDER_REFUND,
                order_id=reward.order_id,
                ref_no=reward.ref_no,
                remark=f"服务器积分扣回：{reason[:120]}",
                allow_negative=True,
            )
            reward.status = PointRewardStatus.REVERSED.value
            reward.reverse_point_log_id = log.id
            reward.reversed_at = now
            reward.remark = reason
            reversed_points += reward.points

    if rewards:
        await db.flush()
    return {
        "cancelled_count": cancelled,
        "reversed_points": reversed_points,
    }


def current_quarter_start(now: datetime | None = None) -> datetime:
    now = now or datetime.now()
    quarter_month = ((now.month - 1) // 3) * 3 + 1
    return datetime(now.year, quarter_month, 1)


async def get_current_quarter_self_spent(db: AsyncSession, user_id: int) -> int:
    from app.models.spend_event import SpendEventTypeConfig, UserSpendEvent

    result = await db.execute(
        select(func.coalesce(func.sum(UserSpendEvent.amount_cents), 0))
        .join(SpendEventTypeConfig, SpendEventTypeConfig.code == UserSpendEvent.event_type)
        .where(
            UserSpendEvent.user_id == user_id,
            UserSpendEvent.created_at >= current_quarter_start(),
            SpendEventTypeConfig.enabled.is_(True),
        )
    )
    return max(0, int(result.scalar_one() or 0))


async def reward_referrer_for_order(db: AsyncSession, order):
    """服务开通成功后给普通推广者创建冻结推广奖励。

    代理身份激活的推广者走代理差价结算，不叠加普通推广奖励。
    """
    from app.models.agent_settlement import AgentSettlement
    from app.models.user import User
    from app.models.user_agent import AgentStatus, UserAgent
    from app.services import agent_service as _agent_svc

    existing = await db.execute(
        select(PointLog).where(
            PointLog.order_id == order.id,
            PointLog.change_type == PointChangeType.REFERRAL_REWARD,
        )
    )
    if existing.scalar_one_or_none():
        return None

    existing_settlement = await db.execute(
        select(AgentSettlement).where(
            AgentSettlement.order_id == order.id,
            AgentSettlement.settlement_type == "member_commission",
        )
    )
    if existing_settlement.scalar_one_or_none():
        return None

    customer_res = await db.execute(select(User).where(User.id == order.user_id))
    customer = customer_res.scalar_one_or_none()
    if not customer or not customer.bound_agent_id or customer.bound_agent_id == customer.id:
        return None

    active_agent_res = await db.execute(
        select(UserAgent).where(
            UserAgent.user_id == customer.bound_agent_id,
            UserAgent.status.in_([AgentStatus.ACTIVE, AgentStatus.OBSERVING]),
        )
    )
    if active_agent_res.scalar_one_or_none():
        return None

    referrer_membership = await get_or_create_membership(db, customer.bound_agent_id)
    member_level = referrer_membership.member_level
    rate = member_level.commission_rate if member_level else 0
    if rate <= 0:
        return None

    pay_amount_cents = max(0, order.total_amount - (order.discount_amount or 0))
    reward_cents = int(pay_amount_cents * rate / 10000)
    points = reward_cents * 10
    if points <= 0:
        return None

    order.member_commission_rate = rate
    order.member_commission_amount = reward_cents

    settlement = AgentSettlement(
        order_id=order.id,
        order_no=order.out_order_no,
        settlement_type="member_commission",
        user_id=customer.bound_agent_id,
        customer_user_id=customer.id,
        standard_amount=order.total_amount,
        customer_pay_amount=pay_amount_cents,
        commission_rate=rate,
        delta_amount=reward_cents,
        points_amount=points,
        status="pending",
        remark=f"普通推广奖励，实付{pay_amount_cents}分，比例{rate / 100:.1f}%，预计{points}积分",
    )
    db.add(settlement)
    order.settlement_status = "pending"
    await db.flush()
    await _agent_svc.settle_order(db, order)
    return settlement


def calculate_points_discount(available_points: int, order_amount_cents: int, max_rate: int = 3500) -> tuple[int, int]:
    """计算积分可抵扣金额

    Returns:
        (可用积分, 抵扣金额分)
    """
    available_points = max(0, available_points)
    # 最大可抵扣比例
    max_discount_cents = int(order_amount_cents * max_rate / 10000)
    # 积分可抵扣金额
    max_points_discount_cents = (available_points // POINTS_TO_CENTS_RATE) * 100

    actual_discount_cents = min(max_discount_cents, max_points_discount_cents)
    actual_points = (actual_discount_cents // 100) * POINTS_TO_CENTS_RATE

    return actual_points, actual_discount_cents


async def use_points_for_order(
    db: AsyncSession,
    user_id: int,
    points: int,
    order_id: int,
    order_no: str,
) -> PointLog:
    """订单使用积分抵扣"""
    return await add_points(
        db=db,
        user_id=user_id,
        delta=-points,
        change_type=PointChangeType.POINTS_USE,
        order_id=order_id,
        ref_no=order_no,
        remark=f"订单积分抵扣 {points} 积分",
    )


async def add_total_spent(db: AsyncSession, user_id: int, amount_cents: int) -> UserMembership:
    """增加累计消费金额，并检查是否升级"""
    membership = await get_or_create_membership(db, user_id)
    membership.total_spent += amount_cents

    # 检查升级
    if not membership.member_level_locked:
        result = await db.execute(
            select(MemberLevel)
            .where(
                MemberLevel.enabled == True,
                MemberLevel.min_total_spent <= membership.total_spent,
            )
            .order_by(MemberLevel.level.desc())
        )
        new_level = result.scalars().first()

        if new_level and (membership.member_level_id != new_level.id):
            from app.services import notification_service

            membership.member_level_id = new_level.id
            membership.upgraded_at = datetime.now()
            await notification_service.safe_notify(
                db,
                user_id=user_id,
                category="membership",
                title="会员等级提升",
                content=f"恭喜！您的会员等级已提升至 LV{new_level.level} {new_level.name}。",
                link="/membership",
            )

    await db.flush()
    await db.refresh(membership)
    return membership


async def get_point_logs(
    db: AsyncSession,
    user_id: int,
    limit: int = 20,
    offset: int = 0,
) -> list[PointLog]:
    """获取用户积分流水"""
    result = await db.execute(
        select(PointLog)
        .where(PointLog.user_id == user_id)
        .order_by(PointLog.created_at.desc())
        .limit(limit)
        .offset(offset)
    )
    return list(result.scalars().all())
