feat: 后端接入微信提现人工审核与资金安全校验

提现申请改为先扣款并进入待审核，审核通过后才发起微信打款，审核拒绝或微信失败时自动退款。

新增运营后台提现列表的关键词搜索、日期筛选、快捷筛选和排序参数，并支持批量通过、批量拒绝、批量刷新查单。

新增微信支付配置健康检查、现金账本校验、自动对账 worker 和单实例保护。

新增数据库唯一索引，约束同一用户未完成提现和同一提现单重复退款，提升并发安全性。

app/admin/repositories/queries.py

@@ -5,9 +5,13 @@
 """
 from __future__ import annotations
 
-from sqlalchemy import Select, func, select
+from datetime import datetime, timedelta, timezone
+from zoneinfo import ZoneInfo
+
+from sqlalchemy import Select, asc, desc, func, or_, select
 from sqlalchemy.orm import Session
 
+from app.models.admin import AdminAuditLog
 from app.models.comparison import ComparisonRecord
 from app.models.feedback import Feedback
 from app.models.user import User
@@ -88,15 +92,114 @@ def list_all_withdraw_orders(
     *,
     user_id: int | None = None,
     status: str | None = None,
+    keyword: str | None = None,
+    date_from: datetime | None = None,
+    date_to: datetime | None = None,
+    date_field: str = "created_at",
+    sort_by: str = "created_at",
+    sort_order: str = "desc",
+    quick_filter: str | None = None,
     limit: int = 20,
     cursor: int | None = None,
 ) -> tuple[list[WithdrawOrder], int | None]:
     stmt = select(WithdrawOrder)
+    needs_user_join = bool(keyword and keyword.strip()) or quick_filter == "high_risk"
+    if needs_user_join:
+        stmt = stmt.outerjoin(User, User.id == WithdrawOrder.user_id)
+
     if user_id is not None:
         stmt = stmt.where(WithdrawOrder.user_id == user_id)
     if status:
         stmt = stmt.where(WithdrawOrder.status == status)
-    return cursor_paginate(db, stmt, WithdrawOrder.id, limit=limit, cursor=cursor)
+
+    kw = (keyword or "").strip()
+    if kw:
+        pattern = f"%{kw}%"
+        conditions = [
+            WithdrawOrder.out_bill_no.ilike(pattern),
+            WithdrawOrder.transfer_bill_no.ilike(pattern),
+            WithdrawOrder.user_name.ilike(pattern),
+            WithdrawOrder.wechat_state.ilike(pattern),
+            WithdrawOrder.fail_reason.ilike(pattern),
+            User.phone.ilike(pattern),
+            User.nickname.ilike(pattern),
+            User.wechat_nickname.ilike(pattern),
+        ]
+        if kw.isdigit():
+            conditions.append(WithdrawOrder.user_id == int(kw))
+        stmt = stmt.where(or_(*conditions))
+
+    date_col = WithdrawOrder.updated_at if date_field == "updated_at" else WithdrawOrder.created_at
+    if date_from is not None:
+        stmt = stmt.where(date_col >= _as_utc_naive(date_from))
+    if date_to is not None:
+        stmt = stmt.where(date_col <= _as_utc_naive(date_to))
+
+    now = datetime.now(timezone.utc).replace(tzinfo=None)
+    today_start = (
+        datetime.now(ZoneInfo("Asia/Shanghai"))
+        .replace(hour=0, minute=0, second=0, microsecond=0)
+        .astimezone(timezone.utc)
+        .replace(tzinfo=None)
+    )
+    if quick_filter == "abnormal":
+        stmt = stmt.where(
+            or_(
+                WithdrawOrder.status.in_(("failed", "rejected")),
+                WithdrawOrder.fail_reason.is_not(None),
+                WithdrawOrder.wechat_state.in_(("FAIL", "CANCELLED", "CLOSED")),
+            )
+        )
+    elif quick_filter == "failed":
+        stmt = stmt.where(WithdrawOrder.status == "failed")
+    elif quick_filter == "overdue_reviewing":
+        stmt = stmt.where(
+            WithdrawOrder.status == "reviewing",
+            WithdrawOrder.created_at <= now - timedelta(minutes=30),
+        )
+    elif quick_filter == "pending_overdue":
+        stmt = stmt.where(
+            WithdrawOrder.status == "pending",
+            WithdrawOrder.updated_at <= now - timedelta(minutes=15),
+        )
+    elif quick_filter == "today":
+        stmt = stmt.where(WithdrawOrder.created_at >= today_start)
+    elif quick_filter == "high_risk":
+        stmt = stmt.where(
+            or_(
+                WithdrawOrder.user_name.is_(None),
+                WithdrawOrder.user_name == "",
+                User.status != "active",
+                User.created_at >= now - timedelta(hours=24),
+                WithdrawOrder.status.in_(("failed", "rejected")),
+                WithdrawOrder.fail_reason.is_not(None),
+            )
+        )
+
+    sort_cols = {
+        "id": WithdrawOrder.id,
+        "created_at": WithdrawOrder.created_at,
+        "updated_at": WithdrawOrder.updated_at,
+        "amount_cents": WithdrawOrder.amount_cents,
+    }
+    sort_col = sort_cols.get(sort_by, WithdrawOrder.created_at)
+    order_fn = asc if sort_order == "asc" else desc
+    id_order = asc(WithdrawOrder.id) if sort_order == "asc" else desc(WithdrawOrder.id)
+    stmt = stmt.order_by(order_fn(sort_col), id_order)
+
+    offset = max(cursor or 0, 0)
+    rows = list(db.execute(stmt.offset(offset).limit(limit + 1)).scalars().all())
+    has_more = len(rows) > limit
+    items = rows[:limit]
+    next_cursor = offset + limit if has_more else None
+    return items, next_cursor
+
+
+def _as_utc_naive(value: datetime) -> datetime:
+    """前端传 ISO 时间;DB 当前按 UTC naive 比较最稳(SQLite/本地开发一致)。"""
+    if value.tzinfo is None:
+        return value
+    return value.astimezone(timezone.utc).replace(tzinfo=None)
 
 
 def list_feedbacks(
@@ -122,6 +225,147 @@ def get_withdraw_by_out_bill_no(db: Session, out_bill_no: str) -> WithdrawOrder
     ).scalar_one_or_none()
 
 
+def withdraw_summary(db: Session) -> dict:
+    """提现审核台顶部统计。金额单位:分。"""
+    rows = db.execute(
+        select(
+            WithdrawOrder.status,
+            func.count(WithdrawOrder.id),
+            func.coalesce(func.sum(WithdrawOrder.amount_cents), 0),
+        ).group_by(WithdrawOrder.status)
+    ).all()
+    by_status = {
+        status: {"count": int(count), "amount_cents": int(amount_cents)}
+        for status, count, amount_cents in rows
+    }
+
+    today_start = (
+        datetime.now(ZoneInfo("Asia/Shanghai"))
+        .replace(hour=0, minute=0, second=0, microsecond=0)
+        .astimezone(timezone.utc)
+    )
+
+    def _today_count(status: str) -> int:
+        return db.execute(
+            select(func.count(WithdrawOrder.id)).where(
+                WithdrawOrder.status == status,
+                WithdrawOrder.updated_at >= today_start,
+            )
+        ).scalar_one()
+
+    today_success_amount = db.execute(
+        select(func.coalesce(func.sum(WithdrawOrder.amount_cents), 0)).where(
+            WithdrawOrder.status == "success",
+            WithdrawOrder.updated_at >= today_start,
+        )
+    ).scalar_one()
+
+    return {
+        "reviewing_count": by_status.get("reviewing", {}).get("count", 0),
+        "reviewing_amount_cents": by_status.get("reviewing", {}).get("amount_cents", 0),
+        "pending_count": by_status.get("pending", {}).get("count", 0),
+        "failed_count": by_status.get("failed", {}).get("count", 0),
+        "today_success_count": _today_count("success"),
+        "today_success_amount_cents": int(today_success_amount),
+        "today_rejected_count": _today_count("rejected"),
+    }
+
+
+def list_withdraw_audit_logs(
+    db: Session, out_bill_no: str, *, limit: int = 20
+) -> list[AdminAuditLog]:
+    return list(
+        db.execute(
+            select(AdminAuditLog)
+            .where(AdminAuditLog.target_type == "withdraw", AdminAuditLog.target_id == out_bill_no)
+            .order_by(AdminAuditLog.id.desc())
+            .limit(limit)
+        ).scalars().all()
+    )
+
+
+def withdraw_risk_flags(
+    order: WithdrawOrder,
+    user: User | None,
+    recent_withdraws: list[WithdrawOrder],
+    cash_balance_cents: int,
+) -> tuple[list[str], int]:
+    flags: list[str] = []
+    if not order.user_name:
+        flags.append("缺少提现实名")
+    if user and user.status != "active":
+        flags.append(f"账号状态:{user.status}")
+    if user and user.created_at:
+        created_at = user.created_at.replace(tzinfo=timezone.utc) if user.created_at.tzinfo is None else user.created_at
+        if datetime.now(timezone.utc) - created_at < timedelta(hours=24):
+            flags.append("新注册用户")
+    failed_or_rejected = sum(1 for item in recent_withdraws if item.status in {"failed", "rejected"})
+    if failed_or_rejected:
+        flags.append(f"历史异常提现{failed_or_rejected}笔")
+    recent_reviewing = sum(1 for item in recent_withdraws if item.status == "reviewing")
+    if recent_reviewing >= 3:
+        flags.append(f"待审核提现偏多:{recent_reviewing}笔")
+    if cash_balance_cents < 0:
+        flags.append("现金余额为负")
+    score = min(100, len(flags) * 20 + failed_or_rejected * 10)
+    return flags, score
+
+
+def withdraw_ledger_check(db: Session) -> dict:
+    cash_balance_total = int(
+        db.execute(select(func.coalesce(func.sum(CoinAccount.cash_balance_cents), 0))).scalar_one()
+    )
+    cash_txn_total = int(
+        db.execute(select(func.coalesce(func.sum(CashTransaction.amount_cents), 0))).scalar_one()
+    )
+
+    orders = list(db.execute(select(WithdrawOrder)).scalars().all())
+    cash_txns = list(
+        db.execute(
+            select(CashTransaction).where(
+                CashTransaction.biz_type.in_(("withdraw", "withdraw_refund"))
+            )
+        ).scalars().all()
+    )
+    withdraw_refs = {txn.ref_id for txn in cash_txns if txn.biz_type == "withdraw"}
+    refund_counts: dict[str, int] = {}
+    for txn in cash_txns:
+        if txn.biz_type == "withdraw_refund" and txn.ref_id:
+            refund_counts[txn.ref_id] = refund_counts.get(txn.ref_id, 0) + 1
+
+    missing_withdraw = 0
+    missing_refund = 0
+    refund_on_non_terminal = 0
+    for order in orders:
+        if order.out_bill_no not in withdraw_refs:
+            missing_withdraw += 1
+        has_refund = refund_counts.get(order.out_bill_no, 0) > 0
+        if order.status in {"failed", "rejected"} and not has_refund:
+            missing_refund += 1
+        if has_refund and order.status not in {"failed", "rejected"}:
+            refund_on_non_terminal += 1
+
+    duplicate_refund = sum(1 for count in refund_counts.values() if count > 1)
+    diff = cash_balance_total - cash_txn_total
+    ok = (
+        diff == 0
+        and missing_withdraw == 0
+        and missing_refund == 0
+        and duplicate_refund == 0
+        and refund_on_non_terminal == 0
+    )
+    return {
+        "ok": ok,
+        "cash_balance_total_cents": cash_balance_total,
+        "cash_transaction_total_cents": cash_txn_total,
+        "balance_diff_cents": diff,
+        "missing_withdraw_txn_count": missing_withdraw,
+        "missing_refund_txn_count": missing_refund,
+        "duplicate_refund_txn_count": duplicate_refund,
+        "refund_txn_on_non_terminal_count": refund_on_non_terminal,
+    }
+
+
 def get_user_overview(db: Session, user_id: int) -> dict | None:
     """用户 360 概览:基础资料 + 钱包余额 + 各项 count。历史明细走各自分页接口(带 user_id 过滤)。"""
     user = db.get(User, user_id)

app/admin/routers/withdraw.py

@@ -6,6 +6,7 @@
 """
 from __future__ import annotations
 
