shaguabijia-app-server/app/repositories/wallet.py

"""钱包 CRUD:账户读取 + 金币入账(共享给签到 / 任务)+ 流水查询。

`grant_coins` 是所有"发金币"的唯一入口:它更新余额快照并写一笔流水,但**不 commit**,
由调用方在同一事务里 commit(签到要同时写 signin_record,任务要同时写 user_task,
保证"记录"和"加金币"原子化,不会只发钱不留痕或反之)。
"""
from __future__ import annotations

import re
import unicodedata
import uuid
from datetime import datetime, timedelta, timezone

from sqlalchemy import select, update
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from app.core import rewards
from app.core.config import settings
from app.core.rewards import COIN_PER_CENT, coins_to_cents
from app.integrations import wxpay
from app.models.user import User
from app.models.wallet import (
    CashTransaction,
    CoinAccount,
    CoinTransaction,
    InviteCashTransaction,
    WechatTransferAuthorization,
    WithdrawOrder,
)

# 微信转账终态:成功 / 失败(失败/取消/关闭都退款)
_WX_STATE_SUCCESS = "SUCCESS"
_WX_STATE_FAILED = {"FAIL", "CANCELLED", "CLOSED"}
_WX_STATE_WAIT_CONFIRM = "WAIT_USER_CONFIRM"  # 用户还没在微信确认页确认
_WITHDRAW_ACTIVE_STATUSES = {"reviewing", "pending"}
# 免确认收款授权状态
_WX_AUTH_ACTIVE = "TAKING_EFFECT"   # 已生效,可免确认转账
_WX_AUTH_CLOSED = "CLOSED"          # 已关闭(用户/商户/风控),需重新开启


class InvalidExchangeAmountError(Exception):
    """兑换金币数非法(<最小额 或 不是整分倍数)。"""


class InsufficientCoinError(Exception):
    """金币余额不足。"""


class InvalidWithdrawAmountError(Exception):
    """提现金额超出允许范围。"""


class WechatNotBoundError(Exception):
    """用户未绑定微信(无 openid),无法提现。"""


class WechatAlreadyBoundError(Exception):
    """该微信(openid)已绑定到其他账号。"""


class InsufficientCashError(Exception):
    """现金余额不足。"""


class WithdrawTooFrequentError(Exception):
    """提现申请过于频繁,或已有未完成提现单。"""


class WithdrawTransferError(Exception):
    """调用微信转账失败(已退回余额)。"""


class WithdrawOrderNotFound(Exception):
    """提现单不存在。"""


class WithdrawNotReviewable(Exception):
    """提现单当前状态不可审核(非 reviewing,可能已被处理过)。"""


def get_or_create_account(
    db: Session, user_id: int, *, commit: bool = True, lock: bool = False
) -> CoinAccount:
    """取用户金币账户,不存在则建一个空账户。

    lock=True 时对已存在的账户行加 SELECT FOR UPDATE(读-算-写余额的调用方串行化,防并发
    双写余额错位,如 admin set 模式连点);默认 False 不改 C 端发奖行为。SQLite 下为 no-op。
    """
    acc = db.get(CoinAccount, user_id, with_for_update=True) if lock else db.get(CoinAccount, user_id)
    if acc is None:
        acc = CoinAccount(
            user_id=user_id,
            coin_balance=0,
            cash_balance_cents=0,
            total_coin_earned=0,
        )
        db.add(acc)
        if commit:
            db.commit()
            db.refresh(acc)
        else:
            db.flush()
    return acc


def grant_coins(
    db: Session,
    user_id: int,
    amount: int,
    *,
    biz_type: str,
    ref_id: str | None = None,
    remark: str | None = None,
) -> tuple[CoinAccount, CoinTransaction]:
    """金币变动入口(正数入账 / 负数出账)。更新余额 + 写流水,不 commit。

    返回 (account, transaction)。调用方负责 commit。
    """
    acc = get_or_create_account(db, user_id, commit=False)
    acc.coin_balance += amount
    if amount > 0:
        acc.total_coin_earned += amount

    txn = CoinTransaction(
        user_id=user_id,
        amount=amount,
        balance_after=acc.coin_balance,
        biz_type=biz_type,
        ref_id=ref_id,
        remark=remark,
        created_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),  # 存北京 wall-clock(客户端原样切片显示)
    )
    db.add(txn)
    db.flush()
    return acc, txn


def grant_cash(
    db: Session,
    user_id: int,
    amount_cents: int,
    *,
    biz_type: str,
    ref_id: str | None = None,
    remark: str | None = None,
) -> tuple[CoinAccount, CashTransaction]:
    """现金变动入口(正数入账 / 负数出账)。更新现金余额 + 写流水,不 commit。

    与 [grant_coins] 同模式(运营手动调现金 / 测试发现金用)。返回 (account, transaction),
    调用方负责 commit。不在此校验扣成负——由调用方(admin router)按业务保护。
    """
    acc = get_or_create_account(db, user_id, commit=False)
    acc.cash_balance_cents += amount_cents

    txn = CashTransaction(
        user_id=user_id,
        amount_cents=amount_cents,
        balance_after_cents=acc.cash_balance_cents,
        biz_type=biz_type,
        ref_id=ref_id,
        remark=remark,
        created_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),  # 存北京 wall-clock
    )
    db.add(txn)
    db.flush()
    return acc, txn


