feat(alembic): 添加合并广告奖励会话与提现安全的迁移脚本

alembic/versions/a8c47fc4dc39_merge_invite_fingerprint_and_ad_reward_.py

@@ -1,26 +0,0 @@
-"""merge invite_fingerprint and ad_reward heads
-
-Revision ID: a8c47fc4dc39
-Revises: invite_fingerprint_table, 044dce6e9b1f
-Create Date: 2026-06-10 01:03:03.699443
-
-"""
-from typing import Sequence, Union
-
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision: str = 'a8c47fc4dc39'
-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:
-    pass
-
-
-def downgrade() -> None:
-    pass

alembic/versions/f8d3b1e60a27_add_debug_trace_and_merge_heads.py

@@ -1,49 +0,0 @@
-"""add user.debug_trace_enabled + comparison_record.trace_url
-
-调试链接权限功能：
-  - user.debug_trace_enabled：运营后台给指定用户开「复制调试链接」权限
-  - comparison_record.trace_url：比价记录页「复制调试链接」的数据（pricebot done 帧给，
-    dir 名含落盘时分秒、前端/server 拼不出，必须落库）
-
-注：本迁移最初基于旧 main（#31，双 head）写、down_revision 曾指向那对父；但 #33 已用
-a8c47fc4dc39 合并了那对双 head，故改为线性接在 a8c47fc4dc39 之后、只负责加两列
-（避免与 a8c47fc4dc39 平行再造一个双 head）。
-
-Revision ID: f8d3b1e60a27
-Revises: a8c47fc4dc39
-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] = "a8c47fc4dc39"
-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

@@ -1,50 +0,0 @@
-"""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')

app/admin/main.py

@@ -21,7 +21,6 @@ from app.admin.routers.dashboard import router as dashboard_router
 from app.admin.routers.ops_stat_config import router as ops_stat_config_router
 from app.admin.routers.feedback import router as feedback_router
 from app.admin.routers.ops_marquee_seed import router as ops_marquee_seed_router
-from app.admin.routers.price_report import router as price_report_router
 from app.admin.routers.users import router as users_router
 from app.admin.routers.wallet import router as wallet_router
 from app.admin.routers.withdraw import router as withdraw_router
@@ -81,7 +80,6 @@ admin_app.include_router(ops_marquee_seed_router)
 admin_app.include_router(users_router)
 admin_app.include_router(wallet_router)
 admin_app.include_router(withdraw_router)
-admin_app.include_router(price_report_router)
 admin_app.include_router(feedback_router)
 admin_app.include_router(admins_router)
 admin_app.include_router(audit_router)

app/admin/repositories/mutations.py

