feat(debug-trace): 调试链接权限 + trace_url 落库与按权限下发

- user 加 debug_trace_enabled、comparison_record 加 trace_url
- 迁移 f8d3b1e60a27：加两列，down_revision 用 tuple 顺带合并既存双 head(invite_fingerprint_table + 044dce6e9b1f)
- UserOut 加 debug_trace_enabled → /me 与登录响应带出
- ComparisonRecordIn/Out 加 trace_url；upsert 落库
- /compare/records 列表与详情按 user.debug_trace_enabled 下发；详情连 raw_payload 里那份一并抹掉，防权限绕过
- admin 加 POST /users/{id}/debug-trace（operator + 审计），列表带该字段

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

.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 回调本服务发金币(客户端不参与发奖)。

.gitignore

@@ -43,3 +43,4 @@ secrets/*
 
 # 运行日志(run.sh 输出, 不入库)
 *.log
+logs/

alembic/versions/f8d3b1e60a27_add_debug_trace_and_merge_heads.py

@@ -0,0 +1,54 @@
+"""add user.debug_trace_enabled + comparison_record.trace_url（并合并既存双 head）
+
+调试链接权限功能：
+  - user.debug_trace_enabled：运营后台给指定用户开「复制调试链接」权限
+  - comparison_record.trace_url：比价记录页「复制调试链接」的数据（pricebot done 帧给，
+    dir 名含落盘时分秒、前端/server 拼不出，必须落库）
+
+⚠️ 本迁移 down_revision 是 **tuple**，同时合并了 clone 时 main 上既存的双 head：
+  - invite_fingerprint_table  （#31 被邀请人列表 + 指纹归因 那条线）
+  - 044dce6e9b1f              （merge_ad_feed_reward_session + withdraw_safety_indexes）
+  这两个头平行未合并，`alembic upgrade head` 会报 Multiple heads。本迁移顺手收口成单 head。
+  若想把「合并双 head」与「加列」拆开，可改成：先一个纯 merge 迁移、再一个加列迁移。
+
+Revision ID: f8d3b1e60a27
+Revises: invite_fingerprint_table, 044dce6e9b1f
+Create Date: 2026-06-10 02:40:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = "f8d3b1e60a27"
+down_revision: Union[str, Sequence[str], None] = (
+    "invite_fingerprint_table",
+    "044dce6e9b1f",
+)
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # sa.false() 渲染成 PG 的 false / SQLite 的 0，两端兼容（避免字符串 "false" 在 SQLite 上存歪）
+    op.add_column(
+        "user",
+        sa.Column(
+            "debug_trace_enabled",
+            sa.Boolean(),
+            nullable=False,
+            server_default=sa.false(),
+        ),
+    )
+    op.add_column(
+        "comparison_record",
+        sa.Column("trace_url", sa.String(length=512), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("comparison_record", "trace_url")
+    op.drop_column("user", "debug_trace_enabled")

alembic/versions/invite_fingerprint_table.py

@@ -0,0 +1,50 @@
+"""invite_fingerprint table (任务 3 指纹归因兜底)
+
+Revision ID: invite_fingerprint_table
+Revises: 0cf18d590b1d
+Create Date: 2026-06-08 14:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'invite_fingerprint_table'
+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_table(
+        'invite_fingerprint',
+        sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column('inviter_user_id', sa.Integer(), nullable=False),
+        sa.Column('ip', sa.String(length=64), nullable=False),
+        sa.Column('device_model', sa.String(length=64), nullable=False, server_default=''),
+        sa.Column('screen', sa.String(length=32), nullable=False, server_default=''),
+        sa.Column('user_agent', sa.Text(), nullable=False, server_default=''),
+        sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.ForeignKeyConstraint(['inviter_user_id'], ['user.id']),
+        sa.PrimaryKeyConstraint('id'),
+    )
+    # 反查主索引:(ip, screen, device_model, created_at)
+    # /bind 兜底分支 WHERE ip=? AND screen=? AND device_model=? AND created_at > now - 7d
+    # ORDER BY created_at DESC LIMIT 1 — SQLite/PG 优化器都能反向扫该索引
+    op.create_index(
+        'ix_invite_fp_match',
+        'invite_fingerprint',
+        ['ip', 'screen', 'device_model', 'created_at'],
+    )
+    # 兜底/统计索引:按 inviter 看"这个邀请人的落地页被谁访问过"
+    op.create_index('ix_invite_fp_inviter', 'invite_fingerprint', ['inviter_user_id'])
+
+
+def downgrade() -> None:
+    op.drop_index('ix_invite_fp_inviter', table_name='invite_fingerprint')
+    op.drop_index('ix_invite_fp_match', table_name='invite_fingerprint')
+    op.drop_table('invite_fingerprint')

alembic/versions/merge_ad_feed_reward_session_and_withdraw_safety_heads.py

@@ -0,0 +1,26 @@
+"""merge ad feed reward session and withdraw safety heads
+
+Revision ID: 044dce6e9b1f
+Revises: ad_feed_reward_session, withdraw_safety_indexes
+Create Date: 2026-06-09 09:52:08.276767
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '044dce6e9b1f'
+down_revision: Union[str, Sequence[str], None] = ('ad_feed_reward_session', 'withdraw_safety_indexes')
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass

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")

app/admin/repositories/mutations.py

@@ -24,6 +24,20 @@ def set_user_status(db: Session, user: User, *, status: str, commit: bool = True
     return user
 
 
+def set_user_debug_trace(
+    db: Session, user: User, *, enabled: bool, commit: bool = True
+) -> User:
+    """开关用户「复制调试链接」权限。同 set_user_status：支持 commit=False 让 router 把
+    业务写 + 审计写放进同一事务。"""
+    user.debug_trace_enabled = enabled
+    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:

app/admin/repositories/queries.py

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

app/admin/routers/users.py

@@ -13,6 +13,7 @@ from app.admin.schemas.user import (
     AdminUserListItem,
     AdminUserOverview,
     GrantCoinsRequest,
+    SetDebugTraceRequest,
     SetUserStatusRequest,
 )
 from app.models.admin import AdminUser
@@ -77,6 +78,28 @@ def set_user_status(
     return OkResponse()
 
 
+@router.post("/{user_id}/debug-trace", response_model=OkResponse, summary="开关调试链接权限")
+def set_user_debug_trace(
+    user_id: int,
+    body: SetDebugTraceRequest,
+    request: Request,
+    admin: Annotated[AdminUser, Depends(require_role("operator"))],
+    db: AdminDb,
+) -> OkResponse:
+    user = user_repo.get_user_by_id(db, user_id)
+    if user is None:
+        raise HTTPException(status_code=404, detail="用户不存在")
+    before = user.debug_trace_enabled
+    # 业务写 + 审计写同一事务(commit=False),最后一起 commit(同 set_user_status)
+    mutations.set_user_debug_trace(db, user, enabled=body.enabled, commit=False)
+    write_audit(
+        db, admin, action="user.debug_trace.set", target_type="user", target_id=user_id,
+        detail={"before": before, "after": body.enabled}, ip=get_client_ip(request), commit=False,
+    )
+    db.commit()
+    return OkResponse()
+
+
 @router.post("/{user_id}/coins", response_model=OkResponse, summary="手动增减金币(带审计)")
 def grant_user_coins(
     user_id: int,

app/admin/routers/withdraw.py

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

app/admin/schemas/user.py

@@ -15,6 +15,7 @@ class AdminUserListItem(BaseModel):
     nickname: str | None = None
     register_channel: str
     status: str
+    debug_trace_enabled: bool = False
     wechat_openid: str | None = None
     created_at: datetime
     last_login_at: datetime
@@ -45,3 +46,7 @@ class SetUserStatusRequest(BaseModel):
     status: Literal["active", "disabled"] = Field(
         ..., description="active=解封 / disabled=封禁(注销 deleted 不走此接口)"
     )
+
+
+class SetDebugTraceRequest(BaseModel):
+    enabled: bool = Field(..., description="是否给该用户开「复制调试链接」权限")

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,用户可见)"

app/api/v1/compare_record.py

@@ -77,10 +77,12 @@ def list_records(
     items, next_cursor = crud_compare.list_records(
         db, user.id, limit=limit, cursor=cursor
     )
-    return ComparisonRecordPage(
-        items=[ComparisonRecordOut.model_validate(it) for it in items],
-        next_cursor=next_cursor,
-    )
+    outs = [ComparisonRecordOut.model_validate(it) for it in items]
+    # 权限闸:未开 debug_trace_enabled 的用户不下发 trace_url(列表页「复制调试链接」靠它)
+    if not user.debug_trace_enabled:
+        for o in outs:
+            o.trace_url = None
+    return ComparisonRecordPage(items=outs, next_cursor=next_cursor)
 
 
 @router.get(
@@ -94,4 +96,12 @@ def get_record(
     rec = crud_compare.get_record(db, user.id, record_id)
     if rec is None:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="record not found")
-    return ComparisonRecordDetailOut.model_validate(rec)
+    out = ComparisonRecordDetailOut.model_validate(rec)
+    # 权限闸:未开 debug_trace_enabled 的用户不下发 trace_url。
+    # ⚠️ raw_payload 是上报体全量(model_dump),里面也藏着一份 trace_url,必须一并抹掉——
+    # 否则无权限用户从详情接口的 raw_payload 绕过权限闸拿到 trace_url。
+    if not user.debug_trace_enabled:
+        out.trace_url = None
+        if isinstance(out.raw_payload, dict):
+            out.raw_payload.pop("trace_url", None)
+    return out

app/api/v1/coupon.py

@@ -183,7 +183,15 @@ def coupon_prompt_should_show(
     device_id: str, db: DbSession
 ) -> CouponPromptShouldShowOut:
     """今天这台设备已 engage(领或拒)过 → should_show=false。客户端据此决定弹不弹
-    (前台 SP 缓存做快速路径,这里是权威)。"""
+    (纯后台判据,客户端不再做前台 SP 缓存判断)。"""
     return CouponPromptShouldShowOut(
         should_show=not coupon_repo.has_engaged_today(db, device_id)
     )
+
+
+@router.post("/prompt/reset", summary="重置今日领券引导窗 engagement(开发测频控用)")
+def coupon_prompt_reset(payload: CouponPromptDismissIn, db: DbSession) -> dict[str, bool]:
+    """删这台设备今天的 engagement → has_engaged_today 变 false,今天又能弹。
+    开发设置「重置今日领券弹窗状态」按钮调。MVP 不鉴权,按 device_id。"""
+    coupon_repo.reset_today_engagement(db, payload.device_id)
+    return {"ok": True}

app/api/v1/invite.py

@@ -1,25 +1,37 @@
 """好友邀请 endpoint。
 
-路由前缀 /api/v1/invite,需 Bearer 鉴权(用户级数据):
-  GET  /me     我的邀请码 + 分享链接 + 已邀人数/已得金币(邀请页展示)
-  POST /bind   把当前登录用户(被邀请人)绑定到某邀请码,注册即生效,双方各发金币
+路由前缀 /api/v1/invite,需 Bearer 鉴权(用户级数据);唯一例外是 /landing-track
+(落地页 dl.html 浏览器访问、无 token,详见任务 3 [[invite-three-tasks]]):
+  GET  /me                我的邀请码 + 分享链接 + 已邀人数/已得金币(邀请页展示)
+  POST /bind              把当前登录用户(被邀请人)绑定到某邀请码;支持三种归因:
+                          ① clipboard:首启读剪贴板拿邀请码 → 上报码
+                          ② manual:用户在邀请页输入邀请码 → 上报码
+                          ③ fingerprint:剪贴板没拿到时上报指纹,后端反查
+  POST /landing-track     落地页 dl.html 访问时上报指纹(剪贴板兜底的服务端一端,无需鉴权)
 
-客户端两种调用 /bind:
-  - 自动:首启读剪贴板拿到邀请码 → 登录后调本接口(channel=clipboard)
-  - 手动:用户在邀请页输入好友邀请码(channel=manual)
-
-绑定的真正逻辑(邀请码生成/反查、幂等、自邀屏蔽、发金币)在 repositories/invite.py。
+绑定的真正逻辑(邀请码生成/反查、幂等、自邀屏蔽、发金币、指纹反查)在
+repositories/invite.py。
 """
 from __future__ import annotations
 
 import logging
+import re
 
-from fastapi import APIRouter
+from fastapi import APIRouter, Request
 
 from app.api.deps import CurrentUser, DbSession
+from app.core import rewards
 from app.core.config import settings
 from app.repositories import invite as invite_repo
-from app.schemas.invite import BindInviteIn, BindInviteOut, InviteInfoOut
+from app.schemas.invite import (
+    BindInviteIn,
+    BindInviteOut,
+    InviteeItem,
+    InviteeListOut,
+    InviteInfoOut,
+    LandingTrackIn,
+    LandingTrackOut,
+)
 
 logger = logging.getLogger("shagua.invite")
 
@@ -31,9 +43,43 @@ _BIND_MESSAGES = {
     "invalid_code": "邀请码无效",
     "self_invite": "不能填写自己的邀请码",
     "not_eligible": "邀请仅对新注册用户生效",
+    "fp_not_found": "未匹配到邀请关系",
 }
 
 
+def _client_ip(request: Request) -> str:
+    """从 HTTP 头拿真实 IP。nginx 反代时走 X-Forwarded-For;裸跑 uvicorn 走 request.client。"""
+    xff = request.headers.get("x-forwarded-for", "")
+    if xff:
+        # X-Forwarded-For 可能是 "client, proxy1, proxy2" 链;取第一个 = 真实客户端
+        return xff.split(",")[0].strip()
+    return request.client.host if request.client else ""
+
+
+# Android UA 形如 '... ; <Model> Build/<id>) AppleWebKit/...';抓 ';' 后到 ' Build/' 前
+# 的 token = Build.MODEL(跨端跟客户端 android.os.Build.MODEL 对齐)。user-agents 库
+# 把 Android 设备归为 'Smartphone' 通用名,抓不到真实型号,只能正则。
+_ANDROID_BUILD_MODEL_RE = re.compile(r";\s*([^;]+?)\s+Build/", re.IGNORECASE)
+
+
+def _parse_device_model(ua: str) -> str:
+    """解析浏览器 UA 拿手机型号(如 '24115RA8EC'、'PJF110')。
+
+    Android:正则抓 UA 里 ';...Build/' 前的 token(== Build.MODEL),跨端可严格匹配。
+    iOS / 解析不到:退到 user-agents 库的通用解析,失败/UA 空 → 返空字符串。
+    """
+    if not ua:
+        return ""
+    m = _ANDROID_BUILD_MODEL_RE.search(ua)
+    if m:
+        return m.group(1).strip()
+    try:
+        from user_agents import parse
+        return (parse(ua).device.model or "").strip()
+    except Exception:  # noqa: BLE001
+        return ""
+
+
 @router.get("/me", response_model=InviteInfoOut, summary="我的邀请码 + 分享链接 + 战绩")
 def my_invite(user: CurrentUser, db: DbSession) -> InviteInfoOut:
     code = invite_repo.ensure_code(db, user)
@@ -48,17 +94,124 @@ def my_invite(user: CurrentUser, db: DbSession) -> InviteInfoOut:
     )
 
 
-@router.post("/bind", response_model=BindInviteOut, summary="绑定邀请人(注册即生效,双方发金币)")
-def bind_invite(req: BindInviteIn, user: CurrentUser, db: DbSession) -> BindInviteOut:
-    result = invite_repo.bind(
-        db, invitee=user, invite_code=req.invite_code, channel=req.channel,
+@router.get("/invitees", response_model=InviteeListOut, summary="我邀请的人列表(分页)")
+def my_invitees(
+    user: CurrentUser, db: DbSession, limit: int = 20, offset: int = 0
+) -> InviteeListOut:
+    """邀请页小窗(取前几条) + 完整列表页(分页加载)共用。
+
+    名字/头像的降级兜底在 repositories/invite.get_invitees 算好,这里只组装响应。
+    limit 夹到 [1,50] 防滥用;offset 不小于 0。
+    """
+    limit = max(1, min(limit, 50))
+    offset = max(0, offset)
+    items, total, has_more = invite_repo.get_invitees(
+        db, user.id, limit=limit, offset=offset,
+    )
+    return InviteeListOut(
+        items=[InviteeItem(**it) for it in items],
+        total=total,
+        has_more=has_more,
+    )
+
+
+@router.post(
+    "/landing-track",
+    response_model=LandingTrackOut,
+    summary="落地页指纹采集(无需鉴权)",
+)
+def landing_track(
+    req: LandingTrackIn, request: Request, db: DbSession
+) -> LandingTrackOut:
+    """B 浏览器打开 dl.html?ref=xxx 时上报指纹。
+
+    服务端从 HTTP 头拿 IP/UA、解析 UA 拿 device_model,跟 req.screen 一起入库。
+    无需鉴权(浏览器没 token)。invalid_code / no_ip 走 silent 返回(不抛 5xx 影响下载流程)。
+    """
+    inviter = invite_repo.resolve_inviter(db, req.ref)
+    if inviter is None or inviter.status != "active":
+        return LandingTrackOut(status="invalid_code")
+
+    ip = _client_ip(request)
+    if not ip:
+        return LandingTrackOut(status="no_ip")
+
+    ua_str = request.headers.get("user-agent", "")
+    device_model = _parse_device_model(ua_str)
+
+    invite_repo.record_fingerprint(
+        db,
+        inviter_user_id=inviter.id,
+        ip=ip,
+        screen=req.screen,
+        device_model=device_model,
+        user_agent=ua_str,
     )
     logger.info(
-        "invite bind invitee=%d code=%s channel=%s -> %s",
-        user.id, req.invite_code, req.channel, result.status,
+        "invite landing-track inviter=%d ip=%s screen=%s model=%s",
+        inviter.id, ip, req.screen, device_model,
     )
+    return LandingTrackOut(status="ok")
+
+
+@router.post("/bind", response_model=BindInviteOut, summary="绑定邀请人(注册即生效,双方发金币)")
+def bind_invite(
+    req: BindInviteIn, user: CurrentUser, db: DbSession, request: Request
+) -> BindInviteOut:
+    code = (req.invite_code or "").strip()
+
+    # 路径 1:有邀请码 → 走原路径(clipboard / manual)
+    if code:
+        result = invite_repo.bind(
+            db, invitee=user, invite_code=code, channel=req.channel,
+        )
+        logger.info(
+            "invite bind invitee=%d code=%s channel=%s -> %s",
+            user.id, code, req.channel, result.status,
+        )
+        return BindInviteOut(
+            status=result.status,
+            coins_awarded=result.invitee_coin,
+            message=_BIND_MESSAGES.get(result.status, result.status),
+        )
+
+    # 路径 2:无邀请码 + 有指纹 → 指纹兜底反查
+    if req.fingerprint is not None:
+        ip = _client_ip(request)
+        inviter = invite_repo.resolve_inviter_by_fingerprint(
+            db,
+            ip=ip,
+            screen=req.fingerprint.screen,
+            device_model=req.fingerprint.device_model,
+            window_days=rewards.INVITE_FP_WINDOW_DAYS,
+        )
+        if inviter is None:
+            logger.info(
+                "invite bind invitee=%d fingerprint not_found ip=%s screen=%s model=%s",
+                user.id, ip, req.fingerprint.screen, req.fingerprint.device_model,
+            )
+            return BindInviteOut(
+                status="fp_not_found",
+                coins_awarded=0,
+                message=_BIND_MESSAGES["fp_not_found"],
+            )
+        # 用反查到的 inviter.invite_code 走原 bind 流程(走原幂等/自邀/新人闸/发币逻辑)
+        result = invite_repo.bind(
+            db, invitee=user, invite_code=inviter.invite_code, channel="fingerprint",
+        )
+        logger.info(
+            "invite bind invitee=%d fingerprint -> inviter=%d code=%s -> %s",
+            user.id, inviter.id, inviter.invite_code, result.status,
+        )
+        return BindInviteOut(
+            status=result.status,
+            coins_awarded=result.invitee_coin,
+            message=_BIND_MESSAGES.get(result.status, result.status),
+        )
+
+    # 路径 3:啥都没传 → 无效请求
     return BindInviteOut(
-        status=result.status,
-        coins_awarded=result.invitee_coin,
-        message=_BIND_MESSAGES.get(result.status, result.status),
+        status="invalid_code",
+        coins_awarded=0,
+        message=_BIND_MESSAGES["invalid_code"],
     )

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 才认;任一道闸拦住即恢复正常审核,

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 查询兜底),但微信要求该字段非空,故启用免确认前必须配置;留空时免确认相关接口返回未配置。

app/core/rewards.py

@@ -85,6 +85,11 @@ INVITE_INVITEE_COINS: int = 10000
 # 自动绑(剪贴板)在首次注册登录后几秒内发生;留 72h 给手动填码兜底。
 INVITE_NEW_USER_WINDOW_HOURS: int = 72
 
+# 指纹归因兜底(剪贴板被覆盖时):落地页 POST /landing-track 存的指纹记录,在此窗口内可被
+# /bind 反查撞库。窗口取舍:太短(24h)覆盖不到"晚上看链接、第二天装"的常见场景;太宽
+# (30d)IP/设备会漂、匹配错率上升。7 天兼顾"看广告→使用"周期与匹配精度。
+INVITE_FP_WINDOW_DAYS: int = 7
+
 
 # ===== 看激励视频 / 信息流广告发金币 =====
 # 金币数值体系约定:eCPM 单位按"元/千次展示"处理,单次收入 = eCPM / 1000 元。

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

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,

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(

app/models/__init__.py

@@ -13,6 +13,7 @@ from app.models.coupon_state import (  # noqa: F401
 )
 from app.models.feedback import Feedback  # noqa: F401
 from app.models.invite import InviteRelation  # noqa: F401
+from app.models.invite_fingerprint import InviteFingerprint  # noqa: F401
 from app.models.meituan_coupon import MeituanCoupon  # noqa: F401
 from app.models.ops_marquee_seed import OpsMarqueeSeed  # noqa: F401
 from app.models.ops_stat_config import OpsStatConfig  # noqa: F401

app/models/comparison.py

@@ -53,6 +53,10 @@ class ComparisonRecord(Base):
     )
     # pricebot 侧 trace_id:关联调试落盘 + 幂等去重键
     trace_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    # 本次比价的公网调试链接(price.shaguabijia.com/traces/{dir}/)。pricebot done 帧给、
+    # 客户端上报带上——dir 名含 pricebot 落盘的时分秒,前端/server 都拼不出,必须存。
+    # 查看接口按 user.debug_trace_enabled 决定返不返回。旧记录 / 未开上云为 None。
+    trace_url: Mapped[str | None] = mapped_column(String(512), nullable=True)
 
     # ===== 源平台(发起比价的那家)=====
     source_platform_id: Mapped[str | None] = mapped_column(String(32), nullable=True)

app/models/invite_fingerprint.py

@@ -0,0 +1,46 @@
+"""邀请指纹归因表(剪贴板归因兜底)。
+
+数据流(对应 [[invite-three-tasks]] 任务 3):
+  1. B 浏览器打开落地页 dl.html?ref=邀请码 → JS POST /api/v1/invite/landing-track
+  2. 后端从 HTTP 头拿 IP/UA,解析 UA 得 device_model,跟 JS 上报的 screen 一起入库
+  3. B 装包首启 → 登录后,若剪贴板没拿到邀请码(被覆盖),客户端再算一次 screen+Build.MODEL
+     上报 → 后端用 (ip, screen, device_model) 反查本表 7 天内最近匹配 → 拿出 inviter_user_id
+     → 走原 bind 路径(channel=fingerprint)
+
+不要索引(IP, created_at) 单独,组合索引 (ip, screen, device_model, created_at desc) 反查更省。
+"""
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import DateTime, ForeignKey, Integer, String, Text, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base
+
+
+class InviteFingerprint(Base):
+    __tablename__ = "invite_fingerprint"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    inviter_user_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("user.id"), index=True, nullable=False
+    )
+    # 落地页访问者 IP(服务端从 HTTP 头拿,X-Forwarded-For 由部署侧 nginx 透传)
+    ip: Mapped[str] = mapped_column(String(64), nullable=False)
+    # 客户端 Build.MODEL / 浏览器 UA 解析出来的型号(如 "PJF110"、"23046PNC9C")
+    device_model: Mapped[str] = mapped_column(String(64), nullable=False, default="")
+    # 屏幕分辨率 "1080x2400"(浏览器 screen.width × height,客户端 DisplayMetrics)
+    screen: Mapped[str] = mapped_column(String(32), nullable=False, default="")
+    # 完整 UA 字符串(留 debug / 排查"匹配不上"用,不直接参与匹配)
+    user_agent: Mapped[str] = mapped_column(Text, nullable=False, default="")
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
+    )
+
+    def __repr__(self) -> str:  # pragma: no cover
+        return (
+            f"<InviteFingerprint inviter={self.inviter_user_id} "
+            f"ip={self.ip} model={self.device_model} screen={self.screen}>"
+        )

app/models/user.py

@@ -11,7 +11,7 @@ from __future__ import annotations
 
 from datetime import datetime, timezone
 
-from sqlalchemy import DateTime, Integer, String, func
+from sqlalchemy import Boolean, DateTime, Integer, String, false, func
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.db.base import Base
@@ -51,6 +51,13 @@ class User(Base):
     # 账号状态:active / disabled / deleted
     status: Mapped[str] = mapped_column(String(20), nullable=False, default="active")
 
+    # 调试链接权限:开了的用户在比价完成弹窗 + 比价记录页能看到「复制调试链接」按钮
+    # (复制 price.shaguabijia.com 的 trace 链接发给开发排障)。运营后台按用户配置;
+    # /me 与登录响应里带出给前端做条件渲染。默认 false。
+    debug_trace_enabled: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=False, server_default=false()
+    )
+
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), server_default=func.now(), nullable=False
     )

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(

app/repositories/comparison.py

@@ -84,6 +84,7 @@ def upsert_record(
         source_package=payload.source_package,
         information=payload.information,
         best_deeplink=payload.best_deeplink,
+        trace_url=payload.trace_url,
         total_dish_count=payload.total_dish_count,
         skipped_dish_count=payload.skipped_dish_count,
         items=[it.model_dump(exclude_none=True) for it in payload.items],

app/repositories/coupon_state.py

@@ -9,7 +9,7 @@ import logging
 from datetime import date, datetime
 from zoneinfo import ZoneInfo
 
-from sqlalchemy import select
+from sqlalchemy import delete, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -65,6 +65,19 @@ def mark_engagement(
         db.rollback()
 
 
+def reset_today_engagement(db: Session, device_id: str) -> int:
+    """删这台设备今天的 engagement(开发设置「重置今日领券弹窗状态」调,测频控用)。
+    删后 has_engaged_today → false,今天又能弹。返回删除行数。"""
+    result = db.execute(
+        delete(CouponPromptEngagement).where(
+            CouponPromptEngagement.device_id == device_id,
+            CouponPromptEngagement.engage_date == today_cn(),
+        )
+    )
+    db.commit()
+    return result.rowcount or 0
+
+
 # ===== 领券记录(coupon_claim_record)=====
 
 def record_claims(

app/repositories/invite.py

@@ -21,6 +21,7 @@ from sqlalchemy.orm import Session
 
 from app.core import rewards
 from app.models.invite import InviteRelation
+from app.models.invite_fingerprint import InviteFingerprint
 from app.models.user import User
 from app.repositories import wallet as crud_wallet
 
@@ -164,3 +165,121 @@ def get_stats(db: Session, inviter_id: int) -> tuple[int, int]:
         .where(InviteRelation.inviter_user_id == inviter_id)
     ).scalar_one()
     return int(count), int(coins)
+
+
+def _mask_phone(phone: str) -> str:
+    """手机号脱敏:138****8888。前端拿不到完整号,展示被邀请人时在此兜底名字。
+
+    11 位标准手机号 → 前 3 + **** + 后 4;非标准(异常/截断)→ 只露后 4 位;太短 → "新用户"。
+    """
+    p = (phone or "").strip()
+    if len(p) == 11:
+        return f"{p[:3]}****{p[-4:]}"
+    if len(p) >= 4:
+        return f"****{p[-4:]}"
+    return "新用户"
+
+
+def get_invitees(
+    db: Session, inviter_id: int, *, limit: int = 20, offset: int = 0
+) -> tuple[list[dict], int, bool]:
+    """查某邀请人的被邀请人列表(按邀请时间倒序, 分页),返回 (items, total, has_more)。
+
+    每条 item = {display_name, avatar_url, coins, invited_at}。降级兜底:
+      名字 = 昵称 → 微信昵称 → 脱敏手机号(很多被邀请人是刚注册新用户、没设资料);
+      头像 = 用户头像 → 微信头像 → None(前端画默认色块)。
+    邀请关系表 join 用户表;total 单独 count(分页时算 has_more)。
+    """
+    total = db.execute(
+        select(func.count())
+        .select_from(InviteRelation)
+        .where(InviteRelation.inviter_user_id == inviter_id)
+    ).scalar_one()
+
+    rows = db.execute(
+        select(InviteRelation, User)
+        .join(User, User.id == InviteRelation.invitee_user_id)
+        .where(InviteRelation.inviter_user_id == inviter_id)
+        .order_by(InviteRelation.created_at.desc())
+        .limit(limit)
+        .offset(offset)
+    ).all()
+
+    items: list[dict] = []
+    for rel, u in rows:
+        items.append({
+            "display_name": u.nickname or u.wechat_nickname or _mask_phone(u.phone),
+            "avatar_url": u.avatar_url or u.wechat_avatar_url or None,
+            "coins": rel.inviter_coin,
+            "invited_at": rel.created_at,
+        })
+    has_more = offset + len(rows) < int(total)
+    return items, int(total), has_more
+
+
+# ===== 指纹归因(剪贴板兜底,见 [[invite-three-tasks]] 任务 3)=====
+
+def record_fingerprint(
+    db: Session,
+    *,
+    inviter_user_id: int,
+    ip: str,
+    screen: str,
+    device_model: str,
+    user_agent: str,
+) -> InviteFingerprint:
+    """落地页 dl.html 访问时调用,写一行指纹记录。
+
+    后续被邀请人登录、剪贴板没拿到邀请码时,/bind 会按 (ip, screen, device_model) 反查本表
+    7 天内最近一条匹配 → 拿出 inviter_user_id 撞库。
+
+    本函数会 commit;调用方(/landing-track endpoint)在此之前无其它写操作。
+    """
+    fp = InviteFingerprint(
+        inviter_user_id=inviter_user_id,
+        ip=ip[:64],
+        screen=screen[:32],
+        device_model=device_model[:64],
+        user_agent=user_agent[:2000] if user_agent else "",  # 截一下防异常长 UA
+    )
+    db.add(fp)
+    db.commit()
+    db.refresh(fp)
+    return fp
+
+
+def resolve_inviter_by_fingerprint(
+    db: Session,
+    *,
+    ip: str,
+    screen: str,
+    device_model: str,
+    window_days: int,
+) -> User | None:
+    """根据指纹反查邀请人(time window 内最近一条匹配)。
+
+    匹配规则:(ip, device_model) 相等 + created_at > now - window_days。
+
+    screen 字段**只入库不反查**:浏览器算物理像素走 `CSS × devicePixelRatio` 路径、
+    Android 走 `DisplayMetrics.widthPixels` 真实硬件值,两端浮点 round 必然 ±1 像素漂移
+    (DPR 不严格是整数)→ 严格匹配注定撞不上。同 device_model 必同 screen(同型号同硬件)
+    → screen 是冗余字段,删它不损精度。撞错只有"同 IP 下同型号手机同时被邀请"才会
+    发生,中国家庭/公司 WiFi 场景概率极低。
+    """
+    if not ip:
+        return None
+    from datetime import datetime, timedelta, timezone
+    cutoff = datetime.now(timezone.utc) - timedelta(days=window_days)
+    fp = db.execute(
+        select(InviteFingerprint)
+        .where(
+            InviteFingerprint.ip == ip,
+            InviteFingerprint.device_model == device_model,
+            InviteFingerprint.created_at > cutoff,
+        )
+        .order_by(InviteFingerprint.created_at.desc())
+        .limit(1)
+    ).scalar_one_or_none()
+    if fp is None:
+        return None
+    return db.get(User, fp.inviter_user_id)

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):
             # 微信明确无此单 → 转账从未创建(如崩溃在扣款后/调用前),退款安全

app/schemas/auth.py

@@ -25,6 +25,8 @@ class UserOut(BaseModel):
     status: str
     created_at: datetime
     last_login_at: datetime
+    # 调试链接权限:前端据此在比价结果弹窗/记录页显示「复制调试链接」按钮。默认 false。
+    debug_trace_enabled: bool = False
 
 
 # ===== Token 通用结构 =====

app/schemas/compare_record.py

@@ -67,6 +67,9 @@ class ComparisonRecordIn(BaseModel):
     status: str | None = Field(None, description="success / failed,可不传由服务端派生")
     # 最优平台商家/商品深链(客户端从 collectedLinks[best_index] 取);「再次比价」直达用
     best_deeplink: str | None = Field(None, description="最优平台深链,再次比价直达")
+    # pricebot done.params.trace_url 原样上报,落库供记录页「复制调试链接」(dir 名含落盘
+    # 时分秒前端拼不出,必须由后端透传)。
+    trace_url: str | None = Field(None, description="本次比价公网调试链接")
 
 
 # ===== 读取出参 =====
@@ -79,6 +82,8 @@ class ComparisonRecordOut(BaseModel):
     id: int
     business_type: str
     trace_id: str
+    # 公网调试链接;仅当 user.debug_trace_enabled 时由端点填充,否则端点层置 None(权限闸)。
+    trace_url: str | None = None
     source_platform_id: str | None = None
     source_platform_name: str | None = None
     source_package: str | None = None

app/schemas/invite.py

@@ -1,6 +1,8 @@
 """邀请相关 API 收发模型。"""
 from __future__ import annotations
 
+from datetime import datetime
+
 from pydantic import BaseModel, Field
 
 
@@ -11,15 +13,61 @@ class InviteInfoOut(BaseModel):
     coins_earned: int               # 累计从邀请获得的金币
 
 
+class LandingTrackIn(BaseModel):
+    """落地页 dl.html 访问时上报的指纹(用于剪贴板归因失败时兜底)。"""
+    ref: str = Field(..., min_length=4, max_length=16, description="邀请码(落地页 ?ref=)")
+    screen: str = Field("", max_length=32, description="屏幕分辨率,如 1080x2400")
+    # IP / UA 服务端从 HTTP 头自动拿,无需 JS 上报
+
+
+class LandingTrackOut(BaseModel):
+    status: str                     # ok / invalid_code / no_ip
+
+
+class FingerprintIn(BaseModel):
+    """客户端登录后剪贴板归因失败时上报的设备指纹。
+
+    跨端匹配字段:浏览器落地页存的同名字段 == 客户端 Build.MODEL / DisplayMetrics 算的值。
+    IP 服务端从 HTTP 头拿,不在这里。
+    """
+    screen: str = Field("", max_length=32)
+    device_model: str = Field("", max_length=64)
+
+
 class BindInviteIn(BaseModel):
-    invite_code: str = Field(..., min_length=4, max_length=16, description="邀请人的邀请码")
+    # invite_code: 可选 — 走指纹兜底时为空(channel=fingerprint)
+    invite_code: str | None = Field(
+        None, max_length=16, description="邀请码;走指纹兜底时为空"
+    )
     channel: str = Field(
         "clipboard", max_length=16,
-        description="归因来源:clipboard(首启读剪贴板自动) / manual(用户手动填)",
+        description="归因来源:clipboard(首启读剪贴板自动) / manual(手动填) / fingerprint(指纹兜底)",
     )
+    # 当 invite_code 为空、channel=fingerprint 时,后端用这组指纹反查
+    fingerprint: FingerprintIn | None = None
 
 
 class BindInviteOut(BaseModel):
-    status: str                     # success / already_bound / invalid_code / self_invite / not_eligible
+    status: str                     # success / already_bound / invalid_code / self_invite / not_eligible / fp_not_found
     coins_awarded: int              # 本次给当前用户(被邀请人)发的金币
     message: str                    # 给前端直接展示的文案
+
+
+class InviteeItem(BaseModel):
+    """被邀请人列表的一条(邀请页小窗 / 完整列表页共用)。
+
+    display_name / avatar_url 的"降级兜底"在后端算好(见 repositories/invite.get_invitees):
+    名字 = 昵称 → 微信昵称 → 脱敏手机号(前端拿不到完整号,必须后端脱敏);
+    头像 = 用户头像 → 微信头像 → null(前端拿到 null 画默认色块)。
+    """
+    display_name: str               # 已兜底好的显示名(真名/脱敏号)
+    avatar_url: str | None = None   # 头像 URL;null = 前端画默认色块
+    coins: int                      # 这次邀请给我(邀请人)发的金币
+    invited_at: datetime            # 邀请绑定时间(前端转"今天/3天前")
+
+
+class InviteeListOut(BaseModel):
+    """GET /invite/invitees 响应:被邀请人列表 + 分页。"""
+    items: list[InviteeItem]
+    total: int                      # 我邀请的总人数(用于"共 N 人")
+    has_more: bool                  # 还有下一页吗(完整列表页滚到底加载更多)

data/media/dl.html

@@ -98,6 +98,27 @@
   var isAndroid = /Android/i.test(ua);
   var ref = new URLSearchParams(location.search).get("ref");  // 邀请码(来自二维码 URL ?ref=)
 
+  // 【任务 3】指纹归因兜底:页面加载即上报访问者指纹,后端存 invite_fingerprint 表。
+  // 当 APK 首启读剪贴板失败(被覆盖)时,客户端会用 (IP+屏幕+UA 解析的手机型号) 反查
+  // 本表 7 天内最近一条匹配 → 撞库出原邀请人 → 走原 bind 流程。
+  // 任何失败都 silent(不影响下载主流程);后端 invalid_code/no_ip 也只返 200。
+  if (ref) {
+    // screen 报【物理像素】= CSS 像素 × devicePixelRatio,跟 Android dm.widthPixels(物理像素)对齐。
+    // 不同设备 DPR 不同(常见 2/2.5/3/3.5),CSS 像素直接报会跟客户端不对齐 → 撞不上库。
+    var _dpr = window.devicePixelRatio || 1;
+    var _sw = Math.round(screen.width * _dpr);
+    var _sh = Math.round(screen.height * _dpr);
+    fetch("/api/v1/invite/landing-track", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({
+        ref: ref,
+        screen: _sw + "x" + _sh,
+        // IP / UA 服务端从 HTTP 头自动拿,无需 JS 上报
+      }),
+    }).catch(function () {});  // silent,绝不阻断下载
+  }
+
   // 把邀请码写进剪贴板,APK 首启时读出来完成归因(deferred deeplink 的关键一步)。
   // 浏览器要求:必须在用户点击手势里调用 + HTTPS 下才允许写。
   function legacyCopy(payload) {                // 老 webview / 无 clipboard API 兜底

pyproject.toml

@@ -34,6 +34,9 @@ dependencies = [
 
     # admin 后台账号密码 hash(用户侧是手机号+验证码登录,不需要密码;admin 才用)
     "bcrypt>=4.0.0",
+
+    # 邀请指纹归因:解析浏览器 UA 拿手机型号(Build.MODEL),跨端匹配用
+    "user-agents>=2.2.0",
 ]
 
 [project.optional-dependencies]

scripts/seed_invitee_avatar_test.py

@@ -0,0 +1,91 @@
+# -*- coding: utf-8 -*-
+"""造测试数据验证邀请列表"真实头像能不能连上"。
+
+A 邀请 B1(上传真实头像)/B2(无头像→色块)/B3(设昵称→色块), 然后:
+  ① 打印 A 的邀请码 + /invitees 返回(看 B1 有 avatar_url、B2 空、B3 昵称);
+  ② 直接 HTTP 访问 B1 的头像地址,确认文件能取到(后端静态托管通)。
+前端 AsyncImage 实际显示,装机后真机用 A 登录看。
+
+跑: <app-server py> scripts/seed_invitee_avatar_test.py"""
+import io
+import struct
+import sys
+import zlib
+
+import httpx
+
+sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
+
+BASE = "http://127.0.0.1:8770"
+
+
+def solid_png(size: int = 96, rgb: tuple = (80, 140, 240)) -> bytes:
+    """生成一张纯色 PNG(蓝色 96x96),够真机看清、魔数是合法 PNG。"""
+    row = b"\x00" + bytes(rgb) * size
+    raw = row * size
+    comp = zlib.compress(raw)
+
+    def chunk(typ: bytes, data: bytes) -> bytes:
+        c = typ + data
+        return struct.pack(">I", len(data)) + c + struct.pack(">I", zlib.crc32(c) & 0xFFFFFFFF)
+
+    sig = b"\x89PNG\r\n\x1a\n"
+    ihdr = struct.pack(">IIBBBBB", size, size, 8, 2, 0, 0, 0)  # 8bit, RGB
+    return sig + chunk(b"IHDR", ihdr) + chunk(b"IDAT", comp) + chunk(b"IEND", b"")
+
+
+def login(c: httpx.Client, phone: str) -> str:
+    c.post(f"{BASE}/api/v1/auth/sms/send", json={"phone": phone})
+    r = c.post(f"{BASE}/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
+    r.raise_for_status()
+    return r.json()["access_token"]
+
+
+def auth(tok: str) -> dict:
+    return {"Authorization": f"Bearer {tok}"}
+
+
+with httpx.Client(timeout=15) as c:
+    # A = 邀请人(真机用它登录看)
+    A_PHONE = "13900008001"
+    a_tok = login(c, A_PHONE)
+    a_code = c.get(f"{BASE}/api/v1/invite/me", headers=auth(a_tok)).json()["invite_code"]
+    print(f"邀请人 A: 手机号={A_PHONE} 验证码=123456 邀请码={a_code}")
+
+    # B1: 绑 A 码 + 上传真实头像
+    b1 = login(c, "13900008002")
+    c.post(f"{BASE}/api/v1/invite/bind", json={"invite_code": a_code}, headers=auth(b1))
+    r = c.post(
+        f"{BASE}/api/v1/user/avatar",
+        headers=auth(b1),
+        files={"file": ("avatar.png", io.BytesIO(solid_png()), "image/png")},
+    )
+    print(f"B1 绑定 + 上传头像 -> HTTP {r.status_code}, avatar_url={r.json().get('avatar_url')!r}")
+
+    # B2: 绑 A 码, 不传头像(测色块兜底)
+    b2 = login(c, "13900008003")
+    c.post(f"{BASE}/api/v1/invite/bind", json={"invite_code": a_code}, headers=auth(b2))
+    print("B2 绑定, 无头像(测色块)")
+
+    # B3: 绑 A 码 + 设昵称(测昵称优先 + 色块)
+    b3 = login(c, "13900008004")
+    c.post(f"{BASE}/api/v1/invite/bind", json={"invite_code": a_code}, headers=auth(b3))
+    c.patch(f"{BASE}/api/v1/user/profile", json={"nickname": "测试昵称"}, headers=auth(b3))
+    print("B3 绑定 + 昵称='测试昵称'")
+
+    # A 的邀请列表
+    inv = c.get(f"{BASE}/api/v1/invite/invitees", headers=auth(a_tok)).json()
+    print(f"\n=== A 的 /invitees (共 {inv['total']} 人) ===")
+    for it in inv["items"]:
+        print(f"  name={it['display_name']!r}  avatar={it['avatar_url']!r}  coins={it['coins']}  time={it['invited_at']}")
+
+    # 关键: 直接 HTTP 取 B1 的头像文件, 确认后端静态托管能 serve(= 前端拿这地址也能加载)
+    b1_avatar = next((it["avatar_url"] for it in inv["items"] if it["avatar_url"]), None)
+    print("\n=== 头像文件 HTTP 可访问性(后端静态托管) ===")
+    if b1_avatar:
+        rr = c.get(f"{BASE}{b1_avatar}")
+        ok = rr.status_code == 200 and (rr.headers.get("content-type", "").startswith("image"))
+        print(f"  GET {BASE}{b1_avatar}")
+        print(f"  -> HTTP {rr.status_code}, content-type={rr.headers.get('content-type')}, {len(rr.content)} bytes  {'✅ 能取到' if ok else '✗ 取不到'}")
+    else:
+        print("  没有带头像的被邀请人?!")

tests/test_invite.py

@@ -1,4 +1,4 @@
-"""好友邀请测试:邀请码、绑定、双方发金币、幂等、自邀/无效码屏蔽。
+"""好友邀请测试:邀请码、绑定、双方发金币、幂等、自邀/无效码屏蔽、指纹兜底归因。
 
 用 sms mock 登录拿 token(同 test_welfare),再跑邀请闭环。
 """
@@ -6,12 +6,16 @@ from __future__ import annotations
 
 from datetime import datetime, timedelta, timezone
 
+from sqlalchemy import select
+
 from app.core.rewards import (
+    INVITE_FP_WINDOW_DAYS,
     INVITE_INVITEE_COINS,
     INVITE_INVITER_COINS,
     INVITE_NEW_USER_WINDOW_HOURS,
 )
 from app.db.session import SessionLocal
+from app.models.invite_fingerprint import InviteFingerprint
 from app.repositories.user import get_user_by_phone
 
 
@@ -158,3 +162,216 @@ def test_old_user_not_eligible(client) -> None:
     assert r.json()["coins_awarded"] == 0
     assert _coin_balance(client, b) == 0
     assert _coin_balance(client, a) == 0
+
+
+# =====================================================================
+# 任务 3:指纹归因(剪贴板兜底,见 brain [[invite-three-tasks]])
+# =====================================================================
+
+def test_landing_track_records_fingerprint(client) -> None:
+    """落地页 POST /landing-track 成功存指纹(无需鉴权)。"""
+    a = _login(client, "13800002040")
+    a_code = _my_code(client, a)
+    # 浏览器无 token 调
+    r = client.post(
+        "/api/v1/invite/landing-track",
+        json={"ref": a_code, "screen": "1080x2400"},
+    )
+    assert r.status_code == 200, r.text
+    assert r.json()["status"] == "ok"
+
+
+def test_landing_track_invalid_code(client) -> None:
+    """无效邀请码 → invalid_code(silent 返,不影响落地页下载流程)。"""
+    r = client.post(
+        "/api/v1/invite/landing-track",
+        json={"ref": "ZZZZZZ", "screen": "1080x2400"},
+    )
+    assert r.status_code == 200, r.text
+    assert r.json()["status"] == "invalid_code"
+
+
+def test_bind_by_fingerprint_success(client) -> None:
+    """B 落地页存指纹 → B 登录后用指纹兜底绑定成功(模拟剪贴板被覆盖丢失)。"""
+    a = _login(client, "13800002041")
+    a_code = _my_code(client, a)
+
+    # 1. B 浏览器(无 token)访问落地页 → 存指纹
+    r1 = client.post(
+        "/api/v1/invite/landing-track",
+        json={"ref": a_code, "screen": "1234x5678"},  # 用独特 screen 避免跟其它测试撞
+    )
+    assert r1.json()["status"] == "ok"
+
+    # 2. B 注册登录(剪贴板被覆盖,没拿到邀请码)
+    b = _login(client, "13800002042")
+
+    # 3. B 不带 invite_code,只带 fingerprint 走兜底
+    r2 = client.post(
+        "/api/v1/invite/bind",
+        json={
+            "channel": "fingerprint",
+            "fingerprint": {"screen": "1234x5678", "device_model": ""},
+        },
+        headers=_auth(b),
+    )
+    assert r2.status_code == 200, r2.text
+    assert r2.json()["status"] == "success"
+    # 双方各发金币
+    assert _coin_balance(client, a) == INVITE_INVITER_COINS
+    assert _coin_balance(client, b) == INVITE_INVITEE_COINS
+
+
+def test_bind_by_fingerprint_not_found(client) -> None:
+    """没有匹配的指纹记录 → fp_not_found,不发奖。"""
+    b = _login(client, "13800002043")
+    r = client.post(
+        "/api/v1/invite/bind",
+        json={
+            "channel": "fingerprint",
+            "fingerprint": {"screen": "0000x9999", "device_model": "no_such_model"},
+        },
+        headers=_auth(b),
+    )
+    assert r.status_code == 200, r.text
+    assert r.json()["status"] == "fp_not_found"
+    assert r.json()["coins_awarded"] == 0
+    assert _coin_balance(client, b) == 0
+
+
+def test_bind_by_fingerprint_outside_window(client) -> None:
+    """指纹记录超出 INVITE_FP_WINDOW_DAYS 窗口 → fp_not_found。
+
+    用独特设备型号隔离:测试环境所有请求 IP 都是 testclient、落地页默认无 UA(解析出的
+    型号="")，而指纹反查按 (IP, 型号) 匹配 → 会串到别的测试残留的、没过期的 model=""
+    指纹上(误判 success)。这里给落地页传一个独特 UA(解析出独特型号 OUTWIN9X),反查只可能
+    命中本测试自己的指纹,"过期就反查不到"才验得准。
+    """
+    a = _login(client, "13800002044")
+    a_code = _my_code(client, a)
+
+    # 落地页存指纹(独特 UA → 独特型号 OUTWIN9X),再手动把 created_at 改到窗口外
+    unique_screen = "2222x3333"
+    unique_ua = "Mozilla/5.0 (Linux; Android 13; OUTWIN9X Build/TQ) AppleWebKit/537.36"
+    r1 = client.post(
+        "/api/v1/invite/landing-track",
+        json={"ref": a_code, "screen": unique_screen},
+        headers={"user-agent": unique_ua},
+    )
+    assert r1.json()["status"] == "ok"
+
+    with SessionLocal() as db:
+        fp = db.execute(
+            select(InviteFingerprint).where(InviteFingerprint.screen == unique_screen)
+        ).scalar_one()
+        fp.created_at = datetime.now(timezone.utc) - timedelta(days=INVITE_FP_WINDOW_DAYS + 1)
+        db.commit()
+
+    b = _login(client, "13800002045")
+    r = client.post(
+        "/api/v1/invite/bind",
+        json={
+            "channel": "fingerprint",
+            "fingerprint": {"screen": unique_screen, "device_model": "OUTWIN9X"},
+        },
+        headers=_auth(b),
+    )
+    assert r.status_code == 200, r.text
+    assert r.json()["status"] == "fp_not_found"
+    assert _coin_balance(client, a) == 0
+    assert _coin_balance(client, b) == 0
+
+
+def test_bind_no_code_no_fingerprint(client) -> None:
+    """既没邀请码也没指纹 → invalid_code(无效请求)。"""
+    b = _login(client, "13800002046")
+    r = client.post(
+        "/api/v1/invite/bind",
+        json={"channel": "fingerprint"},  # 没 invite_code、没 fingerprint
+        headers=_auth(b),
+    )
+    assert r.status_code == 200, r.text
+    assert r.json()["status"] == "invalid_code"
+    assert _coin_balance(client, b) == 0
+
+
+# =====================================================================
+# 被邀请人列表 GET /invitees(邀请页小窗 + 完整列表页共用)
+# =====================================================================
+
+def test_invitees_basic(client) -> None:
+    """A 邀 B、C → 列表返 2 条、total=2、没设资料的名字=脱敏手机号、头像 null、金币对。"""
+    a = _login(client, "13800002050")
+    a_code = _my_code(client, a)
+    for phone in ("13800002051", "13800002052"):
+        u = _login(client, phone)
+        client.post("/api/v1/invite/bind", json={"invite_code": a_code}, headers=_auth(u))
+
+    r = client.get("/api/v1/invite/invitees", headers=_auth(a))
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["total"] == 2
+    assert body["has_more"] is False
+    assert len(body["items"]) == 2
+    # 没设昵称头像 → 名字脱敏手机号、头像 null(不依赖顺序用 set)
+    names = {it["display_name"] for it in body["items"]}
+    assert names == {"138****2051", "138****2052"}
+    assert all(it["avatar_url"] is None for it in body["items"])
+    assert all(it["coins"] == INVITE_INVITER_COINS for it in body["items"])
+
+
+def test_invitees_order_desc(client) -> None:
+    """按邀请时间倒序:最近邀请的在最前(手动拉开时间,避开 SQLite 秒级精度)。"""
+    from app.models.invite import InviteRelation
+    a = _login(client, "13800002060")
+    a_code = _my_code(client, a)
+    b = _login(client, "13800002061")
+    c = _login(client, "13800002062")
+    client.post("/api/v1/invite/bind", json={"invite_code": a_code}, headers=_auth(b))
+    client.post("/api/v1/invite/bind", json={"invite_code": a_code}, headers=_auth(c))
+    with SessionLocal() as db:
+        ub = get_user_by_phone(db, "13800002061")
+        rel_b = db.execute(
+            select(InviteRelation).where(InviteRelation.invitee_user_id == ub.id)
+        ).scalar_one()
+        rel_b.created_at = datetime.now(timezone.utc) - timedelta(hours=2)
+        db.commit()
+    items = client.get("/api/v1/invite/invitees", headers=_auth(a)).json()["items"]
+    assert items[0]["display_name"] == "138****2062"  # C 刚邀,在前
+    assert items[1]["display_name"] == "138****2061"  # B 2 小时前,在后
+
+
+def test_invitees_nickname_priority(client) -> None:
+    """设了昵称的被邀请人 → 名字用昵称(优先于脱敏手机号)。"""
+    a = _login(client, "13800002070")
+    a_code = _my_code(client, a)
+    b = _login(client, "13800002071")
+    client.post("/api/v1/invite/bind", json={"invite_code": a_code}, headers=_auth(b))
+    with SessionLocal() as db:
+        u = get_user_by_phone(db, "13800002071")
+        u.nickname = "小明"
+        db.commit()
+    items = client.get("/api/v1/invite/invitees", headers=_auth(a)).json()["items"]
+    assert items[0]["display_name"] == "小明"
+
+
+def test_invitees_pagination(client) -> None:
+    """分页:limit=1 → 1 条 + has_more=True;offset=1 → 第 2 条 + has_more=False。"""
+    a = _login(client, "13800002080")
+    a_code = _my_code(client, a)
+    for phone in ("13800002081", "13800002082"):
+        u = _login(client, phone)
+        client.post("/api/v1/invite/bind", json={"invite_code": a_code}, headers=_auth(u))
+
+    p1 = client.get("/api/v1/invite/invitees?limit=1&offset=0", headers=_auth(a)).json()
+    assert len(p1["items"]) == 1 and p1["total"] == 2 and p1["has_more"] is True
+    p2 = client.get("/api/v1/invite/invitees?limit=1&offset=1", headers=_auth(a)).json()
+    assert len(p2["items"]) == 1 and p2["has_more"] is False
+
+
+def test_invitees_empty_and_auth(client) -> None:
+    """没邀请人 → 空列表;不带 token → 401。"""
+    a = _login(client, "13800002090")
+    body = client.get("/api/v1/invite/invitees", headers=_auth(a)).json()
+    assert body["total"] == 0 and body["items"] == [] and body["has_more"] is False
+    assert client.get("/api/v1/invite/invitees").status_code == 401