"""激励服务：资产钱包 / 推荐绑定 / 提现（对应《03-接口文档》第 9 章、《02-技术文档》6.6）"""
from datetime import datetime, timedelta
from decimal import Decimal

import redis.asyncio as aioredis
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import biz_config
from app.core import errors
from app.core.response import BizException
from app.models.incentive import AssetTransaction, AssetWallet, Referral, Withdrawal
from app.models.user import User
from app.services import config_service, membership_service

# 资产流水业务类型（与 asset_transactions.biz_type 注释一致）
ASSET_BIZ_REBATE = 1        # 充值返利
ASSET_BIZ_COMPLETE = 2      # 完整观看收益
ASSET_BIZ_LIKE = 3          # 点赞收益
ASSET_BIZ_WITHDRAW = 4      # 提现
ASSET_BIZ_WITHDRAW_BACK = 5 # 提现退回
ASSET_BIZ_CLAWBACK = 6      # 返利扣回
ASSET_BIZ_ADJUST = 7        # 后台调整

# 提现单在途锁兜底过期（正常由后台审核后释放）
WD_LOCK_TTL = 7 * 24 * 3600
# 补绑邀请码时限：注册后 24h 内
BIND_WINDOW_HOURS = 24

REFERRAL_STATUS_TEXT = {0: "待激活", 1: "有效", 2: "无效", 3: "风控冻结"}
WITHDRAW_STATUS_TEXT = {0: "待审核", 1: "已打款", 2: "已拒绝"}


def _fmt(dt: datetime | None) -> str | None:
    return dt.strftime("%Y-%m-%d %H:%M:%S") if dt else None


def _asset_cfg() -> dict:
    return biz_config.get("asset") or {}


def _withdraw_cfg() -> dict:
    return biz_config.get("withdraw") or {}


# ---------- 资产钱包变更（与积分钱包同一套"行锁 + 流水 + 乐观锁"规范） ----------

async def get_or_create_asset_wallet(db: AsyncSession, user_id: int, for_update: bool = False) -> AssetWallet:
    stmt = select(AssetWallet).where(AssetWallet.user_id == user_id)
    if for_update:
        stmt = stmt.with_for_update()
    wallet = (await db.execute(stmt)).scalar_one_or_none()
    if wallet is None:
        wallet = AssetWallet(user_id=user_id)
        db.add(wallet)
        await db.flush()
        if for_update:
            wallet = (await db.execute(stmt)).scalar_one()
    return wallet


async def change_assets(
    db: AsyncSession,
    user_id: int,
    amount: Decimal,
    biz_type: int,
    biz_id: str = "",
    remark: str = "",
    allow_negative: bool = False,
) -> Decimal:
    """变更资产余额，返回变动后余额。

    - amount 正增负减；扣减默认不允许透支（退款/返利扣回场景传 allow_negative=True）
    - 同事务内：行锁 → 校验/更新余额与累计 → 写流水
    """
    wallet = await get_or_create_asset_wallet(db, user_id, for_update=True)
    new_balance = wallet.assets + amount
    if amount < 0 and new_balance < 0 and not allow_negative:
        raise BizException(errors.ASSET_NOT_ENOUGH)

    wallet.assets = new_balance
    wallet.version += 1
    if amount > 0:
        wallet.total_earned += amount
    elif biz_type == ASSET_BIZ_WITHDRAW:
        wallet.total_withdrawn += -amount

    db.add(
        AssetTransaction(
            user_id=user_id,
            change_amount=amount,
            balance_after=new_balance,
            biz_type=biz_type,
            biz_id=biz_id,
            remark=remark,
        )
    )
    await db.flush()
    return new_balance


# ---------- 推荐 ----------