@@ -8,13 +8,9 @@ set_user_status / update_feedback_status 支持 commit=False,让 router 把"业
 """
 from __future__ import annotations
 
-from datetime import datetime
-
 from sqlalchemy.orm import Session
 
-from app.core.rewards import CN_TZ
 from app.models.feedback import Feedback
-from app.models.price_report import PriceReport
 from app.models.user import User
 
 
@@ -28,20 +24,6 @@ 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:
@@ -52,30 +34,3 @@ def update_feedback_status(
     else:
         db.flush()
     return feedback
-
-
-def review_price_report(
-    db: Session,
-    report: PriceReport,
-    *,
-    status: str,
-    reward_coins: int | None = None,
-    reject_reason: str | None = None,
-    commit: bool = True,
-) -> PriceReport:
-    """审核上报:置 approved/rejected + 记审核时间;通过填 reward_coins、拒绝填 reject_reason。
-
-    发金币(wallet.grant_coins)不在这里——由 router 在同一事务里调,涉钱逻辑不重写(见模块头注释)。
-    """
-    report.status = status
-    if reward_coins is not None:
-        report.reward_coins = reward_coins
-    if reject_reason is not None:
-        report.reject_reason = reject_reason
-    report.reviewed_at = datetime.now(CN_TZ).replace(tzinfo=None)  # 北京 wall-clock,同 created_at
-    if commit:
-        db.commit()
-        db.refresh(report)
-    else:
-        db.flush()
-    return report

app/admin/repositories/queries.py

@@ -14,7 +14,6 @@ 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.price_report import PriceReport
 from app.models.user import User
 from app.models.wallet import CashTransaction, CoinAccount, CoinTransaction, WithdrawOrder
 
@@ -396,34 +395,3 @@ def get_user_overview(db: Session, user_id: int) -> dict | None:
         ).scalar_one(),
         "feedback_total": _count(Feedback, Feedback.user_id == user_id),
     }
-
-
-def list_price_reports(
-    db: Session,
-    *,
-    status: str | None = None,
-    user_id: int | None = None,
-    limit: int = 20,
-    cursor: int | None = None,
-) -> tuple[list[PriceReport], int | None]:
-    """上报更低价列表(admin 全量,可按状态/用户筛)。游标同 feedback:id 倒序。"""
-    stmt = select(PriceReport)
-    if status:
-        stmt = stmt.where(PriceReport.status == status)
-    if user_id is not None:
-        stmt = stmt.where(PriceReport.user_id == user_id)
-    return cursor_paginate(db, stmt, PriceReport.id, limit=limit, cursor=cursor)
-
-
-def price_report_summary(db: Session) -> dict:
-    """上报审核台各状态计数(待审核/已通过/已拒绝/合计)。"""
-    rows = db.execute(
-        select(PriceReport.status, func.count(PriceReport.id)).group_by(PriceReport.status)
-    ).all()
-    by_status = {status: int(count) for status, count in rows}
-    return {
-        "pending": by_status.get("pending", 0),
-        "approved": by_status.get("approved", 0),
-        "rejected": by_status.get("rejected", 0),
-        "total": sum(by_status.values()),
-    }

app/admin/routers/price_report.py

@@ -1,103 +0,0 @@
-"""admin 上报更低价审核:列表 + 统计(读)+ 通过(发金币)/拒绝(写,带审计)。
-
-数据由客户端 POST /api/v1/report 写入 price_report 表(提交即 pending);本路由是运营后台
-对它的人工审核窗口。**通过** → 给上报用户钱包发固定金币(PRICE_REPORT_REWARD_COINS):
-改状态 + 发金币(wallet.grant_coins)+ 审计同一事务一起 commit(原子,仿 users.grant_user_coins),
-绝不只改状态不发钱或反之。客户端轮询 GET /api/v1/report/records 自动看到结果(无需推送)。
-"""
-from __future__ import annotations
-
-from typing import Annotated
-
-from fastapi import APIRouter, Depends, HTTPException, Query, Request
-
-from app.admin.audit import write_audit
-from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
-from app.admin.repositories import mutations, queries
-from app.admin.schemas.common import CursorPage, OkResponse
-from app.admin.schemas.price_report import (
-    PriceReportOut,
-    PriceReportRejectRequest,
-    PriceReportSummary,
-)
-from app.core.rewards import PRICE_REPORT_REWARD_COINS
-from app.models.admin import AdminUser
-from app.models.price_report import PriceReport
-from app.repositories import wallet as wallet_repo
-
-router = APIRouter(
-    prefix="/admin/api/price-reports",
-    tags=["admin-price-report"],
-    dependencies=[Depends(get_current_admin)],
-)
-
-
-@router.get("", response_model=CursorPage[PriceReportOut], summary="上报更低价列表(筛选+分页)")
-def list_price_reports(
-    db: AdminDb,
-    status: Annotated[str | None, Query()] = None,
-    user_id: Annotated[int | None, Query()] = None,
-    limit: Annotated[int, Query(ge=1, le=100)] = 20,
-    cursor: Annotated[int | None, Query()] = None,
-) -> CursorPage[PriceReportOut]:
-    items, next_cursor = queries.list_price_reports(
-        db, status=status, user_id=user_id, limit=limit, cursor=cursor,
-    )
-    return CursorPage(
-        items=[PriceReportOut.model_validate(r) for r in items], next_cursor=next_cursor,
-    )
-
-
-@router.get("/summary", response_model=PriceReportSummary, summary="上报审核统计(各状态计数)")
-def price_report_summary(db: AdminDb) -> PriceReportSummary:
-    return PriceReportSummary.model_validate(queries.price_report_summary(db))
-
-
-@router.post("/{report_id}/approve", response_model=OkResponse, summary="通过上报(发固定金币)")
-def approve_price_report(
-    report_id: int,
-    request: Request,
-    admin: Annotated[AdminUser, Depends(require_role("operator"))],
-    db: AdminDb,
-) -> OkResponse:
-    rep = db.get(PriceReport, report_id)
-    if rep is None:
-        raise HTTPException(status_code=404, detail="上报记录不存在")
-    if rep.status != "pending":
-        raise HTTPException(status_code=400, detail=f"该上报已审核过(当前 {rep.status}),不可重复操作")
-    coins = PRICE_REPORT_REWARD_COINS
-    # 改状态 + 发金币 + 审计同一事务(commit=False),最后一起 commit:改了就有痕、发了就留账
-    mutations.review_price_report(db, rep, status="approved", reward_coins=coins, commit=False)
-    wallet_repo.grant_coins(
-        db, rep.user_id, coins,
-        biz_type="price_report_reward", ref_id=str(rep.id), remark="上报更低价审核通过",
-    )
-    write_audit(
-        db, admin, action="price_report.approve", target_type="price_report", target_id=report_id,
-        detail={"reward_coins": coins, "user_id": rep.user_id}, ip=get_client_ip(request), commit=False,
-    )
-    db.commit()
-    return OkResponse()
-
-
-@router.post("/{report_id}/reject", response_model=OkResponse, summary="拒绝上报")
-def reject_price_report(
-    report_id: int,
-    body: PriceReportRejectRequest,
-    request: Request,
-    admin: Annotated[AdminUser, Depends(require_role("operator"))],
-    db: AdminDb,
-) -> OkResponse:
-    rep = db.get(PriceReport, report_id)
-    if rep is None:
-        raise HTTPException(status_code=404, detail="上报记录不存在")
-    if rep.status != "pending":
-        raise HTTPException(status_code=400, detail=f"该上报已审核过(当前 {rep.status}),不可重复操作")
-    reason = body.reason.strip()
-    mutations.review_price_report(db, rep, status="rejected", reject_reason=reason, commit=False)
-    write_audit(
-        db, admin, action="price_report.reject", target_type="price_report", target_id=report_id,
-        detail={"reason": reason, "user_id": rep.user_id}, ip=get_client_ip(request), commit=False,
-    )
-    db.commit()
-    return OkResponse()

app/admin/routers/users.py

@@ -13,7 +13,6 @@ from app.admin.schemas.user import (
     AdminUserListItem,
     AdminUserOverview,
     GrantCoinsRequest,
-    SetDebugTraceRequest,
     SetUserStatusRequest,
 )
 from app.models.admin import AdminUser
@@ -78,28 +77,6 @@ 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/schemas/price_report.py

@@ -1,46 +0,0 @@
-"""admin 上报更低价审核 schemas。"""
-from __future__ import annotations
-
-from datetime import datetime
-
-from pydantic import BaseModel, ConfigDict, Field
-
-
-class PriceReportOut(BaseModel):
-    """admin 列表项:price_report 全字段(审核要看的快照 + 截图 + 状态 + 奖励)。"""
-
-    model_config = ConfigDict(from_attributes=True)
-
-    id: int
-    user_id: int
-    comparison_record_id: int | None = None
-    # 选中比价记录的快照
-    store_name: str | None = None
-    dish_summary: str | None = None
-    original_platform_id: str | None = None
-    original_platform_name: str | None = None
-    original_price_cents: int | None = None
-    # 用户上报的更低价
-    reported_platform_id: str
-    reported_platform_name: str
-    reported_price_cents: int
-    images: list[str] = []
-    # 审核
-    status: str
-    reject_reason: str | None = None
-    reward_coins: int | None = None
-    reviewed_at: datetime | None = None
-    created_at: datetime
-
-
-class PriceReportRejectRequest(BaseModel):
-    reason: str = Field(min_length=1, max_length=256, description="拒绝理由,用户端记录页会看到")
-
-
-class PriceReportSummary(BaseModel):
-    """审核台顶部各状态计数。"""
-
-    pending: int
-    approved: int
-    rejected: int
-    total: int

app/admin/schemas/user.py

@@ -15,7 +15,6 @@ 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
@@ -46,7 +45,3 @@ class SetUserStatusRequest(BaseModel):
     status: Literal["active", "disabled"] = Field(
         ..., description="active=解封 / disabled=封禁(注销 deleted 不走此接口)"
     )
-
-
-class SetDebugTraceRequest(BaseModel):
-    enabled: bool = Field(..., description="是否给该用户开「复制调试链接」权限")

app/api/v1/compare_record.py

@@ -77,12 +77,10 @@ def list_records(
     items, next_cursor = crud_compare.list_records(
         db, user.id, limit=limit, cursor=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)
+    return ComparisonRecordPage(
+        items=[ComparisonRecordOut.model_validate(it) for it in items],
+        next_cursor=next_cursor,
+    )
 
 
 @router.get(
@@ -96,12 +94,4 @@ 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")
-    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
+    return ComparisonRecordDetailOut.model_validate(rec)

app/api/v1/coupon.py

@@ -183,15 +183,7 @@ 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,37 +1,25 @@
 """好友邀请 endpoint。
 