def grant_invite_cash(
    db: Session,
    user_id: int,
    amount_cents: int,
    *,
    biz_type: str,
    ref_id: str | None = None,
    remark: str | None = None,
) -> tuple[CoinAccount, InviteCashTransaction]:
    """邀请奖励金变动入口(正数入账 / 负数出账)。更新 invite_cash_balance_cents + 写
    invite_cash_transaction,不 commit。与金币兑换的 cash_balance_cents **物理隔离**
    (产品红线:邀请奖励金 ≠ 金币现金,两本账不可累加)。返回 (account, transaction),
    调用方负责 commit。不在此校验扣成负——由调用方按业务保护。"""
    acc = get_or_create_account(db, user_id, commit=False)
    acc.invite_cash_balance_cents += amount_cents

    txn = InviteCashTransaction(
        user_id=user_id,
        amount_cents=amount_cents,
        balance_after_cents=acc.invite_cash_balance_cents,
        biz_type=biz_type,
        ref_id=ref_id,
        remark=remark,
        created_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
    )
    db.add(txn)
    db.flush()
    return acc, txn


def list_coin_transactions(
    db: Session,
    user_id: int,
    *,
    limit: int = 20,
    cursor: int | None = None,
) -> tuple[list[CoinTransaction], int | None]:
    """金币流水分页(按 id 倒序,游标式)。

    cursor 为上一页最后一条的 id;返回 (本页列表, next_cursor)。
    next_cursor 为 None 表示没有下一页。
    """
    stmt = select(CoinTransaction).where(CoinTransaction.user_id == user_id)
    if cursor is not None:
        stmt = stmt.where(CoinTransaction.id < cursor)
    stmt = stmt.order_by(CoinTransaction.id.desc()).limit(limit)

    items = list(db.execute(stmt).scalars().all())
    next_cursor = items[-1].id if len(items) == limit else None
    return items, next_cursor


def exchange_coins_to_cash(
    db: Session,
    user_id: int,
    coin_amount: int,
    *,
    remark: str | None = None,
    enforce_min: bool = True,
) -> tuple[CoinAccount, int]:
    """金币兑现金。返回 (account, 本次兑入的分)。

    校验:金额为整分倍数(InvalidExchangeAmountError)、余额充足(InsufficientCoinError);
    `enforce_min=True`(手动兑默认)时还要 >= 后台配置的兑换下限,`enforce_min=False`
    (0 点自动兑「到分全额」用)只要够 1 分(COIN_PER_CENT)即可,不受手动下限约束。
    `remark` 不传则用默认「N金币兑M分」。扣金币 + 加现金,两条流水同事务 commit。
    """
    floor_min = rewards.get_min_exchange_coin(db) if enforce_min else COIN_PER_CENT
    if coin_amount < floor_min or coin_amount % COIN_PER_CENT != 0:
        raise InvalidExchangeAmountError

    acc = get_or_create_account(db, user_id, commit=False)
    if acc.coin_balance < coin_amount:
        raise InsufficientCoinError

    cents = coins_to_cents(coin_amount)
    tx_remark = remark or f"{coin_amount}金币兑{cents}分"

    # 1. 扣金币(写 coin_transaction)
    grant_coins(db, user_id, -coin_amount, biz_type="exchange_out", remark=tx_remark)
    # 2. 加现金(写 cash_transaction)
    acc.cash_balance_cents += cents
    db.add(
        CashTransaction(
            user_id=user_id,
            amount_cents=cents,
            balance_after_cents=acc.cash_balance_cents,
            biz_type="exchange_in",
            remark=tx_remark,
            created_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
        )
    )
    db.commit()
    db.refresh(acc)
    return acc, cents


def _has_exchange_in_on(db: Session, user_id: int, day) -> bool:
    """该用户在指定日期(北京日)是否已有 exchange_in 现金流水。

    自动兑复用 exchange_in 类型,故用「当天是否已有 exchange_in」做幂等键:timer 重触发 /
    手动重跑当天不会重复兑。手动兑入口已下线,不会与之误判混淆(若日后恢复手动兑需改用独立类型)。
    created_at 落库为北京 naive 时间(见 exchange_coins_to_cash),按 [当天 0 点, 次日 0 点) 比较。
    """
    day_start = datetime.combine(day, datetime.min.time())
    next_day = day_start + timedelta(days=1)
    return (
        db.execute(
            select(CashTransaction.id)
            .where(
                CashTransaction.user_id == user_id,
                CashTransaction.biz_type == "exchange_in",
                CashTransaction.created_at >= day_start,
                CashTransaction.created_at < next_day,
            )
            .limit(1)
        ).first()
        is not None
    )


