diff --git a/.env.example b/.env.example index 20ef954..e4278ed 100644 --- a/.env.example +++ b/.env.example @@ -79,6 +79,10 @@ WXPAY_MCH_PRIVATE_KEY_PATH=./secrets/apiclient_key.pem WXPAY_PUBLIC_KEY_ID=PUB_KEY_ID_xxxxxxxx WXPAY_PUBLIC_KEY_PATH=./secrets/pub_key.pem WXPAY_TRANSFER_SCENE_ID=1000 +WXPAY_AUTH_NOTIFY_URL= +WITHDRAW_AUTO_RECONCILE_ENABLED=false +WITHDRAW_AUTO_RECONCILE_INTERVAL_SEC=300 +WITHDRAW_AUTO_RECONCILE_OLDER_THAN_MINUTES=15 # ===== 穿山甲激励视频(服务端发奖回调)===== # 看完激励视频后穿山甲服务器 S2S 回调本服务发金币(客户端不参与发奖)。 diff --git a/.gitignore b/.gitignore index e1ee6ce..bd48023 100644 --- a/.gitignore +++ b/.gitignore @@ -43,3 +43,4 @@ secrets/* # 运行日志(run.sh 输出, 不入库) *.log +logs/ diff --git a/alembic/versions/withdraw_safety_indexes.py b/alembic/versions/withdraw_safety_indexes.py new file mode 100644 index 0000000..5e5ec9f --- /dev/null +++ b/alembic/versions/withdraw_safety_indexes.py @@ -0,0 +1,42 @@ +"""add withdraw concurrency safety indexes + +Revision ID: withdraw_safety_indexes +Revises: 0cf18d590b1d +Create Date: 2026-06-08 16:20:00.000000 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = "withdraw_safety_indexes" +down_revision: Union[str, Sequence[str], None] = "0cf18d590b1d" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_index( + "ux_withdraw_order_user_active", + "withdraw_order", + ["user_id"], + unique=True, + sqlite_where=sa.text("status IN ('reviewing', 'pending')"), + postgresql_where=sa.text("status IN ('reviewing', 'pending')"), + ) + op.create_index( + "ux_cash_transaction_withdraw_refund_ref", + "cash_transaction", + ["ref_id"], + unique=True, + sqlite_where=sa.text("biz_type = 'withdraw_refund' AND ref_id IS NOT NULL"), + postgresql_where=sa.text("biz_type = 'withdraw_refund' AND ref_id IS NOT NULL"), + ) + + +def downgrade() -> None: + op.drop_index("ux_cash_transaction_withdraw_refund_ref", table_name="cash_transaction") + op.drop_index("ux_withdraw_order_user_active", table_name="withdraw_order") diff --git a/app/admin/repositories/queries.py b/app/admin/repositories/queries.py index ea0160c..9601653 100644 --- a/app/admin/repositories/queries.py +++ b/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) diff --git a/app/admin/routers/withdraw.py b/app/admin/routers/withdraw.py index 06940e3..7fcfff6 100644 --- a/app/admin/routers/withdraw.py +++ b/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) diff --git a/app/admin/schemas/wallet.py b/app/admin/schemas/wallet.py index cd6de21..e2e9874 100644 --- a/app/admin/schemas/wallet.py +++ b/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,用户可见)" diff --git a/app/api/v1/wallet.py b/app/api/v1/wallet.py index c0bc017..8b64c23 100644 --- a/app/api/v1/wallet.py +++ b/app/api/v1/wallet.py @@ -176,7 +176,7 @@ def withdraw_info(user: CurrentUser, db: DbSession) -> WithdrawInfoOut: "/withdraw", response_model=WithdrawResultOut, summary="发起提现(扣款建单,待人工审核;审核通过后才打款)", - dependencies=[Depends(rate_limit(20, 60, "withdraw"))], # #6 同 IP 每分钟≤20 次 + dependencies=[Depends(rate_limit(5, 60, "withdraw"))], # IP 级粗限流;用户级未完成单限制在仓库层 ) def withdraw(req: WithdrawRequest, user: CurrentUser, db: DbSession) -> WithdrawResultOut: # 提现发起本身不调微信(打款在审核通过后),但仍要求微信支付已配置——否则审核通过也打不了款,提前拦 @@ -189,12 +189,17 @@ def withdraw(req: WithdrawRequest, user: CurrentUser, db: DbSession) -> Withdraw except crud_wallet.InvalidWithdrawAmountError as e: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, - detail=f"amount_cents must be within [{WITHDRAW_MIN_CENTS}, {WITHDRAW_MAX_CENTS}]", + detail="提现金额不符合要求", ) from e except crud_wallet.WechatNotBoundError as e: - raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="wechat not bound") from e + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="请先绑定微信") from e + except crud_wallet.WithdrawTooFrequentError as e: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="已有提现申请正在审核或打款中,请处理完成后再申请", + ) from e except crud_wallet.InsufficientCashError as e: - raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="insufficient cash balance") from e + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="现金余额不足") from e # 调试直发(非生产):skip_review=true 时跳过人工审核,立即发起微信转账(等价 admin approve)。 # 双闸保护——客户端仅 debug 包下发此 flag,服务端仅非 prod 才认;任一道闸拦住即恢复正常审核, diff --git a/app/core/config.py b/app/core/config.py index 7daebd0..2742f4d 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -102,6 +102,9 @@ class Settings(BaseSettings): WXPAY_PUBLIC_KEY_PATH: str = "./secrets/pub_key.pem" # 微信支付平台公钥 WXPAY_TRANSFER_SCENE_ID: str = "1000" # 转账场景 ID(1000=现金营销) WXPAY_REQUEST_TIMEOUT_SEC: int = 10 + WITHDRAW_AUTO_RECONCILE_ENABLED: bool = False + WITHDRAW_AUTO_RECONCILE_INTERVAL_SEC: int = 300 + WITHDRAW_AUTO_RECONCILE_OLDER_THAN_MINUTES: int = 15 # 免确认收款授权(用户授权免确认模式)的授权结果回调地址,必须公网可访问 HTTPS、不带参数。 # 发起授权 / 首单顺带授权时作为 authorization_notify_url 传给微信。一期不处理回调内容 # (授权状态靠 query 查询兜底),但微信要求该字段非空,故启用免确认前必须配置;留空时免确认相关接口返回未配置。 diff --git a/app/core/withdraw_reconcile_worker.py b/app/core/withdraw_reconcile_worker.py new file mode 100644 index 0000000..eac6dd6 --- /dev/null +++ b/app/core/withdraw_reconcile_worker.py @@ -0,0 +1,118 @@ +"""提现 pending 单自动对账后台任务。""" +from __future__ import annotations + +import asyncio +import contextlib +import logging +import os +import time +from collections.abc import Iterator +from pathlib import Path + +from sqlalchemy.exc import SQLAlchemyError + +from app.core.config import settings +from app.db.session import SessionLocal +from app.integrations.wxpay import WxPayNotConfiguredError +from app.repositories import wallet as wallet_repo + +logger = logging.getLogger("shagua.withdraw_reconcile") +_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "withdraw_reconcile.lock" + + +def _touch_lock() -> None: + with contextlib.suppress(FileNotFoundError): + os.utime(_LOCK_PATH, None) + + +@contextlib.contextmanager +def _single_instance_lock(stale_after_sec: int) -> Iterator[bool]: + """同机多进程保护:同一时间只允许一个自动查单 worker 运行。""" + _LOCK_PATH.parent.mkdir(parents=True, exist_ok=True) + fd: int | None = None + try: + try: + fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY) + except FileExistsError: + try: + age = time.time() - _LOCK_PATH.stat().st_mtime + except FileNotFoundError: + age = stale_after_sec + 1 + if age > stale_after_sec: + with contextlib.suppress(FileNotFoundError): + _LOCK_PATH.unlink() + try: + fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY) + except FileExistsError: + fd = None + + if fd is None: + yield False + return + + os.write(fd, f"pid={os.getpid()} started_at={int(time.time())}\n".encode("ascii")) + yield True + finally: + if fd is not None: + os.close(fd) + with contextlib.suppress(FileNotFoundError): + _LOCK_PATH.unlink() + + +def _reconcile_once(older_than_minutes: int) -> dict: + with SessionLocal() as db: + return wallet_repo.reconcile_pending_withdraws(db, older_than_minutes=older_than_minutes) + + +async def _run_loop() -> None: + interval = max(30, int(settings.WITHDRAW_AUTO_RECONCILE_INTERVAL_SEC)) + older_than = max(1, int(settings.WITHDRAW_AUTO_RECONCILE_OLDER_THAN_MINUTES)) + lock_stale_after = max(interval * 3, 600) + with _single_instance_lock(lock_stale_after) as lock_acquired: + if not lock_acquired: + logger.warning("withdraw auto reconcile skipped: another worker owns lock") + return + + await _run_locked_loop(interval, older_than) + + +async def _run_locked_loop(interval: int, older_than: int) -> None: + logger.info( + "withdraw auto reconcile started interval=%ss older_than=%sm", + interval, + older_than, + ) + try: + while True: + try: + _touch_lock() + result = await asyncio.to_thread(_reconcile_once, older_than) + if result["checked"] or result["resolved"]: + logger.info("withdraw auto reconcile result=%s", result) + except WxPayNotConfiguredError: + logger.warning("withdraw auto reconcile skipped: wxpay not configured") + except SQLAlchemyError: + logger.exception("withdraw auto reconcile db error") + except Exception: # noqa: BLE001 - 后台任务不能因单次异常退出 + logger.exception("withdraw auto reconcile unexpected error") + await asyncio.sleep(interval) + except asyncio.CancelledError: + logger.info("withdraw auto reconcile stopped") + raise + + +def start_withdraw_reconcile_worker() -> asyncio.Task | None: + if not settings.WITHDRAW_AUTO_RECONCILE_ENABLED: + logger.info("withdraw auto reconcile disabled") + return None + if not settings.wxpay_configured: + logger.warning("withdraw auto reconcile enabled but wxpay not configured") + return asyncio.create_task(_run_loop(), name="withdraw-auto-reconcile") + + +async def stop_withdraw_reconcile_worker(task: asyncio.Task | None) -> None: + if task is None: + return + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task diff --git a/app/integrations/wxpay.py b/app/integrations/wxpay.py index 06e0915..9910c3a 100644 --- a/app/integrations/wxpay.py +++ b/app/integrations/wxpay.py @@ -14,7 +14,9 @@ import base64 import json import time import uuid +from pathlib import Path +import certifi import httpx from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import padding @@ -23,6 +25,7 @@ from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey, RSAPubl from app.core.config import settings _API_HOST = "https://api.mch.weixin.qq.com" +_PROJECT_ROOT = Path(__file__).resolve().parents[2] _TRANSFER_PATH = "/v3/fund-app/mch-transfer/transfer-bills" # 免确认收款授权(用户授权免确认模式) _AUTH_PATH = "/v3/fund-app/mch-transfer/user-confirm-authorization" @@ -38,15 +41,30 @@ class WxPayNotConfiguredError(Exception): """微信支付凭证 / 证书缺失,无法调用。""" +def _http_client() -> httpx.Client: + """微信相关 HTTP 客户端。 + + 显式使用当前 Python 环境的 certifi 证书,避免被外部 SSL_CERT_FILE 指到不存在文件。 + 代理仍允许从环境变量读取,方便本地开发。 + """ + return httpx.Client(verify=certifi.where()) + + +def _resolve_config_path(value: str) -> Path: + path = Path(value) + return path if path.is_absolute() else _PROJECT_ROOT / path + + def _load_private_key() -> RSAPrivateKey: global _private_key if _private_key is None: + key_path = _resolve_config_path(settings.WXPAY_MCH_PRIVATE_KEY_PATH) try: - with open(settings.WXPAY_MCH_PRIVATE_KEY_PATH, "rb") as f: + with key_path.open("rb") as f: _private_key = serialization.load_pem_private_key(f.read(), password=None) except FileNotFoundError as e: raise WxPayNotConfiguredError( - f"商户私钥不存在: {settings.WXPAY_MCH_PRIVATE_KEY_PATH}" + f"商户私钥不存在: {key_path}" ) from e return _private_key @@ -54,12 +72,13 @@ def _load_private_key() -> RSAPrivateKey: def _load_public_key() -> RSAPublicKey: global _public_key if _public_key is None: + key_path = _resolve_config_path(settings.WXPAY_PUBLIC_KEY_PATH) try: - with open(settings.WXPAY_PUBLIC_KEY_PATH, "rb") as f: + with key_path.open("rb") as f: _public_key = serialization.load_pem_public_key(f.read()) except FileNotFoundError as e: raise WxPayNotConfiguredError( - f"微信支付平台公钥不存在: {settings.WXPAY_PUBLIC_KEY_PATH}" + f"微信支付平台公钥不存在: {key_path}" ) from e return _public_key @@ -126,7 +145,7 @@ def create_transfer( "Content-Type": "application/json", "Wechatpay-Serial": settings.WXPAY_PUBLIC_KEY_ID, } - with httpx.Client() as client: + with _http_client() as client: resp = client.post( f"{_API_HOST}{_TRANSFER_PATH}", content=body_str, @@ -143,7 +162,7 @@ def query_transfer(out_bill_no: str) -> dict: "Authorization": _build_authorization("GET", path, ""), "Accept": "application/json", } - with httpx.Client() as client: + with _http_client() as client: resp = client.get( f"{_API_HOST}{path}", headers=headers, timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC ) @@ -159,7 +178,7 @@ def cancel_transfer(out_bill_no: str) -> dict: "Accept": "application/json", "Content-Type": "application/json", } - with httpx.Client() as client: + with _http_client() as client: resp = client.post( f"{_API_HOST}{path}", headers=headers, timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC ) @@ -171,16 +190,17 @@ def code_to_userinfo(code: str) -> dict: 返回 {openid, nickname, avatar_url, raw}。失败抛 ValueError。 注意:微信隐私新政下,部分 app 的 sns/userinfo 可能返回脱敏值(昵称"微信用户"/灰头像); nickname/avatar_url 可能为空,调用方需兜底。""" - r1 = httpx.get( - "https://api.weixin.qq.com/sns/oauth2/access_token", - params={ - "appid": settings.WECHAT_APP_ID, - "secret": settings.WECHAT_APP_SECRET, - "code": code, - "grant_type": "authorization_code", - }, - timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC, - ) + with _http_client() as client: + r1 = client.get( + "https://api.weixin.qq.com/sns/oauth2/access_token", + params={ + "appid": settings.WECHAT_APP_ID, + "secret": settings.WECHAT_APP_SECRET, + "code": code, + "grant_type": "authorization_code", + }, + timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC, + ) d1 = r1.json() if "openid" not in d1 or "access_token" not in d1: raise ValueError(f"微信授权失败: {d1.get('errmsg', d1)}") @@ -191,11 +211,12 @@ def code_to_userinfo(code: str) -> dict: raw: dict = {} # userinfo 拉取失败不应让绑定失败(openid 已拿到),吞掉异常只是没昵称头像 try: - r2 = httpx.get( - "https://api.weixin.qq.com/sns/userinfo", - params={"access_token": d1["access_token"], "openid": openid, "lang": "zh_CN"}, - timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC, - ) + with _http_client() as client: + r2 = client.get( + "https://api.weixin.qq.com/sns/userinfo", + params={"access_token": d1["access_token"], "openid": openid, "lang": "zh_CN"}, + timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC, + ) raw = r2.json() nickname = raw.get("nickname") or None avatar_url = raw.get("headimgurl") or None @@ -244,7 +265,7 @@ def apply_transfer_authorization( "Accept": "application/json", "Content-Type": "application/json", } - with httpx.Client() as client: + with _http_client() as client: resp = client.post( f"{_API_HOST}{_AUTH_PATH}", content=body_str, @@ -262,7 +283,7 @@ def query_transfer_authorization(out_authorization_no: str) -> dict: "Authorization": _build_authorization("GET", path, ""), "Accept": "application/json", } - with httpx.Client() as client: + with _http_client() as client: resp = client.get( f"{_API_HOST}{path}", headers=headers, timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC ) @@ -277,7 +298,7 @@ def close_transfer_authorization(out_authorization_no: str) -> dict: "Accept": "application/json", "Content-Type": "application/json", } - with httpx.Client() as client: + with _http_client() as client: resp = client.post( f"{_API_HOST}{path}", headers=headers, timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC ) @@ -319,7 +340,7 @@ def pre_transfer_with_authorization( "Content-Type": "application/json", "Wechatpay-Serial": settings.WXPAY_PUBLIC_KEY_ID, } - with httpx.Client() as client: + with _http_client() as client: resp = client.post( f"{_API_HOST}{_PRE_TRANSFER_AUTH_PATH}", content=body_str, @@ -356,7 +377,7 @@ def transfer_with_authorization( "Content-Type": "application/json", "Wechatpay-Serial": settings.WXPAY_PUBLIC_KEY_ID, } - with httpx.Client() as client: + with _http_client() as client: resp = client.post( f"{_API_HOST}{_TRANSFER_WITH_AUTH_PATH}", content=body_str, diff --git a/app/main.py b/app/main.py index b905b93..fe296f2 100644 --- a/app/main.py +++ b/app/main.py @@ -35,6 +35,10 @@ from app.api.v1.wallet import router as wallet_router from app.api.v1.wxpay import router as wxpay_router from app.core.config import settings from app.core.logging import setup_logging +from app.core.withdraw_reconcile_worker import ( + start_withdraw_reconcile_worker, + stop_withdraw_reconcile_worker, +) setup_logging(debug=settings.APP_DEBUG) logger = logging.getLogger("shagua.main") @@ -50,8 +54,12 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]: settings.APP_DEBUG, settings.DATABASE_URL.split("://", 1)[0], ) - yield - logger.info("shutting down") + reconcile_task = start_withdraw_reconcile_worker() + try: + yield + finally: + await stop_withdraw_reconcile_worker(reconcile_task) + logger.info("shutting down") app = FastAPI( diff --git a/app/models/wallet.py b/app/models/wallet.py index 9aff2ea..8c5b849 100644 --- a/app/models/wallet.py +++ b/app/models/wallet.py @@ -10,7 +10,7 @@ from __future__ import annotations from datetime import datetime -from sqlalchemy import DateTime, ForeignKey, Integer, String, func +from sqlalchemy import DateTime, ForeignKey, Index, Integer, String, func, text from sqlalchemy.orm import Mapped, mapped_column from app.db.base import Base @@ -76,6 +76,15 @@ class WithdrawOrder(Base): """ __tablename__ = "withdraw_order" + __table_args__ = ( + Index( + "ux_withdraw_order_user_active", + "user_id", + unique=True, + sqlite_where=text("status IN ('reviewing', 'pending')"), + postgresql_where=text("status IN ('reviewing', 'pending')"), + ), + ) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) user_id: Mapped[int] = mapped_column( @@ -148,6 +157,15 @@ class CashTransaction(Base): """现金流水(单位:分)。金币兑现金、提现都记在这里。""" __tablename__ = "cash_transaction" + __table_args__ = ( + Index( + "ux_cash_transaction_withdraw_refund_ref", + "ref_id", + unique=True, + sqlite_where=text("biz_type = 'withdraw_refund' AND ref_id IS NOT NULL"), + postgresql_where=text("biz_type = 'withdraw_refund' AND ref_id IS NOT NULL"), + ), + ) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) user_id: Mapped[int] = mapped_column( diff --git a/app/repositories/wallet.py b/app/repositories/wallet.py index 957a25e..8d73dab 100644 --- a/app/repositories/wallet.py +++ b/app/repositories/wallet.py @@ -12,6 +12,7 @@ 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 @@ -31,6 +32,7 @@ from app.models.wallet import ( _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" # 已关闭(用户/商户/风控),需重新开启 @@ -60,6 +62,10 @@ class InsufficientCashError(Exception): """现金余额不足。""" +class WithdrawTooFrequentError(Exception): + """提现申请过于频繁,或已有未完成提现单。""" + + class WithdrawTransferError(Exception): """调用微信转账失败(已退回余额)。""" @@ -283,6 +289,18 @@ def _refund_withdraw( """ if order.status in ("failed", "rejected"): return # 防重复退款(并发/对账与查单/重复拒绝同时触发) + refunded_txn_id = db.execute( + select(CashTransaction.id).where( + CashTransaction.user_id == order.user_id, + CashTransaction.biz_type == "withdraw_refund", + CashTransaction.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) db.add( CashTransaction( @@ -298,7 +316,30 @@ def _refund_withdraw( ) order.status = final_status order.fail_reason = reason[:256] - db.commit() + 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(CashTransaction.id).where( + CashTransaction.user_id == user_id, + CashTransaction.biz_type == "withdraw_refund", + CashTransaction.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: @@ -385,6 +426,15 @@ def create_withdraw( 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) @@ -415,7 +465,26 @@ def create_withdraw( status="reviewing", ) db.add(order) - db.commit() + 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 # 待管理员审核;**不在此处打款** @@ -703,7 +772,15 @@ def refresh_withdraw_status( if order.status != "pending": return order # 已终态,不再查 - result = wxpay.query_transfer(out_bill_no) + 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): # 微信明确无此单 → 转账从未创建(如崩溃在扣款后/调用前),退款安全