OuYingJun1024
a4d214964a
- 数据访问层统一: app/crud/{ad_reward,savings,signin,task,wallet}.py 移入
app/repositories/(与 user.py 同目录),删除 crud/;更新 10 处 import
(api/v1/* + scripts/* + repositories 内部交叉引用 app.crud→app.repositories)
- alembic/versions 9 个迁移文件去掉 hex 前缀,改成可读文件名(如
welfare_tables_coin_account_coin_txn.py);**仅重命名文件**,文件内
revision/down_revision 不动 → 迁移链与已迁移库的 alembic_version 不受影响
(alembic heads/history 验证链完好,单一 head c8d9e0f1a2b3)
- 测试: 37 passed(1 个 coupon 代理失败为连不到 pricebot 上游的环境问题,与本次无关)
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
507 lines
18 KiB
Python
507 lines
18 KiB
Python
"""钱包 CRUD:账户读取 + 金币入账(共享给签到 / 任务)+ 流水查询。
|
|
|
|
`grant_coins` 是所有"发金币"的唯一入口:它更新余额快照并写一笔流水,但**不 commit**,
|
|
由调用方在同一事务里 commit(签到要同时写 signin_record,任务要同时写 user_task,
|
|
保证"记录"和"加金币"原子化,不会只发钱不留痕或反之)。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import select, update
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.rewards import (
|
|
COIN_PER_CENT,
|
|
MIN_EXCHANGE_COIN,
|
|
WITHDRAW_MAX_CENTS,
|
|
WITHDRAW_MIN_CENTS,
|
|
coins_to_cents,
|
|
)
|
|
from app.integrations import wxpay
|
|
from app.models.user import User
|
|
from app.models.wallet import CashTransaction, CoinAccount, CoinTransaction, WithdrawOrder
|
|
|
|
# 微信转账终态:成功 / 失败(失败/取消/关闭都退款)
|
|
_WX_STATE_SUCCESS = "SUCCESS"
|
|
_WX_STATE_FAILED = {"FAIL", "CANCELLED", "CLOSED"}
|
|
_WX_STATE_WAIT_CONFIRM = "WAIT_USER_CONFIRM" # 用户还没在微信确认页确认
|
|
|
|
|
|
class InvalidExchangeAmountError(Exception):
|
|
"""兑换金币数非法(<最小额 或 不是整分倍数)。"""
|
|
|
|
|
|
class InsufficientCoinError(Exception):
|
|
"""金币余额不足。"""
|
|
|
|
|
|
class InvalidWithdrawAmountError(Exception):
|
|
"""提现金额超出允许范围。"""
|
|
|
|
|
|
class WechatNotBoundError(Exception):
|
|
"""用户未绑定微信(无 openid),无法提现。"""
|
|
|
|
|
|
class WechatAlreadyBoundError(Exception):
|
|
"""该微信(openid)已绑定到其他账号。"""
|
|
|
|
|
|
class InsufficientCashError(Exception):
|
|
"""现金余额不足。"""
|
|
|
|
|
|
class WithdrawTransferError(Exception):
|
|
"""调用微信转账失败(已退回余额)。"""
|
|
|
|
|
|
class WithdrawOrderNotFound(Exception):
|
|
"""提现单不存在。"""
|
|
|
|
|
|
def get_or_create_account(db: Session, user_id: int, *, commit: bool = True) -> CoinAccount:
|
|
"""取用户金币账户,不存在则建一个空账户。"""
|
|
acc = 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,
|
|
)
|
|
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
|
|
) -> tuple[CoinAccount, int]:
|
|
"""金币兑现金。返回 (account, 本次兑入的分)。
|
|
|
|
校验:金额 >= MIN_EXCHANGE_COIN 且为整分倍数(InvalidExchangeAmountError),
|
|
余额充足(InsufficientCoinError)。扣金币 + 加现金,两条流水同事务 commit。
|
|
"""
|
|
if coin_amount < MIN_EXCHANGE_COIN 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)
|
|
remark = f"{coin_amount}金币兑{cents}分"
|
|
|
|
# 1. 扣金币(写 coin_transaction)
|
|
grant_coins(db, user_id, -coin_amount, biz_type="exchange_out", remark=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=remark,
|
|
)
|
|
)
|
|
db.commit()
|
|
db.refresh(acc)
|
|
return acc, cents
|
|
|
|
|
|
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 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()
|
|
|
|
|
|
_OUT_BILL_NO_RE = re.compile(r"^[0-9A-Za-z_-]{8,32}$")
|
|
|
|
|
|
def _try_deduct_cash(db: Session, user_id: int, amount_cents: int) -> bool:
|
|
"""原子扣减现金:仅当余额足够时扣,返回是否成功。
|
|
|
|
用带条件的 UPDATE(`WHERE cash_balance_cents >= amount`)避免"读-判断-写"竞态——
|
|
并发/重试时不会两次都通过余额检查导致超额扣款(SQLite 串行写、Postgres 行级,均安全)。
|
|
"""
|
|
res = db.execute(
|
|
update(CoinAccount)
|
|
.where(
|
|
CoinAccount.user_id == user_id,
|
|
CoinAccount.cash_balance_cents >= amount_cents,
|
|
)
|
|
.values(cash_balance_cents=CoinAccount.cash_balance_cents - amount_cents)
|
|
)
|
|
return res.rowcount == 1
|
|
|
|
|
|
def _add_cash(db: Session, user_id: int, amount_cents: int) -> int:
|
|
"""原子增加现金(退款用),返回加后余额。"""
|
|
db.execute(
|
|
update(CoinAccount)
|
|
.where(CoinAccount.user_id == user_id)
|
|
.values(cash_balance_cents=CoinAccount.cash_balance_cents + amount_cents)
|
|
)
|
|
db.flush()
|
|
bal = db.execute(
|
|
select(CoinAccount.cash_balance_cents).where(CoinAccount.user_id == user_id)
|
|
).scalar_one()
|
|
return bal
|
|
|
|
|
|
def _refund_withdraw(db: Session, order: WithdrawOrder, reason: str) -> None:
|
|
"""转账失败/取消:退回现金 + 写 withdraw_refund 流水 + 单子置 failed。幂等:已 failed 不重复退。"""
|
|
if order.status == "failed":
|
|
return # 防重复退款(并发/对账与查单同时触发)
|
|
bal = _add_cash(db, order.user_id, order.amount_cents)
|
|
db.add(
|
|
CashTransaction(
|
|
user_id=order.user_id,
|
|
amount_cents=order.amount_cents,
|
|
balance_after_cents=bal,
|
|
biz_type="withdraw_refund",
|
|
ref_id=order.out_bill_no,
|
|
remark="提现未成功,金额已退回", # 用户可见;技术原因记在 order.fail_reason
|
|
)
|
|
)
|
|
order.status = "failed"
|
|
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,
|
|
*,
|
|
user_name: str | None = None,
|
|
out_bill_no: str | None = None,
|
|
) -> WithdrawOrder:
|
|
"""发起提现(原子扣款 + 幂等 + 模糊失败查单)。
|
|
|
|
#1 用原子条件 UPDATE 扣款,杜绝并发超额。
|
|
#2 out_bill_no 由客户端提供作幂等键:同号重试不会重复转账,直接返回该单当前状态。
|
|
#3 微信调用结果不明时先查单再决定,绝不盲目退款。
|
|
返回提现单(可能含 package_info 供 App 拉起确认页)。
|
|
"""
|
|
if amount_cents < WITHDRAW_MIN_CENTS or amount_cents > WITHDRAW_MAX_CENTS:
|
|
raise InvalidWithdrawAmountError
|
|
|
|
user = db.get(User, user_id)
|
|
openid = user.wechat_openid if user else None
|
|
if not openid:
|
|
raise WechatNotBoundError
|
|
|
|
# #2 幂等:客户端传了合法 out_bill_no 且该单已存在 → 返回现状(pending 的顺手查一下),不重复发起
|
|
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:
|
|
if existing.status == "pending":
|
|
return refresh_withdraw_status(db, user_id, out_bill_no)
|
|
return existing
|
|
else:
|
|
out_bill_no = uuid.uuid4().hex
|
|
|
|
# 账户须存在(原子扣款的 UPDATE 不会建账户)
|
|
get_or_create_account(db, user_id, commit=True)
|
|
|
|
# #1 原子扣款:余额不足时影响行数为 0
|
|
if not _try_deduct_cash(db, user_id, amount_cents):
|
|
db.rollback()
|
|
raise InsufficientCashError
|
|
|
|
bal = db.execute(
|
|
select(CoinAccount.cash_balance_cents).where(CoinAccount.user_id == user_id)
|
|
).scalar_one()
|
|
db.add(
|
|
CashTransaction(
|
|
user_id=user_id,
|
|
amount_cents=-amount_cents,
|
|
balance_after_cents=bal,
|
|
biz_type="withdraw",
|
|
ref_id=out_bill_no,
|
|
remark="提现到微信零钱",
|
|
)
|
|
)
|
|
order = WithdrawOrder(
|
|
user_id=user_id, out_bill_no=out_bill_no, amount_cents=amount_cents, status="pending"
|
|
)
|
|
db.add(order)
|
|
db.commit()
|
|
db.refresh(order)
|
|
|
|
# 2) 调微信转账;结果不明先查单再决定(#3)
|
|
try:
|
|
result = wxpay.create_transfer(openid, amount_cents, out_bill_no, user_name=user_name)
|
|
except Exception as e: # noqa: BLE001 — 超时/网络异常≠失败,查单确认
|
|
_settle_after_ambiguous(db, order, reason=f"转账调用异常: {e}")
|
|
db.refresh(order)
|
|
if order.status == "failed":
|
|
raise WithdrawTransferError(str(e)) from e
|
|
return order
|
|
|
|
if result["status_code"] != 200:
|
|
msg = str(result["data"].get("message") or result["data"])
|
|
_settle_after_ambiguous(db, order, reason=msg)
|
|
db.refresh(order)
|
|
if order.status == "failed":
|
|
raise WithdrawTransferError(msg)
|
|
return order
|
|
|
|
data = result["data"]
|
|
order.wechat_state = data.get("state")
|
|
order.transfer_bill_no = data.get("transfer_bill_no")
|
|
order.package_info = data.get("package_info")
|
|
if data.get("state") == _WX_STATE_SUCCESS:
|
|
order.status = "success"
|
|
db.commit()
|
|
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 # 已终态,不再查
|
|
|
|
result = wxpay.query_transfer(out_bill_no)
|
|
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, *, limit: int = 20, cursor: int | None = None
|
|
) -> tuple[list[WithdrawOrder], int | None]:
|
|
"""提现单分页(按 id 倒序,游标式)。"""
|
|
stmt = select(WithdrawOrder).where(WithdrawOrder.user_id == user_id)
|
|
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
|