async def get_my_referral(db: AsyncSession, redis: aioredis.Redis, user_id: int) -> dict:
    """我的邀请（9.1）：邀请码 + 统计 + 当前会员身份（eligible）"""
    user = await db.get(User, user_id)
    if user is None:
        raise BizException(errors.UNAUTHORIZED, http_status=401)

    base_url = await config_service.get_config(db, redis, "app.invite_base_url", "https://app.example.com/r/")
    total_invitees = (await db.execute(
        select(func.count(Referral.id)).where(Referral.inviter_id == user_id)
    )).scalar() or 0
    activated_count = (await db.execute(
        select(func.count(Referral.id)).where(
            Referral.inviter_id == user_id, Referral.status == 1
        )
    )).scalar() or 0
    # 邀请积分奖励累计（point_transactions biz_type=8）
    from app.models.billing import PointTransaction

    reward_points_total = (await db.execute(
        select(func.coalesce(func.sum(PointTransaction.change_amount), 0)).where(
            PointTransaction.user_id == user_id, PointTransaction.biz_type == 8
        )
    )).scalar() or 0
    # 充值返利资产累计（asset_transactions biz_type=1）
    rebate_assets_total = (await db.execute(
        select(func.coalesce(func.sum(AssetTransaction.change_amount), 0)).where(
            AssetTransaction.user_id == user_id, AssetTransaction.biz_type == ASSET_BIZ_REBATE
        )
    )).scalar() or 0

    return {
        "invite_code": user.invite_code,
        "invite_url": f"{base_url}{user.invite_code}",
        "eligible": await membership_service.is_member(db, user_id),
        "total_invitees": total_invitees,
        "activated_count": activated_count,
        "reward_points_total": int(reward_points_total),
        "rebate_assets_total": float(rebate_assets_total),
    }


async def bind_invite_code(
    db: AsyncSession, redis: aioredis.Redis, user_id: int, invite_code: str, device_id: str | None, ip: str
) -> dict:
    """补绑邀请码（9.2）：注册 24h 内仅一次；无效码/已绑定/绑自己/同设备均 2101"""
    user = await db.get(User, user_id)
    if user is None:
        raise BizException(errors.UNAUTHORIZED, http_status=401)

    bound = (await db.execute(
        select(Referral.id).where(Referral.invitee_id == user_id)
    )).first()
    if bound:
        raise BizException(errors.INVITE_CODE_INVALID, "已绑定过邀请码")
    if user.created_at < datetime.now() - timedelta(hours=BIND_WINDOW_HOURS):
        raise BizException(errors.INVITE_CODE_INVALID, "超过绑定时限（注册后24小时内可绑定）")

    inviter = (await db.execute(
        select(User).where(User.invite_code == invite_code)
    )).scalar_one_or_none()
    if inviter is None:
        raise BizException(errors.INVITE_CODE_INVALID, "邀请码无效")
    if inviter.id == user_id:
        raise BizException(errors.INVITE_CODE_INVALID, "不能绑定自己的邀请码")
    if device_id and await redis.exists(f"ref:device:{device_id}"):
        old = int(await redis.get(f"ref:device:{device_id}") or 0)
        if old and old != user_id:
            raise BizException(errors.INVITE_CODE_INVALID, "当前设备已注册其他账号")

    # 风控规则与注册时一致（邀请人单日超限冻结 / 同 IP 批量注册标记）
    status, risk_flag = 0, ""
    risk_key = f"ref:risk:{inviter.id}:{datetime.now():%Y%m%d}"
    bind_count = await redis.incr(risk_key)
    await redis.expire(risk_key, 48 * 3600)
    max_bind = (biz_config.get("referral") or {}).get("risk_max_bind_per_day", 10)
    if bind_count > max_bind:
        status, risk_flag = 3, "inviter_burst"
    else:
        max_reg = (biz_config.get("referral") or {}).get("risk_max_register_per_ip", 5)
        cnt = (await db.execute(
            select(func.count(User.id)).where(
                User.register_ip == ip, User.created_at > datetime.now() - timedelta(hours=24)
            )
        )).scalar() or 0
        if cnt > max_reg:
            status, risk_flag = 3, "batch_register"

    db.add(Referral(
        inviter_id=inviter.id,
        invitee_id=user_id,
        invite_code=invite_code,
        status=status,
        risk_flag=risk_flag,
        register_ip=ip,
    ))
    await db.commit()
    return {"bound": True, "status": status, "status_text": REFERRAL_STATUS_TEXT[status]}


# ---------- 资产 / 提现 ----------