+from datetime import datetime
 from typing import Annotated
 
 from fastapi import APIRouter, Depends, HTTPException, Query, Request
@@ -14,7 +15,23 @@ from app.admin.audit import write_audit
 from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
 from app.admin.repositories import queries
 from app.admin.schemas.common import CursorPage
-from app.admin.schemas.wallet import ReconcileResult, WithdrawOrderOut, WithdrawRejectRequest
+from app.admin.schemas.admin import AdminAuditLogOut
+from app.admin.schemas.wallet import (
+    CashTxnOut,
+    ReconcileResult,
+    WithdrawBulkRejectRequest,
+    WithdrawBulkRequest,
+    WithdrawBulkResult,
+    WithdrawBulkItemResult,
+    WithdrawDetailOut,
+    WithdrawLedgerCheckOut,
+    WithdrawOrderOut,
+    WithdrawRejectRequest,
+    WithdrawSummaryOut,
+    WithdrawUserSnapshot,
+    WxpayHealthCheckOut,
+)
+from app.core.config import settings
 from app.integrations import wxpay
 from app.models.admin import AdminUser
 from app.repositories import wallet as wallet_repo
@@ -31,17 +48,143 @@ def list_withdraws(
     db: AdminDb,
     user_id: Annotated[int | None, Query()] = None,
     status: Annotated[str | None, Query()] = None,
+    keyword: Annotated[str | None, Query(max_length=100)] = None,
+    date_from: Annotated[datetime | None, Query()] = None,
+    date_to: Annotated[datetime | None, Query()] = None,
+    date_field: Annotated[str, Query(pattern="^(created_at|updated_at)$")] = "created_at",
+    sort_by: Annotated[
+        str, Query(pattern="^(id|created_at|updated_at|amount_cents)$")
+    ] = "created_at",
+    sort_order: Annotated[str, Query(pattern="^(asc|desc)$")] = "desc",
+    quick_filter: Annotated[
+        str | None,
+        Query(pattern="^(abnormal|failed|overdue_reviewing|pending_overdue|today|high_risk)$"),
+    ] = None,
     limit: Annotated[int, Query(ge=1, le=100)] = 20,
     cursor: Annotated[int | None, Query()] = None,
 ) -> CursorPage[WithdrawOrderOut]:
     items, next_cursor = queries.list_all_withdraw_orders(
-        db, user_id=user_id, status=status, limit=limit, cursor=cursor,
+        db,
+        user_id=user_id,
+        status=status,
+        keyword=keyword,
+        date_from=date_from,
+        date_to=date_to,
+        date_field=date_field,
+        sort_by=sort_by,
+        sort_order=sort_order,
+        quick_filter=quick_filter,
+        limit=limit,
+        cursor=cursor,
     )
     return CursorPage(
         items=[WithdrawOrderOut.model_validate(o) for o in items], next_cursor=next_cursor,
     )
 
 