-路由前缀 /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 访问时上报指纹(剪贴板兜底的服务端一端,无需鉴权)
+路由前缀 /api/v1/invite,需 Bearer 鉴权(用户级数据):
+  GET  /me     我的邀请码 + 分享链接 + 已邀人数/已得金币(邀请页展示)
+  POST /bind   把当前登录用户(被邀请人)绑定到某邀请码,注册即生效,双方各发金币
 
-绑定的真正逻辑(邀请码生成/反查、幂等、自邀屏蔽、发金币、指纹反查)在
-repositories/invite.py。
+客户端两种调用 /bind:
+  - 自动:首启读剪贴板拿到邀请码 → 登录后调本接口(channel=clipboard)
+  - 手动:用户在邀请页输入好友邀请码(channel=manual)
+
+绑定的真正逻辑(邀请码生成/反查、幂等、自邀屏蔽、发金币)在 repositories/invite.py。
 """
 from __future__ import annotations
 
 import logging
-import re
 
-from fastapi import APIRouter, Request
+from fastapi import APIRouter
 
 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,
-    InviteeItem,
-    InviteeListOut,
-    InviteInfoOut,
-    LandingTrackIn,
-    LandingTrackOut,
-)
+from app.schemas.invite import BindInviteIn, BindInviteOut, InviteInfoOut
 
 logger = logging.getLogger("shagua.invite")
 
@@ -43,43 +31,9 @@ _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)
@@ -94,124 +48,17 @@ def my_invite(user: CurrentUser, db: DbSession) -> InviteInfoOut:
     )
 
 
-@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,
+@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,
     )
     logger.info(
-        "invite landing-track inviter=%d ip=%s screen=%s model=%s",
-        inviter.id, ip, req.screen, device_model,
+        "invite bind invitee=%d code=%s channel=%s -> %s",
+        user.id, req.invite_code, req.channel, result.status,
     )
-    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="invalid_code",
-        coins_awarded=0,
-        message=_BIND_MESSAGES["invalid_code"],
+        status=result.status,
+        coins_awarded=result.invitee_coin,
+        message=_BIND_MESSAGES.get(result.status, result.status),
     )

app/core/rewards.py

@@ -77,12 +77,6 @@ def record_milestone_reward(milestone: int) -> int:
     return RECORD_MILESTONES[milestone - 1]
 
 
-# ===== 上报更低价(人工审核通过发固定金币)=====
-# 用户上报"某平台比我们算的最低价更便宜"+ 截图,经运营后台人工审核通过后发放的固定金币奖励。
-# 与广告/任务同量级;固定值(产品 2026-06 定),要调直接改这里;客户端记录页按 reward_coins 显示。
-PRICE_REPORT_REWARD_COINS: int = 1000
-
-
 # ===== 邀请好友(注册即生效,邀请人 + 被邀请人各发金币)=====
 # 10000 金币 = 1 元,双方各得 1 元。MVP 先用固定常量(不走 app_config)。
 INVITE_INVITER_COINS: int = 10000
@@ -91,11 +85,6 @@ 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/models/__init__.py

@@ -13,7 +13,6 @@ 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,10 +53,6 @@ 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

@@ -1,46 +0,0 @@
-"""邀请指纹归因表(剪贴板归因兜底)。
-
-数据流(对应 [[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 Boolean, DateTime, Integer, String, false, func
+from sqlalchemy import DateTime, Integer, String, func
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.db.base import Base
@@ -51,13 +51,6 @@ 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/repositories/comparison.py

@@ -84,7 +84,6 @@ 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 delete, select
+from sqlalchemy import select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -65,19 +65,6 @@ 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,7 +21,6 @@ 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
 
@@ -165,121 +164,3 @@ 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/schemas/auth.py

@@ -25,8 +25,6 @@ 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,9 +67,6 @@ 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="本次比价公网调试链接")
 
 
 # ===== 读取出参 =====
@@ -82,8 +79,6 @@ 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,8 +1,6 @@
 """邀请相关 API 收发模型。"""
 from __future__ import annotations
 
-from datetime import datetime
-
 from pydantic import BaseModel, Field
 
 
@@ -13,61 +11,15 @@ 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: 可选 — 走指纹兜底时为空(channel=fingerprint)
-    invite_code: str | None = Field(
-        None, max_length=16, description="邀请码;走指纹兜底时为空"
-    )
+    invite_code: str = Field(..., min_length=4, max_length=16, description="邀请人的邀请码")
     channel: str = Field(
         "clipboard", max_length=16,
-        description="归因来源:clipboard(首启读剪贴板自动) / manual(手动填) / fingerprint(指纹兜底)",
+        description="归因来源:clipboard(首启读剪贴板自动) / manual(用户手动填)",
     )
-    # 当 invite_code 为空、channel=fingerprint 时,后端用这组指纹反查
-    fingerprint: FingerprintIn | None = None
 
 
 class BindInviteOut(BaseModel):
-    status: str                     # success / already_bound / invalid_code / self_invite / not_eligible / fp_not_found
+    status: str                     # success / already_bound / invalid_code / self_invite / not_eligible
     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,27 +98,6 @@
   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,9 +34,6 @@ 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

@@ -1,91 +0,0 @@
-# -*- 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,16 +6,12 @@ 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
 
 
@@ -162,216 +158,3 @@ 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