From c0e6801c695f3f1685eafde9db9ecdc232be677e Mon Sep 17 00:00:00 2001 From: unknown <798648091@qq.com> Date: Sun, 26 Jul 2026 11:36:35 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8A=9F=E8=83=BD=EF=BC=9A=E5=AE=8C=E5=96=84?= =?UTF-8?q?=E6=8F=90=E7=8E=B0=E5=AE=A1=E6=A0=B8=E4=B8=8E=E7=94=A8=E6=88=B7?= =?UTF-8?q?=E9=A3=8E=E9=99=A9=E7=AE=A1=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- alembic/versions/user_manual_risk_fields.py | 76 +++++ app/admin/permissions.py | 5 +- app/admin/repositories/mutations.py | 19 ++ app/admin/repositories/queries.py | 336 ++++++++++---------- app/admin/routers/users.py | 56 +++- app/admin/routers/withdraw.py | 81 +++-- app/admin/schemas/user.py | 18 ++ app/admin/schemas/wallet.py | 49 ++- app/models/user.py | 9 +- app/repositories/wallet.py | 143 ++++++++- tests/test_admin_read.py | 163 ++++++++++ tests/test_admin_roles.py | 8 +- tests/test_admin_write.py | 101 +++++- tests/test_invite_cash_withdraw.py | 45 +++ tests/test_withdraw.py | 74 ++++- tests/test_withdraw_ledger_check.py | 151 --------- 16 files changed, 940 insertions(+), 394 deletions(-) create mode 100644 alembic/versions/user_manual_risk_fields.py delete mode 100644 tests/test_withdraw_ledger_check.py diff --git a/alembic/versions/user_manual_risk_fields.py b/alembic/versions/user_manual_risk_fields.py new file mode 100644 index 0000000..9c3616e --- /dev/null +++ b/alembic/versions/user_manual_risk_fields.py @@ -0,0 +1,76 @@ +"""add manual high-risk flag and note to user + +Revision ID: user_manual_risk_fields +Revises: risk_monitor_generic +""" +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +from alembic import op + +revision: str = "user_manual_risk_fields" +down_revision: str | None = "risk_monitor_generic" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql") +_INVITE_WITHDRAW_PAGE = "invite-withdraws" + + +def _finance_role() -> sa.TableClause: + return sa.table( + "admin_role", + sa.column("name", sa.String), + sa.column("pages", _JSON), + ) + + +def upgrade() -> None: + with op.batch_alter_table("user") as batch_op: + batch_op.add_column( + sa.Column( + "is_high_risk", + sa.Boolean(), + server_default=sa.false(), + nullable=False, + ) + ) + batch_op.add_column(sa.Column("high_risk_note", sa.Text(), nullable=True)) + batch_op.create_index("ix_user_is_high_risk", ["is_high_risk"], unique=False) + role = _finance_role() + conn = op.get_bind() + row = conn.execute( + sa.select(role.c.pages).where(role.c.name == "finance") + ).scalar_one_or_none() + if row is not None and _INVITE_WITHDRAW_PAGE not in (row or []): + conn.execute( + role.update() + .where(role.c.name == "finance") + .values(pages=[*(row or []), _INVITE_WITHDRAW_PAGE]) + ) + + +def downgrade() -> None: + role = _finance_role() + conn = op.get_bind() + row = conn.execute( + sa.select(role.c.pages).where(role.c.name == "finance") + ).scalar_one_or_none() + if row is not None: + conn.execute( + role.update() + .where(role.c.name == "finance") + .values( + pages=[ + page for page in (row or []) if page != _INVITE_WITHDRAW_PAGE + ] + ) + ) + with op.batch_alter_table("user") as batch_op: + batch_op.drop_index("ix_user_is_high_risk") + batch_op.drop_column("high_risk_note") + batch_op.drop_column("is_high_risk") diff --git a/app/admin/permissions.py b/app/admin/permissions.py index e8158c1..6ac62b4 100644 --- a/app/admin/permissions.py +++ b/app/admin/permissions.py @@ -23,7 +23,8 @@ PERMISSION_CATALOG: list[dict] = [ {"key": "cps", "label": "CPS收益"}, ]}, {"group": "奖励审核", "pages": [ - {"key": "withdraws", "label": "提现审核"}, + {"key": "invite-withdraws", "label": "邀请提现审核"}, + {"key": "withdraws", "label": "其他提现审核"}, {"key": "price-reports", "label": "低价审核"}, {"key": "feedbacks", "label": "用户反馈"}, ]}, @@ -59,7 +60,7 @@ BUILTIN_ROLES: list[dict] = [ "cps", "risk-monitor", "device-liveness", "price-reports", "feedbacks", "huawei-review", ]}, {"name": "finance", "label": "财务", "pages": [ - "dashboard", "ad-revenue-report", "cps", "withdraws", + "dashboard", "ad-revenue-report", "cps", "invite-withdraws", "withdraws", ]}, {"name": "tech", "label": "技术", "pages": [ "dashboard", "risk-monitor", "device-liveness", "analytics-health", "config", "ad-revenue", "huawei-review", diff --git a/app/admin/repositories/mutations.py b/app/admin/repositories/mutations.py index f667811..7fd4cf5 100644 --- a/app/admin/repositories/mutations.py +++ b/app/admin/repositories/mutations.py @@ -44,6 +44,25 @@ def set_user_debug_trace( return user +def set_user_risk( + db: Session, + user: User, + *, + is_high_risk: bool, + note: str | None, + commit: bool = True, +) -> User: + """设置人工风险结论和备注，支持与审计日志共用同一事务。""" + user.is_high_risk = is_high_risk + user.high_risk_note = note.strip() if is_high_risk and note else None + if commit: + db.commit() + db.refresh(user) + else: + db.flush() + return user + + def update_feedback_status( db: Session, feedback: Feedback, *, status: str, commit: bool = True ) -> Feedback: diff --git a/app/admin/repositories/queries.py b/app/admin/repositories/queries.py index 75550e7..3a9eb94 100644 --- a/app/admin/repositories/queries.py +++ b/app/admin/repositories/queries.py @@ -22,14 +22,15 @@ from app.models.comparison import ComparisonRecord from app.models.coupon_state import CouponPromptEngagement from app.models.device import DeviceLiveness from app.models.feedback import Feedback +from app.models.invite import InviteRelation from app.models.onboarding import OnboardingCompletion from app.models.price_report import PriceReport +from app.models.savings import SavingsRecord from app.models.user import User from app.models.wallet import ( CashTransaction, CoinAccount, CoinTransaction, - InviteCashTransaction, WithdrawOrder, ) from app.repositories import activity, ad_ecpm @@ -710,16 +711,7 @@ def list_all_withdraw_orders( 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), - ) - ) + stmt = stmt.where(User.is_high_risk.is_(True)) sort_cols = { "id": WithdrawOrder.id, @@ -745,7 +737,9 @@ def _as_utc(value: datetime) -> datetime: return value.astimezone(timezone.utc) -def withdraw_list_enrichment(db: Session, user_ids: list[int]) -> dict[int, dict]: +def withdraw_list_enrichment( + db: Session, user_ids: list[int], *, source: str | None = None +) -> dict[int, dict]: """批量富化提现单列表:按本页 user_id 取 手机号/昵称 + 各自累计成功提现金额(分)。 两条聚合查询搞定(避免逐行 N+1)。累计口径与 get_user_overview 的 withdraw_success_cents @@ -755,25 +749,32 @@ def withdraw_list_enrichment(db: Session, user_ids: list[int]) -> dict[int, dict uniq = list(set(user_ids)) if not uniq: return {} - success_rows = db.execute( - select( - WithdrawOrder.user_id, - func.coalesce(func.sum(WithdrawOrder.amount_cents), 0), - ) - .where(WithdrawOrder.user_id.in_(uniq), WithdrawOrder.status == "success") - .group_by(WithdrawOrder.user_id) - ).all() + success_stmt = select( + WithdrawOrder.user_id, + func.coalesce(func.sum(WithdrawOrder.amount_cents), 0), + ).where(WithdrawOrder.user_id.in_(uniq), WithdrawOrder.status == "success") + if source: + success_stmt = success_stmt.where(WithdrawOrder.source == source) + success_rows = db.execute(success_stmt.group_by(WithdrawOrder.user_id)).all() success_map = {uid: int(total) for uid, total in success_rows} users = db.execute( - select(User.id, User.phone, User.nickname).where(User.id.in_(uniq)) + select( + User.id, + User.phone, + User.nickname, + User.is_high_risk, + User.high_risk_note, + ).where(User.id.in_(uniq)) ).all() return { uid: { "phone": phone, "nickname": nickname, + "is_high_risk": is_high_risk, + "high_risk_note": high_risk_note, "cumulative_success_cents": success_map.get(uid, 0), } - for uid, phone, nickname in users + for uid, phone, nickname, is_high_risk, high_risk_note in users } @@ -879,15 +880,16 @@ def get_withdraw_by_out_bill_no(db: Session, out_bill_no: str) -> WithdrawOrder ).scalar_one_or_none() -def withdraw_summary(db: Session) -> dict: +def withdraw_summary(db: Session, *, source: str | None = None) -> dict: """提现审核台顶部统计。金额单位:分。""" - rows = db.execute( - select( - WithdrawOrder.status, - func.count(WithdrawOrder.id), - func.coalesce(func.sum(WithdrawOrder.amount_cents), 0), - ).group_by(WithdrawOrder.status) - ).all() + summary_stmt = select( + WithdrawOrder.status, + func.count(WithdrawOrder.id), + func.coalesce(func.sum(WithdrawOrder.amount_cents), 0), + ) + if source: + summary_stmt = summary_stmt.where(WithdrawOrder.source == source) + rows = db.execute(summary_stmt.group_by(WithdrawOrder.status)).all() by_status = { status: {"count": int(count), "amount_cents": int(amount_cents)} for status, count, amount_cents in rows @@ -900,19 +902,23 @@ def withdraw_summary(db: Session) -> dict: ) 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", + stmt = select(func.count(WithdrawOrder.id)).where( + WithdrawOrder.status == status, WithdrawOrder.updated_at >= today_start, ) - ).scalar_one() + if source: + stmt = stmt.where(WithdrawOrder.source == source) + return db.execute(stmt).scalar_one() + + today_amount_stmt = select( + func.coalesce(func.sum(WithdrawOrder.amount_cents), 0) + ).where( + WithdrawOrder.status == "success", + WithdrawOrder.updated_at >= today_start, + ) + if source: + today_amount_stmt = today_amount_stmt.where(WithdrawOrder.source == source) + today_success_amount = db.execute(today_amount_stmt).scalar_one() return { "reviewing_count": by_status.get("reviewing", {}).get("count", 0), @@ -925,6 +931,117 @@ def withdraw_summary(db: Session) -> dict: } +def invite_overview( + db: Session, + inviter_user_id: int, + *, + date_from: datetime | None = None, + date_to: datetime | None = None, +) -> dict: + """邀请提现详情：邀请关系统计及受邀用户首次比价/首单信息。 + + 首次记录按自增 id 取最早一条，批量查询避免按受邀用户逐行查询。 + 邀请成功口径为完成首次比价并已发邀请奖励(compare_reward_granted)。 + """ + relation_stmt = ( + select(InviteRelation, User) + .join(User, User.id == InviteRelation.invitee_user_id) + .where(InviteRelation.inviter_user_id == inviter_user_id) + .order_by(InviteRelation.created_at.asc(), InviteRelation.id.asc()) + ) + if date_from is not None: + relation_stmt = relation_stmt.where(User.created_at >= _as_utc_naive(date_from)) + if date_to is not None: + relation_stmt = relation_stmt.where(User.created_at <= _as_utc_naive(date_to)) + relation_rows = db.execute(relation_stmt).all() + invitee_ids = [relation.invitee_user_id for relation, _ in relation_rows] + if not invitee_ids: + return {"invite_total": 0, "invite_success_total": 0, "items": []} + + first_compare_ids = list( + db.execute( + select(func.min(ComparisonRecord.id)) + .where( + ComparisonRecord.user_id.in_(invitee_ids), + ComparisonRecord.status == "success", + ) + .group_by(ComparisonRecord.user_id) + ).scalars() + ) + comparisons = ( + db.execute( + select( + ComparisonRecord.user_id, + ComparisonRecord.store_name, + ComparisonRecord.product_names, + ).where(ComparisonRecord.id.in_(first_compare_ids)) + ).all() + if first_compare_ids + else [] + ) + comparison_map = { + user_id: (store_name, product_names) + for user_id, store_name, product_names in comparisons + } + + first_order_ids = list( + db.execute( + select(func.min(SavingsRecord.id)) + .where( + SavingsRecord.user_id.in_(invitee_ids), + SavingsRecord.source == "compare", + ) + .group_by(SavingsRecord.user_id) + ).scalars() + ) + orders = ( + db.execute( + select( + SavingsRecord.user_id, + SavingsRecord.shop_name, + SavingsRecord.title, + SavingsRecord.dishes, + SavingsRecord.order_amount_cents, + ).where(SavingsRecord.id.in_(first_order_ids)) + ).all() + if first_order_ids + else [] + ) + order_map = { + user_id: ( + shop_name, + title or "、".join(dishes or []), + order_amount_cents, + ) + for user_id, shop_name, title, dishes, order_amount_cents in orders + } + + items = [] + for relation, user in relation_rows: + compare_store, compare_products = comparison_map.get(user.id, (None, None)) + order_store, order_products, order_amount = order_map.get( + user.id, (None, None, None) + ) + items.append( + { + "user_id": user.id, + "phone": user.phone, + "registered_at": user.created_at, + "invite_success": bool(relation.compare_reward_granted), + "first_compare_store": compare_store, + "first_compare_products": compare_products, + "first_order_store": order_store, + "first_order_products": order_products, + "first_order_amount_cents": order_amount, + } + ) + return { + "invite_total": len(items), + "invite_success_total": sum(1 for item in items if item["invite_success"]), + "items": items, + } + + def list_withdraw_audit_logs( db: Session, out_bill_no: str, *, limit: int = 20 ) -> list[AdminAuditLog]: @@ -945,8 +1062,6 @@ def withdraw_risk_flags( 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: @@ -969,125 +1084,6 @@ def withdraw_risk_flags( return flags, score -def _check_withdraw_ledger_side( - orders: list[WithdrawOrder], txns: list, *, withdraw_biz: str, refund_biz: str -) -> dict: - """对某一本账(普通现金 / 邀请奖励金)做提现单 ↔ 流水的交叉校验。 - - orders 已按 source 过滤到本账;txns 是本账流水表里 withdraw_biz/refund_biz 两类流水。 - 规则:每单发起应有一条扣款流水(ref_id=out_bill_no);失败/拒绝单应有且仅一条退款流水; - 非退款终态不应出现退款流水。四个计数全为 0 即本账自洽。 - """ - withdraw_refs = {txn.ref_id for txn in txns if txn.biz_type == withdraw_biz} - refund_counts: dict[str, int] = {} - for txn in txns: - if txn.biz_type == refund_biz 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 - - return { - "missing_withdraw": missing_withdraw, - "missing_refund": missing_refund, - "duplicate_refund": sum(1 for count in refund_counts.values() if count > 1), - "refund_on_non_terminal": refund_on_non_terminal, - } - - -def withdraw_ledger_check(db: Session) -> dict: - """现金账本校验:两本物理隔离的账各自对账(产品红线:coin_cash / invite_cash 不串)。 - - 普通现金:CoinAccount.cash_balance_cents ↔ cash_transaction(withdraw/withdraw_refund); - 邀请奖励金:CoinAccount.invite_cash_balance_cents ↔ invite_cash_transaction - (invite_withdraw/invite_withdraw_refund)。 - 提现单按 source 分流到对应账核对——邀请提现的流水写在 invite_cash_transaction 表, - 绝不能拿去和普通现金流水比(否则每笔邀请提现单都会被误报「缺扣款/缺退款流水」)。 - 分流口径与 create_withdraw 一致:仅 source==invite_cash 走邀请账,其余(含历史空值)归普通现金。 - """ - orders = list(db.execute(select(WithdrawOrder)).scalars().all()) - coin_orders = [o for o in orders if o.source != "invite_cash"] - invite_orders = [o for o in orders if o.source == "invite_cash"] - - # —— 普通现金账(coin_cash) —— - 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() - ) - cash_txns = list( - db.execute( - select(CashTransaction).where( - CashTransaction.biz_type.in_(("withdraw", "withdraw_refund")) - ) - ).scalars().all() - ) - coin = _check_withdraw_ledger_side( - coin_orders, cash_txns, withdraw_biz="withdraw", refund_biz="withdraw_refund" - ) - cash_diff = cash_balance_total - cash_txn_total - - # —— 邀请奖励金账(invite_cash,独立账户 + 独立流水表) —— - invite_balance_total = int( - db.execute( - select(func.coalesce(func.sum(CoinAccount.invite_cash_balance_cents), 0)) - ).scalar_one() - ) - invite_txn_total = int( - db.execute( - select(func.coalesce(func.sum(InviteCashTransaction.amount_cents), 0)) - ).scalar_one() - ) - invite_txns = list( - db.execute( - select(InviteCashTransaction).where( - InviteCashTransaction.biz_type.in_(("invite_withdraw", "invite_withdraw_refund")) - ) - ).scalars().all() - ) - invite = _check_withdraw_ledger_side( - invite_orders, invite_txns, - withdraw_biz="invite_withdraw", refund_biz="invite_withdraw_refund", - ) - invite_diff = invite_balance_total - invite_txn_total - - ok = ( - cash_diff == 0 - and invite_diff == 0 - and all(v == 0 for v in coin.values()) - and all(v == 0 for v in invite.values()) - ) - return { - "ok": ok, - # 普通现金账(coin_cash:金币兑换的现金) - "cash_balance_total_cents": cash_balance_total, - "cash_transaction_total_cents": cash_txn_total, - "balance_diff_cents": cash_diff, - "missing_withdraw_txn_count": coin["missing_withdraw"], - "missing_refund_txn_count": coin["missing_refund"], - "duplicate_refund_txn_count": coin["duplicate_refund"], - "refund_txn_on_non_terminal_count": coin["refund_on_non_terminal"], - # 邀请奖励金账(invite_cash:与普通现金物理隔离,各自对账) - "invite_cash_balance_total_cents": invite_balance_total, - "invite_cash_transaction_total_cents": invite_txn_total, - "invite_balance_diff_cents": invite_diff, - "invite_missing_withdraw_txn_count": invite["missing_withdraw"], - "invite_missing_refund_txn_count": invite["missing_refund"], - "invite_duplicate_refund_txn_count": invite["duplicate_refund"], - "invite_refund_txn_on_non_terminal_count": invite["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) @@ -1147,6 +1143,7 @@ def user_reward_stats( *, date_from: datetime | None = None, date_to: datetime | None = None, + withdraw_source: str | None = None, ) -> dict: """提现详情「用户统计区」10 项。窗口作用于除「现金余额」外的所有项(余额是当前快照)。 @@ -1154,22 +1151,35 @@ def user_reward_stats( 平均 eCPM 用原始分值(分/千次)按记录取算术平均;各「提现」= 该来源累计金币折现。 传统任务 = 窗口内正向金币中,排除广告(reward_video/feed_ad_reward)与人工调整后的折现。 """ + withdraw_source_conds = ( + [WithdrawOrder.source == withdraw_source] if withdraw_source else [] + ) wd_success = db.execute( select(func.coalesce(func.sum(WithdrawOrder.amount_cents), 0)).where( WithdrawOrder.user_id == user_id, WithdrawOrder.status == "success", + *withdraw_source_conds, *_window_conds(WithdrawOrder.created_at, date_from, date_to), ) ).scalar_one() wd_total = db.execute( select(func.count(WithdrawOrder.id)).where( WithdrawOrder.user_id == user_id, + *withdraw_source_conds, *_window_conds(WithdrawOrder.created_at, date_from, date_to), ) ).scalar_one() acc = db.get(CoinAccount, user_id) # 现金余额:当前快照,不随窗口 - cash_balance = acc.cash_balance_cents if acc else 0 + cash_balance = ( + ( + acc.invite_cash_balance_cents + if withdraw_source == "invite_cash" + else acc.cash_balance_cents + ) + if acc + else 0 + ) # 只投影本统计实际使用的列。避免滚动发布或旧本地库尚未补齐无关新列时， # SQLAlchemy 因 select(ORM) 自动展开整表字段而让提现详情整体 500。 diff --git a/app/admin/routers/users.py b/app/admin/routers/users.py index fce483a..bed727e 100644 --- a/app/admin/routers/users.py +++ b/app/admin/routers/users.py @@ -16,6 +16,7 @@ from app.admin.schemas.user import ( GrantCashRequest, GrantCoinsRequest, SetDebugTraceRequest, + SetUserRiskRequest, SetUserStatusRequest, UserCoinRecord, UserRewardStats, @@ -84,12 +85,21 @@ def get_user_reward_stats( db: AdminDb, date_from: Annotated[datetime | None, Query()] = None, date_to: Annotated[datetime | None, Query()] = None, + withdraw_source: Annotated[ + str | None, Query(pattern="^(coin_cash|invite_cash)$") + ] = None, ) -> UserRewardStats: """提现详情抽屉「用户统计区」。date_from/date_to 都不传 = 注册至今(全量)。""" if user_repo.get_user_by_id(db, user_id) is None: raise HTTPException(status_code=404, detail="用户不存在") return UserRewardStats( - **queries.user_reward_stats(db, user_id, date_from=date_from, date_to=date_to) + **queries.user_reward_stats( + db, + user_id, + date_from=date_from, + date_to=date_to, + withdraw_source=withdraw_source, + ) ) @@ -138,6 +148,50 @@ def set_user_status( return OkResponse() +@router.post("/{user_id}/risk", response_model=OkResponse, summary="设置人工高风险标记与备注") +def set_user_risk( + user_id: int, + body: SetUserRiskRequest, + request: Request, + admin: Annotated[ + AdminUser, Depends(require_role("operator", "finance")) + ], + db: AdminDb, +) -> OkResponse: + user = user_repo.get_user_by_id(db, user_id) + if user is None: + raise HTTPException(status_code=404, detail="用户不存在") + before = { + "is_high_risk": user.is_high_risk, + "high_risk_note": user.high_risk_note, + } + mutations.set_user_risk( + db, + user, + is_high_risk=body.is_high_risk, + note=body.note, + commit=False, + ) + write_audit( + db, + admin, + action="user.risk.set", + target_type="user", + target_id=user_id, + detail={ + "before": before, + "after": { + "is_high_risk": user.is_high_risk, + "high_risk_note": user.high_risk_note, + }, + }, + ip=get_client_ip(request), + commit=False, + ) + db.commit() + return OkResponse() + + @router.post("/{user_id}/debug-trace", response_model=OkResponse, summary="开关调试链接权限") def set_user_debug_trace( user_id: int, diff --git a/app/admin/routers/withdraw.py b/app/admin/routers/withdraw.py index fa5035f..f12dc25 100644 --- a/app/admin/routers/withdraw.py +++ b/app/admin/routers/withdraw.py @@ -1,4 +1,4 @@ -"""admin 提现:列表(读)+ 单笔重试查单 + 批量对账(写,带审计)。 +"""admin 提现:列表(读)+ 单笔/批量查单与审核操作(写,带审计)。 提现的钱逻辑(查微信/退款/撤单/幂等)全部复用 app.repositories.wallet,admin 只触发 + 记审计。 这些 wallet 函数内部各自 commit(涉及微信调用),审计在其后单独 commit:操作本身幂等, @@ -14,17 +14,16 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Request 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.admin import AdminAuditLogOut +from app.admin.schemas.common import CursorPage from app.admin.schemas.wallet import ( CashTxnOut, - ReconcileResult, + InviteOverviewOut, + WithdrawBulkItemResult, WithdrawBulkRejectRequest, WithdrawBulkRequest, WithdrawBulkResult, - WithdrawBulkItemResult, WithdrawDetailOut, - WithdrawLedgerCheckOut, WithdrawListItemOut, WithdrawOrderOut, WithdrawRejectRequest, @@ -83,8 +82,16 @@ def list_withdraws( cursor=cursor, ) # 联表带出手机号/昵称 + 累计成功提现(本页 user_id 批量富化,2 条聚合查询,无 N+1)。 - enrichment = queries.withdraw_list_enrichment(db, [o.user_id for o in items]) - _empty = {"phone": None, "nickname": None, "cumulative_success_cents": 0} + enrichment = queries.withdraw_list_enrichment( + db, [o.user_id for o in items], source=source + ) + _empty = { + "phone": None, + "nickname": None, + "is_high_risk": False, + "high_risk_note": None, + "cumulative_success_cents": 0, + } out_items = [ WithdrawListItemOut.model_validate(o).model_copy( update=enrichment.get(o.user_id, _empty) @@ -95,8 +102,13 @@ def list_withdraws( @router.get("/summary", response_model=WithdrawSummaryOut, summary="提现审核台统计") -def withdraws_summary(db: AdminDb) -> WithdrawSummaryOut: - return WithdrawSummaryOut(**queries.withdraw_summary(db)) +def withdraws_summary( + db: AdminDb, + source: Annotated[ + str | None, Query(pattern="^(coin_cash|invite_cash)$") + ] = None, +) -> WithdrawSummaryOut: + return WithdrawSummaryOut(**queries.withdraw_summary(db, source=source)) @router.get( @@ -153,13 +165,13 @@ def withdraw_health_check(db: AdminDb) -> WxpayHealthCheckOut: ) -@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: +def withdraw_detail( + out_bill_no: str, + db: AdminDb, + date_from: Annotated[datetime | None, Query()] = None, + date_to: Annotated[datetime | None, Query()] = None, +) -> WithdrawDetailOut: order = queries.get_withdraw_by_out_bill_no(db, out_bill_no) if order is None: raise HTTPException(status_code=404, detail="提现单不存在") @@ -172,6 +184,8 @@ def withdraw_detail(out_bill_no: str, db: AdminDb) -> WithdrawDetailOut: id=user.id, phone=user.phone, nickname=user.nickname, + is_high_risk=user.is_high_risk, + high_risk_note=user.high_risk_note, status=user.status, wechat_nickname=user.wechat_nickname, wechat_avatar_url=user.wechat_avatar_url, @@ -195,37 +209,36 @@ def withdraw_detail(out_bill_no: str, db: AdminDb) -> WithdrawDetailOut: recent_withdraws, overview["cash_balance_cents"] if overview else 0, ) + detail_enrichment = queries.withdraw_list_enrichment( + db, [order.user_id], source=order.source + ).get(order.user_id, {}) return WithdrawDetailOut( order=WithdrawOrderOut.model_validate(order), user=user_snapshot, + cumulative_success_cents=int( + detail_enrichment.get("cumulative_success_cents", 0) + ), 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], + invite_overview=( + InviteOverviewOut( + **queries.invite_overview( + db, + order.user_id, + date_from=date_from, + date_to=date_to, + ) + ) + if order.source == "invite_cash" + else None + ), ) -# 注意:/reconcile 必须在 /{out_bill_no}/refresh 之前声明(静态路径优先于路径参数) -@router.post("/reconcile", response_model=ReconcileResult, summary="批量对账(扫超时 pending 单)") -def reconcile( - request: Request, - admin: Annotated[AdminUser, Depends(require_role("finance"))], - db: AdminDb, - older_than_minutes: Annotated[int, Query(ge=0)] = 15, -) -> ReconcileResult: - try: - result = wallet_repo.reconcile_pending_withdraws(db, older_than_minutes=older_than_minutes) - except wxpay.WxPayNotConfiguredError as e: - raise HTTPException(status_code=503, detail="微信支付未配置") from e - write_audit( - db, admin, action="withdraw.reconcile", target_type="withdraw", target_id=None, - detail=result, ip=get_client_ip(request), commit=True, - ) - return ReconcileResult(**result) - - def _bulk_result(items: list[WithdrawBulkItemResult]) -> WithdrawBulkResult: success = sum(1 for item in items if item.ok) return WithdrawBulkResult( diff --git a/app/admin/schemas/user.py b/app/admin/schemas/user.py index 7d673a6..aa826da 100644 --- a/app/admin/schemas/user.py +++ b/app/admin/schemas/user.py @@ -16,6 +16,8 @@ class AdminUserListItem(BaseModel): register_channel: str status: str debug_trace_enabled: bool = False + is_high_risk: bool = False + high_risk_note: str | None = None wechat_openid: str | None = None wechat_nickname: str | None = None created_at: datetime @@ -118,3 +120,19 @@ class SetUserStatusRequest(BaseModel): class SetDebugTraceRequest(BaseModel): enabled: bool = Field(..., description="是否给该用户开「复制调试链接」权限") + + +class SetUserRiskRequest(BaseModel): + is_high_risk: bool = Field(..., description="是否标记为高风险用户") + note: str | None = Field( + None, + max_length=500, + description="高风险备注；标记高风险时必填，解除后清空", + ) + + @field_validator("note") + @classmethod + def validate_note(cls, value: str | None, info): + if info.data.get("is_high_risk") and not (value or "").strip(): + raise ValueError("标记高风险时必须填写原因") + return value.strip() if value and value.strip() else None diff --git a/app/admin/schemas/wallet.py b/app/admin/schemas/wallet.py index 7f61d7b..db12793 100644 --- a/app/admin/schemas/wallet.py +++ b/app/admin/schemas/wallet.py @@ -60,6 +60,8 @@ class WithdrawListItemOut(WithdrawOrderOut): phone: str | None = None nickname: str | None = None + is_high_risk: bool = False + high_risk_note: str | None = None cumulative_success_cents: int = 0 # 累计成功提现 = SUM(amount_cents) WHERE status='success' @@ -77,6 +79,8 @@ class WithdrawUserSnapshot(BaseModel): id: int phone: str nickname: str | None = None + is_high_risk: bool = False + high_risk_note: str | None = None status: str wechat_nickname: str | None = None wechat_avatar_url: str | None = None @@ -87,19 +91,34 @@ class WithdrawUserSnapshot(BaseModel): withdraw_success_cents: int +class InviteeDetailOut(BaseModel): + user_id: int + phone: str + registered_at: datetime + invite_success: bool + first_compare_store: str | None = None + first_compare_products: str | None = None + first_order_store: str | None = None + first_order_products: str | None = None + first_order_amount_cents: int | None = None + + +class InviteOverviewOut(BaseModel): + invite_total: int + invite_success_total: int + items: list[InviteeDetailOut] + + class WithdrawDetailOut(BaseModel): order: WithdrawOrderOut user: WithdrawUserSnapshot | None = None + cumulative_success_cents: int = 0 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 + invite_overview: InviteOverviewOut | None = None class WithdrawBulkRequest(BaseModel): @@ -128,26 +147,6 @@ class WithdrawBulkResult(BaseModel): items: list[WithdrawBulkItemResult] -class WithdrawLedgerCheckOut(BaseModel): - ok: bool - # 普通现金账(coin_cash:金币兑换的现金) - 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 - # 邀请奖励金账(invite_cash:与普通现金物理隔离,各自对账)。默认 0 向后兼容。 - invite_cash_balance_total_cents: int = 0 - invite_cash_transaction_total_cents: int = 0 - invite_balance_diff_cents: int = 0 - invite_missing_withdraw_txn_count: int = 0 - invite_missing_refund_txn_count: int = 0 - invite_duplicate_refund_txn_count: int = 0 - invite_refund_txn_on_non_terminal_count: int = 0 - - class WxpayHealthCheckOut(BaseModel): ok: bool wxpay_configured: bool diff --git a/app/models/user.py b/app/models/user.py index 1161ba1..e7fbe4a 100644 --- a/app/models/user.py +++ b/app/models/user.py @@ -11,7 +11,7 @@ from __future__ import annotations from datetime import datetime, timezone -from sqlalchemy import Boolean, DateTime, Integer, String, false, func +from sqlalchemy import Boolean, DateTime, Integer, String, Text, false, func from sqlalchemy.orm import Mapped, mapped_column from app.db.base import Base @@ -62,6 +62,13 @@ class User(Base): Boolean, nullable=False, default=False, server_default=false() ) + # 运营人工风险标记。与自动风控分值分开：这里表达人工复核结论，并保留可编辑备注， + # 供邀请提现、其他提现和用户管理三个页面统一展示。 + is_high_risk: Mapped[bool] = mapped_column( + Boolean, nullable=False, default=False, server_default=false(), index=True + ) + high_risk_note: Mapped[str | None] = mapped_column(Text, nullable=True) + created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), nullable=False ) diff --git a/app/repositories/wallet.py b/app/repositories/wallet.py index 4337752..6dde112 100644 --- a/app/repositories/wallet.py +++ b/app/repositories/wallet.py @@ -6,6 +6,7 @@ """ from __future__ import annotations +import logging import re import unicodedata import uuid @@ -20,7 +21,6 @@ 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.repositories.user import apply_wechat_display_identity from app.models.wallet import ( CashTransaction, CoinAccount, @@ -29,12 +29,55 @@ from app.models.wallet import ( WechatTransferAuthorization, WithdrawOrder, ) +from app.repositories.user import apply_wechat_display_identity from app.services import notification_events +logger = logging.getLogger(__name__) + # 微信转账终态:成功 / 失败(失败/取消/关闭都退款) _WX_STATE_SUCCESS = "SUCCESS" _WX_STATE_FAILED = {"FAIL", "CANCELLED", "CLOSED"} _WX_STATE_WAIT_CONFIRM = "WAIT_USER_CONFIRM" # 用户还没在微信确认页确认 +_WX_FAIL_REASON_LABELS = { + "ACCOUNT_FROZEN": "用户微信账户被冻结", + "ACCOUNT_NOT_EXIST": "用户微信账户不存在", + "BANK_CARD_ACCOUNT_ABNORMAL": "用户银行卡已销户、冻结、作废或挂失", + "BANK_CARD_BANK_INFO_WRONG": "用户登记的银行或分支行信息有误", + "BANK_CARD_CARD_INFO_WRONG": "用户银行卡户名或卡号有误", + "BANK_CARD_COLLECTIONS_ABOVE_QUOTA": "用户银行卡收款达到限额", + "BANK_CARD_PARAM_ERROR": "用户收款银行卡信息错误", + "BANK_CARD_STATUS_ABNORMAL": "用户银行卡状态异常", + "BLOCK_B2C_USERLIMITAMOUNT_BSRULE_MONTH": "用户本月转账收款已达限额", + "BLOCK_B2C_USERLIMITAMOUNT_MONTH": "用户账户存在风险，本月收款受限", + "DAY_RECEIVED_COUNT_EXCEED": "用户当日收款次数已达上限", + "DAY_RECEIVED_QUOTA_EXCEED": "用户当日收款额度已达上限", + "EXCEEDED_ESTIMATED_AMOUNT": "转账金额超过预约金额范围", + "ID_CARD_NOT_CORRECT": "收款人身份证校验不通过", + "MCH_CANCEL": "商户已撤销付款", + "MERCHANT_REJECT": "商户转账验密人已驳回", + "MERCHANT_NOT_CONFIRM": "商户转账验密人超时未确认", + "NAME_NOT_CORRECT": "收款人姓名校验不通过", + "OPENID_INVALID": "用户 OpenID 无效或不属于当前 AppID", + "OTHER_FAIL_REASON_TYPE": "微信返回其他失败原因", + "OVERDUE_CLOSE": "超过微信系统重试期，订单自动关闭", + "PAYEE_ACCOUNT_ABNORMAL": "用户微信账户收款异常", + "PAYER_ACCOUNT_ABNORMAL": "商户账户付款受限", + "PRODUCT_AUTH_CHECK_FAIL": "商户未开通转账权限或权限已冻结", + "REALNAME_ACCOUNT_RECEIVED_QUOTA_EXCEED": "用户微信实名账户收款受限", + "REAL_NAME_CHECK_FAIL": "用户未完成微信实名认证", + "RECEIVE_ACCOUNT_NOT_CONFIGURE": "商户未配置收款用户列表", + "RESERVATION_INFO_NOT_MATCH": "转账信息与预约信息不一致", + "RESERVATION_SCENE_NOT_MATCH": "转账场景与预约场景不一致", + "RESERVATION_STATE_INVALID": "预约转账单状态异常", + "TRANSFER_QUOTA_EXCEED": "用户单笔收款额度已达上限", + "TRANSFER_REMARK_SET_FAIL": "微信转账备注设置失败", + "TRANSFER_RISK": "该笔转账存在风险，已被微信拦截", + "TRANSFER_SCENE_INVALID": "商户未获取当前转账场景", + "TRANSFER_SCENE_UNAVAILABLE": "当前转账场景暂不可用", + "RELATED_ORDER_TRANSFER_AMOUNT_EXCEED": "关联订单累计付款金额超过上限", + "RELATED_ORDER_TRANSFER_COUNT_EXCEED": "关联订单累计付款次数超过上限", + "BUDGET_NOT_ENOUGH": "商户预算资金不足", +} # 占用新人档「一次性」资格的提现状态:进行中(reviewing/pending)或成功打款(success)。 # 被拒/转账失败/解绑退回(rejected/failed,均已退款、钱没到手)不在此列 → 新人档恢复可提 # (2026-07-16 修正:此前判定不看状态,解绑微信退回后 0.1 被误判已用、资格永久锁死)。 @@ -479,6 +522,19 @@ def _try_deduct_cash(db: Session, user_id: int, amount_cents: int, source: str = def _add_cash(db: Session, user_id: int, amount_cents: int, source: str = "coin_cash") -> int: """原子增加指定账户余额(退款用),返回加后余额。source 决定退回哪个账户(两账户隔离)。""" + # Normal withdrawals always create the account before deducting funds. This + # fallback covers legacy rows, hand-written fixtures, and broken migrations: + # a missing balance snapshot must not make a legitimate refund fail with 500. + if db.get(CoinAccount, user_id) is None: + logger.error( + "withdraw refund found missing coin_account; recreating empty account: " + "user_id=%s source=%s amount_cents=%s", + user_id, + source, + amount_cents, + ) + get_or_create_account(db, user_id, commit=False) + col = _balance_col(source) db.execute( update(CoinAccount) @@ -582,6 +638,32 @@ def _wx_not_found(result: dict) -> bool: return "NOT_FOUND" in str(code) +def _wechat_api_error_reason(data: object) -> str: + """Format a non-200 WeChat API response for operator display.""" + if not isinstance(data, dict): + return f"微信发起转账失败：{data}" + code = str(data.get("code") or "").strip() + message = str(data.get("message") or "").strip() + if code and message: + return f"微信发起转账失败：{message}（{code}）" + return f"微信发起转账失败：{message or code or '未知错误'}" + + +def _wechat_terminal_failure_reason(data: dict, state: str) -> str: + """Translate WeChat query ``fail_reason`` while preserving unknown codes.""" + code = str(data.get("fail_reason") or "").strip() + if code: + label = _WX_FAIL_REASON_LABELS.get(code) + if label: + return f"微信转账失败：{label}（{code}）" + return f"微信转账失败：{code}" + if state == "CANCELLED": + return "微信转账已撤销（CANCELLED）" + if state == "CLOSED": + return "微信转账已关闭（CLOSED）" + return f"微信转账失败（状态：{state or 'FAIL'}）" + + def _settle_after_ambiguous(db: Session, order: WithdrawOrder, reason: str) -> None: """转账调用结果不明(超时/异常/非200)时,**先查单再决定**,绝不盲目退款(防退款后又到账)。 - 微信查到 SUCCESS → 钱已出,置 success,不退款 @@ -607,13 +689,17 @@ def _settle_after_ambiguous(db: Session, order: WithdrawOrder, reason: str) -> N state = q["data"].get("state", "") order.wechat_state = state + order.transfer_bill_no = q["data"].get("transfer_bill_no") or order.transfer_bill_no if state == _WX_STATE_SUCCESS: order.status = "success" - order.transfer_bill_no = q["data"].get("transfer_bill_no") db.commit() notification_events.notify_withdraw_success(db, order) # PRD #3 提现到账 elif state in _WX_STATE_FAILED: - _refund_withdraw(db, order, reason=reason) + _refund_withdraw( + db, + order, + reason=_wechat_terminal_failure_reason(q["data"], state), + ) else: order.package_info = q["data"].get("package_info") or order.package_info db.commit() @@ -967,6 +1053,14 @@ def _apply_transfer_result(db: Session, order: WithdrawOrder, data: dict) -> Wit order.package_info = data.get("package_info") # 免确认转账无此字段(None);确认模式带它供拉确认页 if data.get("state") == _WX_STATE_SUCCESS: order.status = "success" + elif data.get("state") in _WX_STATE_FAILED: + _refund_withdraw( + db, + order, + reason=_wechat_terminal_failure_reason(data, str(data.get("state") or "")), + ) + db.refresh(order) + return order db.commit() db.refresh(order) if order.status == "success": # 免确认转账直接到账 → PRD #3 提现到账 @@ -1011,7 +1105,11 @@ def execute_withdraw_transfer(db: Session, order: WithdrawOrder) -> WithdrawOrde return order if result["status_code"] != 200: # 金额安全:查转账单后定夺,绝不盲退(未创建→退款,已创建→按真实状态) - _settle_after_ambiguous(db, order, reason=str(result["data"].get("message") or result["data"])) + _settle_after_ambiguous( + db, + order, + reason=_wechat_api_error_reason(result["data"]), + ) # 授权有效性:回查授权单,微信侧已失效(用户关闭/风控)→标 closed,下次提现自动回退方式一重新授权 _refresh_active_auth(db, order.user_id) db.refresh(order) @@ -1039,7 +1137,11 @@ def execute_withdraw_transfer(db: Session, order: WithdrawOrder) -> WithdrawOrde return order if result["status_code"] != 200: - _settle_after_ambiguous(db, order, reason=str(result["data"].get("message") or result["data"])) + _settle_after_ambiguous( + db, + order, + reason=_wechat_api_error_reason(result["data"]), + ) db.refresh(order) return order @@ -1096,19 +1198,24 @@ def refresh_withdraw_status( ).scalar_one_or_none() if order is None: raise WithdrawOrderNotFound - if order.status != "pending": - return order # 已终态,不再查 + if order.status not in {"pending", "failed"}: + return order + enrich_failed_order = order.status == "failed" try: result = wxpay.query_transfer(out_bill_no) except wxpay.WxPayNotConfiguredError: raise except Exception as exc: # noqa: BLE001 - 查单失败不能把运营后台打成 500 + if enrich_failed_order: + return order order.fail_reason = f"微信查单异常,保持pending: {exc}"[:256] db.commit() db.refresh(order) return order if result["status_code"] != 200: + if enrich_failed_order: + return order if _wx_not_found(result): # 微信明确无此单 → 转账从未创建(如崩溃在扣款后/调用前),退款安全 _refund_withdraw(db, order, reason="微信无此单,已退回") @@ -1117,12 +1224,32 @@ def refresh_withdraw_status( state = result["data"].get("state", "") order.wechat_state = state + order.transfer_bill_no = ( + result["data"].get("transfer_bill_no") or order.transfer_bill_no + ) + if enrich_failed_order: + if state in _WX_STATE_FAILED: + order.fail_reason = _wechat_terminal_failure_reason( + result["data"], state + )[:256] + elif state == _WX_STATE_SUCCESS: + order.fail_reason = ( + "资金状态异常：本地已退款，但微信查单显示已到账，请人工核查" + ) + db.commit() + db.refresh(order) + return order + if state == _WX_STATE_SUCCESS: order.status = "success" db.commit() notification_events.notify_withdraw_success(db, order) # PRD #3 提现到账 elif state in _WX_STATE_FAILED: - _refund_withdraw(db, order, reason=f"微信转账状态 {state}") + _refund_withdraw( + db, + order, + reason=_wechat_terminal_failure_reason(result["data"], state), + ) elif state == _WX_STATE_WAIT_CONFIRM and cancel_if_unconfirmed: # 用户从确认页回来了却仍未确认 → 视为放弃:撤销微信单(防事后确认导致重复打款)后退款。 # 撤单失败(可能已被确认进 ACCEPTED 的竞态)则保持 pending,等下次查询。 diff --git a/tests/test_admin_read.py b/tests/test_admin_read.py index 06b19aa..217998b 100644 --- a/tests/test_admin_read.py +++ b/tests/test_admin_read.py @@ -12,6 +12,9 @@ from app.admin.repositories import admin_user as admin_repo from app.db.session import SessionLocal, engine from app.models.comparison import ComparisonRecord from app.models.feedback import Feedback +from app.models.invite import InviteRelation +from app.models.savings import SavingsRecord +from app.models.user import User from app.models.wallet import CashTransaction, WithdrawOrder from app.repositories import user as user_repo from app.repositories import wallet as wallet_repo @@ -162,6 +165,57 @@ def test_user_reward_detail_does_not_select_unrelated_new_ad_columns( assert records.status_code == 200, records.text +def test_user_reward_stats_can_scope_withdrawals_by_account( + admin_client: TestClient, admin_token: str +) -> None: + uid = _seed_user_with_data("13800000023") + db = SessionLocal() + try: + account = wallet_repo.get_or_create_account(db, uid) + account.cash_balance_cents = 123 + account.invite_cash_balance_cents = 456 + db.add_all( + [ + WithdrawOrder( + user_id=uid, + out_bill_no="rewardstatsinvite00000001", + amount_cents=250, + source="invite_cash", + status="success", + ), + WithdrawOrder( + user_id=uid, + out_bill_no="rewardstatsinvite00000002", + amount_cents=50, + source="invite_cash", + status="reviewing", + ), + ] + ) + db.commit() + finally: + db.close() + + coin = admin_client.get( + f"/admin/api/users/{uid}/reward-stats", + params={"withdraw_source": "coin_cash"}, + headers=_auth(admin_token), + ) + invite = admin_client.get( + f"/admin/api/users/{uid}/reward-stats", + params={"withdraw_source": "invite_cash"}, + headers=_auth(admin_token), + ) + assert coin.status_code == 200, coin.text + assert invite.status_code == 200, invite.text + assert coin.json()["withdraw_success_cents"] == 100 + assert coin.json()["withdraw_total"] == 1 + assert coin.json()["cash_balance_cents"] == 123 + assert invite.json()["withdraw_success_cents"] == 250 + assert invite.json()["withdraw_total"] == 2 + assert invite.json()["cash_balance_cents"] == 456 + + def test_user_filter_by_status(admin_client: TestClient, admin_token: str) -> None: _seed_user_with_data("13800000003") r = admin_client.get("/admin/api/users", params={"status": "active"}, headers=_auth(admin_token)) @@ -307,10 +361,19 @@ def test_withdraw_list_exposes_source_and_filters( db = SessionLocal() try: uid = user_repo.upsert_user_for_login(db, phone="13800009002", register_channel="sms").id + # 直接从 ORM 查询，验证列表富化返回人工风险结论。 + user = db.get(User, uid) + assert user is not None + user.is_high_risk = True + user.high_risk_note = "邀请行为异常" db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}coin0001", amount_cents=100, status="success", source="coin_cash")) db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}invite001", amount_cents=200, status="success", source="invite_cash")) + db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}coinreview", + amount_cents=50, status="reviewing", source="coin_cash")) + db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}invitereview", + amount_cents=80, status="reviewing", source="invite_cash")) db.commit() finally: db.close() @@ -325,6 +388,106 @@ def test_withdraw_list_exposes_source_and_filters( ) items = r.json()["items"] assert items and all(it["source"] == "invite_cash" for it in items) + assert all(it["is_high_risk"] is True for it in items) + assert all(it["high_risk_note"] == "邀请行为异常" for it in items) + # 累计提现按当前审核页账户隔离，不把普通现金的 100 分串进邀请金。 + assert all(it["cumulative_success_cents"] == 200 for it in items) + + invite_summary = admin_client.get( + "/admin/api/withdraws/summary", + params={"source": "invite_cash"}, + headers=_auth(admin_token), + ).json() + other_summary = admin_client.get( + "/admin/api/withdraws/summary", + params={"source": "coin_cash"}, + headers=_auth(admin_token), + ).json() + assert invite_summary["reviewing_count"] >= 1 + assert other_summary["reviewing_count"] >= 1 + assert invite_summary["reviewing_amount_cents"] >= 80 + assert other_summary["reviewing_amount_cents"] >= 50 + + +def test_invite_withdraw_detail_contains_invitee_first_actions( + admin_client: TestClient, admin_token: str +) -> None: + db = SessionLocal() + try: + inviter = user_repo.upsert_user_for_login( + db, phone="13800009031", register_channel="sms" + ) + invitee = user_repo.upsert_user_for_login( + db, phone="13800009032", register_channel="sms" + ) + db.add( + InviteRelation( + inviter_user_id=inviter.id, + invitee_user_id=invitee.id, + status="effective", + compare_reward_granted=True, + compare_reward_cents=50, + ) + ) + db.add( + ComparisonRecord( + user_id=invitee.id, + trace_id="invite-detail-first-compare", + status="success", + store_name="首次比价商家", + product_names="商品甲、商品乙", + ) + ) + db.add( + SavingsRecord( + user_id=invitee.id, + order_amount_cents=1888, + saved_amount_cents=300, + source="compare", + shop_name="首次下单商家", + title="首次下单商品", + dishes=[], + ) + ) + bill = "inviteoverviewbill0001" + db.add( + WithdrawOrder( + user_id=inviter.id, + out_bill_no=bill, + amount_cents=100, + status="reviewing", + source="invite_cash", + ) + ) + db.commit() + finally: + db.close() + + response = admin_client.get( + f"/admin/api/withdraws/{bill}", headers=_auth(admin_token) + ) + assert response.status_code == 200, response.text + overview = response.json()["invite_overview"] + assert overview["invite_total"] == 1 + assert overview["invite_success_total"] == 1 + item = overview["items"][0] + assert item["phone"] == "13800009032" + assert item["first_compare_store"] == "首次比价商家" + assert item["first_compare_products"] == "商品甲、商品乙" + assert item["first_order_store"] == "首次下单商家" + assert item["first_order_products"] == "首次下单商品" + assert item["first_order_amount_cents"] == 1888 + + filtered = admin_client.get( + f"/admin/api/withdraws/{bill}", + params={ + "date_from": "2099-01-01T00:00:00Z", + "date_to": "2099-01-02T00:00:00Z", + }, + headers=_auth(admin_token), + ) + assert filtered.status_code == 200, filtered.text + assert filtered.json()["invite_overview"]["invite_total"] == 0 def test_ad_coin_audit_full_count_truncate_and_only_mismatch( diff --git a/tests/test_admin_roles.py b/tests/test_admin_roles.py index edc5e4b..430ae3e 100644 --- a/tests/test_admin_roles.py +++ b/tests/test_admin_roles.py @@ -190,7 +190,13 @@ def test_builtin_roles_labels_and_pages(admin_client, super_token) -> None: assert roles["finance"]["label"] == "财务" assert roles["tech"]["label"] == "技术" # 页集对齐 Prototypes/dashboard/permissions.md 的 ROLES - assert set(roles["finance"]["pages"]) == {"dashboard", "ad-revenue-report", "cps", "withdraws"} + assert set(roles["finance"]["pages"]) == { + "dashboard", + "ad-revenue-report", + "cps", + "invite-withdraws", + "withdraws", + } assert set(roles["tech"]["pages"]) == { "dashboard", "risk-monitor", "device-liveness", "analytics-health", "config", "ad-revenue", "huawei-review", "event-logs", "audit-logs", diff --git a/tests/test_admin_write.py b/tests/test_admin_write.py index fc6b73c..8f2012f 100644 --- a/tests/test_admin_write.py +++ b/tests/test_admin_write.py @@ -103,6 +103,61 @@ def _seed_price_report(phone: str) -> int: db.close() +def test_set_user_risk_requires_note_and_writes_audit( + admin_client: TestClient, operator_token: str +) -> None: + uid = _seed_user("13900000991") + + missing_note = admin_client.post( + f"/admin/api/users/{uid}/risk", + json={"is_high_risk": True, "note": " "}, + headers=_auth(operator_token), + ) + assert missing_note.status_code == 422 + + marked = admin_client.post( + f"/admin/api/users/{uid}/risk", + json={"is_high_risk": True, "note": "邀请记录存在批量异常"}, + headers=_auth(operator_token), + ) + assert marked.status_code == 200, marked.text + + listed = admin_client.get( + "/admin/api/users", + params={"phone": "13900000991"}, + headers=_auth(operator_token), + ).json()["items"][0] + assert listed["is_high_risk"] is True + assert listed["high_risk_note"] == "邀请记录存在批量异常" + + db = SessionLocal() + try: + audit = db.execute( + select(AdminAuditLog) + .where( + AdminAuditLog.action == "user.risk.set", + AdminAuditLog.target_id == str(uid), + ) + .order_by(AdminAuditLog.id.desc()) + ).scalars().first() + assert audit is not None + assert audit.detail["after"]["is_high_risk"] is True + finally: + db.close() + + cleared = admin_client.post( + f"/admin/api/users/{uid}/risk", + json={"is_high_risk": False, "note": None}, + headers=_auth(operator_token), + ) + assert cleared.status_code == 200 + overview = admin_client.get( + f"/admin/api/users/{uid}", headers=_auth(operator_token) + ).json() + assert overview["user"]["is_high_risk"] is False + assert overview["user"]["high_risk_note"] is None + + # ===== 调金币 ===== def test_grant_coins_writes_txn_and_audit(admin_client: TestClient, finance_token: str) -> None: @@ -579,15 +634,47 @@ def test_withdraw_refresh_404(admin_client: TestClient, finance_token: str) -> N ).status_code == 404 -def test_withdraw_reconcile(admin_client: TestClient, finance_token: str, monkeypatch) -> None: +def test_withdraw_refresh_enriches_existing_failed_reason( + admin_client: TestClient, finance_token: str, monkeypatch +) -> None: + uid = _seed_user("13900000017") + with SessionLocal() as db: + db.add( + WithdrawOrder( + user_id=uid, + out_bill_no="adminfailedreason01", + amount_cents=100, + status="failed", + wechat_state="FAIL", + fail_reason="微信转账状态 FAIL", + ) + ) + db.commit() + from app.repositories import wallet as wr + monkeypatch.setattr( - wr.wxpay, "query_transfer", - lambda obn: {"status_code": 200, "data": {"state": "SUCCESS"}}, + wr.wxpay, + "query_transfer", + lambda out_bill_no: { + "status_code": 200, + "data": { + "state": "FAIL", + "fail_reason": "TRANSFER_RISK", + "transfer_bill_no": "wx_failed_reason_01", + }, + }, ) - r = admin_client.post( - "/admin/api/withdraws/reconcile", params={"older_than_minutes": 0}, + + response = admin_client.post( + "/admin/api/withdraws/adminfailedreason01/refresh", headers=_auth(finance_token), ) - assert r.status_code == 200, r.text - assert "checked" in r.json() and "resolved" in r.json() + + assert response.status_code == 200, response.text + assert response.json()["status"] == "failed" + assert response.json()["wechat_state"] == "FAIL" + assert response.json()["transfer_bill_no"] == "wx_failed_reason_01" + assert response.json()["fail_reason"] == ( + "微信转账失败：该笔转账存在风险，已被微信拦截（TRANSFER_RISK）" + ) diff --git a/tests/test_invite_cash_withdraw.py b/tests/test_invite_cash_withdraw.py index bfa5096..4301ef2 100644 --- a/tests/test_invite_cash_withdraw.py +++ b/tests/test_invite_cash_withdraw.py @@ -127,6 +127,51 @@ def test_invite_cash_reject_refunds_invite_account(client, monkeypatch) -> None: db.close() +def test_invite_cash_reject_recreates_missing_account(client, monkeypatch) -> None: + """A legacy or hand-written order without CoinAccount can still be refunded.""" + _patch_userinfo(monkeypatch, "openid_ic_missing_account") + token = _login(client, "13800004012") + _seed_balances(client, token, "13800004012", cash=0, invite_cash=200) + client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token)) + + response = client.post( + "/api/v1/wallet/withdraw", + json={"amount_cents": 200, "source": "invite_cash"}, + headers=_auth(token), + ) + assert response.status_code == 200, response.text + bill = response.json()["out_bill_no"] + + with SessionLocal() as db: + user = db.execute( + select(User).where(User.phone == "13800004012") + ).scalar_one() + db.delete(db.get(CoinAccount, user.id)) + db.commit() + + _reject(bill, "missing account fixture") + + with SessionLocal() as db: + user = db.execute( + select(User).where(User.phone == "13800004012") + ).scalar_one() + account = db.get(CoinAccount, user.id) + order = crud_wallet._get_withdraw_or_raise(db, bill) + refunds = db.execute( + select(InviteCashTransaction).where( + InviteCashTransaction.user_id == user.id, + InviteCashTransaction.biz_type == "invite_withdraw_refund", + InviteCashTransaction.ref_id == bill, + ) + ).scalars().all() + + assert account is not None + assert account.invite_cash_balance_cents == 200 + assert order.status == "rejected" + assert order.fail_reason == "missing account fixture" + assert len(refunds) == 1 and refunds[0].amount_cents == 200 + + def test_two_accounts_withdraw_independent(client, monkeypatch) -> None: """两账户各提各的不串:提 invite_cash(拒绝→验证退款回原账户),再提 cash,各扣各账户互不串。""" _patch_userinfo(monkeypatch, "openid_ic_3") diff --git a/tests/test_withdraw.py b/tests/test_withdraw.py index 88a0eb9..9a597dd 100644 --- a/tests/test_withdraw.py +++ b/tests/test_withdraw.py @@ -9,7 +9,7 @@ from sqlalchemy import select from app.db.session import SessionLocal from app.models.user import User -from app.models.wallet import CoinAccount, WithdrawOrder, WechatTransferAuthorization +from app.models.wallet import CoinAccount, WechatTransferAuthorization, WithdrawOrder from app.repositories import wallet as crud_wallet @@ -232,6 +232,78 @@ def test_withdraw_approve_transfer_fail_refunds(client, monkeypatch) -> None: assert r.json()["items"][0]["status"] == "failed" +def test_approved_withdraw_query_fail_exposes_wechat_reason(client, monkeypatch) -> None: + """审核通过后微信终态失败时，保存官方失败码对应原因并退回余额。""" + monkeypatch.setattr( + "app.integrations.wxpay.code_to_userinfo", + lambda code: { + "openid": "openid_query_fail", + "nickname": None, + "avatar_url": None, + "raw": {}, + }, + ) + monkeypatch.setattr( + "app.integrations.wxpay.create_transfer", + lambda openid, amount_fen, out_bill_no, user_name=None: { + "status_code": 200, + "data": { + "state": "WAIT_USER_CONFIRM", + "package_info": "pkg_query_fail", + "transfer_bill_no": "tb_query_fail", + }, + }, + ) + token = _login(client, "13800002025") + _seed_cash(client, token, "13800002025", 100) + client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token)) + + response = client.post( + "/api/v1/wallet/withdraw", + json={"amount_cents": 50}, + headers=_auth(token), + ) + bill = response.json()["out_bill_no"] + _approve(bill) + + monkeypatch.setattr( + "app.integrations.wxpay.query_transfer", + lambda out_bill_no: { + "status_code": 200, + "data": { + "state": "FAIL", + "fail_reason": "PAYEE_ACCOUNT_ABNORMAL", + "transfer_bill_no": "tb_query_fail", + }, + }, + ) + response = client.get( + "/api/v1/wallet/withdraw/status", + params={"out_bill_no": bill}, + headers=_auth(token), + ) + + assert response.status_code == 200, response.text + assert response.json()["status"] == "failed" + assert response.json()["wechat_state"] == "FAIL" + assert response.json()["fail_reason"] == ( + "微信转账失败：用户微信账户收款异常（PAYEE_ACCOUNT_ABNORMAL）" + ) + with SessionLocal() as db: + order = db.execute( + select(WithdrawOrder).where(WithdrawOrder.out_bill_no == bill) + ).scalar_one() + assert order.transfer_bill_no == "tb_query_fail" + account = client.get("/api/v1/wallet/account", headers=_auth(token)) + assert account.json()["cash_balance_cents"] == 100 + + +def test_unknown_wechat_fail_reason_code_is_preserved() -> None: + assert crud_wallet._wechat_terminal_failure_reason( + {"fail_reason": "NEW_WECHAT_REASON"}, "FAIL" + ) == "微信转账失败：NEW_WECHAT_REASON" + + def test_withdraw_idempotent_same_bill_no(client, monkeypatch) -> None: """#2 同 out_bill_no 重试:只转一次,第二次返回同一单,余额只扣一次。""" monkeypatch.setattr("app.integrations.wxpay.code_to_userinfo", lambda code: {"openid": "openid_idem", "nickname": None, "avatar_url": None, "raw": {}}) diff --git a/tests/test_withdraw_ledger_check.py b/tests/test_withdraw_ledger_check.py deleted file mode 100644 index a9054d9..0000000 --- a/tests/test_withdraw_ledger_check.py +++ /dev/null @@ -1,151 +0,0 @@ -"""提现现金账本校验(admin ledger-check)测试:两本物理隔离的账各自对账。 - -历史盲区:`withdraw_ledger_check` 曾拿全部提现单去和**普通现金流水**(cash_transaction)比对, -而 source=invite_cash 的提现单流水其实在 invite_cash_transaction 表,导致每笔邀请提现单都被 -误报「缺扣款/缺退款流水」。这里用真实提现 API 造单 + before/after 差值断言锁定修复: - 1) 邀请提现单不再污染普通现金账的缺流水计数; - 2) 邀请账户已被纳入对账(能抓到它自己的缺流水); - 3) 普通现金账的原有对账未被改坏。 - -conftest 的库是 session 级共享、测试间不清,故一律用 before/after 差值,只反映本用例造的数据。 -""" -from __future__ import annotations - -from sqlalchemy import delete, select - -from app.admin.repositories.queries import withdraw_ledger_check -from app.db.session import SessionLocal -from app.models.user import User -from app.models.wallet import CoinAccount, InviteCashTransaction -from app.repositories import wallet as crud_wallet - - -def _login(client, phone: str) -> str: - client.post("/api/v1/auth/sms/send", json={"phone": phone}) - r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"}) - assert r.status_code == 200, r.text - return r.json()["access_token"] - - -def _auth(token: str) -> dict[str, str]: - return {"Authorization": f"Bearer {token}"} - - -def _patch_userinfo(monkeypatch, openid: str) -> None: - monkeypatch.setattr( - "app.integrations.wxpay.code_to_userinfo", - lambda code: {"openid": openid, "nickname": None, "avatar_url": None, "raw": {}}, - ) - - -def _seed_balances(client, token: str, phone: str, *, cash: int = 0, invite_cash: int = 0) -> None: - client.get("/api/v1/wallet/account", headers=_auth(token)) - db = SessionLocal() - try: - user = db.execute(select(User).where(User.phone == phone)).scalar_one() - acc = db.get(CoinAccount, user.id) - acc.cash_balance_cents = cash - acc.invite_cash_balance_cents = invite_cash - db.commit() - finally: - db.close() - - -def _reject(bill: str, reason: str = "测试拒绝") -> None: - db = SessionLocal() - try: - crud_wallet.reject_withdraw(db, bill, reason) - finally: - db.close() - - -def _ledger() -> dict: - db = SessionLocal() - try: - return withdraw_ledger_check(db) - finally: - db.close() - - -def test_rejected_invite_withdraw_not_flagged_missing(client, monkeypatch) -> None: - """核心回归:一笔被拒绝的 invite_cash 提现单,扣款/退款流水都在 invite_cash_transaction, - 不应让普通现金账的缺扣款/缺退款计数增加(修复前每笔会各 +1)。""" - before = _ledger() - - _patch_userinfo(monkeypatch, "openid_lc_1") - token = _login(client, "13800005001") - _seed_balances(client, token, "13800005001", cash=0, invite_cash=500) - client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token)) - r = client.post( - "/api/v1/wallet/withdraw", - json={"amount_cents": 200, "source": "invite_cash"}, - headers=_auth(token), - ) - assert r.status_code == 200, r.text - _reject(r.json()["out_bill_no"]) # rejected + 退款流水落 invite_cash_transaction - - after = _ledger() - - # 普通现金账不该因这笔 invite 单产生缺流水(修复前会各 +1 → 就是页面上看到的误报) - assert after["missing_withdraw_txn_count"] == before["missing_withdraw_txn_count"] - assert after["missing_refund_txn_count"] == before["missing_refund_txn_count"] - # 邀请账扣款 + 退款流水齐全,邀请账自身也不该缺 - assert after["invite_missing_withdraw_txn_count"] == before["invite_missing_withdraw_txn_count"] - assert after["invite_missing_refund_txn_count"] == before["invite_missing_refund_txn_count"] - - -def test_invite_ledger_detects_missing_withdraw_txn(client, monkeypatch) -> None: - """删掉一笔 invite 提现单的扣款流水 → 邀请账缺扣款计数 +1、ok=False, - 证明邀请账户已真正纳入对账(修复前邀请账完全不校验、永远报不出问题)。""" - _patch_userinfo(monkeypatch, "openid_lc_2") - token = _login(client, "13800005002") - _seed_balances(client, token, "13800005002", cash=0, invite_cash=500) - client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token)) - r = client.post( - "/api/v1/wallet/withdraw", - json={"amount_cents": 200, "source": "invite_cash"}, - headers=_auth(token), - ) - bill = r.json()["out_bill_no"] - - before = _ledger() - db = SessionLocal() - try: - db.execute( - delete(InviteCashTransaction).where( - InviteCashTransaction.ref_id == bill, - InviteCashTransaction.biz_type == "invite_withdraw", - ) - ) - db.commit() - finally: - db.close() - after = _ledger() - - assert ( - after["invite_missing_withdraw_txn_count"] - == before["invite_missing_withdraw_txn_count"] + 1 - ) - assert after["ok"] is False - - -def test_coin_cash_withdraw_still_reconciled(client, monkeypatch) -> None: - """普通现金 coin_cash 提现单齐全时不新增缺流水(确保分账改造没弄坏原有普通现金对账)。""" - _patch_userinfo(monkeypatch, "openid_lc_3") - token = _login(client, "13800005003") - _seed_balances(client, token, "13800005003", cash=500, invite_cash=0) - client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token)) - - before = _ledger() - r = client.post( - "/api/v1/wallet/withdraw", - # 50 分 = 0.5 元档(7-9 起 coin_cash 只能提预设档位) - json={"amount_cents": 50, "source": "coin_cash"}, - headers=_auth(token), - ) - assert r.status_code == 200, r.text - after = _ledger() - - # 普通现金提现扣款流水随单写入 cash_transaction,缺扣款计数不变;邀请账更不受影响 - assert after["missing_withdraw_txn_count"] == before["missing_withdraw_txn_count"] - assert after["invite_missing_withdraw_txn_count"] == before["invite_missing_withdraw_txn_count"]