+@router.get("/summary", response_model=WithdrawSummaryOut, summary="提现审核台统计")
+def withdraws_summary(db: AdminDb) -> WithdrawSummaryOut:
+    return WithdrawSummaryOut(**queries.withdraw_summary(db))
+
+
+@router.get("/health-check", response_model=WxpayHealthCheckOut, summary="提现配置健康检查")
+def withdraw_health_check() -> WxpayHealthCheckOut:
+    private_path = wxpay._resolve_config_path(settings.WXPAY_MCH_PRIVATE_KEY_PATH)  # noqa: SLF001
+    public_path = wxpay._resolve_config_path(settings.WXPAY_PUBLIC_KEY_PATH)  # noqa: SLF001
+    issues: list[str] = []
+
+    private_loadable = False
+    public_loadable = False
+    try:
+        wxpay._load_private_key()  # noqa: SLF001
+        private_loadable = True
+    except Exception as e:  # noqa: BLE001
+        issues.append(f"商户私钥不可用:{e}")
+    try:
+        wxpay._load_public_key()  # noqa: SLF001
+        public_loadable = True
+    except Exception as e:  # noqa: BLE001
+        issues.append(f"微信支付平台公钥不可用:{e}")
+
+    if not settings.wxpay_configured:
+        issues.append("微信支付基础配置不完整")
+    if not settings.WXPAY_AUTH_NOTIFY_URL:
+        issues.append("免确认授权回调地址未配置")
+    if not settings.WITHDRAW_AUTO_RECONCILE_ENABLED:
+        issues.append("自动对账未开启")
+
+    return WxpayHealthCheckOut(
+        ok=not issues,
+        wxpay_configured=settings.wxpay_configured,
+        wxpay_auth_configured=settings.wxpay_auth_configured,
+        private_key_path=str(private_path),
+        private_key_exists=private_path.exists(),
+        private_key_loadable=private_loadable,
+        public_key_path=str(public_path),
+        public_key_exists=public_path.exists(),
+        public_key_loadable=public_loadable,
+        auth_notify_url_configured=bool(settings.WXPAY_AUTH_NOTIFY_URL),
+        auto_reconcile_enabled=settings.WITHDRAW_AUTO_RECONCILE_ENABLED,
+        auto_reconcile_interval_sec=settings.WITHDRAW_AUTO_RECONCILE_INTERVAL_SEC,
+        auto_reconcile_older_than_minutes=settings.WITHDRAW_AUTO_RECONCILE_OLDER_THAN_MINUTES,
+        issues=issues,
+    )
+
+
+@router.get("/ledger-check", response_model=WithdrawLedgerCheckOut, summary="提现资金账本校验")
+def withdraw_ledger_check(db: AdminDb) -> WithdrawLedgerCheckOut:
+    return WithdrawLedgerCheckOut(**queries.withdraw_ledger_check(db))
+
+
+@router.get("/{out_bill_no}", response_model=WithdrawDetailOut, summary="提现单详情")
+def withdraw_detail(out_bill_no: str, db: AdminDb) -> WithdrawDetailOut:
+    order = queries.get_withdraw_by_out_bill_no(db, out_bill_no)
+    if order is None:
+        raise HTTPException(status_code=404, detail="提现单不存在")
+
+    overview = queries.get_user_overview(db, order.user_id)
+    user_snapshot = None
+    if overview is not None:
+        user = overview["user"]
+        user_snapshot = WithdrawUserSnapshot(
+            id=user.id,
+            phone=user.phone,
+            nickname=user.nickname,
+            status=user.status,
+            wechat_nickname=user.wechat_nickname,
+            wechat_avatar_url=user.wechat_avatar_url,
+            created_at=user.created_at,
+            last_login_at=user.last_login_at,
+            cash_balance_cents=overview["cash_balance_cents"],
+            withdraw_total=overview["withdraw_total"],
+            withdraw_success_cents=overview["withdraw_success_cents"],
+        )
+
+    recent_withdraws, _ = queries.list_all_withdraw_orders(
+        db, user_id=order.user_id, limit=5, cursor=None,
+    )
+    recent_cash_transactions, _ = queries.list_all_cash_transactions(
+        db, user_id=order.user_id, limit=8, cursor=None,
+    )
+    audit_logs = queries.list_withdraw_audit_logs(db, out_bill_no, limit=10)
+    risk_flags, risk_score = queries.withdraw_risk_flags(
+        order,
+        overview["user"] if overview else None,
+        recent_withdraws,
+        overview["cash_balance_cents"] if overview else 0,
+    )
+
+    return WithdrawDetailOut(
+        order=WithdrawOrderOut.model_validate(order),
+        user=user_snapshot,
+        risk_flags=risk_flags,
+        risk_score=risk_score,
+        recent_withdraws=[WithdrawOrderOut.model_validate(o) for o in recent_withdraws],
+        recent_cash_transactions=[CashTxnOut.model_validate(t) for t in recent_cash_transactions],
+        audit_logs=[AdminAuditLogOut.model_validate(log) for log in audit_logs],
+    )
+
+
 # 注意:/reconcile 必须在 /{out_bill_no}/refresh 之前声明(静态路径优先于路径参数)
 @router.post("/reconcile", response_model=ReconcileResult, summary="批量对账(扫超时 pending 单)")
 def reconcile(
@@ -61,6 +204,159 @@ def reconcile(
     return ReconcileResult(**result)
 
 
+def _bulk_result(items: list[WithdrawBulkItemResult]) -> WithdrawBulkResult:
+    success = sum(1 for item in items if item.ok)
+    return WithdrawBulkResult(
+        total=len(items),
+        success=success,
+        failed=len(items) - success,
+        items=items,
+    )
+
+
+@router.post("/bulk/refresh", response_model=WithdrawBulkResult, summary="批量刷新查单")
+def bulk_refresh_withdraws(
+    body: WithdrawBulkRequest,
+    request: Request,
+    admin: Annotated[AdminUser, Depends(require_role("finance"))],
+    db: AdminDb,
+) -> WithdrawBulkResult:
+    results: list[WithdrawBulkItemResult] = []
+    ip = get_client_ip(request)
+    for out_bill_no in body.out_bill_nos:
+        try:
+            order = queries.get_withdraw_by_out_bill_no(db, out_bill_no)
+            if order is None:
+                raise wallet_repo.WithdrawOrderNotFound
+            refreshed = wallet_repo.refresh_withdraw_status(
+                db, order.user_id, out_bill_no, cancel_if_unconfirmed=True,
+            )
+            write_audit(
+                db, admin, action="withdraw.refresh", target_type="withdraw",
+                target_id=out_bill_no,
+                detail={
+                    "status": refreshed.status,
+                    "wechat_state": refreshed.wechat_state,
+                    "fail_reason": refreshed.fail_reason,
+                    "bulk": True,
+                },
+                ip=ip, commit=True,
+            )
+            results.append(
+                WithdrawBulkItemResult(
+                    out_bill_no=out_bill_no, ok=True, status=refreshed.status
+                )
+            )
+        except wallet_repo.WithdrawOrderNotFound:
+            db.rollback()
+            results.append(
+                WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error="提现单不存在")
+            )
+        except wxpay.WxPayNotConfiguredError:
+            db.rollback()
+            results.append(
+                WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error="微信支付未配置")
+            )
+        except Exception as e:  # noqa: BLE001 - 批量操作单笔失败不打断整批
+            db.rollback()
+            results.append(
+                WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error=f"系统异常: {e}")
+            )
+    return _bulk_result(results)
+
+
+@router.post("/bulk/approve", response_model=WithdrawBulkResult, summary="批量审核通过并打款")
+def bulk_approve_withdraws(
+    body: WithdrawBulkRequest,
+    request: Request,
+    admin: Annotated[AdminUser, Depends(require_role("finance"))],
+    db: AdminDb,
+) -> WithdrawBulkResult:
+    results: list[WithdrawBulkItemResult] = []
+    ip = get_client_ip(request)
+    for out_bill_no in body.out_bill_nos:
+        try:
+            order = wallet_repo.approve_withdraw(db, out_bill_no)
+            write_audit(
+                db, admin, action="withdraw.approve", target_type="withdraw",
+                target_id=out_bill_no,
+                detail={
+                    "status": order.status,
+                    "wechat_state": order.wechat_state,
+                    "amount_cents": order.amount_cents,
+                    "bulk": True,
+                },
+                ip=ip, commit=True,
+            )
+            results.append(
+                WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=True, status=order.status)
+            )
+        except wallet_repo.WithdrawOrderNotFound:
+            db.rollback()
+            results.append(
+                WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error="提现单不存在")
+            )
+        except wallet_repo.WithdrawNotReviewable as e:
+            db.rollback()
+            results.append(
+                WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error=str(e))
+            )
+        except wxpay.WxPayNotConfiguredError:
+            db.rollback()
+            results.append(
+                WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error="微信支付未配置")
+            )
+        except Exception as e:  # noqa: BLE001 - 批量操作单笔失败不打断整批
+            db.rollback()
+            results.append(
+                WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error=f"系统异常: {e}")
+            )
+    return _bulk_result(results)
+
+
+@router.post("/bulk/reject", response_model=WithdrawBulkResult, summary="批量审核拒绝并退款")
+def bulk_reject_withdraws(
+    body: WithdrawBulkRejectRequest,
+    request: Request,
+    admin: Annotated[AdminUser, Depends(require_role("finance"))],
+    db: AdminDb,
+) -> WithdrawBulkResult:
+    results: list[WithdrawBulkItemResult] = []
+    ip = get_client_ip(request)
+    for out_bill_no in body.out_bill_nos:
+        try:
+            order = wallet_repo.reject_withdraw(db, out_bill_no, body.reason)
+            write_audit(
+                db, admin, action="withdraw.reject", target_type="withdraw",
+                target_id=out_bill_no,
+                detail={
+                    "reason": body.reason,
+                    "amount_cents": order.amount_cents,
+                    "bulk": True,
+                },
+                ip=ip, commit=True,
+            )
+            results.append(
+                WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=True, status=order.status)
+            )
+        except wallet_repo.WithdrawOrderNotFound:
+            db.rollback()
+            results.append(
+                WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error="提现单不存在")
+            )
+        except wallet_repo.WithdrawNotReviewable as e:
+            db.rollback()
+            results.append(
+                WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error=str(e))
+            )
+        except Exception as e:  # noqa: BLE001 - 批量操作单笔失败不打断整批
+            db.rollback()
+            results.append(
+                WithdrawBulkItemResult(out_bill_no=out_bill_no, ok=False, error=f"系统异常: {e}")
+            )
+    return _bulk_result(results)
+
+
 @router.post("/{out_bill_no}/refresh", response_model=WithdrawOrderOut, summary="单笔重试查单")
 def refresh_withdraw(
     out_bill_no: str,
@@ -79,7 +375,11 @@ def refresh_withdraw(
         raise HTTPException(status_code=503, detail="微信支付未配置") from e
     write_audit(
         db, admin, action="withdraw.refresh", target_type="withdraw", target_id=out_bill_no,
-        detail={"status": refreshed.status, "wechat_state": refreshed.wechat_state},
+        detail={
+            "status": refreshed.status,
+            "wechat_state": refreshed.wechat_state,
+            "fail_reason": refreshed.fail_reason,
+        },
         ip=get_client_ip(request), commit=True,
     )
     return WithdrawOrderOut.model_validate(refreshed)

app/admin/schemas/wallet.py

@@ -5,6 +5,8 @@ from datetime import datetime
 
 from pydantic import BaseModel, ConfigDict, Field
 
+from app.admin.schemas.admin import AdminAuditLogOut
+
 
 class CoinTxnOut(BaseModel):
     model_config = ConfigDict(from_attributes=True)
@@ -48,11 +50,99 @@ class WithdrawOrderOut(BaseModel):
     updated_at: datetime
 
 
+class WithdrawSummaryOut(BaseModel):
+    reviewing_count: int
+    reviewing_amount_cents: int
+    pending_count: int
+    failed_count: int
+    today_success_count: int
+    today_success_amount_cents: int
+    today_rejected_count: int
+
+
+class WithdrawUserSnapshot(BaseModel):
+    id: int
+    phone: str
+    nickname: str | None = None
+    status: str
+    wechat_nickname: str | None = None
+    wechat_avatar_url: str | None = None
+    created_at: datetime
+    last_login_at: datetime
+    cash_balance_cents: int
+    withdraw_total: int
+    withdraw_success_cents: int
+
+
+class WithdrawDetailOut(BaseModel):
+    order: WithdrawOrderOut
+    user: WithdrawUserSnapshot | None = None
+    risk_flags: list[str]
+    risk_score: int
+    recent_withdraws: list[WithdrawOrderOut]
+    recent_cash_transactions: list[CashTxnOut]
+    audit_logs: list[AdminAuditLogOut]
+
+
 class ReconcileResult(BaseModel):
     checked: int
     resolved: int
 
 
+class WithdrawBulkRequest(BaseModel):
+    out_bill_nos: list[str] = Field(
+        ..., min_length=1, max_length=50, description="提现商户单号列表"
+    )
+
+
+class WithdrawBulkRejectRequest(WithdrawBulkRequest):
+    reason: str = Field(
+        ..., min_length=1, max_length=200, description="批量拒绝理由(用户可见)"
+    )
+
+
+class WithdrawBulkItemResult(BaseModel):
+    out_bill_no: str
+    ok: bool
+    status: str | None = None
+    error: str | None = None
+
+
+class WithdrawBulkResult(BaseModel):
+    total: int
+    success: int
+    failed: int
+    items: list[WithdrawBulkItemResult]
+
+
+class WithdrawLedgerCheckOut(BaseModel):
+    ok: bool
+    cash_balance_total_cents: int
+    cash_transaction_total_cents: int
+    balance_diff_cents: int
+    missing_withdraw_txn_count: int
+    missing_refund_txn_count: int
+    duplicate_refund_txn_count: int
+    refund_txn_on_non_terminal_count: int
+
+
+class WxpayHealthCheckOut(BaseModel):
+    ok: bool
+    wxpay_configured: bool
+    wxpay_auth_configured: bool
+    private_key_path: str
+    private_key_exists: bool
+    private_key_loadable: bool
+    public_key_path: str
+    public_key_exists: bool
+    public_key_loadable: bool
+    auth_notify_url_configured: bool
+    auto_reconcile_enabled: bool
+    auto_reconcile_interval_sec: int
+    auto_reconcile_older_than_minutes: int
+    issues: list[str]
+
+
 class WithdrawRejectRequest(BaseModel):
     reason: str = Field(
         ..., min_length=1, max_length=200, description="拒绝理由(写入 fail_reason,用户可见)"