async def get_my_assets(db: AsyncSession, redis: aioredis.Redis, user_id: int) -> dict:
    """资产钱包（9.3）"""
    wallet = await get_or_create_asset_wallet(db, user_id)
    per_yuan = Decimal(str(_asset_cfg().get("assets_per_yuan", 10)))
    has_pending = bool(await redis.exists(f"wd:pending:{user_id}")) or bool(
        (await db.execute(
            select(Withdrawal.id).where(Withdrawal.user_id == user_id, Withdrawal.status == 0)
        )).first()
    )
    return {
        "assets": float(wallet.assets),
        "assets_value_yuan": float(wallet.assets / per_yuan),
        "total_earned": float(wallet.total_earned),
        "total_withdrawn": float(wallet.total_withdrawn),
        "min_withdraw_assets": int(_withdraw_cfg().get("min_assets", 100)),
        "has_pending_withdraw": has_pending,
    }


async def list_asset_records(
    db: AsyncSession, user_id: int, page: int, page_size: int, biz_type: int | None
) -> dict:
    """资产流水（9.4）"""
    stmt = select(AssetTransaction).where(AssetTransaction.user_id == user_id)
    if biz_type is not None:
        stmt = stmt.where(AssetTransaction.biz_type == biz_type)
    total = (await db.execute(select(func.count()).select_from(stmt.subquery()))).scalar() or 0
    rows = (
        await db.execute(
            stmt.order_by(AssetTransaction.id.desc()).offset((page - 1) * page_size).limit(page_size)
        )
    ).scalars().all()
    return {
        "total": total,
        "list": [
            {
                "id": r.id,
                "change_amount": float(r.change_amount),
                "balance_after": float(r.balance_after),
                "biz_type": r.biz_type,
                "remark": r.remark,
                "created_at": _fmt(r.created_at),
            }
            for r in rows
        ],
    }


async def apply_withdraw(
    db: AsyncSession, redis: aioredis.Redis, user_id: int, assets: Decimal, channel: int, account: str
) -> dict:
    """申请提现（9.5）：在途锁防并发 → 事务内冻结扣减 + 建单；审核拒绝由后台退回并释放锁"""
    min_assets = int(_withdraw_cfg().get("min_assets", 100))
    if assets < min_assets:
        raise BizException(errors.WITHDRAW_TOO_LOW)

    lock_key = f"wd:pending:{user_id}"
    if not await redis.set(lock_key, 1, nx=True, ex=WD_LOCK_TTL):
        raise BizException(errors.WITHDRAW_PENDING)
    try:
        # 双保险：DB 层存在待审核单也拦截（锁丢失场景）
        if (await db.execute(
            select(Withdrawal.id).where(Withdrawal.user_id == user_id, Withdrawal.status == 0)
        )).first():
            raise BizException(errors.WITHDRAW_PENDING)

        per_yuan = Decimal(str(_asset_cfg().get("assets_per_yuan", 10)))
        amount_fen = int((assets / per_yuan) * 100)  # 10资产=1元 → 1资产=10分
        # 冻结扣减（余额不足抛 2102，事务回滚）
        await change_assets(
            db, user_id, -assets, ASSET_BIZ_WITHDRAW,
            remark=f"申请提现冻结（{account}）",
        )
        wd = Withdrawal(
            withdraw_no=membership_service.gen_order_no("W"),
            user_id=user_id,
            assets=assets,
            amount_fen=amount_fen,
            channel=channel,
            account=account,
        )
        db.add(wd)
        await db.commit()
    except Exception:
        await redis.delete(lock_key)  # 建单失败释放锁
        raise

    return {
        "withdraw_no": wd.withdraw_no,
        "assets": float(wd.assets),
        "amount_yuan": float(Decimal(wd.amount_fen) / 100),
        "status": 0,
        "status_text": WITHDRAW_STATUS_TEXT[0],
    }


async def list_withdraw_records(db: AsyncSession, user_id: int, page: int, page_size: int) -> dict:
    """提现记录（9.6）"""
    stmt = select(Withdrawal).where(Withdrawal.user_id == user_id)
    total = (await db.execute(select(func.count()).select_from(stmt.subquery()))).scalar() or 0
    rows = (
        await db.execute(
            stmt.order_by(Withdrawal.id.desc()).offset((page - 1) * page_size).limit(page_size)
        )
    ).scalars().all()
    return {
        "total": total,
        "list": [
            {
                "withdraw_no": w.withdraw_no,
                "assets": float(w.assets),
                "amount_yuan": float(Decimal(w.amount_fen) / 100),
                "channel": w.channel,
                "status": w.status,
                "status_text": WITHDRAW_STATUS_TEXT.get(w.status, ""),
                "created_at": _fmt(w.created_at),
            }
            for w in rows
        ],
    }