def daily_auto_exchange(db: Session) -> dict:
    """0 点定时任务:把每个用户金币「到分全额」兑成现金(复用 exchange_in 流水类型)。

    - **到分全额**:coin_amount = floor(余额 / 100) * 100;余额不足 1 分(<100)的零头留到下次。
    - **幂等**:当天已有 exchange_in 的用户跳过(见 [_has_exchange_in_on])。
    - **逐用户独立事务**:单个用户异常 rollback 不影响其他人;exchange_coins_to_cash 内部按用户 commit。
    返回统计 dict(scanned/converted/skipped_done/skipped_dust/failed/total_cents)。
    """
    # ⚠️「今天」写死北京时(rewards.cn_today() = datetime.now(CN_TZ).date(), CN_TZ=+8),与服务器/用户
    #   时区无关(用户 2026-07-01 硬约束:兑换一律北京 0 点日切)。**禁止**改成 date.today() / 裸
    #   datetime.now().date()——那会跟随进程本地时区,服务器非 CST 时会在错误的"天"兑/重复兑。
    today = rewards.cn_today()
    stats = {
        "scanned": 0, "converted": 0, "skipped_done": 0,
        "skipped_dust": 0, "failed": 0, "total_cents": 0,
    }
    user_ids = db.execute(
        select(CoinAccount.user_id).where(CoinAccount.coin_balance >= COIN_PER_CENT)
    ).scalars().all()

    for user_id in user_ids:
        stats["scanned"] += 1
        try:
            if _has_exchange_in_on(db, user_id, today):
                stats["skipped_done"] += 1
                continue
            acc = get_or_create_account(db, user_id, commit=False)
            coin_amount = (acc.coin_balance // COIN_PER_CENT) * COIN_PER_CENT
            if coin_amount < COIN_PER_CENT:
                stats["skipped_dust"] += 1
                continue
            _, cents = exchange_coins_to_cash(
                db, user_id, coin_amount, remark="0 点自动兑现金", enforce_min=False
            )
            stats["converted"] += 1
            stats["total_cents"] += cents
        except Exception:  # noqa: BLE001 - 批处理不因单用户异常中断
            db.rollback()
            stats["failed"] += 1

    return stats


def list_cash_transactions(
    db: Session,
    user_id: int,
    *,
    limit: int = 20,
    cursor: int | None = None,
) -> tuple[list[CashTransaction], int | None]:
    """现金流水分页(按 id 倒序,游标式)。"""
    stmt = select(CashTransaction).where(CashTransaction.user_id == user_id)
    if cursor is not None:
        stmt = stmt.where(CashTransaction.id < cursor)
    stmt = stmt.order_by(CashTransaction.id.desc()).limit(limit)

    items = list(db.execute(stmt).scalars().all())
    next_cursor = items[-1].id if len(items) == limit else None
    return items, next_cursor


# ===== 提现(现金 → 微信零钱) =====


def bind_wechat_openid(db: Session, user_id: int, code: str) -> dict:
    """用微信授权 code 换 openid + 昵称头像并绑定到用户。失败抛 ValueError。
    返回 {openid, nickname, avatar_url}(昵称头像可能为 None,见 wxpay.code_to_userinfo)。"""
    info = wxpay.code_to_userinfo(code)  # 失败抛 ValueError
    user = db.get(User, user_id)
    if user is None:
        raise WithdrawOrderNotFound  # 理论上当前用户必存在

    # #5 一个微信只能绑一个账号:若该 openid 已属于别的用户,拒绝(防多账号薅羊毛/重复提现)
    other = db.execute(
        select(User.id).where(User.wechat_openid == info["openid"], User.id != user_id)
    ).scalar_one_or_none()
    if other is not None:
        raise WechatAlreadyBoundError

    user.wechat_openid = info["openid"]
    user.wechat_nickname = info["nickname"]
    user.wechat_avatar_url = info["avatar_url"]
    db.commit()
    return info


def get_cash_balance_cents(db: Session, user_id: int) -> int:
    """只读查现金余额(分)。账户不存在返回 0,**不创建账户**——注销前置校验用,
    不该给一个即将被删除的用户凭空建一行空账户(get_or_create_account 会建)。"""
    val = db.execute(
        select(CoinAccount.cash_balance_cents).where(CoinAccount.user_id == user_id)
    ).scalar_one_or_none()
    return val or 0


def has_reviewing_withdraw(db: Session, user_id: int) -> bool:
    """用户是否有「待审核(reviewing)」的提现单。

    只看 reviewing:这类单还没打款,用户确认解绑时会被立即退回现金
    (refund_reviewing_withdraws_on_unbind),解绑前据此提示用户知情确认。
    pending 单转账已在途(execute 已过 openid 闸),解绑影响不到、照常打款,故不计入。
    """
    return (
        db.execute(
            select(WithdrawOrder.id)
            .where(
                WithdrawOrder.user_id == user_id,
                WithdrawOrder.status == "reviewing",
            )
            .limit(1)
        ).first()
        is not None
    )


def unbind_wechat_openid(db: Session, user_id: int) -> None:
    """解绑微信:清空 openid。已绑定才清,未绑定为幂等空操作。"""
    user = db.get(User, user_id)
    if user is None:
        raise WithdrawOrderNotFound
    if user.wechat_openid is not None:
        user.wechat_openid = None
        user.wechat_nickname = None
        user.wechat_avatar_url = None
        db.commit()


def refund_reviewing_withdraws_on_unbind(db: Session, user_id: int) -> int:
    """解绑微信时,把该用户所有「待审核(reviewing)」提现单立即退回现金并置终态。

    reviewing 单还没打款(钱在 create_withdraw 时已从现金扣进待审核单),解绑后这笔无法
    打款到微信,直接退回现金最干净:用户即时拿回钱、后台不再挂死单(不必等管理员审核)。
    复用 _refund_withdraw(退现金 + 流水 + 幂等)。返回退掉的单数。
    """
    orders = (
        db.execute(
            select(WithdrawOrder).where(
                WithdrawOrder.user_id == user_id,
                WithdrawOrder.status == "reviewing",
            )
        )
        .scalars()
        .all()
    )
    for order in orders:
        _refund_withdraw(
            db,
            order,
            reason="用户解绑微信,待审核提现自动退回",
            remark="解绑微信,提现已退回现金余额",
        )
    return len(orders)


_OUT_BILL_NO_RE = re.compile(r"^[0-9A-Za-z_-]{8,32}$")


def _balance_col(source: str):
    """提现账户来源 → CoinAccount 余额列。invite_cash=邀请奖励金;否则金币兑换的现金。
    两账户物理隔离,提现扣款/退款都按 source 走对应列,互不串。"""
    return (
        CoinAccount.invite_cash_balance_cents
        if source == "invite_cash"
        else CoinAccount.cash_balance_cents
    )


def _try_deduct_cash(db: Session, user_id: int, amount_cents: int, source: str = "coin_cash") -> bool:
    """原子扣减指定账户余额:仅当余额足够时扣,返回是否成功。

    用带条件的 UPDATE(`WHERE <col> >= amount`)避免"读-判断-写"竞态——并发/重试时不会两次
    都通过余额检查导致超额扣款(SQLite 串行写、Postgres 行级,均安全)。source 决定扣
    cash_balance_cents(coin_cash) 还是 invite_cash_balance_cents(invite_cash)。
    """
    col = _balance_col(source)
    res = db.execute(
        update(CoinAccount)
        .where(CoinAccount.user_id == user_id, col >= amount_cents)
        .values({col: col - amount_cents})
    )
    return res.rowcount == 1


def _add_cash(db: Session, user_id: int, amount_cents: int, source: str = "coin_cash") -> int:
    """原子增加指定账户余额(退款用),返回加后余额。source 决定退回哪个账户(两账户隔离)。"""
    col = _balance_col(source)
    db.execute(
        update(CoinAccount)
        .where(CoinAccount.user_id == user_id)
        .values({col: col + amount_cents})
    )
    db.flush()
    bal = db.execute(
        select(col).where(CoinAccount.user_id == user_id)
    ).scalar_one()
    return bal


def _refund_withdraw(
    db: Session, order: WithdrawOrder, reason: str, *, final_status: str = "failed",
    remark: str | None = None,
) -> None:
    """退回现金 + 写 withdraw_refund 流水 + 单子置终态。幂等:已是退款终态不重复退。

    final_status:
      - "failed"  (默认):微信转账失败/取消 → 自动退款
      - "rejected":管理员审核拒绝 → 退款(走 reject_withdraw)
    两种都已退款,用 in 判定防重复退(并发/对账/拒绝重复点)。
    remark:给用户可见的现金流水文案;省略则按 final_status 取默认(解绑退回等场景可自定义)。
    """
    if order.status in ("failed", "rejected"):
        return  # 防重复退款(并发/对账与查单/重复拒绝同时触发)
    # 账户隔离:按 order.source 退回对应账户 + 写对应退款流水表
    is_invite = order.source == "invite_cash"
    txn_model = InviteCashTransaction if is_invite else CashTransaction
    refund_biz = "invite_withdraw_refund" if is_invite else "withdraw_refund"
    refunded_txn_id = db.execute(
        select(txn_model.id).where(
            txn_model.user_id == order.user_id,
            txn_model.biz_type == refund_biz,
            txn_model.ref_id == order.out_bill_no,
        ).limit(1)
    ).scalar_one_or_none()
    if refunded_txn_id is not None:
        order.status = final_status
        order.fail_reason = reason[:256]
        db.commit()
        return
    bal = _add_cash(db, order.user_id, order.amount_cents, order.source)
    db.add(
        txn_model(
            user_id=order.user_id,
            amount_cents=order.amount_cents,
            balance_after_cents=bal,
            biz_type=refund_biz,
            ref_id=order.out_bill_no,
            # 用户可见文案区分"未成功(自动退)"vs"审核未通过";技术原因记在 order.fail_reason
            remark=(
                remark
                or ("提现审核未通过,金额已退回" if final_status == "rejected" else "提现未成功,金额已退回")
            ),
            created_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
        )
    )
    order.status = final_status
    order.fail_reason = reason[:256]
    out_bill_no = order.out_bill_no
    user_id = order.user_id
    try:
        db.commit()
    except IntegrityError:
        # 并发退款兜底:唯一退款流水已被另一事务写入时,回滚本事务的加钱和流水,
        # 再把订单状态补到终态。这样无论拒绝/查单/对账怎么并发,金额最多退一次。
        db.rollback()
        refunded_txn_id = db.execute(
            select(txn_model.id).where(
                txn_model.user_id == user_id,
                txn_model.biz_type == refund_biz,
                txn_model.ref_id == out_bill_no,
            ).limit(1)
        ).scalar_one_or_none()
        if refunded_txn_id is None:
            raise
        fresh_order = db.execute(
            select(WithdrawOrder).where(WithdrawOrder.out_bill_no == out_bill_no)
        ).scalar_one_or_none()
        if fresh_order is not None and fresh_order.status not in ("failed", "rejected"):
            fresh_order.status = final_status
            fresh_order.fail_reason = reason[:256]
            db.commit()


def _wx_not_found(result: dict) -> bool:
    """微信查单/转账响应是否表示'此单不存在'(说明转账从未创建,退款安全)。"""
    if result.get("status_code") == 404:
        return True
    data = result.get("data")
    code = data.get("code", "") if isinstance(data, dict) else ""
    return "NOT_FOUND" in str(code)


def _settle_after_ambiguous(db: Session, order: WithdrawOrder, reason: str) -> None:
    """转账调用结果不明(超时/异常/非200)时,**先查单再决定**,绝不盲目退款(防退款后又到账)。
      - 微信查到 SUCCESS        → 钱已出,置 success,不退款
      - 微信查到 FAIL/取消/关闭  → 退款
      - 微信查到在途/待确认       → 保持 pending(带上 package_info/state),交给查单/对账后续处理
      - 微信明确无此单           → 从未创建,退款安全
      - 查询本身也失败           → 保持 pending,留给对账任务(reconcile)兜底,绝不退款
    """
    try:
        q = wxpay.query_transfer(order.out_bill_no)
    except Exception:  # noqa: BLE001 — 查询都失败,保持 pending 等对账
        order.fail_reason = (reason + " | 查单异常,保持pending")[:256]
        db.commit()
        return

    if q["status_code"] != 200:
        if _wx_not_found(q):
            _refund_withdraw(db, order, reason=reason + " (微信无此单,已退回)")
        else:
            order.fail_reason = (reason + " | 查单非200,保持pending")[:256]
            db.commit()
        return

    state = q["data"].get("state", "")
    order.wechat_state = state
    if state == _WX_STATE_SUCCESS:
        order.status = "success"
        order.transfer_bill_no = q["data"].get("transfer_bill_no")
        db.commit()
    elif state in _WX_STATE_FAILED:
        _refund_withdraw(db, order, reason=reason)
    else:
        order.package_info = q["data"].get("package_info") or order.package_info
        db.commit()


def create_withdraw(
    db: Session,
    user_id: int,
    amount_cents: int,
    *,
    source: str = "coin_cash",
    user_name: str | None = None,
    out_bill_no: str | None = None,
    allow_sub_min: bool = False,
) -> WithdrawOrder:
    """发起提现:原子扣款 + 建单 reviewing(待人工审核),**不打款**。

    打款推迟到管理员在 admin 后台审核通过(approve_withdraw → execute_withdraw_transfer)才发起;
    拒绝则退回现金(reject_withdraw)。
    #1 用原子条件 UPDATE 扣款,杜绝并发超额——且"待审核期间钱已扣减",防止用户拿同一笔余额
       重复发起多笔提现(审核拒绝再退回)。
    #2 out_bill_no 客户端幂等键:同号重试返回该单现状(reviewing 等审核),不重复扣款建单。
    实名 user_name 在此存下(WithdrawOrder.user_name),供异步审核打款时传给微信(达额需实名)。

    allow_sub_min:放行低于"提现最低额"的小额(用于 0.01 元调试提现)。仅由 endpoint 在
    `skip_review and not is_prod`(debug 包 + 非生产双闸)时置 True;仍受 schema gt=0 与 max 上限约束。
    生产恒为 False → 最低额校验照常,绝不可能提 0.01。
    """
    min_c = 0 if allow_sub_min else rewards.get_withdraw_min_cents(db)
    if amount_cents < min_c or amount_cents > rewards.get_withdraw_max_cents(db):
        raise InvalidWithdrawAmountError

    # 提现即要求已绑微信:否则审核通过也打不了款,提前拦更友好
    user = db.get(User, user_id)
    openid = user.wechat_openid if user else None
    if not openid:
        raise WechatNotBoundError

    # #2 幂等:该单已存在 → 直接返回现状(reviewing 等审核 / 已处理态原样返回),不重复扣款。
    # 不再像旧版对 pending 查微信:本函数不再涉及微信,查单交给 /withdraw/status 与对账任务。
    if out_bill_no and _OUT_BILL_NO_RE.match(out_bill_no):
        existing = db.execute(
            select(WithdrawOrder).where(
                WithdrawOrder.out_bill_no == out_bill_no, WithdrawOrder.user_id == user_id
            )
        ).scalar_one_or_none()
        if existing is not None:
            return existing
    else:
        out_bill_no = uuid.uuid4().hex

    active_order_id = db.execute(
        select(WithdrawOrder.id).where(
            WithdrawOrder.user_id == user_id,
            WithdrawOrder.status.in_(_WITHDRAW_ACTIVE_STATUSES),
        ).limit(1)
    ).scalar_one_or_none()
    if active_order_id is not None:
        raise WithdrawTooFrequentError

    # 账户须存在(原子扣款的 UPDATE 不会建账户)
    get_or_create_account(db, user_id, commit=True)

    # #1 原子扣款:余额不足时影响行数为 0(按 source 扣对应账户)
    if not _try_deduct_cash(db, user_id, amount_cents, source):
        db.rollback()
        raise InsufficientCashError

    is_invite = source == "invite_cash"
    txn_model = InviteCashTransaction if is_invite else CashTransaction
    withdraw_biz = "invite_withdraw" if is_invite else "withdraw"
    bal = db.execute(
        select(_balance_col(source)).where(CoinAccount.user_id == user_id)
    ).scalar_one()
    db.add(
        txn_model(
            user_id=user_id,
            amount_cents=-amount_cents,
            balance_after_cents=bal,
            biz_type=withdraw_biz,
            ref_id=out_bill_no,
            remark="提现到微信零钱(待审核)",
            created_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
        )
    )
    order = WithdrawOrder(
        user_id=user_id,
        out_bill_no=out_bill_no,
        amount_cents=amount_cents,
        source=source,
        user_name=user_name,
        status="reviewing",
    )
    db.add(order)
    try:
        db.commit()
    except IntegrityError:
        db.rollback()
        existing = db.execute(
            select(WithdrawOrder).where(
                WithdrawOrder.out_bill_no == out_bill_no, WithdrawOrder.user_id == user_id
            )
        ).scalar_one_or_none()
        if existing is not None:
            return existing
        active_order_id = db.execute(
            select(WithdrawOrder.id).where(
                WithdrawOrder.user_id == user_id,
                WithdrawOrder.status.in_(_WITHDRAW_ACTIVE_STATUSES),
            ).limit(1)
        ).scalar_one_or_none()
        if active_order_id is not None:
            raise WithdrawTooFrequentError from None
        raise
    db.refresh(order)
    return order  # 待管理员审核;**不在此处打款**


# ===== 免确认收款授权(用户授权免确认模式)=====
# 用户授权一次后,后续提现走 transfer_with_authorization 免逐笔确认直接到账。
# WechatTransferAuthorization 一个用户一条;out_authorization_no 我方生成,authorization_id 微信生效后返回。


def _auth_display_name(user: User | None) -> str:
    """微信授权页展示的"开通账号"昵称(≤32,utf8)。

    微信对 user_display_name 校验极严:**连空格和标点(. 等)都算"控制字符"拒收**——实测
    'wonderable ai'(带空格)被 400 拒、'wonderableai'/'周周'/'傻瓜比价用户8888' 才过。
    故只保留 文字(Unicode L*,中英文/CJK)与数字(N*),其余(emoji/符号/空格/标点/控制符/
    零宽连接符/变体选择符/星形面字符)一律剔除;清空则兜底手机号尾号。
    """
    raw = ((user.nickname if user else None) or (user.wechat_nickname if user else None) or "").strip()
    kept = [
        ch for ch in raw
        if ord(ch) <= 0xFFFF and unicodedata.category(ch)[0] in ("L", "N")
    ]
    name = "".join(kept).strip()
    if not name and user and user.phone:
        tail = user.phone[-4:] if len(user.phone) >= 4 else user.phone
        name = f"傻瓜比价用户{tail}"
    return (name or "傻瓜比价用户")[:32]


def _refresh_active_auth(db: Session, user_id: int) -> None:
    """免确认转账失败后回查授权有效性(权威判定,不靠猜错误码):
      微信侧明确非生效(用户在微信关闭 / 风控 / 无此单)→ 标 closed,下次提现自动回退方式一重新授权;
      仍生效(失败实为商户余额不足等与授权无关的原因)→ 保持 active,下次免确认重试。
    仅对本地 active 记录查询;查询本身失败(5xx/网络)不改状态,下次再核,避免误关有效授权。"""
    auth = db.get(WechatTransferAuthorization, user_id)
    if auth is None or auth.state != "active" or not settings.wxpay_configured:
        return
    try:
        res = wxpay.query_transfer_authorization(auth.out_authorization_no)
    except Exception:  # noqa: BLE001 — 查询失败保持 active,下次再核
        return
    if res["status_code"] == 200:
        if res["data"].get("state") != _WX_AUTH_ACTIVE:  # CLOSED / 其他非生效 → 失效
            auth.state = "closed"
            db.commit()
    elif _wx_not_found(res):  # 明确无此授权单 → 失效
        auth.state = "closed"
        db.commit()
    # 其他查询错误(5xx 等):不改状态,保持 active,下次再核


def sync_transfer_auth(db: Session, user_id: int) -> WechatTransferAuthorization | None:
    """同步待确认授权最新状态(仅 pending 时查微信):
      TAKING_EFFECT → active + 落 authorization_id;CLOSED/超期无此单 → closed。返回最新记录(或 None)。
    非 pending(active/closed)直接返回不查;查询异常或缺凭证则保持原状。"""
    auth = db.get(WechatTransferAuthorization, user_id)
    if auth is None or auth.state != "pending" or not settings.wxpay_configured:
        return auth
    try:
        res = wxpay.query_transfer_authorization(auth.out_authorization_no)
    except Exception:  # noqa: BLE001 — 查询失败保持 pending,下次再同步
        return auth
    if res["status_code"] != 200:
        if _wx_not_found(res):  # 超期未确认被关闭 → closed;其他错误保持 pending
            auth.state = "closed"
            db.commit()
        return auth
    state = res["data"].get("state", "")
    if state == _WX_AUTH_ACTIVE:
        auth.state = "active"
        auth.authorization_id = res["data"].get("authorization_id") or auth.authorization_id
        db.commit()
    elif state == _WX_AUTH_CLOSED:
        auth.state = "closed"
        db.commit()
    # WAIT_USER_CONFIRM:保持 pending,等用户确认
    return auth


def _ensure_pending_auth_no(db: Session, user_id: int, openid: str) -> str:
    """为"方式一(转账顺带授权)"准备待确认授权单号:
      已有 pending → 复用其 out_authorization_no(避免重复申请刷满"同场景≤5"上限);
      无 / 已 closed → 生成新单号并 upsert 成 pending。返回 out_authorization_no。"""
    auth = db.get(WechatTransferAuthorization, user_id)
    if auth is not None and auth.state == "pending":
        if auth.openid != openid:  # 换绑过 → 刷新快照
            auth.openid = openid
            db.commit()
        return auth.out_authorization_no
    new_no = uuid.uuid4().hex  # 32 位,符合 [0-9A-Za-z_-]{8,32}
    if auth is None:
        db.add(
            WechatTransferAuthorization(
                user_id=user_id, openid=openid, out_authorization_no=new_no,
                authorization_id=None, state="pending",
            )
        )
    else:  # closed → 重新开启,换新单号
        auth.openid = openid
        auth.out_authorization_no = new_no
        auth.authorization_id = None
        auth.state = "pending"
    db.commit()
    return new_no


def apply_transfer_auth(db: Session, user_id: int) -> dict:
    """方式二:显式开启免确认收款(申请授权,不转账)。
    返回 {already_active, package_info, mch_id, app_id};already_active=True 表示已开启无需再授权。
    未绑微信抛 WechatNotBoundError;微信返回非 200 抛 WithdrawTransferError。"""
    user = db.get(User, user_id)
    openid = user.wechat_openid if user else None
    if not openid:
        raise WechatNotBoundError

    auth = db.get(WechatTransferAuthorization, user_id)
    if auth is not None and auth.state == "active" and auth.authorization_id:
        return {
            "already_active": True, "package_info": None,
            "mch_id": settings.WXPAY_MCH_ID, "app_id": settings.WECHAT_APP_ID,
        }

    out_auth_no = _ensure_pending_auth_no(db, user_id, openid)
    result = wxpay.apply_transfer_authorization(
        out_auth_no, openid, _auth_display_name(user), settings.WXPAY_AUTH_NOTIFY_URL
    )
    if result["status_code"] != 200:
        raise WithdrawTransferError(str(result["data"].get("message") or result["data"]))
    return {
        "already_active": False,
        "package_info": result["data"].get("package_info"),
        "mch_id": settings.WXPAY_MCH_ID,
        "app_id": settings.WECHAT_APP_ID,
    }


def close_transfer_auth(db: Session, user_id: int) -> None:
    """关闭免确认收款(解除授权)。best-effort 调微信解除 + 本地置 closed。"""
    auth = db.get(WechatTransferAuthorization, user_id)
    if auth is None:
        return
    if auth.state != "closed" and settings.wxpay_configured:
        try:
            wxpay.close_transfer_authorization(auth.out_authorization_no)
        except Exception:  # noqa: BLE001 — 微信解除失败不阻塞本地置 closed(用户也可在微信侧自行关闭)
            pass
    auth.state = "closed"
    db.commit()


def _apply_transfer_result(db: Session, order: WithdrawOrder, data: dict) -> WithdrawOrder:
    """落微信转账应答到提现单:记 state/转账单号/package_info,SUCCESS 即 success。"""
    order.wechat_state = data.get("state")
    order.transfer_bill_no = data.get("transfer_bill_no")
    order.package_info = data.get("package_info")  # 免确认转账无此字段(None);确认模式带它供拉确认页
    if data.get("state") == _WX_STATE_SUCCESS:
        order.status = "success"
    db.commit()
    db.refresh(order)
    return order


def execute_withdraw_transfer(db: Session, order: WithdrawOrder) -> WithdrawOrder:
    """审核通过后真正发起微信转账(复用原"转账 + 模糊失败查单"逻辑,绝不盲目退款)。**不抛异常**。

    免确认收款(用户授权免确认模式)分叉:
      ① 已有生效授权(active)→ transfer_with_authorization 免确认转账,直接到账,无 package_info。
      ② 无生效授权:
         - 已配 WXPAY_AUTH_NOTIFY_URL → pre_transfer_with_authorization(方式一):转账 + 顺带申请授权,
           返回 WAIT_USER_CONFIRM + package_info,用户在确认这笔收款时一并授权,下次起免确认。
         - 未配回调地址(未启用免确认)→ 退化为原 create_transfer 确认模式,行为同改造前。
    免确认转账失败 → 先 _settle_after_ambiguous 保金额安全,再 _refresh_active_auth 回查授权,失效则标 closed(下次回退方式一)。
    用户审核期间解绑微信 → 退款 failed。结果不明(超时/非200)→ 先查单再决定,绝不盲目退款。
    """
    user = db.get(User, order.user_id)
    openid = user.wechat_openid if user else None
    if not openid:
        _refund_withdraw(db, order, reason="用户已解绑微信,无法打款")
        db.refresh(order)
        return order

    # 先同步待确认授权:捕获首单确认后微信已生效的 authorization_id(pending→active),或失效(→closed)
    auth = sync_transfer_auth(db, order.user_id)

    order.status = "pending"  # 进入打款在途(转账调用前崩溃留 pending,交对账兜底)
    db.commit()

    # ① 有生效授权 → 免确认转账(无需用户确认,直接到账)
    if auth is not None and auth.state == "active" and auth.authorization_id:
        try:
            result = wxpay.transfer_with_authorization(
                auth.authorization_id, order.amount_cents, order.out_bill_no, user_name=order.user_name
            )
        except Exception as e:  # noqa: BLE001 — 超时/网络异常≠失败,查单确认
            _settle_after_ambiguous(db, order, reason=f"免确认转账调用异常: {e}")
            db.refresh(order)
            return order
        if result["status_code"] != 200:
            # 金额安全:查转账单后定夺,绝不盲退(未创建→退款,已创建→按真实状态)
            _settle_after_ambiguous(db, order, reason=str(result["data"].get("message") or result["data"]))
            # 授权有效性:回查授权单,微信侧已失效(用户关闭/风控)→标 closed,下次提现自动回退方式一重新授权
            _refresh_active_auth(db, order.user_id)
            db.refresh(order)
            return order
        return _apply_transfer_result(db, order, result["data"])

    # ② 无生效授权 → 启用免确认则转账+顺带授权(方式一),否则退化为原确认模式
    try:
        if settings.WXPAY_AUTH_NOTIFY_URL:
            out_auth_no = _ensure_pending_auth_no(db, order.user_id, openid)
            result = wxpay.pre_transfer_with_authorization(
                openid, order.amount_cents, order.out_bill_no,
                out_authorization_no=out_auth_no,
                user_display_name=_auth_display_name(user),
                notify_url=settings.WXPAY_AUTH_NOTIFY_URL,
                user_name=order.user_name,
            )
        else:
            result = wxpay.create_transfer(
                openid, order.amount_cents, order.out_bill_no, user_name=order.user_name
            )
    except Exception as e:  # noqa: BLE001 — 超时/网络异常≠失败,查单确认
        _settle_after_ambiguous(db, order, reason=f"转账调用异常: {e}")
        db.refresh(order)
        return order

    if result["status_code"] != 200:
        _settle_after_ambiguous(db, order, reason=str(result["data"].get("message") or result["data"]))
        db.refresh(order)
        return order

    return _apply_transfer_result(db, order, result["data"])


def _get_withdraw_or_raise(db: Session, out_bill_no: str) -> WithdrawOrder:
    """按 out_bill_no 取提现单(admin 跨用户,不限 user_id)。不存在抛 WithdrawOrderNotFound。"""
    order = db.execute(
        select(WithdrawOrder).where(WithdrawOrder.out_bill_no == out_bill_no)
    ).scalar_one_or_none()
    if order is None:
        raise WithdrawOrderNotFound
    return order


def approve_withdraw(db: Session, out_bill_no: str) -> WithdrawOrder:
    """管理员审核通过:校验 reviewing → 发起微信转账。返回最终 order(pending/success/failed)。

    幂等防误操作:非 reviewing(已通过/已拒绝/已打款)抛 WithdrawNotReviewable,不会重复打款。
    """
    order = _get_withdraw_or_raise(db, out_bill_no)
    if order.status != "reviewing":
        raise WithdrawNotReviewable(f"提现单状态为 {order.status},非待审核,不能通过")
    return execute_withdraw_transfer(db, order)


def reject_withdraw(db: Session, out_bill_no: str, reason: str) -> WithdrawOrder:
    """管理员审核拒绝:校验 reviewing → 退回现金 + 置 rejected。

    非 reviewing 抛 WithdrawNotReviewable(防对已打款单误退导致重复退款)。
    """
    order = _get_withdraw_or_raise(db, out_bill_no)
    if order.status != "reviewing":
        raise WithdrawNotReviewable(f"提现单状态为 {order.status},非待审核,不能拒绝")
    _refund_withdraw(db, order, reason=reason or "审核未通过", final_status="rejected")
    db.refresh(order)
    return order


def refresh_withdraw_status(
    db: Session, user_id: int, out_bill_no: str, *, cancel_if_unconfirmed: bool = False
) -> WithdrawOrder:
    """查微信单状态并归一化:SUCCESS→success;FAIL/CANCELLED/CLOSED→退款+failed;其余保持 pending。

    cancel_if_unconfirmed:查到 WAIT_USER_CONFIRM 时,是否当作"用户放弃"撤销并退款。
      - True:用户从确认页返回后查单、或对账老单时传 → 撤销+退款。
      - False(默认):幂等重试只想读状态时传 → 保留 pending,不能误撤用户还想确认的单。
    """
    order = db.execute(
        select(WithdrawOrder).where(
            WithdrawOrder.out_bill_no == out_bill_no, WithdrawOrder.user_id == user_id
        )
    ).scalar_one_or_none()
    if order is None:
        raise WithdrawOrderNotFound
    if order.status != "pending":
        return order  # 已终态,不再查

    try:
        result = wxpay.query_transfer(out_bill_no)
    except wxpay.WxPayNotConfiguredError:
        raise
    except Exception as exc:  # noqa: BLE001 - 查单失败不能把运营后台打成 500
        order.fail_reason = f"微信查单异常,保持pending: {exc}"[:256]
        db.commit()
        db.refresh(order)
        return order
    if result["status_code"] != 200:
        if _wx_not_found(result):
            # 微信明确无此单 → 转账从未创建(如崩溃在扣款后/调用前),退款安全
            _refund_withdraw(db, order, reason="微信无此单,已退回")
            db.refresh(order)
        return order  # 其他查询失败不改状态,保持 pending 等下次

    state = result["data"].get("state", "")
    order.wechat_state = state
    if state == _WX_STATE_SUCCESS:
        order.status = "success"
        db.commit()
    elif state in _WX_STATE_FAILED:
        _refund_withdraw(db, order, reason=f"微信转账状态 {state}")
    elif state == _WX_STATE_WAIT_CONFIRM and cancel_if_unconfirmed:
        # 用户从确认页回来了却仍未确认 → 视为放弃:撤销微信单(防事后确认导致重复打款)后退款。
        # 撤单失败(可能已被确认进 ACCEPTED 的竞态)则保持 pending,等下次查询。
        cancel = wxpay.cancel_transfer(out_bill_no)
        if cancel["status_code"] in (200, 202):
            _refund_withdraw(db, order, reason="用户未确认,已撤销提现")
        else:
            db.commit()
    else:
        db.commit()  # ACCEPTED / PROCESSING 等真在途,仍 pending
    db.refresh(order)
    return order


def reconcile_pending_withdraws(db: Session, *, older_than_minutes: int = 15) -> dict:
    """对账:扫描超过 N 分钟仍 pending 的提现单,逐单查微信归一化(成功/退款/撤销未确认)。
    防止"扣了款但转账没发起/没确认"的孤儿单永久锁住用户余额。建议用 cron/systemd timer 定时跑。
    返回 {checked, resolved} 计数。"""
    cutoff = datetime.now(timezone.utc) - timedelta(minutes=older_than_minutes)
    orders = list(
        db.execute(
            select(WithdrawOrder).where(
                WithdrawOrder.status == "pending", WithdrawOrder.created_at < cutoff
            )
        ).scalars().all()
    )
    resolved = 0
    for o in orders:
        try:
            # 老单视同"用户已放弃确认",查到 WAIT_USER_CONFIRM 也撤销退款
            refreshed = refresh_withdraw_status(
                db, o.user_id, o.out_bill_no, cancel_if_unconfirmed=True
            )
            if refreshed.status != "pending":
                resolved += 1
        except Exception:  # noqa: BLE001 — 单笔失败不影响其余,下轮再试
            db.rollback()
            continue
    return {"checked": len(orders), "resolved": resolved}


def list_withdraw_orders(
    db: Session, user_id: int, *, source: str | None = None, limit: int = 20, cursor: int | None = None
) -> tuple[list[WithdrawOrder], int | None]:
    """提现单分页(按 id 倒序,游标式)。source 非空时只返回该账户来源的单(coin_cash / invite_cash)。"""
    stmt = select(WithdrawOrder).where(WithdrawOrder.user_id == user_id)
    if source is not None:
        stmt = stmt.where(WithdrawOrder.source == source)
    if cursor is not None:
        stmt = stmt.where(WithdrawOrder.id < cursor)
    stmt = stmt.order_by(WithdrawOrder.id.desc()).limit(limit)
    items = list(db.execute(stmt).scalars().all())
    next_cursor = items[-1].id if len(items) == limit else None
    return items, next_cursor