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

- rewards: 激励视频实发改按 calculate_ad_reward_coin(eCPM, 当日第N次) 公式;AD_REWARD_COIN/MAX_AD_REWARD_COIN 降为历史兼容口径;新增 SIGNIN_BOOST_COIN=2000

- 签到膨胀: Day1-13 看完激励视频额外发固定金币、Day14 不允许 (signin.py / signin-boost)

- ad_session_id 幂等: ad_ecpm/ad_reward/ad_feed_reward 记录加 ad_session_id 列 + 唯一索引(新迁移 coin_reward_phase2 / ad_feed_reward_session)

- ad.py + schemas + repositories: ecpm-report / feed-reward / reward-status / test-grant 改造;config_schema 增 signin_boost_coin;admin stats overview 补充

- 同步 docs/api + docs/database + docs/integrations/pangle 及 tests(test_ad_reward / test_welfare)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

Co-authored-by: OuYingJun1024 <1034284404@qq.com>
Reviewed-on: #28
Co-authored-by: ouzhou <ouzhou@wonderable.ai>
Co-committed-by: ouzhou <ouzhou@wonderable.ai>

.env.example

@@ -40,6 +40,9 @@ MT_CPS_APP_KEY=
 MT_CPS_APP_SECRET=
 # 默认渠道追踪标识(sid),用于区分不同 app 的 CPS 数据
 MT_CPS_DEFAULT_SID=sgbjia
+# 美团调用走的代理。⚠️ 本机/内网开发直连美团会 SSL EOF,必须填本地代理(如 http://127.0.0.1:7897);
+# 线上国内服务器留空(=直连)。留空且本机直连失败时 /feed、/coupons、/top-sales 会返回空。
+MT_CPS_PROXY=
 
 # ===== Pricebot 上游 (领券/比价业务透传目标) =====
 # 客户端调本服务的 /api/v1/coupon/step 等,我们透传到 pricebot-backend。
@@ -76,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

@@ -23,7 +23,13 @@ dist/
 *.db-journal
 *.db-shm
 *.db-wal
-data/
+# data/ 整体不入库(运行时数据/上传文件/大二进制)。例外:邀请落地页 dl.html——
+# 它既是生产落地页又是本地测试资产,纳入 git 便于同事一致测试
+# (见 docs/邀请功能-实现原理与本地测试.md)。其余(avatars/ / *.apk / app.db 等)仍忽略。
+data/*
+!data/media/
+data/media/*
+!data/media/dl.html
 
 secrets/*
 !secrets/.gitkeep
@@ -37,3 +43,4 @@ secrets/*
 
 # 运行日志(run.sh 输出, 不入库)
 *.log
+logs/

alembic/versions/0cf18d590b1d_merge_coupon_state_and_invite_heads.py

@@ -0,0 +1,26 @@
+"""merge coupon_state and invite heads
+
+Revision ID: 0cf18d590b1d
+Revises: coupon_state_tables, invite_code_and_relation
+Create Date: 2026-06-08 05:49:29.604571
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '0cf18d590b1d'
+down_revision: Union[str, Sequence[str], None] = ('coupon_state_tables', 'invite_code_and_relation')
+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/11a1d08c6f55_merge_meituan_coupon_与_ops_price_.py

@@ -0,0 +1,26 @@
+"""merge meituan_coupon 与 ops/price_observation 迁移头
+
+Revision ID: 11a1d08c6f55
+Revises: meituan_coupon_table, opsrename01
+Create Date: 2026-06-08 00:06:01.633796
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '11a1d08c6f55'
+down_revision: Union[str, Sequence[str], None] = ('meituan_coupon_table', 'opsrename01')
+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/ad_feed_reward_session_id.py

@@ -0,0 +1,38 @@
+"""add ad_session_id to ad feed reward record
+
+Revision ID: ad_feed_reward_session
+Revises: coin_reward_phase2
+Create Date: 2026-06-09
+"""
+
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "ad_feed_reward_session"
+down_revision = "coin_reward_phase2"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "ad_feed_reward_record",
+        sa.Column("ad_session_id", sa.String(length=64), nullable=True),
+    )
+    op.create_index(
+        op.f("ix_ad_feed_reward_record_ad_session_id"),
+        "ad_feed_reward_record",
+        ["ad_session_id"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        op.f("ix_ad_feed_reward_record_ad_session_id"),
+        table_name="ad_feed_reward_record",
+    )
+    op.drop_column("ad_feed_reward_record", "ad_session_id")

alembic/versions/coin_reward_points_phase2.py

@@ -0,0 +1,71 @@
+"""coin reward points phase2
+
+Revision ID: coin_reward_phase2
+Revises: 0cf18d590b1d
+Create Date: 2026-06-08
+"""
+
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "coin_reward_phase2"
+down_revision = "0cf18d590b1d"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "ad_ecpm_record",
+        sa.Column("ad_session_id", sa.String(length=64), nullable=True),
+    )
+    op.create_index(
+        op.f("ix_ad_ecpm_record_ad_session_id"),
+        "ad_ecpm_record",
+        ["ad_session_id"],
+        unique=False,
+    )
+    op.create_index(
+        "uq_ad_ecpm_record_session",
+        "ad_ecpm_record",
+        ["ad_session_id"],
+        unique=True,
+    )
+
+    op.add_column(
+        "ad_reward_record",
+        sa.Column(
+            "reward_scene",
+            sa.String(length=32),
+            nullable=False,
+            server_default="reward_video",
+        ),
+    )
+    op.add_column(
+        "ad_reward_record",
+        sa.Column("ad_session_id", sa.String(length=64), nullable=True),
+    )
+    op.add_column(
+        "ad_reward_record",
+        sa.Column("ecpm_raw", sa.String(length=32), nullable=True),
+    )
+    op.create_index(
+        op.f("ix_ad_reward_record_ad_session_id"),
+        "ad_reward_record",
+        ["ad_session_id"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(op.f("ix_ad_reward_record_ad_session_id"), table_name="ad_reward_record")
+    op.drop_column("ad_reward_record", "ecpm_raw")
+    op.drop_column("ad_reward_record", "ad_session_id")
+    op.drop_column("ad_reward_record", "reward_scene")
+
+    op.drop_index("uq_ad_ecpm_record_session", table_name="ad_ecpm_record")
+    op.drop_index(op.f("ix_ad_ecpm_record_ad_session_id"), table_name="ad_ecpm_record")
+    op.drop_column("ad_ecpm_record", "ad_session_id")

alembic/versions/coupon_state_tables.py

@@ -0,0 +1,77 @@
+"""coupon state tables (领券今日状态:弹窗频控 engagement + 领券记录 claim)
+
+Revision ID: coupon_state_tables
+Revises: 11a1d08c6f55
+Create Date: 2026-06-08 00:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'coupon_state_tables'
+down_revision: Union[str, Sequence[str], None] = '11a1d08c6f55'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+# PG 上为 JSONB,其它(SQLite dev)为 JSON——与模型层 with_variant 对齐
+_JSON = sa.JSON().with_variant(postgresql.JSONB(), 'postgresql')
+
+
+def upgrade() -> None:
+    # 领券记录(资产层,当前不参与去重判断)
+    op.create_table(
+        'coupon_claim_record',
+        sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column('device_id', sa.String(length=64), nullable=False),
+        sa.Column('user_id', sa.Integer(), nullable=True),
+        sa.Column('coupon_id', sa.String(length=64), nullable=False),
+        sa.Column('claim_date', sa.Date(), nullable=False),
+        sa.Column('status', sa.String(length=24), nullable=False),
+        sa.Column('vendor', sa.String(length=48), nullable=True),
+        sa.Column('coupon_name', sa.String(length=128), nullable=True),
+        sa.Column('claimed_count', sa.Integer(), nullable=True),
+        sa.Column('trace_id', sa.String(length=64), nullable=True),
+        sa.Column('reason', sa.String(length=255), nullable=True),
+        sa.Column('extra', _JSON, nullable=True),
+        sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
+        sa.PrimaryKeyConstraint('id'),
+        sa.UniqueConstraint('device_id', 'coupon_id', 'claim_date', name='uq_coupon_claim_device_coupon_date'),
+    )
+    with op.batch_alter_table('coupon_claim_record', schema=None) as batch_op:
+        batch_op.create_index('ix_coupon_claim_device_date', ['device_id', 'claim_date'], unique=False)
+        batch_op.create_index(batch_op.f('ix_coupon_claim_record_user_id'), ['user_id'], unique=False)
+        batch_op.create_index(batch_op.f('ix_coupon_claim_record_trace_id'), ['trace_id'], unique=False)
+
+    # 弹窗频控(今天 engage 过就不再弹)
+    op.create_table(
+        'coupon_prompt_engagement',
+        sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column('device_id', sa.String(length=64), nullable=False),
+        sa.Column('user_id', sa.Integer(), nullable=True),
+        sa.Column('engage_date', sa.Date(), nullable=False),
+        sa.Column('engage_type', sa.String(length=16), nullable=False),
+        sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
+        sa.PrimaryKeyConstraint('id'),
+        sa.UniqueConstraint('device_id', 'engage_date', name='uq_coupon_engage_device_date'),
+    )
+    with op.batch_alter_table('coupon_prompt_engagement', schema=None) as batch_op:
+        batch_op.create_index(batch_op.f('ix_coupon_prompt_engagement_user_id'), ['user_id'], unique=False)
+
+
+def downgrade() -> None:
+    with op.batch_alter_table('coupon_prompt_engagement', schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f('ix_coupon_prompt_engagement_user_id'))
+    op.drop_table('coupon_prompt_engagement')
+
+    with op.batch_alter_table('coupon_claim_record', schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f('ix_coupon_claim_record_trace_id'))
+        batch_op.drop_index(batch_op.f('ix_coupon_claim_record_user_id'))
+        batch_op.drop_index('ix_coupon_claim_device_date')
+    op.drop_table('coupon_claim_record')

alembic/versions/invite_code_and_relation.py

@@ -0,0 +1,54 @@
+"""user.invite_code + invite_relation table
+
+Revision ID: invite_code_and_relation
+Revises: 11a1d08c6f55
+Create Date: 2026-06-08 00:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'invite_code_and_relation'
+down_revision: Union[str, Sequence[str], None] = '11a1d08c6f55'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # user 加邀请码列(唯一索引;nullable 允许多个 NULL=未生成的用户)
+    with op.batch_alter_table('user', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('invite_code', sa.String(length=16), nullable=True))
+        batch_op.create_index('ix_user_invite_code', ['invite_code'], unique=True)
+
+    # 邀请关系表
+    op.create_table(
+        'invite_relation',
+        sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column('inviter_user_id', sa.Integer(), nullable=False),
+        sa.Column('invitee_user_id', sa.Integer(), nullable=False),
+        sa.Column('channel', sa.String(length=16), nullable=False, server_default='clipboard'),
+        sa.Column('status', sa.String(length=16), nullable=False, server_default='effective'),
+        sa.Column('inviter_coin', sa.Integer(), nullable=False, server_default='0'),
+        sa.Column('invitee_coin', sa.Integer(), nullable=False, server_default='0'),
+        sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+        sa.ForeignKeyConstraint(['inviter_user_id'], ['user.id']),
+        sa.ForeignKeyConstraint(['invitee_user_id'], ['user.id']),
+        sa.PrimaryKeyConstraint('id'),
+    )
+    op.create_index('ix_invite_relation_inviter_user_id', 'invite_relation', ['inviter_user_id'])
+    op.create_index('ix_invite_relation_invitee_user_id', 'invite_relation', ['invitee_user_id'], unique=True)
+    op.create_index('ix_invite_relation_created_at', 'invite_relation', ['created_at'])
+
+
+def downgrade() -> None:
+    op.drop_index('ix_invite_relation_created_at', table_name='invite_relation')
+    op.drop_index('ix_invite_relation_invitee_user_id', table_name='invite_relation')
+    op.drop_index('ix_invite_relation_inviter_user_id', table_name='invite_relation')
+    op.drop_table('invite_relation')
+    with op.batch_alter_table('user', schema=None) as batch_op:
+        batch_op.drop_index('ix_user_invite_code')
+        batch_op.drop_column('invite_code')

alembic/versions/meituan_coupon_table.py

@@ -0,0 +1,62 @@
+"""meituan_coupon table(美团 CPS 券本地缓存,供销量/佣金排序从库里捞、本地排序)
+
+Revision ID: meituan_coupon_table
+Revises: withdraw_review_ad_watch
+Create Date: 2026-06-06 12:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'meituan_coupon_table'
+down_revision: Union[str, Sequence[str], None] = 'withdraw_review_ad_watch'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        'meituan_coupon',
+        sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column('source', sa.String(length=16), nullable=False),
+        sa.Column('platform', sa.Integer(), nullable=False),
+        sa.Column('biz_line', sa.Integer(), nullable=True),
+        sa.Column('city_id', sa.String(length=32), nullable=False),
+        sa.Column('product_view_sign', sa.String(length=128), nullable=False),
+        sa.Column('sku_view_id', sa.String(length=128), nullable=True),
+        sa.Column('name', sa.String(length=256), nullable=True),
+        sa.Column('brand_name', sa.String(length=128), nullable=True),
+        sa.Column('sell_price_cents', sa.Integer(), nullable=True),
+        sa.Column('original_price_cents', sa.Integer(), nullable=True),
+        sa.Column('head_url', sa.String(length=512), nullable=True),
+        sa.Column('sale_volume', sa.String(length=32), nullable=True),
+        sa.Column('sale_volume_num', sa.Integer(), nullable=True),
+        sa.Column('commission_percent', sa.Float(), nullable=True),
+        sa.Column('commission_amount_cents', sa.Integer(), nullable=True),
+        sa.Column('poi_name', sa.String(length=128), nullable=True),
+        sa.Column('available_poi_num', sa.Integer(), nullable=True),
+        sa.Column('delivery_distance_m', sa.Float(), nullable=True),
+        sa.Column('dedup_key', sa.String(length=64), nullable=False),
+        sa.Column('raw', sa.JSON().with_variant(postgresql.JSONB(), 'postgresql'), nullable=False),
+        sa.Column('first_seen', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
+        sa.Column('last_seen', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
+        sa.PrimaryKeyConstraint('id'),
+        sa.UniqueConstraint('source', 'product_view_sign', name='uq_meituan_coupon_source_sign'),
+    )
+    op.create_index('ix_meituan_coupon_source', 'meituan_coupon', ['source'])
+    op.create_index('ix_meituan_coupon_city_id', 'meituan_coupon', ['city_id'])
+    op.create_index('ix_meituan_coupon_brand_name', 'meituan_coupon', ['brand_name'])
+    op.create_index('ix_meituan_coupon_sale_volume_num', 'meituan_coupon', ['sale_volume_num'])
+    op.create_index('ix_meituan_coupon_commission_percent', 'meituan_coupon', ['commission_percent'])
+    op.create_index('ix_meituan_coupon_dedup_key', 'meituan_coupon', ['dedup_key'])
+    op.create_index('ix_meituan_coupon_last_seen', 'meituan_coupon', ['last_seen'])
+
+
+def downgrade() -> None:
+    op.drop_table('meituan_coupon')

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/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/repositories/stats.py

@@ -10,8 +10,11 @@ from datetime import datetime, timedelta, timezone
 from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
+from app.models.ad_feed_reward import AdFeedRewardRecord
+from app.models.ad_reward import AdRewardRecord
 from app.models.comparison import ComparisonRecord
 from app.models.feedback import Feedback
+from app.models.signin import SigninBoostRecord, SigninRecord
 from app.models.user import User
 from app.models.wallet import CoinTransaction, WithdrawOrder
 
@@ -71,6 +74,37 @@ def dashboard_overview(db: Session) -> dict:
         "coins": {
             # 累计发放金币(coin_transaction 里所有 amount>0 之和;负数是兑换/扣减不计)
             "granted_total": _sum(CoinTransaction.amount, CoinTransaction.amount > 0),
+            "reward_video_coin_total": _sum(
+                CoinTransaction.amount,
+                CoinTransaction.amount > 0,
+                CoinTransaction.biz_type.in_(("reward_video", "ad_reward")),
+            ),
+            "reward_video_watch_count": _count(
+                AdRewardRecord,
+                AdRewardRecord.reward_scene == "reward_video",
+                AdRewardRecord.status == "granted",
+            ),
+            "feed_ad_coin_total": _sum(
+                CoinTransaction.amount,
+                CoinTransaction.amount > 0,
+                CoinTransaction.biz_type == "feed_ad_reward",
+            ),
+            "feed_ad_watch_count": _count(
+                AdFeedRewardRecord,
+                AdFeedRewardRecord.status == "granted",
+            ),
+            "signin_coin_total": _sum(
+                CoinTransaction.amount,
+                CoinTransaction.amount > 0,
+                CoinTransaction.biz_type == "signin",
+            ),
+            "signin_count": _count(SigninRecord),
+            "signin_boost_coin_total": _sum(
+                CoinTransaction.amount,
+                CoinTransaction.amount > 0,
+                CoinTransaction.biz_type == "signin_boost",
+            ),
+            "signin_boost_watch_count": _count(SigninBoostRecord),
         },
         "cash": {
             "withdraw_success_cents": _sum(

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/dashboard.py

@@ -14,6 +14,14 @@ class DashboardUsers(BaseModel):
 
 
 class DashboardCoins(BaseModel):
+    reward_video_coin_total: int = 0
+    reward_video_watch_count: int = 0
+    feed_ad_coin_total: int = 0
+    feed_ad_watch_count: int = 0
+    signin_coin_total: int = 0
+    signin_count: int = 0
+    signin_boost_coin_total: int = 0
+    signin_boost_watch_count: int = 0
     granted_total: int
 
 

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/ad.py

@@ -10,6 +10,7 @@
 from __future__ import annotations
 
 import logging
+import json
 import uuid
 
 from fastapi import APIRouter, Depends, HTTPException, Request, status
@@ -23,6 +24,7 @@ from app.repositories import ad_ecpm as crud_ecpm
 from app.repositories import ad_feed_reward as crud_feed
 from app.repositories import ad_reward as crud_ad
 from app.repositories import ad_watch as crud_watch
+from app.repositories import signin as crud_signin
 from app.schemas.ad import (
     AdRewardStatusOut,
     EcpmReportIn,
@@ -30,6 +32,7 @@ from app.schemas.ad import (
     FeedRewardIn,
     FeedRewardOut,
     PangleCallbackOut,
+    TestGrantIn,
     TestGrantOut,
     WatchReportIn,
     WatchReportOut,
@@ -44,6 +47,23 @@ REASON_OK = 0
 REASON_BAD_PARAMS = 1  # 验签过但缺 trans_id / user_id 非数字
 REASON_UNKNOWN_USER = 2  # user_id 不存在(可能伪造)
 
+REWARD_SCENE_REWARD_VIDEO = "reward_video"
+REWARD_SCENE_SIGNIN_BOOST = "signin_boost"
+SUPPORTED_REWARD_SCENES = {REWARD_SCENE_REWARD_VIDEO, REWARD_SCENE_SIGNIN_BOOST}
+
+
+def _parse_extra(raw_extra: str | None) -> dict[str, str]:
+    """解析客户端 setMediaExtra 透传的 JSON;旧格式/异常返回空 dict。"""
+    if not raw_extra:
+        return {}
+    try:
+        data = json.loads(raw_extra)
+    except (TypeError, ValueError):
+        return {}
+    if not isinstance(data, dict):
+        return {}
+    return {str(k): str(v) for k, v in data.items() if v is not None}
+
 
 @router.get(
     "/pangle-callback",
@@ -52,9 +72,9 @@ REASON_UNKNOWN_USER = 2  # user_id 不存在(可能伪造)
     dependencies=[Depends(rate_limit(300, 60, "pangle-callback"))],
 )
 def pangle_callback(request: Request, db: DbSession) -> PangleCallbackOut:
-    """穿山甲 GroMore 在激励视频播完后回调,带 user_id / trans_id / reward_amount / sign 等 query 参数。
+    """穿山甲 GroMore 在激励视频播完后回调,带 user_id / trans_id / ecpm / sign 等 query 参数。
 
-    流程:开关/密钥就绪 → 验签(SHA256(m-key:trans_id)) → 解析 user_id/reward_amount → 幂等发金币。
+    流程:开关/密钥就绪 → 验签(SHA256(m-key:trans_id)) → 解析 user_id/场景/eCPM → 幂等发金币。
     验签失败 403(留给真请求重试);参数缺/坏或 user 不存在 → is_verify=false + reason(不发,不重试)。
     granted / capped → is_verify=true + reason=0。
     """
@@ -77,19 +97,86 @@ def pangle_callback(request: Request, db: DbSession) -> PangleCallbackOut:
         return PangleCallbackOut(is_verify=False, reason=REASON_BAD_PARAMS)
 
     user_id = int(raw_user_id)
-    coin = rewards.resolve_ad_reward_coin(db, params.get("reward_amount"))
     raw = "&".join(f"{k}={v}" for k, v in sorted(params.items()) if k != "sign")
-    try:
-        rec = crud_ad.grant_ad_reward(
-            db, user_id, trans_id, coin=coin,
-            reward_name=params.get("reward_name"), raw=raw[:1024],
+    extra = {}
+    for extra_key in ("extra", "gromoreExtra", "gromore_extra"):
+        extra.update(_parse_extra(params.get(extra_key)))
+    reward_scene = extra.get("reward_scene") or REWARD_SCENE_REWARD_VIDEO
+    ad_session_id = extra.get("ad_session_id")
+    ecpm = params.get("ecpm")
+
+    existing = crud_ad.find_by_trans(db, trans_id)
+    if existing is not None:
+        logger.info(
+            "pangle callback idempotent user_id=%d trans_id=%s status=%s scene=%s",
+            user_id, trans_id, existing.status, existing.reward_scene,
         )
+        return PangleCallbackOut(is_verify=True, reason=REASON_OK)
+
+    try:
+        if reward_scene not in SUPPORTED_REWARD_SCENES:
+            rec = crud_ad.record_external_reward(
+                db, user_id, trans_id, coin=0, reward_scene=reward_scene[:32],
+                ad_session_id=ad_session_id, ecpm=ecpm,
+                reward_name=params.get("reward_name"), raw=raw[:1024],
+                status="unknown_scene",
+            )
+            logger.warning(
+                "pangle callback unknown scene user_id=%d trans_id=%s scene=%s",
+                user_id, trans_id, reward_scene,
+            )
+            return PangleCallbackOut(is_verify=False, reason=REASON_BAD_PARAMS)
+        if reward_scene == REWARD_SCENE_SIGNIN_BOOST:
+            try:
+                boost, _balance = crud_signin.boost_today_signin(
+                    db, user_id, ad_ref_id=trans_id, commit=False
+                )
+            except crud_signin.NotSignedTodayError:
+                db.rollback()
+                rec = crud_ad.record_external_reward(
+                    db, user_id, trans_id, coin=0, reward_scene=reward_scene,
+                    ad_session_id=ad_session_id, ecpm=ecpm,
+                    reward_name=params.get("reward_name"), raw=raw[:1024],
+                    status="not_signed",
+                )
+            except crud_signin.AlreadyBoostedError:
+                db.rollback()
+                rec = crud_ad.record_external_reward(
+                    db, user_id, trans_id, coin=0, reward_scene=reward_scene,
+                    ad_session_id=ad_session_id, ecpm=ecpm,
+                    reward_name=params.get("reward_name"), raw=raw[:1024],
+                    status="already_boosted",
+                )
+            except crud_signin.LastCycleDayBoostBlockedError:
+                db.rollback()
+                rec = crud_ad.record_external_reward(
+                    db, user_id, trans_id, coin=0, reward_scene=reward_scene,
+                    ad_session_id=ad_session_id, ecpm=ecpm,
+                    reward_name=params.get("reward_name"), raw=raw[:1024],
+                    status="last_day",
+                )
+            else:
+                rec = crud_ad.record_external_reward(
+                    db, user_id, trans_id, coin=boost.coin_awarded,
+                    reward_scene=reward_scene, ad_session_id=ad_session_id, ecpm=ecpm,
+                    reward_name=params.get("reward_name"), raw=raw[:1024],
+                    commit=False,
+                )
+                db.commit()
+                db.refresh(rec)
+        else:
+            rec = crud_ad.grant_ad_reward(
+                db, user_id, trans_id, ecpm=ecpm, ad_session_id=ad_session_id,
+                reward_scene=REWARD_SCENE_REWARD_VIDEO,
+                reward_name=params.get("reward_name"), raw=raw[:1024],
+            )
     except crud_ad.UnknownUserError:
         logger.warning("pangle callback unknown user_id=%d trans_id=%s", user_id, trans_id)
         return PangleCallbackOut(is_verify=False, reason=REASON_UNKNOWN_USER)
 
     logger.info(
-        "ad reward user_id=%d trans_id=%s status=%s coin=%d", user_id, trans_id, rec.status, rec.coin
+        "ad reward user_id=%d trans_id=%s scene=%s status=%s coin=%d",
+        user_id, trans_id, rec.reward_scene, rec.status, rec.coin,
     )
     # granted / capped 均算"已处理":is_verify=true 不让穿山甲重试(capped 只是没加币)
     return PangleCallbackOut(is_verify=True, reason=REASON_OK)
@@ -153,11 +240,12 @@ def ecpm_report(payload: EcpmReportIn, user: CurrentUser, db: DbSession) -> Ecpm
     crud_ecpm.create_ecpm_record(
         db, user.id,
         ad_type=payload.ad_type, ecpm_raw=payload.ecpm,
+        ad_session_id=payload.ad_session_id,
         adn=payload.adn, slot_id=payload.slot_id,
     )
     logger.info(
-        "ad ecpm report user_id=%d type=%s ecpm=%s adn=%s slot=%s",
-        user.id, payload.ad_type, payload.ecpm, payload.adn, payload.slot_id,
+        "ad ecpm report user_id=%d type=%s session=%s ecpm=%s adn=%s slot=%s",
+        user.id, payload.ad_type, payload.ad_session_id, payload.ecpm, payload.adn, payload.slot_id,
     )
     return EcpmReportOut(ok=True)
 
@@ -168,7 +256,7 @@ def ecpm_report(payload: EcpmReportIn, user: CurrentUser, db: DbSession) -> Ecpm
     summary="[仅本地联调]模拟穿山甲回调发奖",
     dependencies=[Depends(rate_limit(60, 60, "ad-test-grant"))],
 )
-def test_grant(user: CurrentUser, db: DbSession) -> TestGrantOut:
+def test_grant(user: CurrentUser, db: DbSession, payload: TestGrantIn | None = None) -> TestGrantOut:
     """⚠️ 仅本地联调用:没部署公网、穿山甲 S2S 回调打不到本地时,客户端(debug 包)看完广告后
     调这个接口,直接走与回调相同的发奖逻辑(幂等 + 每日上限),验证"看广告→金币到账"全链路。
 
@@ -178,18 +266,58 @@ def test_grant(user: CurrentUser, db: DbSession) -> TestGrantOut:
     if not settings.AD_REWARD_TEST_GRANT_ENABLED:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="not found")
 
-    # 每次新 trans_id,模拟一次独立的穿山甲发奖回调(幂等键各不相同 → 每次都发,直到当日上限)
+    reward_scene = (payload.reward_scene if payload is not None else REWARD_SCENE_REWARD_VIDEO)
+    if reward_scene not in SUPPORTED_REWARD_SCENES:
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="bad reward_scene")
+
+    # 每次新 trans_id,模拟一次独立的穿山甲发奖回调(幂等键各不相同 → 每次都发,直到当日上限/今日膨胀一次)
     trans_id = f"test-{user.id}-{uuid.uuid4().hex}"
-    try:
-        rec = crud_ad.grant_ad_reward(
-            db, user.id, trans_id, reward_name="测试发奖", raw="client debug test-grant"
-        )
-    except crud_ad.UnknownUserError as e:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="user not found") from e
+    if reward_scene == REWARD_SCENE_SIGNIN_BOOST:
+        try:
+            boost, _balance = crud_signin.boost_today_signin(
+                db, user.id, ad_ref_id=trans_id, commit=False
+            )
+        except crud_signin.NotSignedTodayError:
+            db.rollback()
+            rec = crud_ad.record_external_reward(
+                db, user.id, trans_id, coin=0, reward_scene=reward_scene,
+                raw="client debug test-grant signin_boost", status="not_signed",
+            )
+        except crud_signin.AlreadyBoostedError:
+            db.rollback()
+            rec = crud_ad.record_external_reward(
+                db, user.id, trans_id, coin=0, reward_scene=reward_scene,
+                raw="client debug test-grant signin_boost", status="already_boosted",
+            )
+        except crud_signin.LastCycleDayBoostBlockedError:
+            db.rollback()
+            rec = crud_ad.record_external_reward(
+                db, user.id, trans_id, coin=0, reward_scene=reward_scene,
+                raw="client debug test-grant signin_boost", status="last_day",
+            )
+        else:
+            rec = crud_ad.record_external_reward(
+                db, user.id, trans_id, coin=boost.coin_awarded,
+                reward_scene=reward_scene, reward_name="测试签到膨胀",
+                raw="client debug test-grant signin_boost", commit=False,
+            )
+            db.commit()
+            db.refresh(rec)
+    else:
+        try:
+            rec = crud_ad.grant_ad_reward(
+                db, user.id, trans_id, ecpm="200", reward_name="测试发奖",
+                raw="client debug test-grant ecpm=200",
+            )
+        except crud_ad.UnknownUserError as e:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="user not found") from e
 
     (used, limit, coin_per, round_count, cooldown_until,
      _watched, _watch_limit) = crud_ad.today_status(db, user.id)
-    logger.info("ad TEST grant user_id=%d status=%s coin=%d", user.id, rec.status, rec.coin)
+    logger.info(
+        "ad TEST grant user_id=%d scene=%s status=%s coin=%d",
+        user.id, reward_scene, rec.status, rec.coin,
+    )
     return TestGrantOut(
         granted=(rec.status == "granted"),
         status=rec.status,
@@ -220,6 +348,7 @@ def feed_reward(payload: FeedRewardIn, user: CurrentUser, db: DbSession) -> Feed
         client_event_id=payload.client_event_id,
         ecpm=payload.ecpm,
         duration_seconds=payload.duration_seconds,
+        ad_session_id=payload.ad_session_id,
         adn=payload.adn,
         slot_id=payload.slot_id,
     )

app/api/v1/coupon.py

@@ -16,15 +16,66 @@ from typing import Any
 
 import httpx
 from fastapi import APIRouter, HTTPException, Request, status
+from fastapi.concurrency import run_in_threadpool
 
+from app.api.deps import DbSession
 from app.core.config import settings
 from app.core.pricebot_router import pick_pricebot
+from app.db.session import SessionLocal
+from app.repositories import coupon_state as coupon_repo
+from app.schemas.coupon_state import CouponPromptDismissIn, CouponPromptShouldShowOut
 
 logger = logging.getLogger("shagua.coupon")
 
 router = APIRouter(prefix="/api/v1/coupon", tags=["coupon"])
 
 
+def _to_int(v: object) -> int | None:
+    """user_id 协议是字符串(登录态才带),转 int 存库;缺失/非法 → None。"""
+    if v is None:
+        return None
+    try:
+        return int(v)  # type: ignore[arg-type]
+    except (TypeError, ValueError):
+        return None
+
+
+def _extract_coupon_results(resp_json: dict) -> list[dict]:
+    """从 pricebot 响应抽本帧券结果,并**按 coupon_id 去重**。
+
+    ⚠️ done/单券帧同时带 last_coupon_result(最后一张)+ action.params.coupon_results
+    (全量含最后一张)→ 最后一张出现两次。必须去重:同批里同 coupon_id 两次,
+    record_claims 在 autoflush=False 下两次 select 都查不到刚 add 的行 → 重复 add →
+    commit 撞唯一约束 IntegrityError 回滚整批(done/单券记录全丢)。
+    coupon_results(全量权威)覆盖 last_coupon_result。"""
+    by_id: dict[str, dict] = {}
+    last = resp_json.get("last_coupon_result")
+    if isinstance(last, dict) and last.get("coupon_id"):
+        by_id[last["coupon_id"]] = last
+    params = (resp_json.get("action") or {}).get("params") or {}
+    cr = params.get("coupon_results")
+    if isinstance(cr, list):
+        for x in cr:
+            if isinstance(x, dict) and x.get("coupon_id"):
+                by_id[x["coupon_id"]] = x
+    return list(by_id.values())
+
+
+def _mark_engagement_blocking(
+    device_id: str, user_id: int | None, engage_type: str
+) -> None:
+    """独立 session 写 engagement(async 端点经 run_in_threadpool 调,不阻塞事件循环)。"""
+    with SessionLocal() as db:
+        coupon_repo.mark_engagement(db, device_id, user_id, engage_type)
+
+
+def _record_claims_blocking(
+    device_id: str, user_id: int | None, trace_id: str | None, results: list[dict]
+) -> None:
+    with SessionLocal() as db:
+        coupon_repo.record_claims(db, device_id, user_id, trace_id, results)
+
+
 @router.post("/step", summary="领券任务步进 (透传到 pricebot)")
 async def coupon_step(
     request: Request,
@@ -45,6 +96,21 @@ async def coupon_step(
     if not isinstance(meta, dict):
         meta = {}
 
+    device_id = meta.get("device_id")
+    user_id = _to_int(meta.get("user_id"))  # 登录态才带;判断不靠它,资产留痕用
+    trace_id = meta.get("trace_id")
+
+    # 领券任务首帧(step=0)= 用户已发起领券 → 记一条今日 engagement(claim_started),
+    # 今天这台设备不再弹引导窗(对齐前台"点一键领取即 markEngaged")。写库失败绝不能
+    # 连累领券主流程,整段吞掉。
+    if device_id and meta.get("step") == 0:
+        try:
+            await run_in_threadpool(
+                _mark_engagement_blocking, device_id, user_id, "claim_started"
+            )
+        except Exception as e:  # noqa: BLE001
+            logger.warning("coupon engagement write failed: %s", e)
+
     # 按 trace_id 一致性 hash 选 pricebot 实例(同一领券任务的所有帧落同一进程)
     base = pick_pricebot(meta.get("trace_id"))
     url = f"{base.rstrip('/')}/api/coupon/step"
@@ -80,4 +146,44 @@ async def coupon_step(
             detail=f"pricebot upstream returned {resp.status_code}",
         )
 
-    return resp.json()
+    resp_json = resp.json()
+
+    # 领券结果沉淀(资产):中间帧 last_coupon_result + done 帧 coupon_results 幂等写库。
+    # 当前只记录、不参与"要不要领"判断(MVP 先不去重)。写库失败不影响返回。
+    if device_id:
+        results = _extract_coupon_results(resp_json)
+        if results:
+            try:
+                await run_in_threadpool(
+                    _record_claims_blocking, device_id, user_id, trace_id, results
+                )
+            except Exception as e:  # noqa: BLE001
+                logger.warning("coupon claim write failed: %s", e)
+
+    return resp_json
+
+
+@router.post("/prompt/dismiss", summary="用户拒绝/关闭领券引导窗(记今日已 engage)")
+def coupon_prompt_dismiss(payload: CouponPromptDismissIn, db: DbSession) -> dict[str, bool]:
+    """客户端点关闭引导窗时调用 → 记一条今日 engagement(dismissed),今天不再弹。
+
+    server 在透传链路里看不到"用户拒绝"(拒绝不发起领券),故必须客户端通知。
+    MVP 不鉴权,按 device_id 记。
+    """
+    coupon_repo.mark_engagement(db, payload.device_id, payload.user_id, "dismissed")
+    return {"ok": True}
+
+
+@router.get(
+    "/prompt/should-show",
+    response_model=CouponPromptShouldShowOut,
+    summary="切到外卖 App 时是否还应弹领券引导窗",
+)
+def coupon_prompt_should_show(
+    device_id: str, db: DbSession
+) -> CouponPromptShouldShowOut:
+    """今天这台设备已 engage(领或拒)过 → should_show=false。客户端据此决定弹不弹
+    (前台 SP 缓存做快速路径,这里是权威)。"""
+    return CouponPromptShouldShowOut(
+        should_show=not coupon_repo.has_engaged_today(db, device_id)
+    )

app/api/v1/invite.py

@@ -0,0 +1,64 @@
+"""好友邀请 endpoint。
+
+路由前缀 /api/v1/invite,需 Bearer 鉴权(用户级数据):
+  GET  /me     我的邀请码 + 分享链接 + 已邀人数/已得金币(邀请页展示)
+  POST /bind   把当前登录用户(被邀请人)绑定到某邀请码,注册即生效,双方各发金币
+
+客户端两种调用 /bind:
+  - 自动:首启读剪贴板拿到邀请码 → 登录后调本接口(channel=clipboard)
+  - 手动:用户在邀请页输入好友邀请码(channel=manual)
+
+绑定的真正逻辑(邀请码生成/反查、幂等、自邀屏蔽、发金币)在 repositories/invite.py。
+"""
+from __future__ import annotations
+
+import logging
+
+from fastapi import APIRouter
+
+from app.api.deps import CurrentUser, DbSession
+from app.core.config import settings
+from app.repositories import invite as invite_repo
+from app.schemas.invite import BindInviteIn, BindInviteOut, InviteInfoOut
+
+logger = logging.getLogger("shagua.invite")
+
+router = APIRouter(prefix="/api/v1/invite", tags=["invite"])
+
+_BIND_MESSAGES = {
+    "success": "邀请绑定成功,金币已到账",
+    "already_bound": "你已绑定过邀请人",
+    "invalid_code": "邀请码无效",
+    "self_invite": "不能填写自己的邀请码",
+    "not_eligible": "邀请仅对新注册用户生效",
+}
+
+
+@router.get("/me", response_model=InviteInfoOut, summary="我的邀请码 + 分享链接 + 战绩")
+def my_invite(user: CurrentUser, db: DbSession) -> InviteInfoOut:
+    code = invite_repo.ensure_code(db, user)
+    invited, coins = invite_repo.get_stats(db, user.id)
+    sep = "&" if "?" in settings.INVITE_LANDING_URL else "?"
+    share_url = f"{settings.INVITE_LANDING_URL}{sep}ref={code}"
+    return InviteInfoOut(
+        invite_code=code,
+        share_url=share_url,
+        invited_count=invited,
+        coins_earned=coins,
+    )
+
+
+@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 bind invitee=%d code=%s channel=%s -> %s",
+        user.id, req.invite_code, req.channel, result.status,
+    )
+    return BindInviteOut(
+        status=result.status,
+        coins_awarded=result.invitee_coin,
+        message=_BIND_MESSAGES.get(result.status, result.status),
+    )

app/api/v1/meituan.py

@@ -7,10 +7,14 @@ from __future__ import annotations
 import logging
 from concurrent.futures import ThreadPoolExecutor, as_completed
 
-from fastapi import APIRouter, HTTPException
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import select
+from sqlalchemy.orm import Session
 
 from app.core.config import settings
+from app.db.session import get_db
 from app.integrations.meituan import MeituanCpsError, get_referral_link, query_coupon
+from app.models.meituan_coupon import MeituanCoupon
 from app.schemas.meituan import (
     CouponCard,
     CouponListRequest,
@@ -19,6 +23,7 @@ from app.schemas.meituan import (
     FeedResponse,
     ReferralLinkRequest,
     ReferralLinkResponse,
+    TopSalesRequest,
 )
 
 logger = logging.getLogger("shagua.meituan")
@@ -79,13 +84,21 @@ def _interleave(waimai: list[dict], daodian: list[dict]) -> list[CouponCard]:
     return items
 
 
-@router.post("/feed", response_model=FeedResponse, summary="混合feed(外卖+到店交叉, 无限流)")
+def _commission_pct(card: CouponCard) -> float:
+    """'1.4%' → 1.4;解析失败按 0(会被智能推荐过滤掉)。"""
+    try:
+        return float(card.commission_rate.rstrip("%"))
+    except (ValueError, AttributeError):
+        return 0.0
+
+
+@router.post("/feed", response_model=FeedResponse, summary="混合feed(外卖+到店交叉);tab=rec智能推荐/distance距离最近")
 def feed(req: FeedRequest) -> FeedResponse:
     if not settings.mt_cps_configured:
         return FeedResponse(items=[], has_next=False, page=req.page)
-    page_idx = req.page - 1
     lon, lat = req.longitude, req.latitude
-    logger.info("[feed] page=%s lon=%.6f lat=%.6f", req.page, lon, lat)
+    tab = (req.tab or "").strip()
+    logger.info("[feed] tab=%s page=%s lon=%.6f lat=%.6f", tab or "(default)", req.page, lon, lat)
 
     def _fetch_topic(platform: int, biz_line: int | None, topic: int) -> list[dict]:
         try:
@@ -97,6 +110,28 @@ def feed(req: FeedRequest) -> FeedResponse:
         except MeituanCpsError:
             return []
 
+    # 距离最近:拉齐全部榜单轮次,合并去重,后端在【完整池子】上全局按距离由近及远排,一次性返回。
+    # (距离排序必须在完整池上做、不能逐页排——这正是之前放前端不合理的根因。)
+    if tab == "distance":
+        with ThreadPoolExecutor(max_workers=len(_TOPIC_ROUNDS) * 2) as pool:
+            futs = []
+            for wm_topic, dd_topic in _TOPIC_ROUNDS:
+                futs.append(pool.submit(_fetch_topic, 1, None, wm_topic))
+                futs.append(pool.submit(_fetch_topic, 2, 1, dd_topic))
+            raws = [f.result() for f in futs]
+        seen: set[str] = set()
+        cards: list[CouponCard] = []
+        for raw_list in raws:
+            for it in raw_list:
+                card = CouponCard.from_raw(it)
+                if card.product_view_sign and card.product_view_sign not in seen:
+                    seen.add(card.product_view_sign)
+                    cards.append(card)
+        cards.sort(key=lambda c: c.distance_meters if c.distance_meters is not None else float("inf"))
+        return FeedResponse(items=cards, has_next=False, page=1)
+
+    # 智能推荐(rec,默认):沿用逐轮分页的混合 feed,后端过滤掉佣金率 < 3%。
+    page_idx = req.page - 1
     if page_idx >= len(_TOPIC_ROUNDS):
         return FeedResponse(items=[], has_next=False, page=req.page)
 
@@ -107,6 +142,8 @@ def feed(req: FeedRequest) -> FeedResponse:
         waimai, daodian = f_wm.result(), f_dd.result()
 
     items = _interleave(waimai, daodian)
+    if tab == "rec":
+        items = [c for c in items if _commission_pct(c) >= 3.0]
     has_next = page_idx + 1 < len(_TOPIC_ROUNDS)
     return FeedResponse(items=items, has_next=has_next, page=req.page)
 
@@ -130,3 +167,37 @@ def referral_link(req: ReferralLinkRequest) -> ReferralLinkResponse:
     link_map = raw.get("referralLinkMap") or {}
     link = raw.get("data") or link_map.get("1") or link_map.get("3") or next(iter(link_map.values()), "")
     return ReferralLinkResponse(link=link, link_map=link_map)
+
+
+@router.post("/top-sales", response_model=CouponListResponse,
+             summary="销量最高(从离线库 meituan_coupon 按销量降序 + 跨源去重,不实时打美团)")
+def top_sales(req: TopSalesRequest, db: Session = Depends(get_db)) -> CouponListResponse:
+    # 只取有销量档的(美团很多券没销量,排不了);按销量降序、同销量再按佣金降序
+    stmt = select(MeituanCoupon).where(MeituanCoupon.sale_volume_num.isnot(None))
+    if req.platform is not None:
+        stmt = stmt.where(MeituanCoupon.platform == req.platform)
+    stmt = stmt.order_by(
+        MeituanCoupon.sale_volume_num.desc(),
+        MeituanCoupon.commission_percent.desc(),
+    )
+    rows = db.execute(stmt).scalars().all()
+
+    # 跨源去重(dedup_key = 品牌|名|价):按销量降序遍历,每个 dedup_key 只留第一条(=销量最高那条)。
+    # 用 raw(整条原始返回)重建 CouponCard,字段与实时接口完全一致,前端无需改渲染。
+    seen: set[str] = set()
+    cards: list[CouponCard] = []
+    for row in rows:
+        if row.dedup_key in seen:
+            continue
+        seen.add(row.dedup_key)
+        try:
+            card = CouponCard.from_raw(row.raw or {})
+        except Exception:  # noqa: BLE001
+            continue
+        if card.product_view_sign:
+            cards.append(card)
+
+    start = (req.page - 1) * req.page_size
+    page_items = cards[start:start + req.page_size]
+    has_next = start + req.page_size < len(cards)
+    return CouponListResponse(items=page_items, has_next=has_next, search_id=None)

app/api/v1/signin.py

@@ -12,7 +12,9 @@ import logging
 from fastapi import APIRouter, HTTPException, status
 
 from app.api.deps import CurrentUser, DbSession
+from app.repositories import ad_reward as crud_ad
 from app.repositories import signin as crud_signin
+from app.repositories import wallet as crud_wallet
 from app.schemas.welfare import (
     SigninBoostRequest,
     SigninBoostResultOut,
@@ -54,14 +56,21 @@ def do_signin(user: CurrentUser, db: DbSession) -> SigninResultOut:
 def boost_signin(
     payload: SigninBoostRequest, user: CurrentUser, db: DbSession
 ) -> SigninBoostResultOut:
-    try:
-        record, balance = crud_signin.boost_today_signin(
-            db, user.id, ad_ref_id=payload.ad_ref_id
-        )
-    except crud_signin.NotSignedTodayError as e:
-        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="not signed today") from e
-    except crud_signin.AlreadyBoostedError as e:
-        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="already boosted today") from e
+    if not payload.ad_ref_id:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="ad reward required")
+    ad_rec = crud_ad.find_by_trans(db, payload.ad_ref_id)
+    if (
+        ad_rec is None
+        or ad_rec.user_id != user.id
+        or ad_rec.reward_scene != "signin_boost"
+        or ad_rec.status != "granted"
+    ):
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="ad reward not verified")
+    record = crud_signin.boost_by_ad_ref(db, user.id, payload.ad_ref_id)
+    if record is None:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="signin boost not granted")
+    acc = crud_wallet.get_or_create_account(db, user.id)
+    balance = acc.coin_balance
 
     logger.info(
         "signin boost ok user_id=%d date=%s coin=%d",

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

@@ -81,6 +81,9 @@ class Settings(BaseSettings):
     MT_CPS_HOST: str = "https://media.meituan.com"
     MT_CPS_TIMEOUT_SEC: int = 15
     MT_CPS_DEFAULT_SID: str = "sgbjia"
+    # 美团调用走的代理。本机开发直连美团会 SSL EOF,需填 http://127.0.0.1:7897;
+    # 线上国内服务器留空(=直连)。见 .env.example 与 integrations/meituan.py。
+    MT_CPS_PROXY: str = ""
 
     @property
     def mt_cps_configured(self) -> bool:
@@ -99,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 查询兜底),但微信要求该字段非空,故启用免确认前必须配置;留空时免确认相关接口返回未配置。
@@ -170,6 +176,11 @@ class Settings(BaseSettings):
     MEDIA_URL_PREFIX: str = "/media"
     AVATAR_MAX_BYTES: int = 5 * 1024 * 1024  # 头像最大 5MB
 
+    # ===== 邀请好友 =====
+    # 分享落地页(二维码 / 分享链接指向这里;扫码 → 落地页 → 引导浏览器下载 APK)。
+    # MVP 落地页就放 app-server 的 /media 静态目录下(dl.html)。
+    INVITE_LANDING_URL: str = "https://app-api.shaguabijia.com/media/dl.html"
+
     # ===== CORS =====
     CORS_ALLOW_ORIGINS: str = ""
 

app/core/config_schema.py

@@ -42,7 +42,7 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
     "ad_reward_coin": {
         "default": r.AD_REWARD_COIN, "label": "看广告单次金币",
         "group": "看广告", "type": "int",
-        "help": "看完一个激励视频发的金币;须与穿山甲后台代码位'奖励数量'保持一致。",
+        "help": "历史兼容/测试展示值;正式发放按 eCPM 公式计算。",
     },
     "ad_daily_limit": {
         "default": r.DAILY_AD_REWARD_LIMIT, "label": "看广告每日上限(次)",
@@ -50,7 +50,7 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
     },
     "ad_max_coin": {
         "default": r.MAX_AD_REWARD_COIN, "label": "看广告单次金币上限",
-        "group": "看广告", "type": "int", "help": "夹紧穿山甲回调里异常的 reward_amount,防刷爆。",
+        "group": "看广告", "type": "int", "help": "历史 reward_amount 口径保留项;正式发放按 eCPM 公式计算。",
     },
     "ad_round_count": {
         "default": r.VIDEO_ROUND_REQUIRED_COUNT, "label": "每轮看广告次数",
@@ -60,4 +60,9 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
         "default": r.VIDEO_ROUND_COOLDOWN_SECONDS, "label": "广告关闭后冷却(秒)",
         "group": "看广告", "type": "int", "help": "点击退出广告后,下次点击观看前的冷却时间,默认 3 秒。",
     },
+    "signin_boost_coin": {
+        "default": r.SIGNIN_BOOST_COIN, "label": "签到膨胀固定金币",
+        "group": "签到", "type": "int",
+        "help": "Day1-Day13 签到后看完激励视频额外发放的固定金币;Day14 不展示也不允许膨胀。",
+    },
 }

app/core/rewards.py

@@ -77,6 +77,15 @@ def record_milestone_reward(milestone: int) -> int:
     return RECORD_MILESTONES[milestone - 1]
 
 
+# ===== 邀请好友(注册即生效,邀请人 + 被邀请人各发金币)=====
+# 10000 金币 = 1 元,双方各得 1 元。MVP 先用固定常量(不走 app_config)。
+INVITE_INVITER_COINS: int = 10000
+INVITE_INVITEE_COINS: int = 10000
+# "新用户闸":被邀请人必须在注册后此窗口内绑定才发奖(挡存量老用户互相填码薅羊毛)。
+# 自动绑(剪贴板)在首次注册登录后几秒内发生;留 72h 给手动填码兜底。
+INVITE_NEW_USER_WINDOW_HOURS: int = 72
+
+
 # ===== 看激励视频 / 信息流广告发金币 =====
 # 金币数值体系约定:eCPM 单位按"元/千次展示"处理,单次收入 = eCPM / 1000 元。
 AD_ECPM_FACTOR_TABLE: tuple[tuple[float, int, int | None], ...] = (
@@ -135,14 +144,14 @@ def calculate_ad_reward_coin(ecpm: str | int | float | None, today_count_after_t
     return max(0, round(yuan * COIN_PER_YUAN))
 
 
+SIGNIN_BOOST_COIN: int = 2000
+
+
 # ===== 看激励视频发金币(穿山甲 S2S 服务端回调发奖)=====
-# 看完一个激励视频发的金币(666 金币 ≈¥0.0666,汇率 10000 金币=1 元)。
-# 作用:① 回调缺/坏 reward_amount 时的回退值;② 客户端进度接口展示的"单次预告金币";
-#       ③ test-grant 本地联调的发奖额。
-# 真实发放以穿山甲回调带回的 reward_amount 为准(见 resolve_ad_reward_coin),后台应把
-# 代码位"奖励数量"配成与本值一致(=666),保证"广告内展示 / 进度预告 / 实际到账"三者一致。
+# 历史固定金币口径保留项。正式激励视频实发按 calculate_ad_reward_coin(eCPM, 当日第 N 次)
+# 计算;该值只用于旧接口兼容、配置页展示和本地联调兜底。
 AD_REWARD_COIN: int = 666
-# 单次发奖金币上限:夹紧穿山甲回调里异常的 reward_amount(如后台多打一个 0),防刷爆余额。
+# 历史 reward_amount 口径保留项。正式激励视频实发按 eCPM 公式计算。
 MAX_AD_REWARD_COIN: int = 1000
 # 每用户每日发奖次数上限。产品口径:一天最多看 500 次广告。
 DAILY_AD_REWARD_LIMIT: int = 500
@@ -160,7 +169,7 @@ VIDEO_ROUND_COOLDOWN_SECONDS: int = 3
 
 
 def resolve_ad_reward_coin(db, reward_amount: str | int | None) -> int:  # noqa: ANN001
-    """把穿山甲回调的 reward_amount(后台代码位配的"奖励数量")解析成本次发放金币。
+    """历史 reward_amount 口径解析函数,保留给旧脚本/旧配置兼容。
 
     缺失 / 非数字 / ≤0 → 回退配置单次金币;超过配置单次上限 → 夹紧(均从 app_config 读)。
     注:reward_amount 不参与穿山甲 sign(只签 trans_id),但伪造回调需先伪造合法 sign
@@ -227,3 +236,7 @@ def get_ad_round_count(db) -> int:  # noqa: ANN001
 
 def get_ad_cooldown_sec(db) -> int:  # noqa: ANN001
     return int(_cfg(db, "ad_cooldown_sec"))
+
+
+def get_signin_boost_coin(db) -> int:  # noqa: ANN001
+    return int(_cfg(db, "signin_boost_coin"))

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/meituan.py

@@ -62,8 +62,12 @@ def _call(path: str, body_obj: dict[str, Any]) -> dict[str, Any]:
     }
 
     url = f"{settings.MT_CPS_HOST}{path}"
+    # 美团调用走 MT_CPS_PROXY(本机开发直连会 SSL EOF,必须走代理;线上留空=直连)。
+    # trust_env=False:不读进程环境的 HTTP_PROXY,只认配置,避免被错误/失效代理误导。
+    proxy = settings.MT_CPS_PROXY or None
     try:
-        resp = httpx.post(url, content=body, headers=headers, timeout=settings.MT_CPS_TIMEOUT_SEC)
+        with httpx.Client(proxy=proxy, trust_env=False, timeout=settings.MT_CPS_TIMEOUT_SEC) as client:
+            resp = client.post(url, content=body, headers=headers)
     except httpx.HTTPError as e:
         logger.exception("[MT] http error calling %s", url)
         raise MeituanCpsError(f"meituan http error: {e}") from e

app/integrations/pangle.py

@@ -8,9 +8,9 @@
 我们客户端是 useMediation(true) 融合,回调走 GroMore 广告位层级,**不是**联盟代码位层级):
 **sign = SHA256("{m-key}:{trans_id}")** 的十六进制串。注意:
 - 是**普通 SHA256**,不是 HMAC,也不是 RSA;
-- **只对 `m-key:trans_id` 这一个字符串**签名,user_id / reward_amount 等其余参数**不参与**签名
+- **只对 `m-key:trans_id` 这一个字符串**签名,user_id / ecpm / extra 等其余参数**不参与**签名
   (它们的可信度由"能算出正确 sign = 知道 m-key"间接保证——伪造者没有 m-key 就连合法 sign
-  都造不出,自然也无法注入假 user_id/reward_amount)。
+  都造不出,自然也无法注入假 user_id/ecpm/extra)。
 m-key(安全密钥)在穿山甲后台「GroMore 聚合管理 → 搜广告位ID → 编辑」处获取,配到
 `settings.PANGLE_REWARD_SECRET`。(联盟代码位层级用的是另一套 Security Key + isValid 响应,我们不走那条。)
 """

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

@@ -22,6 +22,7 @@ from app.api.v1.compare_record import router as compare_record_router
 from app.api.v1.coupon import router as coupon_router
 from app.api.internal.price import router as internal_price_router
 from app.api.v1.feedback import router as feedback_router
+from app.api.v1.invite import router as invite_router
 from app.api.v1.meituan import router as meituan_router
 from app.api.v1.order import router as order_router
 from app.api.v1.platform import router as platform_router
@@ -34,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")
@@ -49,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(
@@ -79,6 +88,7 @@ def health() -> dict[str, str]:
 app.include_router(auth_router)
 app.include_router(user_router)
 app.include_router(feedback_router)
+app.include_router(invite_router)
 app.include_router(coupon_router)
 app.include_router(compare_router)
 app.include_router(compare_record_router)

app/models/__init__.py

@@ -7,7 +7,13 @@ from app.models.admin import AdminAuditLog, AdminUser  # noqa: F401
 from app.models.app_config import AppConfig  # noqa: F401
 from app.models.comparison import ComparisonRecord  # noqa: F401
 from app.models.comparison_milestone import ComparisonMilestoneClaim  # noqa: F401
+from app.models.coupon_state import (  # noqa: F401
+    CouponClaimRecord,
+    CouponPromptEngagement,
+)
 from app.models.feedback import Feedback  # noqa: F401
+from app.models.invite import InviteRelation  # 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
 from app.models.price_observation import PriceObservation  # noqa: F401

app/models/ad_ecpm.py

@@ -1,11 +1,8 @@
 """广告展示 eCPM 上报记录(内部收益统计/对账)。
 
-每条 = 客户端一次广告展示(`onAdShow`)后读到的 eCPM 信息。和发奖记录
-[ad_reward.AdRewardRecord] 是**两条独立的数据流**:
-- 发奖走穿山甲 S2S 回调(后端 → 有 trans_id、无 ecpm);
-- eCPM 走客户端上报(客户端 → 有 ecpm、无 trans_id)。
-两者没有公共键,无法逐条一一对应,所以本表用于**按用户/按天聚合收益**口径的对账,
-不做"这条发奖 = 这条 ecpm"的精确关联。穿山甲后台报表才是结算权威,本表是细粒度补充。
+每条 = 客户端一次广告展示(`onAdShow`)后读到的 eCPM 信息。`ad_session_id`
+由客户端生成,并通过激励视频 extra 透传给 S2S 回调,用于把"展示 eCPM"与
+"奖励完成"绑定。没有 session id 的旧上报仍可作为按天对账补充。
 
 ⚠️ `ecpm_raw` 原样存客户端上报的字符串——eCPM 单位(分 / 元)截至 2026-05-31 尚未最终确认,
 确认后再加一列解析好的数值;在此之前对账按"待定单位"处理。
@@ -14,7 +11,7 @@ from __future__ import annotations
 
 from datetime import datetime
 
-from sqlalchemy import DateTime, ForeignKey, Integer, String, func
+from sqlalchemy import DateTime, ForeignKey, Integer, String, UniqueConstraint, func
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.db.base import Base
@@ -22,6 +19,9 @@ from app.db.base import Base
 
 class AdEcpmRecord(Base):
     __tablename__ = "ad_ecpm_record"
+    __table_args__ = (
+        UniqueConstraint("ad_session_id", name="uq_ad_ecpm_record_session"),
+    )
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
     user_id: Mapped[int] = mapped_column(
@@ -29,11 +29,13 @@ class AdEcpmRecord(Base):
     )
     # 广告类型:reward_video(激励视频) / draw(Draw 信息流) 等;不强行统一代码位,各类型各自上报
     ad_type: Mapped[str] = mapped_column(String(32), nullable=False)
+    # 客户端生成的一次广告会话 id;激励视频 S2S 回调 extra 会透传同值
+    ad_session_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
     # 实际投放的 ADN(穿山甲 getShowEcpm().getSdkName(),如 pangle / gdt)
     adn: Mapped[str | None] = mapped_column(String(32), nullable=True)
     # 实际展示用的代码位(底层 mediation rit,非客户端配置位)
     slot_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
-    # 客户端上报的 eCPM 原始字符串(单位待确认,原样存)
+    # 客户端上报的 eCPM 原始字符串(单位:元/千次展示,原样存)
     ecpm_raw: Mapped[str] = mapped_column(String(32), nullable=False)
     # 北京时间日期串 'YYYY-MM-DD',按它等值做"按天聚合"(不在 SQL 里做跨时区 date 比较)
     report_date: Mapped[str] = mapped_column(String(10), index=True, nullable=False)

app/models/ad_feed_reward.py

@@ -20,6 +20,7 @@ class AdFeedRewardRecord(Base):
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
     client_event_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    ad_session_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
     user_id: Mapped[int] = mapped_column(
         Integer, ForeignKey("user.id"), index=True, nullable=False
     )

app/models/ad_reward.py

@@ -1,8 +1,8 @@
 """看激励视频发奖记录(穿山甲 S2S 回调)。
 
-每条 = 穿山甲一次发奖回调。`trans_id`(穿山甲交易号)唯一,做幂等键:穿山甲会重试回调,
-同号只发一次金币。`reward_date`(北京时间日期串)给"每日上限"计数用——按日期串等值查,
-不在 SQL 里做跨时区 date 比较(SQLite 上不可靠)。审计 / 对账时整张表可逐笔回溯。
+每条 = 穿山甲/GroMore 一次服务端激励回调。`trans_id`(交易号)唯一,做幂等键:
+穿山甲会重试回调,同号只处理一次。`reward_scene` 区分福利页激励视频、签到膨胀等
+不同奖励场景,避免统计和每日上限互相污染。
 """
 from __future__ import annotations
 
@@ -25,8 +25,14 @@ class AdRewardRecord(Base):
     )
     # 实发金币(capped 时为 0)
     coin: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
-    # granted(已发) / capped(当日超限未发)
+    # granted(已发) / capped(当日超限未发) / ecpm_missing(缺 eCPM 未发)
     status: Mapped[str] = mapped_column(String(16), nullable=False, default="granted")
+    # reward_video(福利页看视频) / signin_boost(签到膨胀)
+    reward_scene: Mapped[str] = mapped_column(String(32), nullable=False, default="reward_video")
+    # 客户端生成并通过 extra 透传的广告会话 id
+    ad_session_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
+    # 本次发奖采用的 eCPM 原始值(回调自带或按 ad_session_id 匹配的客户端上报)
+    ecpm_raw: Mapped[str | None] = mapped_column(String(32), nullable=True)
     # 北京时间日期串 'YYYY-MM-DD',按它等值统计当日发奖次数
     reward_date: Mapped[str] = mapped_column(String(10), index=True, nullable=False)
     # 穿山甲上报的奖励名(参考,不作发奖依据)
@@ -39,4 +45,7 @@ class AdRewardRecord(Base):
     )
 
     def __repr__(self) -> str:  # pragma: no cover
-        return f"<AdRewardRecord trans_id={self.trans_id} user_id={self.user_id} {self.status} coin={self.coin}>"
+        return (
+            f"<AdRewardRecord trans_id={self.trans_id} user_id={self.user_id} "
+            f"scene={self.reward_scene} {self.status} coin={self.coin}>"
+        )

app/models/coupon_state.py

@@ -0,0 +1,130 @@
+"""领券今日状态(弹窗频控 + 领券记录)两张表。
+
+- `coupon_prompt_engagement`:按 (device, 自然日) 记"今天是否对领券引导窗表达过意向"
+  ——点「一键领取」(claim_started) 或 点拒绝/关闭 (dismissed) 都算。切到外卖 App 时
+  据此决定弹不弹:今天 engage 过就不再弹。判断维度是 **device_id**——券发到的是设备上
+  登录的那个外卖账号,device 比 user 更贴近"哪个登录环境",且 device_id 全链路现成、
+  不依赖领券鉴权。
+
+- `coupon_claim_record`:按 (device, 券, 自然日) 记每张券的领取结果,纯沉淀(资产/画像/
+  排查)。当前**不参与**"要不要领"的判断(MVP 先不去重:今天 engage 过就不弹,A 路径
+  主动领则全跑)。留作以后做按券去重 / CPS 归因 / 用户画像的数据源。
+
+口径:
+- 日期 = Asia/Shanghai 的自然日(claim_date / engage_date)。
+- user_id 领券登录态有就记(资产),可空、不进唯一键、不阻塞判断。
+- device_id 客户端生成存 SP,卸载重装会变 → 重装当新设备重新弹一次(产品预期)。
+"""
+from __future__ import annotations
+
+from datetime import date, datetime
+
+from sqlalchemy import (
+    JSON,
+    Date,
+    DateTime,
+    Index,
+    Integer,
+    String,
+    UniqueConstraint,
+    func,
+)
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base
+
+# PG 上用 JSONB,SQLite(本地/测试)退化为通用 JSON(同 price_observation / comparison)。
+_JSON = JSON().with_variant(JSONB(), "postgresql")
+
+
+class CouponClaimRecord(Base):
+    """单张券一天一条领取记录(资产层,当前不做去重判断)。"""
+
+    __tablename__ = "coupon_claim_record"
+    __table_args__ = (
+        # 同设备、同券、同一天只一条:领券每帧 last_coupon_result + done 帧 coupon_results
+        # 会重复上报同一张券,靠它幂等 upsert。
+        UniqueConstraint(
+            "device_id", "coupon_id", "claim_date",
+            name="uq_coupon_claim_device_coupon_date",
+        ),
+        # 去重/统计查询按 (device, 日) 取一天所有券。
+        Index("ix_coupon_claim_device_date", "device_id", "claim_date"),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+
+    # 判断/聚合维度。client getOrCreateDeviceId 生成,重装会变。
+    device_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    # 登录态有就记(资产/画像),可空、不进唯一键。
+    user_id: Mapped[int | None] = mapped_column(Integer, index=True, nullable=True)
+
+    coupon_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    # Asia/Shanghai 自然日。每日可领的券(签到/天天红包)靠这天然每天一条。
+    claim_date: Mapped[date] = mapped_column(Date, nullable=False)
+
+    # success / already_claimed / failed / skipped(原样取 pricebot coupon 结果)
+    status: Mapped[str] = mapped_column(String(24), nullable=False)
+    vendor: Mapped[str | None] = mapped_column(String(48), nullable=True)
+    coupon_name: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    # 这张领到几张(pricebot display_count;给不出时为 None)
+    claimed_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    # 哪次任务领的,回指 pricebot work_logs / 排查
+    trace_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
+    # failed / skipped 原因
+    reason: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    # 杂项兜底:券的结构化信息(面额/入口/关键节点摘要等),免得加字段就迁移。
+    # ⚠️ 别塞原始无障碍树(几十 KB → 行膨胀);原始大树看 trace_id 指过去的 work_logs。
+    extra: Mapped[dict | None] = mapped_column(_JSON, nullable=True)
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), onupdate=func.now(),
+        nullable=False,
+    )
+
+    def __repr__(self) -> str:  # pragma: no cover
+        return (
+            f"<CouponClaimRecord device={self.device_id} coupon={self.coupon_id} "
+            f"date={self.claim_date} status={self.status}>"
+        )
+
+
+class CouponPromptEngagement(Base):
+    """按 (device, 自然日) 记"今天是否对领券引导窗表达过意向"——弹窗频控源。"""
+
+    __tablename__ = "coupon_prompt_engagement"
+    __table_args__ = (
+        # 一台设备一天一条:今天 engage 过(领或拒)就不再弹。
+        UniqueConstraint(
+            "device_id", "engage_date",
+            name="uq_coupon_engage_device_date",
+        ),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+
+    device_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    user_id: Mapped[int | None] = mapped_column(Integer, index=True, nullable=True)
+    # Asia/Shanghai 自然日。
+    engage_date: Mapped[date] = mapped_column(Date, nullable=False)
+    # claim_started(点了一键领取)/ dismissed(点了拒绝/关闭)。仅记录区分,
+    # 判断只看"今天有没有这条",type 不影响弹不弹。
+    engage_type: Mapped[str] = mapped_column(String(16), nullable=False)
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), onupdate=func.now(),
+        nullable=False,
+    )
+
+    def __repr__(self) -> str:  # pragma: no cover
+        return (
+            f"<CouponPromptEngagement device={self.device_id} "
+            f"date={self.engage_date} type={self.engage_type}>"
+        )

app/models/invite.py

@@ -0,0 +1,46 @@
+"""好友邀请关系表(注册即生效)。
+
+一行 = 一次成功的邀请绑定。`invitee_user_id` 唯一 = 一个被邀请人只能被归因一次
+(天然幂等键,防重复发奖,仿 ad_reward.trans_id 思路)。`channel` 记归因来源
+(clipboard 自动 / manual 手动填),便于后续分析"剪贴板自动捕获率"。
+
+邀请码本身是 `user.invite_code`(一人一稳定码,懒生成),不在这张表里。
+"""
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import DateTime, ForeignKey, Integer, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base
+
+
+class InviteRelation(Base):
+    __tablename__ = "invite_relation"
+
+    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
+    )
+    # 被邀请人唯一:一个人只能被邀一次(幂等键,防重复发奖)
+    invitee_user_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("user.id"), unique=True, index=True, nullable=False
+    )
+    # 归因来源:clipboard(剪贴板自动) / manual(手动填码)
+    channel: Mapped[str] = mapped_column(String(16), nullable=False, default="clipboard")
+    # 状态:effective(注册即生效)。预留:以后改"完成首单才生效"时用 pending/effective
+    status: Mapped[str] = mapped_column(String(16), nullable=False, default="effective")
+    # 本次给邀请人 / 被邀请人各发的金币(记账留痕,便于对账)
+    inviter_coin: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    invitee_coin: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+
+    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"<InviteRelation inviter={self.inviter_user_id} "
+            f"invitee={self.invitee_user_id} {self.status}>"
+        )

app/models/meituan_coupon.py

@@ -0,0 +1,93 @@
+"""美团 CPS 券本地缓存表(meituan_coupon)。
+
+把美团联盟 CPS 的券定时抓进本地库,供「销量 / 佣金排序」等查询**直接从库里捞、本地排序**,
+不再每次实时打美团接口——美团搜索/供给对销量排序支持差(实测乱序)、且有 402 限流和召回上限。
+
+数据来源(source,北京试点):
+  - search_waimai : 到家/外卖, 搜索关键词「外卖」, 翻到尽头
+  - search_meishi : 到家/外卖, 搜索关键词「美食」, 翻到尽头
+  - store_supply  : 到店, 多业务线供给(到餐+到综+酒店+门票), 翻到尽头
+
+每 5min(本地测试)/ 1h(线上)全量抓一次,按 (source, product_view_sign) upsert 存最新态,
+last_seen 每轮刷新(可据此清理长期未再出现的陈旧券)。
+
+去重说明:`product_view_sign` / `sku_view_id` 都是「按召回渠道生成」的,跨渠道(搜索 vs 供给)
+会变,**不能当商品全局唯一 id**。因此:
+  - 存储:按 (source, product_view_sign) upsert(同源短周期内 sign 稳定)。
+  - 查询:用 `dedup_key = md5(brand|name|price)` 跨源去重,再按 `sale_volume_num` / `commission_percent` 排序。
+
+价格统一存「分」(cents),与 price_report / comparison_record 一致。`raw` 保留整条原始返回
+(字段越详细越好,避免后续漏字段还要重抓)。
+"""
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import JSON, DateTime, Float, Integer, String, UniqueConstraint, func
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base
+
+
+class MeituanCoupon(Base):
+    __tablename__ = "meituan_coupon"
+    __table_args__ = (
+        UniqueConstraint("source", "product_view_sign", name="uq_meituan_coupon_source_sign"),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+
+    # ===== 来源 / 分类 =====
+    # source: search_waimai | search_meishi | store_supply
+    source: Mapped[str] = mapped_column(String(16), index=True, nullable=False)
+    platform: Mapped[int] = mapped_column(Integer, nullable=False)              # 1 到家/外卖, 2 到店
+    biz_line: Mapped[int | None] = mapped_column(Integer, nullable=True)        # 到店: 1到餐 2到综 3酒店 4门票
+    city_id: Mapped[str] = mapped_column(String(32), index=True, nullable=False)
+
+    # ===== 召回 id(按渠道生成,跨渠道会变;product_view_sign 用于换推广链) =====
+    product_view_sign: Mapped[str] = mapped_column(String(128), nullable=False)
+    sku_view_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
+
+    # ===== 商品本体 =====
+    name: Mapped[str | None] = mapped_column(String(256), nullable=True)
+    brand_name: Mapped[str | None] = mapped_column(String(128), index=True, nullable=True)
+    sell_price_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    original_price_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    head_url: Mapped[str | None] = mapped_column(String(512), nullable=True)
+
+    # ===== 销量(美团只给粗档位:热销1w+;num=排序用的下界数值,如 1w+ → 10000) =====
+    sale_volume: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    sale_volume_num: Mapped[int | None] = mapped_column(Integer, index=True, nullable=True)
+
+    # ===== 佣金(percent 如 1.4 表示 1.4%;amount 为分) =====
+    commission_percent: Mapped[float | None] = mapped_column(Float, index=True, nullable=True)
+    commission_amount_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
+
+    # ===== 门店 / 距离 =====
+    poi_name: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    available_poi_num: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    delivery_distance_m: Mapped[float | None] = mapped_column(Float, nullable=True)
+
+    # ===== 跨源去重键 + 原始返回 =====
+    dedup_key: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
+    raw: Mapped[dict] = mapped_column(
+        JSON().with_variant(JSONB(), "postgresql"), nullable=False, default=dict
+    )
+
+    # ===== 抓取时间窗 =====
+    first_seen: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
+    last_seen: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False
+    )
+
+    def __repr__(self) -> str:  # pragma: no cover
+        return (
+            f"<MeituanCoupon id={self.id} source={self.source} "
+            f"name={self.name!r} sale={self.sale_volume} comm={self.commission_percent}>"
+        )

app/models/user.py

@@ -33,6 +33,12 @@ class User(Base):
     nickname: Mapped[str | None] = mapped_column(String(64), nullable=True)
     avatar_url: Mapped[str | None] = mapped_column(String(512), nullable=True)
 
+    # 邀请码:每个用户一个稳定短码(懒生成,见 repositories/invite.ensure_code),
+    # 分享链接 / 二维码里带它。unique 允许多个 NULL(未生成的用户)。
+    invite_code: Mapped[str | None] = mapped_column(
+        String(16), unique=True, index=True, nullable=True
+    )
+
     # 微信开放平台 openid(微信登录绑定后存,提现转账到零钱用)。同一 appid 下唯一;
     # unique 索引保证一个微信只能绑一个账号(nullable,允许多个 NULL=未绑定)。
     wechat_openid: Mapped[str | None] = mapped_column(

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/ad_ecpm.py

@@ -7,6 +7,7 @@ user 存在(JWT),故不做 UnknownUser 校验。best-effort 上报:丢一两条
 from __future__ import annotations
 
 from sqlalchemy import func, select
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
 from app.core.rewards import cn_today
@@ -19,24 +20,52 @@ def create_ecpm_record(
     *,
     ad_type: str,
     ecpm_raw: str,
+    ad_session_id: str | None = None,
     adn: str | None = None,
     slot_id: str | None = None,
 ) -> AdEcpmRecord:
     """落一条 eCPM 上报记录。report_date 用北京时间当天,供按天聚合。"""
+    if ad_session_id:
+        existing = find_by_session(db, user_id=user_id, ad_session_id=ad_session_id)
+        if existing is not None:
+            return existing
     rec = AdEcpmRecord(
         user_id=user_id,
         ad_type=ad_type,
+        ad_session_id=ad_session_id,
         adn=adn,
         slot_id=slot_id,
         ecpm_raw=ecpm_raw,
         report_date=cn_today().isoformat(),
     )
     db.add(rec)
-    db.commit()
+    try:
+        db.commit()
+    except IntegrityError:
+        db.rollback()
+        if ad_session_id:
+            existing = find_by_session(db, user_id=user_id, ad_session_id=ad_session_id)
+            if existing is not None:
+                return existing
+        raise
     db.refresh(rec)
     return rec
 
 
+def find_by_session(
+    db: Session, *, user_id: int, ad_session_id: str | None
+) -> AdEcpmRecord | None:
+    """按广告会话找 eCPM。旧客户端无 ad_session_id 时返回 None。"""
+    if not ad_session_id:
+        return None
+    return db.execute(
+        select(AdEcpmRecord).where(
+            AdEcpmRecord.user_id == user_id,
+            AdEcpmRecord.ad_session_id == ad_session_id,
+        )
+    ).scalar_one_or_none()
+
+
 def count_today(db: Session, user_id: int) -> int:
     """该用户今日(北京时间)上报的 eCPM 条数,排查/对账辅助用。"""
     return db.execute(

app/repositories/ad_feed_reward.py

@@ -63,6 +63,7 @@ def grant_feed_reward(
     client_event_id: str,
     ecpm: str,
     duration_seconds: int,
+    ad_session_id: str | None = None,
     adn: str | None = None,
     slot_id: str | None = None,
 ) -> AdFeedRewardRecord:
@@ -82,6 +83,7 @@ def grant_feed_reward(
             reward_date=today,
             duration_seconds=safe_duration,
             unit_count=unit_count,
+            ad_session_id=ad_session_id,
             ecpm_raw=ecpm,
             adn=adn,
             slot_id=slot_id,
@@ -103,6 +105,7 @@ def grant_feed_reward(
         reward_date=today,
         duration_seconds=safe_duration,
         unit_count=unit_count,
+        ad_session_id=ad_session_id,
         ecpm_raw=ecpm,
         adn=adn,
         slot_id=slot_id,

app/repositories/ad_reward.py

@@ -5,8 +5,8 @@
   2. trans_id 唯一 → 同一交易号二次回调不重复发(穿山甲会重试)。
   3. 当日发奖次数 ≥ 上限 → 记一条 status='capped' 但不发金币。
 
-发金币复用 `wallet.grant_coins`(biz_type='ad_reward', ref_id=trans_id),与发奖记录同事务,
-保证"记一笔 + 加金币"原子化。单次金币 / 每日上限 / 每轮冷却 都从 app_config 读(运营后台可改)。
+普通福利页激励视频按 eCPM 公式发金币;缺 eCPM 时记 status='ecpm_missing' 但不发金币。
+签到膨胀等其他激励视频场景复用本表记录 S2S 幂等,实际发币由各自业务仓储完成。
 """
 from __future__ import annotations
 
@@ -21,6 +21,7 @@ from app.core.ad_cooldown import compute_cooldown
 from app.core.rewards import DAILY_AD_WATCH_SECONDS_LIMIT, cn_today
 from app.models.ad_reward import AdRewardRecord
 from app.models.user import User
+from app.repositories import ad_ecpm as crud_ecpm
 from app.repositories import wallet as crud_wallet
 from app.repositories.ad_watch import watched_seconds_today
 
@@ -35,6 +36,11 @@ def _find_by_trans(db: Session, trans_id: str) -> AdRewardRecord | None:
     ).scalar_one_or_none()
 
 
+def find_by_trans(db: Session, trans_id: str) -> AdRewardRecord | None:
+    """按 S2S 交易号查询处理记录,供路由在分场景前做幂等短路。"""
+    return _find_by_trans(db, trans_id)
+
+
 def _granted_today(db: Session, user_id: int, reward_date: str) -> int:
     return db.execute(
         select(func.count())
@@ -43,6 +49,7 @@ def _granted_today(db: Session, user_id: int, reward_date: str) -> int:
             AdRewardRecord.user_id == user_id,
             AdRewardRecord.reward_date == reward_date,
             AdRewardRecord.status == "granted",
+            AdRewardRecord.reward_scene == "reward_video",
         )
     ).scalar_one()
 
@@ -52,16 +59,13 @@ def grant_ad_reward(
     user_id: int,
     trans_id: str,
     *,
-    coin: int | None = None,
+    ecpm: str | None = None,
+    ad_session_id: str | None = None,
+    reward_scene: str = "reward_video",
     reward_name: str | None = None,
     raw: str | None = None,
 ) -> AdRewardRecord:
-    """看广告发奖(幂等 + 每日限额)。返回发奖记录(status=granted/capped)。
-
-    coin 为本次发放金币(由调用方按穿山甲回调 reward_amount 解析,见
-    rewards.resolve_ad_reward_coin);None → 读配置 get_ad_reward_coin(test-grant / 缺省场景)。
-    user_id 不存在抛 UnknownUserError(不建账户,防伪造 user_id 刷出脏数据)。
-    """
+    """福利页激励视频发奖(幂等 + 每日限额 + eCPM 公式)。"""
     # #2 幂等:同 trans_id 已处理过 → 原样返回,不重复发
     existing = _find_by_trans(db, trans_id)
     if existing is not None:
@@ -83,24 +87,86 @@ def grant_ad_reward(
         rec = AdRewardRecord(
             trans_id=trans_id, user_id=user_id, coin=0, status="capped",
             reward_date=today, reward_name=reward_name, raw=raw,
+            reward_scene=reward_scene, ad_session_id=ad_session_id, ecpm_raw=ecpm,
         )
         return _commit_record(db, rec, trans_id)
 
-    if coin is None:
-        coin = rewards.get_ad_reward_coin(db)
+    ecpm_raw = ecpm
+    if not ecpm_raw and ad_session_id:
+        ecpm_rec = crud_ecpm.find_by_session(db, user_id=user_id, ad_session_id=ad_session_id)
+        ecpm_raw = ecpm_rec.ecpm_raw if ecpm_rec is not None else None
+
+    if not ecpm_raw:
+        rec = AdRewardRecord(
+            trans_id=trans_id, user_id=user_id, coin=0, status="ecpm_missing",
+            reward_date=today, reward_name=reward_name, raw=raw,
+            reward_scene=reward_scene, ad_session_id=ad_session_id, ecpm_raw=None,
+        )
+        return _commit_record(db, rec, trans_id)
+
+    coin = rewards.calculate_ad_reward_coin(ecpm_raw, _granted_today(db, user_id, today) + 1)
 
     # 发金币 + 记一笔,同事务
     crud_wallet.grant_coins(
         db, user_id, coin,
-        biz_type="ad_reward", ref_id=trans_id, remark="看视频奖励金币",
+        biz_type="reward_video", ref_id=trans_id, remark="看视频奖励金币",
     )
     rec = AdRewardRecord(
         trans_id=trans_id, user_id=user_id, coin=coin, status="granted",
         reward_date=today, reward_name=reward_name, raw=raw,
+        reward_scene=reward_scene, ad_session_id=ad_session_id, ecpm_raw=ecpm_raw,
     )
     return _commit_record(db, rec, trans_id)
 
 
+def record_external_reward(
+    db: Session,
+    user_id: int,
+    trans_id: str,
+    *,
+    coin: int,
+    reward_scene: str,
+    ad_session_id: str | None = None,
+    ecpm: str | None = None,
+    reward_name: str | None = None,
+    raw: str | None = None,
+    status: str = "granted",
+    commit: bool = True,
+) -> AdRewardRecord:
+    """记录非普通看视频场景的 S2S 回调幂等,发币由调用方业务仓储完成。"""
+    existing = _find_by_trans(db, trans_id)
+    if existing is not None:
+        return existing
+    if db.get(User, user_id) is None:
+        raise UnknownUserError
+    rec = AdRewardRecord(
+        trans_id=trans_id,
+        user_id=user_id,
+        coin=coin,
+        status=status,
+        reward_date=cn_today().isoformat(),
+        reward_name=reward_name,
+        raw=raw,
+        reward_scene=reward_scene,
+        ad_session_id=ad_session_id,
+        ecpm_raw=ecpm,
+    )
+    db.add(rec)
+    if commit:
+        try:
+            db.commit()
+        except IntegrityError:
+            db.rollback()
+            existing = _find_by_trans(db, trans_id)
+            if existing is not None:
+                return existing
+            raise
+        db.refresh(rec)
+    else:
+        db.flush()
+    return rec
+
+
 def _commit_record(db: Session, rec: AdRewardRecord, trans_id: str) -> AdRewardRecord:
     """提交发奖记录;并发下同 trans_id 撞唯一约束时回滚并返回已存在的那条(幂等兜底)。"""
     db.add(rec)
@@ -125,6 +191,7 @@ def _granted_times_today_desc(db: Session, user_id: int, reward_date: str) -> li
                 AdRewardRecord.user_id == user_id,
                 AdRewardRecord.reward_date == reward_date,
                 AdRewardRecord.status == "granted",
+                AdRewardRecord.reward_scene == "reward_video",
             )
             .order_by(AdRewardRecord.created_at.desc())
         ).scalars()
@@ -153,7 +220,7 @@ def today_status(
     return (
         len(granted_desc),
         rewards.get_ad_daily_limit(db),
-        rewards.get_ad_reward_coin(db),
+        0,
         state.round_count,
         state.cooldown_until,
         watched_seconds_today(db, user_id, today=today),

app/repositories/comparison.py

@@ -5,9 +5,12 @@
 """
 from __future__ import annotations
 
+from datetime import datetime
+
 from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
+from app.core.rewards import CN_TZ
 from app.models.comparison import ComparisonRecord
 from app.models.savings import SavingsRecord
 from app.schemas.compare_record import ComparisonRecordIn
@@ -104,7 +107,15 @@ def upsert_record(
         db.refresh(existing)
         return existing
 
-    rec = ComparisonRecord(user_id=user_id, trace_id=payload.trace_id, **fields)
+    # created_at 显式存 naive 北京 wall-clock(同 savings_record):客户端「比价记录」页
+    # formatRecordTime 原样切片 created_at 字符串、不转时区,而默认 server_default=func.now()
+    # 在 SQLite 下返回 UTC → 直接慢 8h。覆盖分支不动 created_at(保留首次创建时间)。
+    rec = ComparisonRecord(
+        user_id=user_id,
+        trace_id=payload.trace_id,
+        created_at=datetime.now(CN_TZ).replace(tzinfo=None),
+        **fields,
+    )
     db.add(rec)
     db.commit()
     db.refresh(rec)

app/repositories/coupon_state.py

@@ -0,0 +1,129 @@
+"""领券今日状态读写:弹窗频控(engagement)+ 领券记录(claim)。
+
+写操作按唯一键幂等 upsert,自带 commit + 并发 IntegrityError 兜底(对齐 price_observation)。
+日期口径 = Asia/Shanghai 的自然日。
+"""
+from __future__ import annotations
+
+import logging
+from datetime import date, datetime
+from zoneinfo import ZoneInfo
+
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from app.models.coupon_state import CouponClaimRecord, CouponPromptEngagement
+
+logger = logging.getLogger("shagua.coupon_state")
+
+_CN_TZ = ZoneInfo("Asia/Shanghai")
+
+
+def today_cn() -> date:
+    """Asia/Shanghai 的自然日(领券判断的"今天")。"""
+    return datetime.now(_CN_TZ).date()
+
+
+# ===== 弹窗频控(coupon_prompt_engagement)=====
+
+def has_engaged_today(db: Session, device_id: str) -> bool:
+    """这台设备今天是否已对领券引导窗表达过意向(领或拒)。有 = 不再弹。"""
+    row = db.execute(
+        select(CouponPromptEngagement.id).where(
+            CouponPromptEngagement.device_id == device_id,
+            CouponPromptEngagement.engage_date == today_cn(),
+        )
+    ).first()
+    return row is not None
+
+
+def mark_engagement(
+    db: Session, device_id: str, user_id: int | None, engage_type: str
+) -> None:
+    """记今日意向(claim_started / dismissed)。(device, 今天) 唯一,幂等 upsert。"""
+    today = today_cn()
+    row = db.execute(
+        select(CouponPromptEngagement).where(
+            CouponPromptEngagement.device_id == device_id,
+            CouponPromptEngagement.engage_date == today,
+        )
+    ).scalar_one_or_none()
+    if row is not None:
+        row.engage_type = engage_type
+        if user_id is not None:
+            row.user_id = user_id
+    else:
+        db.add(CouponPromptEngagement(
+            device_id=device_id, user_id=user_id,
+            engage_date=today, engage_type=engage_type,
+        ))
+    try:
+        db.commit()
+    except IntegrityError:
+        # 并发下另一请求刚插了同 (device, 日) → 唯一约束撞,回滚忽略(本就幂等)。
+        db.rollback()
+
+
+# ===== 领券记录(coupon_claim_record)=====
+
+def record_claims(
+    db: Session,
+    device_id: str,
+    user_id: int | None,
+    trace_id: str | None,
+    results: list[dict],
+) -> int:
+    """一批券领取结果幂等写入,返回写入(新增 + 更新)条数。
+
+    results 单项取自 pricebot 的 last_coupon_result / done.coupon_results,识别字段:
+    coupon_id(必需)/ status(必需)/ name / vendor / reason /(display_count)。
+    (device, coupon_id, 今天) 唯一:重复上报同张券走更新(status 以最后一次为准)。
+    """
+    today = today_cn()
+    written = 0
+    seen: set[str] = set()  # 同批去重防御:autoflush=False 下同 coupon_id 重复会两次 add → 撞唯一约束回滚整批
+    for r in results:
+        coupon_id = r.get("coupon_id")
+        status = r.get("status")
+        if not coupon_id or not status or coupon_id in seen:
+            continue  # 脏数据 / 同批重复跳过
+        seen.add(coupon_id)
+        count = r.get("display_count")
+        if count is None:
+            count = r.get("claimed_count")
+        row = db.execute(
+            select(CouponClaimRecord).where(
+                CouponClaimRecord.device_id == device_id,
+                CouponClaimRecord.coupon_id == coupon_id,
+                CouponClaimRecord.claim_date == today,
+            )
+        ).scalar_one_or_none()
+        if row is not None:
+            row.status = status
+            row.reason = r.get("reason")
+            if user_id is not None:
+                row.user_id = user_id
+            if count is not None:
+                row.claimed_count = count
+            row.extra = r
+        else:
+            db.add(CouponClaimRecord(
+                device_id=device_id, user_id=user_id,
+                coupon_id=coupon_id, claim_date=today,
+                status=status, vendor=r.get("vendor"), coupon_name=r.get("name"),
+                claimed_count=count, trace_id=trace_id, reason=r.get("reason"),
+                extra=r,
+            ))
+        written += 1
+    if written == 0:
+        return 0
+    try:
+        db.commit()
+    except IntegrityError:
+        db.rollback()
+        logger.warning(
+            "coupon_claim 并发幂等冲突 device=%s trace=%s,回滚", device_id, trace_id
+        )
+        return 0
+    return written

app/repositories/invite.py

@@ -0,0 +1,166 @@
+"""好友邀请 CRUD(注册即生效,邀请人 + 被邀请人各发金币)。
+
+防重复发奖三道(仿 ad_reward / 提现的资金安全思路):
+  1. invitee_user_id 唯一 → 一个被邀请人只能被绑定一次(幂等键)。
+  2. 自邀屏蔽 → inviter == invitee 直接拒。
+  3. 现成的手机号唯一(每个被邀请人 = 一个真实手机号账号)= 天然限制刷量规模。
+
+发金币复用 wallet.grant_coins(grant 只 flush 不 commit),与建关系记录在**同一事务**
+commit,保证"建关系 + 双方加金币"原子。奖励额 = rewards.INVITE_INVITER_COINS /
+INVITE_INVITEE_COINS。
+"""
+from __future__ import annotations
+
+import secrets
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import func, select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from app.core import rewards
+from app.models.invite import InviteRelation
+from app.models.user import User
+from app.repositories import wallet as crud_wallet
+
+# 邀请码字符集:去掉易混字符(0/O/1/I/L/B/8/S/5/Z/2),用户口述/手输不易错
+_CODE_ALPHABET = "ACDEFGHJKMNPQRTUVWXY34679"
+_CODE_LEN = 6
+
+
+def _gen_code() -> str:
+    return "".join(secrets.choice(_CODE_ALPHABET) for _ in range(_CODE_LEN))
+
+
+def ensure_code(db: Session, user: User) -> str:
+    """保证 user 有邀请码(懒生成),返回它。唯一约束碰撞则换码重试。
+
+    ⚠️ 本函数会 db.commit() 整个 session(与 grant_coins"只 flush 不 commit"约定相反)。
+    当前只在 GET /invite/me 里调(该请求此前无其它写,提交范围干净)。若将来在带其它未提交
+    写的请求里复用本函数,会被它提前 commit——届时应改成 flush + 由调用方 commit。
+    """
+    if user.invite_code:
+        return user.invite_code
+    for _ in range(8):
+        user.invite_code = _gen_code()
+        try:
+            db.commit()
+            db.refresh(user)
+            return user.invite_code
+        except IntegrityError:
+            db.rollback()
+            user = db.get(User, user.id)  # 重取(rollback 后实例已过期),继续换码
+    raise RuntimeError("生成邀请码连续碰撞,请重试")
+
+
+def resolve_inviter(db: Session, invite_code: str) -> User | None:
+    """邀请码 → 邀请人(大小写不敏感)。"""
+    code = (invite_code or "").strip().upper()
+    if not code:
+        return None
+    return db.execute(
+        select(User).where(User.invite_code == code)
+    ).scalar_one_or_none()
+
+
+def _relation_of_invitee(db: Session, invitee_id: int) -> InviteRelation | None:
+    return db.execute(
+        select(InviteRelation).where(InviteRelation.invitee_user_id == invitee_id)
+    ).scalar_one_or_none()
+
+
+def _is_new_user(user: User) -> bool:
+    """被邀请人是否为"新注册"(created_at 在 INVITE_NEW_USER_WINDOW_HOURS 窗口内)。
+
+    "新用户闸":只奖刚注册的人,挡存量老用户互相填码薅羊毛。兼容 PG(tz-aware)与
+    SQLite(naive,按 UTC 解释)。
+    """
+    created = user.created_at
+    if created is None:
+        return False
+    if created.tzinfo is None:
+        created = created.replace(tzinfo=timezone.utc)
+    age = datetime.now(timezone.utc) - created
+    return age <= timedelta(hours=rewards.INVITE_NEW_USER_WINDOW_HOURS)
+
+
+@dataclass
+class BindResult:
+    status: str                          # success / already_bound / invalid_code / self_invite / not_eligible
+    relation: InviteRelation | None = None
+    invitee_coin: int = 0                # 本次给被邀请人发的金币(success 时 >0)
+
+
+def bind(
+    db: Session, *, invitee: User, invite_code: str, channel: str = "clipboard"
+) -> BindResult:
+    """把 invitee 绑定到 invite_code 对应的邀请人,注册即生效 + 双方发金币。
+
+    幂等:invitee 已被绑过 → already_bound(不重复发奖)。
+    """
+    # 幂等:已绑过直接返回(不重复发奖)
+    existing = _relation_of_invitee(db, invitee.id)
+    if existing is not None:
+        return BindResult("already_bound", existing)
+
+    inviter = resolve_inviter(db, invite_code)
+    if inviter is None or inviter.status != "active":
+        return BindResult("invalid_code")
+    if inviter.id == invitee.id:
+        return BindResult("self_invite")
+    # 新用户闸:被邀请人必须是"新注册"(窗口内)才发奖,挡存量老用户互相填码薅羊毛
+    if not _is_new_user(invitee):
+        return BindResult("not_eligible")
+
+    inviter_coin = rewards.INVITE_INVITER_COINS
+    invitee_coin = rewards.INVITE_INVITEE_COINS
+
+    rel = InviteRelation(
+        inviter_user_id=inviter.id,
+        invitee_user_id=invitee.id,
+        channel=(channel or "clipboard")[:16],
+        status="effective",
+        inviter_coin=inviter_coin,
+        invitee_coin=invitee_coin,
+    )
+    db.add(rel)
+    # 双方发金币(同事务,与建关系一起 commit)。ref_id 互指对方便于对账。
+    crud_wallet.grant_coins(
+        db, inviter.id, inviter_coin,
+        biz_type="invite_inviter", ref_id=str(invitee.id), remark="邀请好友奖励",
+    )
+    crud_wallet.grant_coins(
+        db, invitee.id, invitee_coin,
+        biz_type="invite_invitee", ref_id=str(inviter.id), remark="新人受邀奖励",
+    )
+    try:
+        db.commit()
+    except IntegrityError:
+        # 并发:同一 invitee 另一个请求先建了关系 → 回滚返回已存在(幂等兜底)
+        db.rollback()
+        existing = _relation_of_invitee(db, invitee.id)
+        if existing is not None:
+            return BindResult("already_bound", existing)
+        raise
+    except Exception:
+        # 其它 commit 失败(DB 故障 / PG 序列化冲突等):显式回滚,保证"建关系 + 双方发币"
+        # 原子(要么全成要么全无),不依赖 get_db 关闭时的隐式回滚,语义更硬。
+        db.rollback()
+        raise
+    db.refresh(rel)
+    return BindResult("success", rel, invitee_coin)
+
+
+def get_stats(db: Session, inviter_id: int) -> tuple[int, int]:
+    """返回 (已成功邀请人数, 累计从邀请获得的金币)。"""
+    count = db.execute(
+        select(func.count())
+        .select_from(InviteRelation)
+        .where(InviteRelation.inviter_user_id == inviter_id)
+    ).scalar_one()
+    coins = db.execute(
+        select(func.coalesce(func.sum(InviteRelation.inviter_coin), 0))
+        .where(InviteRelation.inviter_user_id == inviter_id)
+    ).scalar_one()
+    return int(count), int(coins)

app/repositories/report.py

@@ -1,9 +1,12 @@
 """price_report 表读写。"""
 from __future__ import annotations
 
+from datetime import datetime
+
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
+from app.core.rewards import CN_TZ
 from app.models.price_report import PriceReport
 
 
@@ -35,6 +38,7 @@ def create_report(
         reported_price_cents=reported_price_cents,
         images=images,
         status="pending",
+        created_at=datetime.now(CN_TZ).replace(tzinfo=None),  # 存北京 wall-clock(同 savings/comparison)
     )
     db.add(rep)
     db.commit()

app/repositories/savings.py

@@ -179,6 +179,10 @@ def create_from_report(
         client_event_id=req.client_event_id,
         device_id=req.device_id,
         source="compare",
+        # created_at 显式存 naive 北京 wall-clock(与 demo 行、聚合 _local_date 的 naive 分支一致)。
+        # 不用列默认 server_default=func.now()——SQLite 下它返回 UTC,会让明细页时间早 8 小时。
+        # ⚠️ 生产 PG(timestamptz)对 naive 的解释与 SQLite 不同,迁前端到 PG 时需确认时区一致。
+        created_at=datetime.now(CN_TZ).replace(tzinfo=None),
     )
     db.add(rec)
     db.commit()

app/repositories/signin.py

@@ -32,6 +32,10 @@ class AlreadyBoostedError(Exception):
     """今天签到奖励已经膨胀过。"""
 
 
+class LastCycleDayBoostBlockedError(Exception):
+    """14 天循环最后一天不允许签到膨胀。"""
+
+
 @dataclass
 class SigninStep:
     day: int  # 1..14
@@ -148,13 +152,27 @@ def _today_record(db: Session, user_id: int) -> SigninRecord | None:
     ).scalar_one_or_none()
 
 
+def boost_by_ad_ref(
+    db: Session, user_id: int, ad_ref_id: str
+) -> SigninBoostRecord | None:
+    """按广告交易号查签到膨胀记录。S2S 发奖后客户端确认用。"""
+    return db.execute(
+        select(SigninBoostRecord).where(
+            SigninBoostRecord.user_id == user_id,
+            SigninBoostRecord.ad_ref_id == ad_ref_id,
+        )
+    ).scalar_one_or_none()
+
+
 def boost_today_signin(
-    db: Session, user_id: int, *, ad_ref_id: str | None = None
+    db: Session, user_id: int, *, ad_ref_id: str | None = None, commit: bool = True
 ) -> tuple[SigninBoostRecord, int]:
-    """签到后看广告膨胀:补发一笔等额签到金币。返回 (膨胀记录, 补发后余额)。"""
+    """签到后看广告膨胀:固定补发配置金币。返回 (膨胀记录, 补发后余额)。"""
     record = _today_record(db, user_id)
     if record is None:
         raise NotSignedTodayError
+    if record.cycle_day == SIGNIN_CYCLE_LEN:
+        raise LastCycleDayBoostBlockedError
 
     today = record.signin_date
     existing = db.execute(
@@ -169,17 +187,20 @@ def boost_today_signin(
     boost = SigninBoostRecord(
         user_id=user_id,
         signin_date=today,
-        coin_awarded=record.coin_awarded,
+        coin_awarded=rewards.get_signin_boost_coin(db),
         ad_ref_id=ad_ref_id,
     )
     db.add(boost)
     try:
         acc, _ = crud_wallet.grant_coins(
-            db, user_id, record.coin_awarded,
-            biz_type="signin_boost", ref_id=today.isoformat(),
+            db, user_id, boost.coin_awarded,
+            biz_type="signin_boost", ref_id=ad_ref_id or today.isoformat(),
             remark=f"签到膨胀 第{record.cycle_day}天",
         )
-        db.commit()
+        if commit:
+            db.commit()
+        else:
+            db.flush()
     except IntegrityError as e:
         db.rollback()
         raise AlreadyBoostedError from e

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):
     """调用微信转账失败(已退回余额)。"""
 
@@ -116,6 +122,7 @@ def grant_coins(
         biz_type=biz_type,
         ref_id=ref_id,
         remark=remark,
+        created_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),  # 存北京 wall-clock(客户端原样切片显示)
     )
     db.add(txn)
     db.flush()
@@ -173,6 +180,7 @@ def exchange_coins_to_cash(
             balance_after_cents=acc.cash_balance_cents,
             biz_type="exchange_in",
             remark=remark,
+            created_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
         )
     )
     db.commit()
@@ -281,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(
@@ -291,11 +311,35 @@ def _refund_withdraw(
             ref_id=order.out_bill_no,
             # 用户可见文案区分"未成功(自动退)"vs"审核未通过";技术原因记在 order.fail_reason
             remark="提现审核未通过,金额已退回" if final_status == "rejected" else "提现未成功,金额已退回",
+            created_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
         )
     )
     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:
@@ -382,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)
 
@@ -401,6 +454,7 @@ def create_withdraw(
             biz_type="withdraw",
             ref_id=out_bill_no,
             remark="提现到微信零钱(待审核)",
+            created_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
         )
     )
     order = WithdrawOrder(
@@ -411,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  # 待管理员审核;**不在此处打款**
 
@@ -699,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/ad.py

@@ -28,7 +28,7 @@ class AdRewardStatusOut(BaseModel):
     used_today: int = Field(..., description="今日已成功发奖次数")
     daily_limit: int = Field(..., description="每日发奖次数上限")
     remaining: int = Field(..., description="今日剩余可领次数")
-    coin_per_ad: int = Field(..., description="看完一个激励视频发的金币")
+    coin_per_ad: int = Field(..., description="历史兼容字段;正式发放按 eCPM 动态计算,当前返回 0")
     round_count: int = Field(
         ..., description="本轮已看次数,当前 round_size=1,客户端可由 cooldown_until 判断短冷却"
     )
@@ -46,11 +46,15 @@ class AdRewardStatusOut(BaseModel):
 class EcpmReportIn(BaseModel):
     """客户端上报一次广告展示的 eCPM(内部收益统计/对账)。
 
-    user_id 不在 body 里——由 JWT 取(Bearer),防伪造。ecpm 原样上报字符串(单位待确认)。
+    user_id 不在 body 里——由 JWT 取(Bearer),防伪造。ecpm 原样上报字符串,后端按元/千次展示处理。
     """
 
     ad_type: str = Field(..., description="广告类型:reward_video(激励视频) / draw(Draw 信息流) 等")
-    ecpm: str = Field(..., description="穿山甲 getShowEcpm().getEcpm() 原始字符串,单位待确认,原样上报")
+    ecpm: str = Field(..., description="穿山甲 getShowEcpm().getEcpm() 原始字符串,按元/千次展示处理")
+    ad_session_id: str | None = Field(
+        None, min_length=8, max_length=64,
+        description="客户端生成的一次广告会话 id;激励视频 S2S extra 会透传同值",
+    )
     adn: str | None = Field(None, description="实际投放 ADN(getSdkName),如 pangle")
     slot_id: str | None = Field(None, description="实际展示代码位(底层 mediation rit)")
 
@@ -78,16 +82,27 @@ class WatchReportOut(BaseModel):
     watch_seconds_remaining: int = Field(..., description="今日剩余可观看秒数")
 
 
+class TestGrantIn(BaseModel):
+    """[仅本地联调]模拟发奖入参。"""
+
+    reward_scene: str = Field(
+        "reward_video",
+        description="模拟发奖场景:reward_video(普通激励视频) / signin_boost(签到膨胀)",
+    )
+
+
 class TestGrantOut(BaseModel):
     """[仅本地联调]模拟发奖结果。带上今日进度,客户端可直接据此刷新展示。"""
 
     granted: bool = Field(..., description="本次是否真的发了金币(达每日上限则 False)")
-    status: str = Field(..., description="granted / capped")
+    status: str = Field(
+        ..., description="granted / capped / not_signed / already_boosted / last_day / unknown_scene"
+    )
     coin: int = Field(..., description="本次发放金币(capped 时为 0)")
     used_today: int = Field(..., description="今日已成功发奖次数")
     daily_limit: int = Field(..., description="每日发奖次数上限")
     remaining: int = Field(..., description="今日剩余可领次数")
-    coin_per_ad: int = Field(..., description="看完一个激励视频发的金币")
+    coin_per_ad: int = Field(..., description="历史兼容字段;正式发放按 eCPM 动态计算,当前返回 0")
     round_count: int = Field(..., description="本轮已看次数,详见 AdRewardStatusOut")
     cooldown_until: datetime | None = Field(
         None, description="本轮冷却结束时间(UTC),详见 AdRewardStatusOut"
@@ -101,6 +116,9 @@ class FeedRewardIn(BaseModel):
     """
 
     client_event_id: str = Field(..., min_length=8, max_length=64, description="客户端生成的幂等事件 id")
+    ad_session_id: str | None = Field(
+        None, min_length=8, max_length=64, description="客户端生成的一次信息流广告会话 id"
+    )
     ecpm: str = Field(..., description="本条信息流广告 eCPM,按元/千次展示处理")
     duration_seconds: int = Field(..., ge=0, description="本条广告实际展示/播放秒数")
     adn: str | None = Field(None, description="实际投放 ADN")

app/schemas/coupon_state.py

@@ -0,0 +1,21 @@
+"""领券今日状态端点的收发模型。"""
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+
+class CouponPromptDismissIn(BaseModel):
+    """客户端拒绝/关闭领券引导窗的通知体。
+
+    server 据此记一条今日 engagement(dismissed)→ 今天这台设备不再弹引导窗。
+    MVP 不鉴权,按 device_id 判断;user_id 登录态带上就一并记(资产),可空。
+    """
+
+    device_id: str
+    user_id: int | None = None
+
+
+class CouponPromptShouldShowOut(BaseModel):
+    """切到外卖 App 时是否还应弹领券引导窗。今天已 engage(领或拒)过 → false。"""
+
+    should_show: bool

app/schemas/invite.py

@@ -0,0 +1,25 @@
+"""邀请相关 API 收发模型。"""
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class InviteInfoOut(BaseModel):
+    invite_code: str                # 我的邀请码
+    share_url: str                  # 落地页链接(含 ?ref=),前端据此生成二维码 + 复制分享
+    invited_count: int              # 已成功邀请人数
+    coins_earned: int               # 累计从邀请获得的金币
+
+
+class BindInviteIn(BaseModel):
+    invite_code: str = Field(..., min_length=4, max_length=16, description="邀请人的邀请码")
+    channel: str = Field(
+        "clipboard", max_length=16,
+        description="归因来源:clipboard(首启读剪贴板自动) / manual(用户手动填)",
+    )
+
+
+class BindInviteOut(BaseModel):
+    status: str                     # success / already_bound / invalid_code / self_invite / not_eligible
+    coins_awarded: int              # 本次给当前用户(被邀请人)发的金币
+    message: str                    # 给前端直接展示的文案

app/schemas/meituan.py

@@ -134,6 +134,12 @@ class FeedRequest(BaseModel):
     latitude: float = Field(..., description="纬度")
     page: int = Field(1, ge=1)
     page_size: int = Field(20, ge=1, le=20)
+    # 筛选/排序口径,后端据此处理后返回(前端不再自己筛/排):
+    #  rec      = 智能推荐:榜单混合 feed 去掉佣金率 < 3%(分页)
+    #  distance = 距离最近:拉齐全部轮次后全局按距离由近及远(一次性返回, has_next=False)
+    #  留空/其它 = 原混合 feed 不筛(老客户端兼容,新 app 会显式传 tab)
+    # (销量最高 sales 走 /coupons 同城热销,不在本接口)
+    tab: str = Field("", description="rec 智能推荐 / distance 距离最近 / 空=不筛(兼容)")
 
 
 class FeedResponse(BaseModel):
@@ -142,6 +148,13 @@ class FeedResponse(BaseModel):
     page: int = 1
 
 
+class TopSalesRequest(BaseModel):
+    """销量最高 tab:从离线库 meituan_coupon 按销量降序取(不实时打美团)。"""
+    page: int = Field(1, ge=1)
+    page_size: int = Field(20, ge=1, le=50)
+    platform: int | None = Field(None, description="可选: 1只外卖 / 2只到店; 不填=全部(全城销量)")
+
+
 # ───────────────── 换链 请求 / 响应 ─────────────────
 
 class ReferralLinkRequest(BaseModel):

data/media/dl.html

@@ -0,0 +1,147 @@
+<!DOCTYPE html>
+<html lang="zh-CN">
+<head>
+<meta charset="utf-8">
+<meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1, user-scalable=no">
+<meta name="format-detection" content="telephone=no">
+<title>傻瓜比价 · 下载</title>
+<style>
+  * { margin:0; padding:0; box-sizing:border-box; -webkit-tap-highlight-color:transparent; }
+  html,body { height:100%; }
+  body {
+    font-family:-apple-system,BlinkMacSystemFont,"PingFang SC","Microsoft YaHei",sans-serif;
+    background:linear-gradient(165deg,#FF7A3D 0%,#FF3B30 52%,#E0245E 100%);
+    color:#fff; min-height:100%; display:flex; flex-direction:column;
+    align-items:center; justify-content:center; padding:40px 26px; text-align:center;
+    overflow-x:hidden;
+  }
+  .logo {
+    width:104px; height:104px; border-radius:26px; background:#fff;
+    display:flex; align-items:center; justify-content:center; font-size:52px;
+    box-shadow:0 14px 34px rgba(0,0,0,.22); margin-bottom:24px;
+  }
+  h1 { font-size:30px; font-weight:800; letter-spacing:1px; }
+  .slogan { margin-top:12px; font-size:16px; line-height:1.7; opacity:.95; max-width:300px; }
+  .feats { margin-top:26px; display:flex; flex-direction:column; gap:12px; width:100%; max-width:320px; }
+  .feat { background:rgba(255,255,255,.16); border-radius:14px; padding:13px 16px; font-size:15px; display:flex; align-items:center; gap:10px; }
+  .feat b { font-weight:700; }
+  .btn {
+    margin-top:34px; width:100%; max-width:320px; border:none; cursor:pointer;
+    background:#fff; color:#FF3B30; font-size:19px; font-weight:800;
+    padding:17px 0; border-radius:999px; box-shadow:0 10px 26px rgba(0,0,0,.22);
+    display:flex; align-items:center; justify-content:center; gap:9px;
+  }
+  .btn:active { transform:translateY(1px); opacity:.92; }
+  .hint { margin-top:16px; font-size:13px; opacity:.85; }
+  .foot { margin-top:30px; font-size:12px; opacity:.6; line-height:1.6; max-width:320px; }
+
+  /* 微信内"去浏览器打开"引导蒙层 */
+  #wxmask {
+    display:none; position:fixed; inset:0; z-index:9999;
+    background:rgba(0,0,0,.86); padding:18px;
+  }
+  #wxmask.show { display:block; }
+  .arrow { position:absolute; top:8px; right:14px; width:120px; }
+  .wxtip { position:absolute; top:150px; right:18px; left:18px; text-align:right; }
+  .wxtip .big { font-size:21px; font-weight:800; line-height:1.5; }
+  .wxtip .big em { color:#FFD24D; font-style:normal; }
+  .wxtip .sub { margin-top:14px; font-size:15px; line-height:1.8; opacity:.9; }
+  .wxsteps { margin-top:26px; text-align:left; background:rgba(255,255,255,.1); border-radius:14px; padding:18px 18px; font-size:15px; line-height:2; }
+  .wxsteps .n { display:inline-block; width:22px; height:22px; line-height:22px; text-align:center; border-radius:50%; background:#FFD24D; color:#333; font-weight:800; font-size:13px; margin-right:8px; }
+  .closebar { position:absolute; bottom:30px; left:0; right:0; text-align:center; font-size:14px; opacity:.7; }
+</style>
+</head>
+<body>
+  <div class="logo">🛒</div>
+  <h1>傻瓜比价</h1>
+  <div class="slogan">买什么都先比一比<br>自动帮你找全网最低价</div>
+
+  <div class="feats">
+    <div class="feat">🍔 <span>点外卖前一键比价,<b>美团/京东/淘宝</b>到手价一目了然</span></div>
+    <div class="feat">🎟️ <span>自动领遍各平台<b>红包券</b>,能省的一分不漏</span></div>
+    <div class="feat">💰 <span>省下的钱看得见,还能<b>赚金币提现</b></span></div>
+  </div>
+
+  <button class="btn" id="dlbtn">⬇️ 下载安装包</button>
+  <div class="hint" id="hint">Android 安卓版 · 安装包约 24 MB</div>
+
+  <div class="foot">
+    安装时如提示「未知来源」,请允许后继续安装。<br>
+    本页为内部测试页。
+  </div>
+
+  <!-- 微信内引导:跳出微信去浏览器 -->
+  <div id="wxmask">
+    <svg class="arrow" viewBox="0 0 120 130" fill="none" xmlns="http://www.w3.org/2000/svg">
+      <path d="M30 120 C 30 70, 55 40, 95 28" stroke="#FFD24D" stroke-width="6" stroke-linecap="round" fill="none" stroke-dasharray="2 13"/>
+      <path d="M95 28 L 78 30 M95 28 L 92 46" stroke="#FFD24D" stroke-width="6" stroke-linecap="round"/>
+    </svg>
+    <div class="wxtip">
+      <div class="big">点击右上角 <em>···</em><br>选择「<em>在浏览器打开</em>」</div>
+      <div class="sub">微信里无法直接下载安装包<br>需在系统浏览器中完成下载</div>
+      <div class="wxsteps">
+        <div><span class="n">1</span>点右上角的 ··· 菜单</div>
+        <div><span class="n">2</span>选择「在浏览器打开」</div>
+        <div><span class="n">3</span>在浏览器里点「下载安装包」</div>
+      </div>
+    </div>
+    <div class="closebar" id="wxclose">我知道了 ✕</div>
+  </div>
+
+<script>
+  // APK 下载地址跟随当前页面 origin:本地测试时页面由笔记本 app-server(LAN:8770)托管 →
+  // 下载也走 LAN;生产由 app-api.shaguabijia.com 托管 → 下载走生产域名。无需按机器改 IP。
+  var APK_URL = location.origin + "/media/shaguabijia.apk";
+  var ua = navigator.userAgent || "";
+  var isWeChat = /MicroMessenger/i.test(ua);
+  var isIOS = /iPhone|iPad|iPod/i.test(ua);
+  var isAndroid = /Android/i.test(ua);
+  var ref = new URLSearchParams(location.search).get("ref");  // 邀请码(来自二维码 URL ?ref=)
+
+  // 把邀请码写进剪贴板,APK 首启时读出来完成归因(deferred deeplink 的关键一步)。
+  // 浏览器要求:必须在用户点击手势里调用 + HTTPS 下才允许写。
+  function legacyCopy(payload) {                // 老 webview / 无 clipboard API 兜底
+    try {
+      var ta = document.createElement("textarea");
+      ta.value = payload; ta.style.position = "fixed"; ta.style.top = "-1000px"; ta.style.opacity = "0";
+      document.body.appendChild(ta); ta.focus(); ta.select();
+      document.execCommand("copy"); document.body.removeChild(ta);
+    } catch (e) {}
+  }
+  function copyInviteCode() {                   // 返回 Promise(完成后才下载,避免异步写入被打断)
+    if (!ref) return Promise.resolve();
+    var payload = "SGBJ_INVITE:" + ref;
+    legacyCopy(payload);                         // 同步兜底:在用户手势内立刻 execCommand 写一次(最可靠)
+    if (navigator.clipboard && navigator.clipboard.writeText) {
+      return navigator.clipboard.writeText(payload).catch(function () {});  // 现代 API 锦上添花,失败无妨
+    }
+    return Promise.resolve();
+  }
+  // 确保剪贴板写完(或最多等 400ms 兜底,防写入卡住)再触发下载。
+  function startDownload() {
+    var fired = false;
+    function go() { if (!fired) { fired = true; window.location.href = APK_URL; } }
+    copyInviteCode().then(go);
+    setTimeout(go, 400);
+  }
+
+  var hint = document.getElementById("hint");
+  if (isIOS) hint.textContent = "检测到 iPhone · iOS 版请前往 App Store";
+
+  var mask = document.getElementById("wxmask");
+  function showMask(){ mask.classList.add("show"); }
+  function hideMask(){ mask.classList.remove("show"); }
+  document.getElementById("wxclose").addEventListener("click", hideMask);
+
+  // 微信里一进页面就提示去浏览器打开(下载在微信内必被拦)
+  if (isWeChat) showMask();
+
+  document.getElementById("dlbtn").addEventListener("click", function(){
+    if (isWeChat) { showMask(); return; }           // 微信内:引导去浏览器
+    if (isIOS)    { alert("iOS 版即将上线,请前往 App Store 搜索「傻瓜比价」"); return; }
+    // 普通浏览器:先把邀请码写进剪贴板(写完或 400ms 兜底),再下载 APK(首启读回完成归因)
+    startDownload();
+  });
+</script>
+</body>
+</html>

docs/README.md

@@ -10,6 +10,7 @@
 | [数据库迁移.md](./数据库迁移.md) | **Alembic 迁移指南**:clone 后如何建表、日常升级、新增迁移、迁移文件命名约定。想"把数据库跑起来 / 改表结构"看这份。 |
 | [待办与技术债.md](./待办与技术债.md) | **待办与技术债账本**:记"现在先简化、以后要补"的事 + 跨前后端技术债(P1 鉴权/用户绑定、引擎移植待办等),免遗忘。想知道"还欠什么、以后要补什么"看这份。 |
 | [看广告赚金币上线清单.md](./看广告赚金币上线清单.md) | **看广告发奖上线 checklist**:跨前后端,记上线前必做(GroMore 回调配置、清理调试脚手架、端到端验收)。上线"看广告赚金币"前对照这份。 |
+| [邀请功能-实现原理与本地测试.md](./邀请功能-实现原理与本地测试.md) | **好友邀请(invite-mvp)实现原理 + 跨前后端具体实现 + 本地内网测试方法**:剪贴板 deferred-deeplink 归因、双方各发 1 万金币、落地页 dl.html、起本地后端 + 编 debug 包 + 两台真机走全链路 + 上线前还差什么。要本地测 / 接手邀请功能看这份。 |
 
 ## api/ 目录是怎么组织的(传送门式)
 

docs/api/ad-ecpm-report.md

@@ -8,7 +8,8 @@
 | 字段 | 类型 | 必填 | 说明 |
 |---|---|---|---|
 | `ad_type` | str | 是 | 广告类型:`reward_video`(激励视频) / `draw`(Draw 信息流) 等 |
-| `ecpm` | str | 是 | 穿山甲 `getShowEcpm().getEcpm()` 原始字符串,**单位待确认(分/元)**,原样上报 |
+| `ad_session_id` | str\|null | 否 | 客户端生成的广告会话 ID;需和穿山甲 `extra.ad_session_id` 一致,用于 S2S 缺 eCPM 时匹配 |
+| `ecpm` | str | 是 | 穿山甲 `getShowEcpm().getEcpm()` 原始字符串,后端按“元/千次展示”参与金币公式 |
 | `adn` | str\|null | 否 | 实际投放 ADN(`getSdkName`),如 `pangle` |
 | `slot_id` | str\|null | 否 | 实际展示代码位(底层 mediation rit,非客户端配置位) |
 
@@ -24,7 +25,8 @@
 ## 说明
 客户端在广告**展示后**(`onAdShow` 读 `getShowEcpm()`)调用,把本次展示的 eCPM 落库做**内部收益统计/对账**。
 
-- **best-effort**:客户端 fire-and-forget,丢一两条不影响业务;穿山甲后台报表是结算权威兜底。
+- 普通激励视频发奖会先用 S2S 回调自带 `ecpm`;若缺失,再按 `ad_session_id` 读取本接口上报的 eCPM;两边都没有则不发并记录异常。
+- **best-effort**:客户端 fire-and-forget,但普通激励视频若 S2S 缺 eCPM,这条上报会成为发奖依据。
 - 落 `ad_ecpm_record` 表,`report_date` 用北京时间当天,供「按用户/按天聚合」对账。
 - **与发奖是两条独立流**:发奖走 [ad-pangle-callback](./ad-pangle-callback.md)(穿山甲 S2S,有 `trans_id`、无 ecpm),本接口客户端上报(有 ecpm、无 `trans_id`)。两者无公共键,**不逐条一一对应**,只做按用户/按天聚合口径。
 - eCPM 是**每千次展示预估**,单条展示预估收益 ≈ `ecpm / 1000`,且为客户端预估口径,非最终结算。

docs/api/ad-feed-reward.md

@@ -11,6 +11,7 @@
 | 字段 | 类型 | 必填 | 说明 |
 |---|---|---:|---|
 | `client_event_id` | string | 是 | 客户端生成的幂等事件 id，8-64 字符 |
+| `ad_session_id` | string\|null | 否 | 客户端生成的一次信息流广告会话 id，用于对账/排查 |
 | `ecpm` | string | 是 | 本条信息流广告 eCPM，按“元/千次展示”处理 |
 | `duration_seconds` | int | 是 | 实际展示/播放秒数 |
 | `adn` | string\|null | 否 | 实际投放 ADN |

docs/api/ad-pangle-callback.md

@@ -14,8 +14,8 @@ GroMore 以 GET 回调,关键参数:
 | `user_id` | string | 客户端 `setUserID` 传入的用户标识（须为数字 = 本系统 user.id） |
 | `trans_id` | string | 交易号（**幂等键** + **唯一参与签名的字段**） |
 | `reward_name` | string | 奖励名（广告位配置,入库备注） |
-| `reward_amount` | int | 奖励数量（广告位配置）→ **本次发放金币**。缺/坏/≤0 回退 `AD_REWARD_COIN`,超 `MAX_AD_REWARD_COIN` 夹紧 |
-| `extra` | string | 客户端透传的 customData（可空,入库备注） |
+| `ecpm` | string\|null | GroMore 回调携带的 eCPM。普通激励视频优先用它计算金币 |
+| `extra` / `gromoreExtra` / `gromore_extra` | string | 客户端透传 JSON。支持 `ad_session_id`、`reward_scene`;`reward_scene=signin_boost` 表示签到膨胀 |
 | `mediation_rit` | string | 代码位 ID（GroMore 带,目前仅入 raw 备查） |
 | `prime_rit` | string | 广告位 ID（同上） |
 | `adn_name` | string | 实际出广告的 ADN 名（同上,可用于收益分析） |
@@ -37,4 +37,9 @@ GroMore 以 GET 回调,关键参数:
 - `503` 回调未配置（`pangle_callback_configured=false`）
 
 ## 说明
-**发奖唯一可信入口**:验签 → 取 `user_id` → 按 `reward_amount` 解析金币 → 幂等发金币（按 `trans_id` 去重 + 每日上限）。客户端不参与发奖,被破解也刷不到钱。验签过但参数缺/坏或 user 不存在 → 不发（`is_verify=false` + `reason`）;granted / capped → `is_verify=true` + `reason=0`。
+**发奖唯一可信入口**:验签 → 取 `user_id`/`extra` → 按 `reward_scene` 分流 → 幂等处理（按 `trans_id` 去重）。客户端不直接发奖,被破解也刷不到钱。
+
+- `reward_scene=reward_video` 或缺省:普通激励视频。金币按 `eCPM / 1000 * eCPM因子 * 当日次数因子 * 10000` 计算;若回调没有 `ecpm`,会按 `extra.ad_session_id` 查客户端 `/ad/ecpm-report` 的上报值;两边都没有 eCPM 时不发币,记录 `status=ecpm_missing`。
+- `reward_scene=signin_boost`:签到膨胀。要求用户当天已签到且不是 Day14;看完视频固定发 `2000` 金币,写 `signin_boost_record` 与 `coin_transaction.biz_type=signin_boost`。
+- 未知 `reward_scene`:不发币,记录 `status=unknown_scene`,返回 `is_verify=false/reason=1`。
+- 验签过但参数缺/坏或 user 不存在 → 不发（`is_verify=false` + `reason`）;granted / capped / ecpm_missing / 业务不满足已记录 → `is_verify=true` + `reason=0`。

docs/api/ad-reward-status.md

@@ -13,7 +13,7 @@
 | `used_today` | int | 今日已成功发奖次数 |
 | `daily_limit` | int | 每日发奖次数上限 |
 | `remaining` | int | 今日剩余可领次数 |
-| `coin_per_ad` | int | 看完一个激励视频发的金币 |
+| `coin_per_ad` | int | 历史兼容字段;正式发放按 eCPM 动态计算,当前返回 0 |
 | `round_count` | int | 本轮已看次数;当前 `round_size=1`,广告关闭后进入短冷却 |
 | `cooldown_until` | datetime\|null | 3 秒短冷却结束时间(UTC ISO);null 表示不在冷却 |
 | `watched_seconds_today` | int | 今日已上报的激励视频观看秒数;当前仅兼容/排查用 |
@@ -22,7 +22,7 @@
 
 ## 说明
 福利页「看视频赚金币」用:
-- 展示「今日还能看 N 次」「看一次得 M 金币」(`remaining`/`coin_per_ad`)
+- 展示「今日还能看 N 次」(`remaining`);单次金币正式按 eCPM 动态计算,不再用 `coin_per_ad` 预估
 - 任务行 CTA 4 态推导:`adLoading` → Loading;`remaining==0` → Capped("明天再来");`cooldown_until` 在未来 → CoolingDown(显示 MM:SS 倒计时,不可点);否则 Normal("去赚取")
 - 弹窗 limit note:`used_today >= daily_limit` 显示「今日视频已到限额，明天再来」;`cooldown_until!=null` 显示「广告冷却中，3秒后再来」
 

docs/api/ad-test-grant.md

@@ -5,7 +5,11 @@
 > ⚠️ **仅本地联调**,受 `AD_REWARD_TEST_GRANT_ENABLED` 开关控制,**生产必须关闭**（默认 False → 一律 404）。
 
 ## 入参
-无（用户由 token 确定）。
+请求体可省略；用户由 token 确定。
+
+| 字段 | 类型 | 必填 | 默认 | 说明 |
+|---|---|---|---|---|
+| `reward_scene` | string | 否 | `reward_video` | 模拟发奖场景。`reward_video`=普通激励视频；`signin_boost`=签到膨胀 |
 
 ## 出参
 响应 `200`:`TestGrantOut`
@@ -13,15 +17,17 @@
 | 字段 | 类型 | 说明 |
 |---|---|---|
 | `granted` | bool | 本次是否真发了金币（达每日上限则 false） |
-| `status` | string | `granted` / `capped`（达上限） |
+| `status` | string | `granted` / `capped`（达上限） / `not_signed` / `already_boosted` / `last_day` / `unknown_scene` |
 | `coin` | int | 本次发放金币（capped 时 0） |
 | `used_today` | int | 今日已成功发奖次数 |
 | `daily_limit` | int | 每日发奖次数上限 |
 | `remaining` | int | 今日剩余可领次数 |
-| `coin_per_ad` | int | 看完一个激励视频发的金币 |
+| `coin_per_ad` | int | 历史兼容字段;正式发放按 eCPM 动态计算,当前返回 0 |
 
 ## 错误码
 - `404` 开关未开（伪装不存在） / 用户不存在
 
 ## 说明
-没公网、穿山甲 S2S 回调打不到本地时,debug 客户端看完广告后调它,直接走与 [ad-pangle-callback](./ad-pangle-callback.md) 相同的发奖逻辑（每次新 `trans_id`,幂等 + 每日上限）。它让已登录客户端能自助发奖 = 绕过反作弊,**严禁在生产开启**。
+没公网、穿山甲 S2S 回调打不到本地时,debug 客户端看完广告后调它,直接走与 [ad-pangle-callback](./ad-pangle-callback.md) 相同的发奖逻辑（每次新 `trans_id`,幂等 + 每日上限/今日膨胀一次）。
+
+`reward_scene=signin_boost` 时复用签到膨胀业务规则:必须当天已签到、非第 14 天、当天未膨胀过,成功后写入 `signin_boost` 金币流水。它让已登录客户端能自助发奖 = 绕过反作弊,**严禁在生产开启**。

docs/api/admin-stats-overview.md

@@ -31,6 +31,14 @@
 | 字段 | 类型 | 说明 |
 |---|---|---|
 | `granted_total` | int | 累计发放金币(coin_transaction 中所有 `amount > 0` 之和;负数兑换/扣减不计) |
+| `reward_video_coin_total` | int | 普通激励视频累计发放金币(`biz_type=reward_video/ad_reward`) |
+| `reward_video_watch_count` | int | 普通激励视频有效发奖视频数(`ad_reward_record.reward_scene=reward_video,status=granted`) |
+| `feed_ad_coin_total` | int | 信息流广告累计发放金币(`biz_type=feed_ad_reward`) |
+| `feed_ad_watch_count` | int | 信息流广告有效完成视频数(`ad_feed_reward_record.status=granted`) |
+| `signin_coin_total` | int | 签到累计发放金币(`biz_type=signin`) |
+| `signin_count` | int | 签到次数(`signin_record`) |
+| `signin_boost_coin_total` | int | 签到膨胀累计发放金币(`biz_type=signin_boost`) |
+| `signin_boost_watch_count` | int | 签到膨胀有效视频数(`signin_boost_record`) |
 
 **DashboardCash**
 | 字段 | 类型 | 说明 |
@@ -64,6 +72,6 @@
 ## 说明
 - 全部为全局只读聚合(count / sum / DAU / 成功率),不改任何数据。
 - `new_today` / `dau` 按**北京时区**(UTC+8)切天,其余金额/计数无时区概念。
-- 金额单位:`*_cents` 为分;金币(`granted_total`)为个数。
+- 金额单位:`*_cents` 为分;金币(`*_coin_total`/`granted_total`)为个数。
 - CPS 收入数据源未接入(referral-link 只换链接,转化/佣金未回收),`cps` 恒为 `{available:false, note:...}`,前端显示"待接入"。
 - 关联表:[user](../database/user.md) / [coin_transaction](../database/coin_transaction.md) / [withdraw_order](../database/withdraw_order.md) / [comparison_record](../database/comparison_record.md) / [feedback](../database/feedback.md)。

docs/api/signin-boost.md

@@ -1,6 +1,6 @@
 # POST /api/v1/signin/boost — 签到后看广告膨胀金币
 
-用户当天已签到后，看完一条激励视频，补发一笔等额签到金币。
+用户 Day1-Day13 当天已签到后，看完一条激励视频，由穿山甲 S2S 回调固定补发 2000 金币。本接口只用于 S2S 发奖后的确认。
 
 ## 鉴权
 
@@ -10,7 +10,7 @@
 
 | 字段 | 类型 | 必填 | 说明 |
 |---|---|---:|---|
-| `ad_ref_id` | string\|null | 否 | 广告会话/交易号。当前开发期可空，后续接 S2S 后可回填 |
+| `ad_ref_id` | string | 是 | 穿山甲 S2S 回调的 `trans_id`。回调需先以 `extra.reward_scene=signin_boost` 完成发奖 |
 
 ## 响应
 
@@ -23,10 +23,11 @@
 ## 错误
 
 - `401`: 未登录
-- `409`: 当天未签到，或当天已经膨胀过
+- `409`: 缺少/无效广告回调记录，非本人广告，回调未发奖，当天未签到，Day14，或当天已经膨胀过
 
 ## 数据写入
 
-- `signin_boost_record` 新增一行，用 `(user_id, signin_date)` 唯一约束防重复。
-- `coin_account` 增加余额。
-- `coin_transaction` 写入 `biz_type=signin_boost`。
+- 本接口不直接发奖;实际写入发生在 `/ad/pangle-callback` 的 `reward_scene=signin_boost` 分支。
+- 回调写 `signin_boost_record` 新增一行，用 `(user_id, signin_date)` 唯一约束防重复。
+- 回调使 `coin_account` 增加固定 `2000` 金币。
+- 回调写入 `coin_transaction.biz_type=signin_boost`。

docs/database/OVERVIEW.md

@@ -64,7 +64,7 @@
 | 金币兑现金 `POST /wallet/exchange` | `coin_account`(U) + `coin_transaction`(C `exchange_out` −) + `cash_transaction`(C `exchange_in` +) | 同事务 |
 | 发起提现 `POST /wallet/withdraw` | `withdraw_order`(C `reviewing`) + `coin_account`(U 扣现金) + `cash_transaction`(C `withdraw` −) | 同事务,**不打款** |
 | 查提现状态 / 用户取消 `GET /wallet/withdraw/status` | `withdraw_order`(U) + 失败→`cash_transaction`(C `withdraw_refund` +) | |
-| 穿山甲发奖 S2S 回调 `POST /ad/pangle-callback` | `ad_reward_record`(C)+ granted→`coin_account`(U)+`coin_transaction`(C `ad_reward`) | `trans_id` 幂等 |
+| 穿山甲发奖 S2S 回调 `POST /ad/pangle-callback` | `ad_reward_record`(C)+ granted→`coin_account`(U)+`coin_transaction`(C `reward_video`/`signin_boost`) | `trans_id` 幂等 |
 | 看广告时长上报 `POST /ad/watch-report` | `ad_watch_log`(C) | |
 | 广告 eCPM 上报 `POST /ad/ecpm-report` | `ad_ecpm_record`(C) | |
 | 信息流广告结算 `POST /ad/feed-reward` | `ad_feed_reward_record`(C)+ granted→`coin_account`(U)+`coin_transaction`(C `feed_ad_reward`) | `client_event_id` 幂等 |
@@ -105,7 +105,7 @@
   | `signin` | 当天日期串(= `signin_record.signin_date` 的 ISO `YYYY-MM-DD`) | + |
   | `signin_boost` | 当天日期串(= `signin_boost_record.signin_date` 的 ISO `YYYY-MM-DD`) | + |
   | `task_<key>` | `user_task.task_key` | + |
-  | `ad_reward` | `ad_reward_record.trans_id` | + |
+  | `reward_video` / `ad_reward`(历史) | `ad_reward_record.trans_id` | + |
   | `feed_ad_reward` | `ad_feed_reward_record.client_event_id` | + |
   | `exchange_out` | null(兑现金,无单据) | − |
   | `admin_grant` / `admin_deduct` | null(原因记在 `remark`=`admin:<reason>`) | + / − |
@@ -118,7 +118,7 @@
   | `exchange_in` | null | + |
 
 - **`comparison_record.store_name` ≈ `savings_record.shop_name`**:无 id 关联,按**店名字符串相等**给比价记录打「已下单」标记(瞬态,不写库)。两边店名同源 = 比价意图识别阶段的门店 query,语义=**店级**(同店比价多次会一并标已下单)。
-- **广告流互不关联**:`ad_reward_record` / `ad_watch_log` / `ad_ecpm_record` / `ad_feed_reward_record` 之间**无公共键**,各自只按 `(user_id, 日期串)` 聚合。别试图 join 它们逐条对应。
+- **广告流会话关联**:`ad_reward_record.ad_session_id` 可与 `ad_ecpm_record.ad_session_id` 对齐;`ad_watch_log` 仍是旧版兼容统计,不逐条参与发奖。
 - **里程碑解锁进度不存库**:`comparison_milestone_claim` 只记「哪几档已领」;进度 = `comparison_record` 里 `status='success'` 的 `count`。
 
 ### ER 关系(文字版)

docs/database/ad_ecpm_record.md

@@ -2,7 +2,7 @@
 
 > 模型 `app/models/ad_ecpm.py` · 仓库 `app/repositories/ad_ecpm.py` · 接口 [ad-ecpm-report](../api/ad-ecpm-report.md) · [← 索引](./README.md) · [总览](./OVERVIEW.md)
 
-每条 = 客户端一次广告展示(`onAdShow`)后读到的 eCPM。看广告三条数据流之一(**收益对账**),与 `ad_reward`(发奖,有 trans_id 无 ecpm)、`ad_watch_log`(时长)并列,**无公共键**——只用于**按用户/按天聚合**估算内部广告收益,不做"这条发奖 = 这条展示"的逐条精确关联。穿山甲后台报表才是结算权威,本表是细粒度补充。
+每条 = 客户端一次广告展示(`onAdShow`)后读到的 eCPM。`ad_session_id` 可与 S2S 回调 `extra.ad_session_id` 对齐:当普通激励视频回调没有 `ecpm` 时,后端会用它匹配本表作为发奖 eCPM 依据。
 
 ## 用在哪 / 增删改查
 - **C(插入)**:`POST /ad/ecpm-report`(`create_ecpm_record`)。每次广告展示上报一条;best-effort,丢一两条不影响业务。鉴权接口已确保 user 存在。
@@ -15,18 +15,19 @@
 | `id` | Integer | PK, autoincrement | |
 | `user_id` | Integer | FK→user.id, index, NOT NULL | 归属用户 |
 | `ad_type` | String(32) | NOT NULL | 广告类型,取值如 `reward_video`(激励视频)/ `draw`(Draw 信息流);各类型各自上报,不强行统一代码位 |
+| `ad_session_id` | String(64) | UNIQUE, index, nullable | 客户端广告会话 ID;用于普通激励视频在 S2S 缺 `ecpm` 时匹配发奖 |
 | `adn` | String(32) | nullable | 实际投放 ADN(`getShowEcpm().getSdkName()`,如 `pangle`/`gdt`) |
 | `slot_id` | String(64) | nullable | 实际展示用代码位(底层 mediation rit,非客户端配置位) |
-| `ecpm_raw` | String(32) | NOT NULL | 客户端上报的 eCPM **原始串**(单位待确认,原样存) |
+| `ecpm_raw` | String(32) | NOT NULL | 客户端上报的 eCPM **原始串**;后端按“元/千次展示”参与金币公式 |
 | `report_date` | String(10) | index, NOT NULL | 北京时间日期串 `YYYY-MM-DD`,按它做按天聚合 |
 | `created_at` | DateTime(tz) | server_default now(), index | 时间 |
 
 ## 关系 / Join Key
 - `user_id` → `user.id`(多对一)。
-- 与 `ad_reward_record` / `ad_watch_log` **无公共键**(独立数据流),只按 `(user_id, report_date)` 聚合。
+- `ad_session_id` → 可关联 `ad_reward_record.ad_session_id`。
 
 ## 索引与约束
-- PK `id`;index `user_id`、`report_date`、`created_at`。
+- PK `id`;index `user_id`、`report_date`、`created_at`、`ad_session_id`;UNIQUE `ad_session_id`。
 
 ## 注意
-- ⚠️ `ecpm_raw` 单位(分/元)截至 2026-05-31 未最终确认;确认后再加一列解析好的数值,在此之前对账按"待定单位"处理。
+- S2S 回调自带 `ecpm` 优先;缺失时才使用同 `ad_session_id` 的客户端上报值。两者都没有时普通激励视频不发币,`ad_reward_record.status=ecpm_missing`。

docs/database/ad_feed_reward_record.md

@@ -8,6 +8,7 @@
 |---|---|---|---|
 | `id` | Integer | PK | 自增主键 |
 | `client_event_id` | String(64) | UNIQUE, NOT NULL | 客户端幂等事件 id |
+| `ad_session_id` | String(64) | index, nullable | 客户端生成的一次信息流广告会话 id |
 | `user_id` | Integer | FK → `user.id`, index, NOT NULL | 用户 |
 | `reward_date` | String(10) | index, NOT NULL | 北京时间日期 `YYYY-MM-DD` |
 | `duration_seconds` | Integer | NOT NULL | 实际展示/播放秒数 |

docs/database/ad_reward_record.md

@@ -2,10 +2,10 @@
 
 > 模型 `app/models/ad_reward.py` · 仓库 `app/repositories/ad_reward.py` · 接口 [ad-pangle-callback](../api/ad-pangle-callback.md) / [ad-reward-status](../api/ad-reward-status.md) / [ad-test-grant](../api/ad-test-grant.md) · [← 索引](./README.md) · [总览](./OVERVIEW.md)
 
-每条 = 穿山甲一次**服务端发奖回调**(用户看完激励视频)。是看广告三条数据流之一(**发奖**;另两条:`ad_watch_log` 时长、`ad_ecpm_record` 收益,三者无公共键)。`trans_id` 唯一做幂等键(穿山甲会重试,同号只发一次)。`reward_date`(北京时间日期串)给"每日上限"计数用。
+每条 = 穿山甲一次**服务端激励回调**。`trans_id` 唯一做幂等键(穿山甲会重试,同号只处理一次)。`reward_scene` 区分普通激励视频、签到膨胀等场景;`reward_date`(北京时间日期串)给普通激励视频"每日上限"计数用。
 
 ## 用在哪 / 增删改查
-- **C(插入)**:`POST /ad/pangle-callback`(穿山甲 S2S,经 SHA256 验签;`grant_ad_reward`)或 `POST /ad/test-grant`(本地联调)。三道闸:① 验签不过 → API 层 403,不进库;② `trans_id` 已存在 → 原样返回不重复发;③ **当日发奖次数(`DAILY_AD_REWARD_LIMIT`,默认 500)到顶** → 记一行 `status='capped'`、`coin=0`、不发币。旧的观看总时长闸仅在 `DAILY_AD_WATCH_SECONDS_LIMIT > 0` 时启用,当前默认停用。否则 `granted` + `grant_coins(biz_type='ad_reward', ref_id=trans_id)` 加币,同事务。
+- **C(插入)**:`POST /ad/pangle-callback`(穿山甲 S2S,经 SHA256 验签;`grant_ad_reward` 或场景业务处理)或 `POST /ad/test-grant`(本地联调)。普通激励视频三道闸:① 验签不过 → API 层 403,不进库;② `trans_id` 已存在 → 原样返回不重复发;③ **当日发奖次数(`DAILY_AD_REWARD_LIMIT`,默认 500)到顶** → 记一行 `status='capped'`、`coin=0`、不发币。否则按 eCPM 公式发币。
 - **U / D**:无。
 - **R**:`GET /ad/reward-status`(看广告页:今日已发次数/上限、单次金币、本轮已看/冷却结束、今日已看时长/上限);审计/对账整表回溯。
 
@@ -13,10 +13,13 @@
 | 列 | 类型 | 约束 / 默认 | 说明(取值 / join) |
 |---|---|---|---|
 | `id` | Integer | PK, autoincrement | |
-| `trans_id` | String(64) | UNIQUE, index, NOT NULL | 穿山甲交易号(幂等键)。**被 `coin_transaction.ref_id` 引用**(biz_type=ad_reward) |
+| `trans_id` | String(64) | UNIQUE, index, NOT NULL | 穿山甲交易号(幂等键)。**被 `coin_transaction.ref_id` 引用**(biz_type=reward_video/signin_boost 等) |
 | `user_id` | Integer | FK→user.id, index, NOT NULL | 归属用户(回调 media_extra 带回;不存在抛 UnknownUserError) |
-| `coin` | Integer | NOT NULL, default 0 | 实发金币;`capped` 时为 0 |
-| `status` | String(16) | NOT NULL, default `granted` | 取值:`granted`(已发)/ `capped`(当日次数超限,记录但不发) |
+| `reward_scene` | String(32) | NOT NULL, default `reward_video` | 奖励场景:`reward_video` 普通激励视频;`signin_boost` 签到膨胀 |
+| `ad_session_id` | String(64) | index, nullable | 客户端广告会话 ID,来自 `extra.ad_session_id`;用于匹配 `ad_ecpm_record` |
+| `ecpm_raw` | String(32) | nullable | 本次发奖采用的 eCPM 原始值;可来自 S2S `ecpm` 或客户端上报 |
+| `coin` | Integer | NOT NULL, default 0 | 实发金币;`capped`/`ecpm_missing`/业务不满足时为 0 |
+| `status` | String(16) | NOT NULL, default `granted` | 取值:`granted`(已发)/ `capped`(当日次数超限)/ `ecpm_missing`(缺 eCPM)/ `not_signed`/`already_boosted`/`last_day` |
 | `reward_date` | String(10) | index, NOT NULL | 北京时间日期串 `YYYY-MM-DD`,按它等值统计当日发奖次数 |
 | `reward_name` | String(64) | nullable | 穿山甲上报奖励名(参考,不作发奖依据) |
 | `raw` | String(1024) | nullable | 回调原始参数(审计排查) |
@@ -24,12 +27,13 @@
 
 ## 关系 / Join Key
 - `user_id` → `user.id`(多对一)。
-- `trans_id` ← 被 `coin_transaction.ref_id` 引用(`granted` 那条发币流水);`capped` 行不发币、无对应流水。
-- 与 `ad_watch_log` / `ad_ecpm_record` **无公共键**;当前不逐条关联 eCPM 和发奖回调。
+- `trans_id` ← 被 `coin_transaction.ref_id` 引用(`granted` 那条发币流水);未发币状态无对应流水。
+- `ad_session_id` → 可关联 `ad_ecpm_record.ad_session_id`。
 
 ## 索引与约束
-- PK `id`;UNIQUE+index `trans_id`;index `user_id`、`reward_date`、`created_at`。
+- PK `id`;UNIQUE+index `trans_id`;index `user_id`、`reward_date`、`created_at`、`ad_session_id`。
 
 ## 注意
-- 当前公网回调尚未接入前,激励视频精确 eCPM 绑定发奖先搁置;实发金币仍以穿山甲回调 `reward_amount` 为准(`resolve_ad_reward_coin` 解析,缺/坏回退默认、超上限夹紧),单次金币/每日上限/单次上限均从 `app_config` 读(运营后台可改)。
+- 普通激励视频按 eCPM 公式发奖;若 S2S 与客户端会话上报都缺 eCPM,记录 `status='ecpm_missing'`、`coin=0`,不发币。
+- 签到膨胀复用本表记录 S2S 幂等,实发固定 `2000` 金币由 `signin_boost_record`/`coin_transaction.biz_type=signin_boost` 承载。
 - 并发同 `trans_id` 撞唯一约束 → catch IntegrityError 回滚返回已存在那条(幂等兜底)。

docs/database/coin_transaction.md

@@ -12,7 +12,7 @@
   | 签到 `POST /signin/do` | `signin` | + | 当天日期串(= `signin_record.signin_date` ISO) |
   | 签到后看广告膨胀 `POST /signin/boost` | `signin_boost` | + | 当天日期串(= `signin_boost_record.signin_date` ISO) |
   | 领任务 `POST /tasks/claim` | `task_<key>`(如 `task_enable_notification`) | + | `user_task.task_key` |
-  | 看广告发奖回调 `POST /ad/pangle-callback` | `ad_reward` | + | `ad_reward_record.trans_id` |
+  | 普通激励视频 S2S 回调 `POST /ad/pangle-callback` | `reward_video`(历史兼容:`ad_reward`) | + | `ad_reward_record.trans_id` |
   | 信息流广告结算 `POST /ad/feed-reward` | `feed_ad_reward` | + | `ad_feed_reward_record.client_event_id` |
   | 金币兑现金 `POST /wallet/exchange` | `exchange_out` | − | null(配套 `cash_transaction.exchange_in`) |
   | admin 手动加金币 | `admin_grant` | + | null(`remark`=`admin:<reason>`) |
@@ -29,14 +29,14 @@
 | `user_id` | Integer | FK→user.id, index, NOT NULL | 归属用户 |
 | `amount` | Integer | NOT NULL | 本笔变动金币;**正=入账(赚),负=出账(花/兑换)** |
 | `balance_after` | Integer | NOT NULL | 本笔后金币余额(= 当时 `coin_account.coin_balance`,对账用) |
-| `biz_type` | String(32) | NOT NULL | 取值见上表:`signin` / `signin_boost` / `task_<key>` / `ad_reward` / `feed_ad_reward` / `exchange_out` / `admin_grant` / `admin_deduct`。无 DB 枚举约束,靠写入方约定 |
+| `biz_type` | String(32) | NOT NULL | 取值见上表:`signin` / `signin_boost` / `task_<key>` / `reward_video` / `ad_reward`(历史) / `feed_ad_reward` / `exchange_out` / `admin_grant` / `admin_deduct`。无 DB 枚举约束,靠写入方约定 |
 | `ref_id` | String(64) | nullable | **关联业务键,指向随 `biz_type` 变(见上表)**;无关联时为 null |
 | `remark` | String(128) | nullable | 备注(如签到「每日签到 第N天」、admin「admin:<reason>」) |
 | `created_at` | DateTime(tz) | server_default now(), index | 时间 |
 
 ## 关系 / Join Key
 - `user_id` → `user.id`(多对一)。
-- `ref_id` 是**软关联**(无 FK),目标表随 `biz_type`:`signin`→签到日 / `signin_boost`→签到日 / `task_<key>`→`user_task.task_key` / `ad_reward`→`ad_reward_record.trans_id` / `feed_ad_reward`→`ad_feed_reward_record.client_event_id` / 其余 null。
+- `ref_id` 是**软关联**(无 FK),目标表随 `biz_type`:`signin`→签到日 / `signin_boost`→`ad_reward_record.trans_id` / `task_<key>`→`user_task.task_key` / `reward_video`/`ad_reward`→`ad_reward_record.trans_id` / `feed_ad_reward`→`ad_feed_reward_record.client_event_id` / 其余 null。
 
 ## 索引与约束
 - PK `id`;index `user_id`、`created_at`。

docs/database/signin_boost_record.md

@@ -1,6 +1,6 @@
 # signin_boost_record — 签到膨胀记录
 
-App 用户当天签到后，看完激励视频可把签到奖励膨胀一次。本表记录膨胀动作，并用唯一约束防重复补发。
+App 用户 Day1-Day13 当天签到后，看完激励视频可固定膨胀 2000 金币一次。Day14 不展示也不允许膨胀。本表记录膨胀动作，并用唯一约束防重复补发。
 
 ## 字段
 
@@ -9,8 +9,8 @@ App 用户当天签到后，看完激励视频可把签到奖励膨胀一次。
 | `id` | Integer | PK | 自增主键 |
 | `user_id` | Integer | FK → `user.id`, index, NOT NULL | 用户 |
 | `signin_date` | Date | NOT NULL | 被膨胀的签到日期，北京时间 |
-| `coin_awarded` | Integer | NOT NULL | 本次补发金币 |
-| `ad_ref_id` | String(64) | nullable | 广告会话/交易号，开发期可空 |
+| `coin_awarded` | Integer | NOT NULL | 本次补发金币,默认固定 2000 |
+| `ad_ref_id` | String(64) | nullable | 穿山甲 S2S 回调 `trans_id` |
 | `created_at` | DateTime | NOT NULL | 创建时间 |
 
 ## 约束
@@ -19,4 +19,4 @@ App 用户当天签到后，看完激励视频可把签到奖励膨胀一次。
 
 ## 关联
 
-- 膨胀成功时写 `coin_transaction.biz_type=signin_boost`，`ref_id=signin_date`。
+- 膨胀成功时写 `coin_transaction.biz_type=signin_boost`，`ref_id=ad_ref_id`。

docs/integrations/pangle.md

@@ -9,7 +9,7 @@
 **`sign = SHA256("{m-key}:{trans_id}")` 的十六进制串。**
 
 - 是**普通 SHA256**，不是 HMAC，也不是 RSA。
-- **只对 `m-key:trans_id` 这一个字符串**签名，`user_id` / `reward_amount` 等其余参数**不参与**签名。它们的可信度由"能算出正确 sign = 知道 m-key"间接保证——伪造者没有 m-key 连合法 sign 都造不出，自然无法注入假 `user_id`。
+- **只对 `m-key:trans_id` 这一个字符串**签名，`user_id` / `ecpm` / `extra` 等其余参数**不参与**签名。它们的可信度由"能算出正确 sign = 知道 m-key"间接保证——伪造者没有 m-key 连合法 sign 都造不出，自然无法注入假 `user_id`。
 - 这是穿山甲 **GroMore「广告位 → 服务端激励回调」官方规范**（supportcenter/26240）。我们客户端是 `useMediation(true)` 融合，回调走 **GroMore 广告位层级**，不是联盟代码位层级。
 
 ## 函数

docs/邀请功能-实现原理与本地测试.md

@@ -0,0 +1,139 @@
+# 好友邀请(invite-mvp)— 实现原理与本地测试
+
+> 跨 **shaguabijia-app-server**(后端)+ **shaguabijia-app-android**(前端)的好友邀请功能。
+> 本文给三件事:**做什么 / 怎么实现 / 怎么在本地内网测**(交接给同事用)。
+> 代码引用一律用文件路径 + 符号名(不写行号,重构后行号会失效)。
+
+---
+
+## 一、功能是什么
+
+**注册即生效**:被邀请人(B)用邀请人(A)的邀请码完成绑定 → **A、B 各得 1 万金币(= 1 元,可提现)**。
+
+---
+
+## 二、实现原理(数据流)
+
+归因(B 装的这次算不算 A 邀的)有**两条通道**:
+
+- **剪贴板自动**(deferred deeplink,生产主路径):落地页把邀请码写进剪贴板,APK 首启读回。
+- **手动填码**(兜底):B 在邀请页直接输入 A 的邀请码。
+
+剪贴板自动这条的完整跨库链路:
+
+```
+A 端(已登录)
+  └─ GET /api/v1/invite/me → 拿到 invite_code + share_url(=落地页?ref=<码>)
+  └─ 分享 share_url(微信/任意渠道)
+        ↓
+B 端(系统浏览器打开 share_url = .../media/dl.html?ref=<码>)
+  └─ 点「下载安装包」
+        ├─ dl.html 把 "SGBJ_INVITE:<码>" 写进剪贴板(execCommand)
+        └─ 跳 APK_URL 下载安装包
+        ↓
+B 安装并首启 App
+  └─ InviteCapture 读剪贴板,识别 "SGBJ_INVITE:" 前缀 → 暂存 pending 邀请码
+  └─ B 登录成功 → AuthRepository 触发 InviteRepository.bindPendingFromClipboard()
+        └─ POST /api/v1/invite/bind { invite_code, channel="clipboard" }
+        ↓
+后端 repositories/invite.py bind()
+  └─ 过四道防线 → 建 invite_relation + 给 A、B 各发金币(同事务原子提交)
+```
+
+手动填码这条:B 在邀请页输码 → `InviteRepository.bindManual()` → `POST /bind { channel="manual" }` → 同一个 `bind()`。
+
+---
+
+## 三、具体实现
+
+### 3.1 后端(shaguabijia-app-server)
+
+| 关注点 | 位置 |
+|---|---|
+| 端点 | `app/api/v1/invite.py`:`GET /api/v1/invite/me`(返回 `invite_code` + `share_url` + 战绩)、`POST /api/v1/invite/bind`(绑定,`channel` = `clipboard` / `manual`)。**均需 Bearer 鉴权**。 |
+| share_url 构造 | `invite.py` 的 `my_invite`:`settings.INVITE_LANDING_URL + "?ref=" + code`。`INVITE_LANDING_URL` 在 `app/core/config.py`(默认 `https://app-api.shaguabijia.com/media/dl.html`)。 |
+| 业务逻辑 | `app/repositories/invite.py`:`ensure_code`(懒生成 6 位邀请码,去混淆字符集,唯一约束碰撞则换码)/ `resolve_inviter`(邀请码→邀请人,大小写不敏感)/ `bind`(下面详述)/ `get_stats`(已邀人数 + 累计金币)。 |
+| 数据模型 | `app/models/invite.py` 的 `InviteRelation`(`inviter_user_id` / `invitee_user_id` / `channel` / `status` / `inviter_coin` / `invitee_coin` / `created_at`)+ `app/models/user.py` 的 `User.invite_code` 列。 |
+| 迁移 | `alembic/versions/invite_code_and_relation.py`:给 `user` 加 `invite_code`(唯一索引)+ 建 `invite_relation` 表。`down_revision = 11a1d08c6f55`。 |
+| 收发模型 | `app/schemas/invite.py`:`InviteInfoOut` / `BindInviteIn` / `BindInviteOut`。 |
+| 奖励常量 | `app/core/rewards.py`:`INVITE_INVITER_COINS` / `INVITE_INVITEE_COINS`(各 10000 = 1 元)、`INVITE_NEW_USER_WINDOW_HOURS`(72)。 |
+
+**`bind()` 的四道防线(防重复 / 防刷,看 `repositories/invite.py`):**
+
+1. **被邀请人唯一**:`invitee_user_id` 唯一约束 → 一个 B 只能被绑一次(幂等键,重复返回 `already_bound`,不重复发奖)。
+2. **自邀屏蔽**:`inviter == invitee` → `self_invite`。
+3. **新人闸**:`_is_new_user`(B 的 `created_at` 在 `INVITE_NEW_USER_WINDOW_HOURS` = 72h 内)才发奖,挡存量老用户互相填码薅羊毛 → 否则 `not_eligible`。
+4. **手机号唯一**(天然限量):每个 B = 一个真实手机号账号。
+
+发金币复用 `repositories/wallet.py` 的 `grant_coins`,与建关系记录在**同一事务**提交,保证"建关系 + 双方加金币"原子。
+
+### 3.2 前端(shaguabijia-app-android)
+
+| 关注点 | 位置 |
+|---|---|
+| 网络层 | `data/api/InviteApi.kt` + `InviteDto.kt`:`me()` / `bind(BindInviteRequest)`。 |
+| 门面 | `data/invite/InviteRepository.kt`:`myInvite()` / `bindManual(code)` / `bindPendingFromClipboard()`。 |
+| 剪贴板捕获 | `util/InviteCapture.kt`:首启读剪贴板找 `SGBJ_INVITE:` 前缀(常量 `CLIP_PREFIX`),命中即暂存到 SP(`pending_invite_code`);**未命中不烧机会**,跨多次启动最多试 `MAX_ATTEMPTS`(5)次,之后靠手动填码兜底。 |
+| 自动绑定触发 | 登录成功后由 `data/auth/AuthRepository.kt` 调 `InviteRepository.bindPendingFromClipboard()`(best-effort,不抛异常)。 |
+| UI | `ui/invite/InviteScreen.kt` + `InviteViewModel.kt`:展示码 / 二维码(`util/QrCodeGen.kt`)/ 战绩 + 手动填码入口。 |
+
+### 3.3 落地页 dl.html(`data/media/dl.html`,由后端 `/media` 静态托管)
+
+- 读 URL 的 `?ref=<邀请码>`。
+- 点「下载安装包」时:`legacyCopy()` 用 `document.execCommand('copy')` **同步**把 `"SGBJ_INVITE:<码>"` 写进剪贴板(execCommand **不要求安全上下文**,内网 HTTP 也能写;`navigator.clipboard` 只是 HTTPS 下锦上添花、失败无妨),再跳 `APK_URL` 下载。
+- `APK_URL = location.origin + "/media/shaguabijia.apk"`:**跟随页面 host**——本地测试时页面由笔记本(LAN:8770)托管 → 下载也走 LAN;生产由 `app-api.shaguabijia.com` 托管 → 走生产域名。**无需按机器改 IP**。
+- 微信内显示「去浏览器打开」引导(微信内 execCommand / apk 下载都受限,必须跳系统浏览器)。
+
+> dl.html 已纳入 git(见本仓 `.gitignore` 对 `data/media/dl.html` 的例外)。`data/media/shaguabijia.apk`(大二进制)、`avatars/` 等仍忽略。
+
+---
+
+## 四、本地内网测试方法(重点)
+
+**拓扑**:同一 WiFi 下,1 台笔记本(起后端 + 托管落地页/APK)+ 2 台安卓真机(A 邀请人 / B 被邀请人)。落地页和 APK 都由笔记本 app-server 的 `/media` 在局域网托管,一处搞定。
+
+### 4.1 起后端
+
+1. 本机 PostgreSQL 起着;`conda activate price`。
+2. `.env` 改两项(**本地测试覆盖,测完改回**):
+   - `SMS_MOCK=true` —— 任意手机号 + 任意 6 位码秒注册,方便建 A/B 两个号(生产是 `false` 走真极光)。
+   - `INVITE_LANDING_URL=http://<笔记本LAN_IP>:8770/media/dl.html` —— 让 `/invite/me` 返回的 share_url 走内网。
+3. `alembic upgrade head` 建表。
+   - ⚠️ **切分支后若报 `Can't locate revision`**(DB 版本指针停在别分支的迁移,例如先测过 coupon-daily-claim-gate,DB 停在 `coupon_state_tables`,而本分支没这个文件):用 `alembic stamp <共同祖先> --purge` 把指针重置到共同祖先(本分支祖先是 `11a1d08c6f55`)→ 再 `alembic upgrade head` 只跑 invite 这一条建表。**别删多余表**。详见 `docs/数据库迁移.md`。
+4. `./run.sh`(监听 `0.0.0.0:8770`)。**笔记本防火墙放行入站 8770**。
+
+### 4.2 落地页 + APK(都由 `/media` 在 LAN 托管)
+
+- **落地页**:`data/media/dl.html` 已在 git、host-agnostic,**无需改**(clone 出来就在)。
+- **APK**:编一个 debug 包放到 `data/media/shaguabijia.apk`:
+  1. 前端 `local.properties` 填 `debug.api.host=<笔记本LAN_IP>` / `debug.api.port=8770`(debug 包的 `BASE_URL` 据此**编译期**烧入,见 `app/build.gradle.kts`;debug 包还放行明文 HTTP)。
+  2. `./gradlew assembleDebug`。
+  3. `cp app/build/outputs/apk/debug/app-debug.apk <app-server>/data/media/shaguabijia.apk`。
+
+### 4.3 跑流程
+
+1. A 真机装 debug 包 → 登录 → 邀请页拿到 6 位码 / share_url(= `http://<LAN>:8770/media/dl.html?ref=<码>`)。
+2. A 把 share_url 发出去 → B 真机**用系统浏览器**打开 → 点「下载」(写剪贴板 + 下 APK)→ 安装。
+3. B 首启 → `InviteCapture` 读剪贴板拿码(Android 12+ 会弹"已粘贴"提示,正常)→ B 用**新手机号**登录 → 自动 `bind(clipboard)` → **A、B 各 +1 万金币,`invite_relation` 入库**。
+4. **手动填码路径**:B 在邀请页直接输 A 的码(`channel=manual`),不依赖剪贴板——剪贴板失败时的兜底,值得单独测。
+
+### 4.4 测试硬约束 / 坑(都是机制,不是 bug)
+
+- **B 必须用新手机号**:`invitee_user_id` 唯一,一个 B 只能绑一次;反复测要换号(或手删 `invite_relation` 那行 + 回滚金币)。
+- **72h 新人闸**:B 注册后 72 小时内绑才发奖(刚注册肯定满足)。
+- **A ≠ B**:自邀被屏蔽。
+- **B 从点下载到首启 App 之间别复制别的东西**:剪贴板会被覆盖 → 归因丢(剪贴板 deferred deeplink 的固有脆弱性)。
+- **笔记本 IP 别变**:debug 包把 `BASE_URL` 的 IP 烧死在编译期,DHCP 一换就连不上 → 给笔记本固定个 LAN IP。
+
+---
+
+## 五、上线前还差什么(本地测通 ≠ 可上线)
+
+本地这一关证明的是"机制能跑通",离生产可上线还差:
+
+1. ⚠️ **alembic 多 head(不修 100% 翻车)**:`coupon_state_tables` 和 `invite_code_and_relation` 都挂在 `11a1d08c6f55` 下、是**兄弟迁移**。invite 一合 main,main 上就有两个 head,prod `alembic upgrade head` 撞 `Multiple head revisions` → 服务起不来。**合 main 前 / 部署前必须建一个 merge 迁移**把两条链并起来(参考 repo 里已有的 `*_merge_*_heads.py`)。本地单分支单 head 测不出。
+2. **release 包 ≠ debug 包**:用户拿到的是 release(签名不同、强制 HTTPS、若开 minify 还有 R8 混淆),本地测的是 debug。上线前用 **release 包真机**验全流程。
+3. **风控 / 薅羊毛**:1 万金币可提现,**邀请人无总数上限**,接码平台批量注册新号绑同一个码即可刷。上线前加 **inviter 上限 + 基础风控**。
+4. **仅验了 happy path**:错误分支文案(`invalid_code` / `already_bound` / `self_invite` / `not_eligible`)、手动填码兜底、stats 展示更新、并发绑定都没真测。
+5. **配置回归**:`SMS_MOCK` / `INVITE_LANDING_URL` 是本地覆盖,生产走真极光 + 默认生产落地页。
+6. **(dl.html 纳入 git 后的部署细节)**:生产机上已有未跟踪的 `data/media/dl.html`(之前手动放的)。若走 git 部署,首次 pull 前需先移走生产那份,否则报 `untracked working tree files would be overwritten`。

scripts/pull_meituan_coupons.py

@@ -0,0 +1,331 @@
+"""美团 CPS 券定时抓取入库(北京试点)。
+
+把 3 路券抓进 meituan_coupon 表,供「销量/佣金排序」从库里捞、本地排序,不再实时打美团:
+  1. search_waimai : 到家/外卖, 搜「外卖」          翻到尽头
+  2. search_meishi : 到家/外卖, 搜「美食」          翻到尽头
+  3. store_supply  : 到店, 多业务线供给(到餐+到综+酒店+门票) 翻到尽头
+
+按 (source, product_view_sign) upsert 存最新态;last_seen 每轮刷新。带文件锁,防止
+上一轮没跑完下一轮又起(本地 5~10min、跨进程 cron 都安全)。
+
+用法:
+    # 单轮(打通验证 / 给 cron 用,线上每 1h 一次)
+    python -m scripts.pull_meituan_coupons --once
+
+    # 本地循环(默认每 10min 一轮)
+    python -m scripts.pull_meituan_coupons --loop --interval 600
+"""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import os
+import re
+import sys
+import time
+from datetime import datetime, timedelta, timezone
+
+# Windows 控制台按 UTF-8 输出中文/¥
+try:
+    sys.stdout.reconfigure(encoding="utf-8")  # type: ignore[attr-defined]
+except Exception:  # noqa: BLE001
+    pass
+
+from sqlalchemy import delete, func, select
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+
+from app.db.session import SessionLocal
+from app.integrations.meituan import MeituanCpsError, _call
+from app.models.meituan_coupon import MeituanCoupon
+
+CITY_BEIJING = "WKV2HMXUEK634WP64CUCUQGM64"
+QUERY_PATH = "/cps_open/common/api/v1/query_coupon"
+PAGE_SIZE = 20
+MAX_PAGES = 80          # 单路安全上限(搜索 ~52 页、供给 ~70 页)
+PAGE_SLEEP = 0.35       # 页间配速,缓解 402
+RETRY = 7
+LOCK_FILE = "data/.meituan_etl.lock"
+LOCK_STALE_SEC = 30 * 60   # 锁超过 30min 视为陈旧(进程异常退出残留),自动接管
+
+SOURCES = [
+    {"code": "search_waimai", "label": "外卖·搜外卖", "kind": "search", "platform": 1, "keyword": "外卖"},
+    {"code": "search_meishi", "label": "外卖·搜美食", "kind": "search", "platform": 1, "keyword": "美食"},
+    {"code": "store_supply", "label": "到店·多业务线供给", "kind": "supply",
+     "platform": 2, "biz_lines": [1, 2, 3, 4]},
+]
+
+
+# ───────────────────────── 美团调用 ─────────────────────────
+
+def _call_retry(body: dict) -> dict | None:
+    """打美团,402/频繁退避重试;其它错误打印并放弃本页。"""
+    for a in range(RETRY):
+        try:
+            return _call(QUERY_PATH, body)
+        except MeituanCpsError as e:
+            msg = str(e)
+            if "402" in msg or "频繁" in msg:
+                time.sleep(2.5 * (a + 1))
+                continue
+            print(f"   [warn] meituan: {msg[:80]}")
+            return None
+        except Exception as e:  # noqa: BLE001
+            print(f"   [warn] {type(e).__name__}: {str(e)[:60]}")
+            time.sleep(2.0 * (a + 1))
+    return None
+
+
+def _pull_search(platform: int, keyword: str) -> list[dict]:
+    rows: list[dict] = []
+    sid = None
+    pg = 1
+    while pg <= MAX_PAGES:
+        body = {"platform": platform, "searchText": keyword, "cityId": CITY_BEIJING, "pageSize": PAGE_SIZE}
+        if sid:
+            body["searchId"] = sid
+        else:
+            body["pageNo"] = pg
+        r = _call_retry(body)
+        if not r:
+            break
+        sid = r.get("searchId")
+        data = r.get("data") or []
+        rows.extend(data)
+        if not r.get("hasNext") or not data:
+            break
+        pg += 1
+        time.sleep(PAGE_SLEEP)
+    return rows
+
+
+def _pull_supply(platform: int, biz_lines: list[int]) -> list[dict]:
+    rows: list[dict] = []
+    sid = None
+    biz_param = [{"bizLine": b} for b in biz_lines]
+    for _ in range(MAX_PAGES):
+        body = {
+            "multipleSupplyList": [{"platform": platform, "bizLineParamList": biz_param}],
+            "cityId": CITY_BEIJING,
+            "sortField": 2,  # 供给查询 sortField 必填;我们入库后本地再排,这里给个默认
+            "pageSize": PAGE_SIZE,
+        }
+        if sid:
+            body["searchId"] = sid
+        r = _call_retry(body)
+        if not r:
+            break
+        sid = r.get("searchId")
+        data = r.get("data") or []
+        rows.extend(data)
+        if not r.get("hasNext") or not data:
+            break
+        time.sleep(PAGE_SLEEP)
+    return rows
+
+
+# ───────────────────────── 解析 ─────────────────────────
+
+_SV_RE = re.compile(r"(\d+(?:\.\d+)?)\s*(w|万|k)?", re.I)
+
+
+def _sale_volume_num(s: str | None) -> int | None:
+    """'热销1w+' → 10000, '热销500+' → 500(取下界,排序用)。"""
+    if not s:
+        return None
+    m = _SV_RE.search(str(s))
+    if not m:
+        return None
+    v = float(m.group(1))
+    u = (m.group(2) or "").lower()
+    if u in ("w", "万"):
+        v *= 10000
+    elif u == "k":
+        v *= 1000
+    return int(v)
+
+
+def _to_cents(yuan) -> int | None:
+    if yuan in (None, "", "null"):
+        return None
+    try:
+        return round(float(yuan) * 100)
+    except (TypeError, ValueError):
+        return None
+
+
+def _parse_item(item: dict, source: dict) -> dict | None:
+    cpd = item.get("couponPackDetail") or {}
+    br = item.get("brandInfo") or {}
+    ci = item.get("commissionInfo") or {}
+    poi = item.get("availablePoiInfo") or {}
+    dp = item.get("deliverablePoiInfo") or {}
+
+    sign = cpd.get("productViewSign") or cpd.get("skuViewId")
+    if not sign:
+        return None
+
+    name = cpd.get("name") or ""
+    brand = br.get("brandName") or ""
+    price_cents = _to_cents(cpd.get("sellPrice"))
+
+    comm_pct = None
+    if ci.get("commissionPercent") is not None:
+        try:
+            comm_pct = float(ci["commissionPercent"]) / 100.0  # 140 → 1.4(%)
+        except (TypeError, ValueError):
+            comm_pct = None
+
+    dist = dp.get("deliveryDistance")
+    try:
+        dist = float(dist) if dist not in (None, "", "null") else None
+    except (TypeError, ValueError):
+        dist = None
+
+    dedup_raw = f"{brand}|{name}|{price_cents}"
+    return {
+        "source": source["code"],
+        "platform": source["platform"],
+        "biz_line": item.get("bizLine") or cpd.get("bizLine"),
+        "city_id": CITY_BEIJING,
+        "product_view_sign": str(sign)[:128],
+        "sku_view_id": cpd.get("skuViewId"),
+        "name": (name[:256] or None),
+        "brand_name": (brand[:128] or None),
+        "sell_price_cents": price_cents,
+        "original_price_cents": _to_cents(cpd.get("originalPrice")),
+        "head_url": ((cpd.get("headUrl") or "").split("@")[0][:512] or None),
+        "sale_volume": cpd.get("saleVolume"),
+        "sale_volume_num": _sale_volume_num(cpd.get("saleVolume")),
+        "commission_percent": comm_pct,
+        "commission_amount_cents": _to_cents(ci.get("commission")),
+        "poi_name": ((dp.get("poiName") or "")[:128] or None),
+        "available_poi_num": poi.get("availablePoiNum"),
+        "delivery_distance_m": dist,
+        "dedup_key": hashlib.md5(dedup_raw.encode("utf-8")).hexdigest(),
+        "raw": item,
+    }
+
+
+# ───────────────────────── 入库(upsert) ─────────────────────────
+
+def _upsert(db, rows: list[dict], now: datetime) -> tuple[int, int]:
+    """按 (source, product_view_sign) upsert。返回 (入库去重后条数, 本源内重复条数)。"""
+    if not rows:
+        return 0, 0
+    dedup: dict[tuple, dict] = {}
+    for r in rows:
+        dedup[(r["source"], r["product_view_sign"])] = r  # 同轮内同键保留最后一条
+    payload = list(dedup.values())
+    for r in payload:
+        r["last_seen"] = now
+        r["updated_at"] = now
+
+    chunk = 500
+    for i in range(0, len(payload), chunk):
+        part = payload[i:i + chunk]
+        stmt = pg_insert(MeituanCoupon).values(part)
+        update_cols = {
+            c: getattr(stmt.excluded, c)
+            for c in part[0]
+            if c not in ("source", "product_view_sign", "first_seen")
+        }
+        stmt = stmt.on_conflict_do_update(
+            constraint="uq_meituan_coupon_source_sign", set_=update_cols
+        )
+        db.execute(stmt)
+        db.commit()
+    return len(payload), len(rows) - len(payload)
+
+
+# ───────────────────────── 运行锁 ─────────────────────────
+
+def _acquire_lock() -> bool:
+    os.makedirs(os.path.dirname(LOCK_FILE) or ".", exist_ok=True)
+    try:
+        fd = os.open(LOCK_FILE, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
+        os.write(fd, f"{os.getpid()} {time.time()}".encode())
+        os.close(fd)
+        return True
+    except FileExistsError:
+        try:
+            with open(LOCK_FILE) as f:
+                parts = f.read().split()
+            ts = float(parts[1]) if len(parts) > 1 else 0.0
+            if time.time() - ts > LOCK_STALE_SEC:
+                os.remove(LOCK_FILE)
+                return _acquire_lock()
+        except Exception:  # noqa: BLE001
+            pass
+        return False
+
+
+def _release_lock() -> None:
+    try:
+        os.remove(LOCK_FILE)
+    except OSError:
+        pass
+
+
+# ───────────────────────── 主流程 ─────────────────────────
+
+def run_once(prune_hours: int = 24) -> None:
+    if not _acquire_lock():
+        print(f"[{datetime.now():%H:%M:%S}] 上一轮还在跑(锁占用),跳过本轮")
+        return
+    t0 = time.time()
+    now = datetime.now(timezone.utc)
+    db = SessionLocal()
+    try:
+        total = 0
+        for src in SOURCES:
+            ts = time.time()
+            if src["kind"] == "search":
+                items = _pull_search(src["platform"], src["keyword"])
+            else:
+                items = _pull_supply(src["platform"], src["biz_lines"])
+            parsed = [p for p in (_parse_item(it, src) for it in items) if p]
+            up, dup = _upsert(db, parsed, now)
+            total += up
+            print(f"  {src['label']:18} 抓{len(items):5} 解析{len(parsed):5} "
+                  f"入库{up:5} (本源去重{dup:3}) {time.time() - ts:4.0f}s")
+        # 清理长期未再出现的陈旧券(美团 sign 轮换 / 券下架后的残留),默认 24h 宽限
+        if prune_hours and prune_hours > 0:
+            cutoff = now - timedelta(hours=prune_hours)
+            pruned = db.execute(
+                delete(MeituanCoupon).where(MeituanCoupon.last_seen < cutoff)
+            ).rowcount
+            db.commit()
+            if pruned:
+                print(f"  清理陈旧券(>{prune_hours}h 未再出现): {pruned} 条")
+        cnt = db.execute(select(func.count()).select_from(MeituanCoupon)).scalar()
+        print(f"[{datetime.now():%H:%M:%S}] 本轮完成: 入库 {total} 条, 表总计 {cnt} 行, "
+              f"用时 {time.time() - t0:.0f}s")
+    finally:
+        db.close()
+        _release_lock()
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser(description="美团 CPS 券定时抓取入库")
+    ap.add_argument("--once", action="store_true", help="只跑一轮(默认)")
+    ap.add_argument("--loop", action="store_true", help="循环跑")
+    ap.add_argument("--interval", type=int, default=600, help="循环间隔秒(默认 600=10min)")
+    ap.add_argument("--prune-hours", type=int, default=24,
+                    help="清理超过 N 小时未再出现的陈旧券(默认 24;0=不清理)")
+    args = ap.parse_args()
+
+    if args.loop:
+        print(f"循环模式: 每 {args.interval}s 一轮 (Ctrl-C 退出)")
+        while True:
+            try:
+                run_once(args.prune_hours)
+            except Exception as e:  # noqa: BLE001
+                print(f"[{datetime.now():%H:%M:%S}] 本轮异常: {type(e).__name__}: {e}")
+                _release_lock()
+            time.sleep(args.interval)
+    else:
+        run_once(args.prune_hours)
+
+
+if __name__ == "__main__":
+    main()

tests/test_ad_reward.py

@@ -10,9 +10,7 @@ from sqlalchemy import select
 from app.core.config import settings
 from app.integrations import pangle
 from app.core.rewards import (
-    AD_REWARD_COIN,
     DAILY_AD_REWARD_LIMIT,
-    MAX_AD_REWARD_COIN,
     VIDEO_ROUND_COOLDOWN_SECONDS,
     VIDEO_ROUND_REQUIRED_COUNT,
     calculate_ad_reward_coin,
@@ -57,41 +55,34 @@ def _coin_balance(client, token: str) -> int:
 
 
 def test_callback_grants_coins(client) -> None:
-    """验签通过的回调(无 reward_amount → 回退 AD_REWARD_COIN)→ 发金币到账 + 计数 +1。"""
+    """验签通过且带 eCPM → 按数值公式发金币到账 + 计数 +1。"""
     phone = "13800003001"
     token = _login(client, phone)
     uid = _user_id(phone)
 
-    r = _callback(client, _signed(uid, "trans_a1", reward_name="金币"))
+    r = _callback(client, _signed(uid, "trans_a1", reward_name="金币", ecpm="200"))
     assert r.status_code == 200, r.text
     assert r.json() == {"is_verify": True, "reason": 0}
 
-    assert _coin_balance(client, token) == AD_REWARD_COIN
+    assert _coin_balance(client, token) == calculate_ad_reward_coin("200", 1)
 
     st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
     assert st["used_today"] == 1
     assert st["daily_limit"] == DAILY_AD_REWARD_LIMIT
     assert st["remaining"] == DAILY_AD_REWARD_LIMIT - 1
-    assert st["coin_per_ad"] == AD_REWARD_COIN
+    assert st["coin_per_ad"] == 0
 
 
-def test_callback_uses_reward_amount(client) -> None:
-    """带合法 reward_amount → 按它发金币(而非常量);异常值回退/夹紧。"""
+def test_callback_without_ecpm_records_exception(client) -> None:
+    """S2S 和客户端会话都没有 eCPM → 不发金币,只记录异常状态。"""
     phone = "13800003011"
     token = _login(client, phone)
     uid = _user_id(phone)
 
-    # 合法值:发 250
-    assert _callback(client, _signed(uid, "ra_ok", reward_amount="250")).status_code == 200
-    assert _coin_balance(client, token) == 250
-
-    # ≤0 / 非数字 → 回退 AD_REWARD_COIN(累加 AD_REWARD_COIN,当前 666)
-    assert _callback(client, _signed(uid, "ra_zero", reward_amount="0")).status_code == 200
-    assert _coin_balance(client, token) == 250 + AD_REWARD_COIN
-
-    # 超上限 → 夹紧到 MAX_AD_REWARD_COIN
-    assert _callback(client, _signed(uid, "ra_big", reward_amount="999999")).status_code == 200
-    assert _coin_balance(client, token) == 250 + AD_REWARD_COIN + MAX_AD_REWARD_COIN
+    assert _callback(client, _signed(uid, "ecpm_missing")).status_code == 200
+    assert _coin_balance(client, token) == 0
+    st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
+    assert st["used_today"] == 0
 
 
 def test_callback_idempotent(client) -> None:
@@ -100,11 +91,11 @@ def test_callback_idempotent(client) -> None:
     token = _login(client, phone)
     uid = _user_id(phone)
 
-    p = _signed(uid, "trans_dup")
+    p = _signed(uid, "trans_dup", ecpm="200")
     assert _callback(client, p).status_code == 200
     assert _callback(client, p).status_code == 200  # 重试
 
-    assert _coin_balance(client, token) == AD_REWARD_COIN  # 没翻倍
+    assert _coin_balance(client, token) == calculate_ad_reward_coin("200", 1)  # 没翻倍
     st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
     assert st["used_today"] == 1
 
@@ -134,15 +125,15 @@ def test_daily_count_cap(client, monkeypatch) -> None:
     uid = _user_id(phone)
 
     for i in range(3):
-        r = _callback(client, _signed(uid, f"trans_cap_{i}"))
+        r = _callback(client, _signed(uid, f"trans_cap_{i}", ecpm="200"))
         assert r.status_code == 200, r.text
 
     # 第 4 次:capped,仍 is_verify(不让穿山甲重试),但不加币
-    r = _callback(client, _signed(uid, "trans_cap_over"))
+    r = _callback(client, _signed(uid, "trans_cap_over", ecpm="200"))
     assert r.status_code == 200
     assert r.json() == {"is_verify": True, "reason": 0}
 
-    assert _coin_balance(client, token) == 3 * AD_REWARD_COIN
+    assert _coin_balance(client, token) == sum(calculate_ad_reward_coin("200", i) for i in range(1, 4))
     st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
     assert st["used_today"] == 3
     assert st["remaining"] == 0
@@ -169,7 +160,7 @@ def test_daily_watch_time_cap(client, monkeypatch) -> None:
 
     # 此时回调发奖 → 时长闸命中 → capped(is_verify 仍 true,不让重试),不加金币
     before = _coin_balance(client, token)
-    r = _callback(client, _signed(uid, "trans_time_cap"))
+    r = _callback(client, _signed(uid, "trans_time_cap", ecpm="200"))
     assert r.status_code == 200
     assert r.json() == {"is_verify": True, "reason": 0}
     assert _coin_balance(client, token) == before  # 未加币
@@ -209,7 +200,7 @@ def test_reward_status_mid_round(client) -> None:
     phone = "13800003102"
     token = _login(client, phone)
     uid = _user_id(phone)
-    _callback(client, _signed(uid, "trans_mid_1"))
+    _callback(client, _signed(uid, "trans_mid_1", ecpm="200"))
     st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
     assert st["used_today"] == 1
     assert st["round_count"] == 0
@@ -224,7 +215,7 @@ def test_reward_status_round_complete_enters_cooldown(client) -> None:
     token = _login(client, phone)
     uid = _user_id(phone)
     for i in range(VIDEO_ROUND_REQUIRED_COUNT):
-        _callback(client, _signed(uid, f"trans_rc_{i}"))
+        _callback(client, _signed(uid, f"trans_rc_{i}", ecpm="200"))
 
     st = client.get("/api/v1/ad/reward-status", headers=_auth(token)).json()
     assert st["used_today"] == VIDEO_ROUND_REQUIRED_COUNT
@@ -248,7 +239,7 @@ def test_reward_status_cooldown_expired(client) -> None:
     token = _login(client, phone)
     uid = _user_id(phone)
     for i in range(VIDEO_ROUND_REQUIRED_COUNT):
-        _callback(client, _signed(uid, f"trans_exp_{i}"))
+        _callback(client, _signed(uid, f"trans_exp_{i}", ecpm="200"))
 
     # 把当日所有 granted 记录的 created_at 推到 11 分钟前(覆盖冷却末尾那条)
     db = SessionLocal()

tests/test_invite.py

@@ -0,0 +1,160 @@
+"""好友邀请测试:邀请码、绑定、双方发金币、幂等、自邀/无效码屏蔽。
+
+用 sms mock 登录拿 token(同 test_welfare),再跑邀请闭环。
+"""
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+from app.core.rewards import (
+    INVITE_INVITEE_COINS,
+    INVITE_INVITER_COINS,
+    INVITE_NEW_USER_WINDOW_HOURS,
+)
+from app.db.session import SessionLocal
+from app.repositories.user import get_user_by_phone
+
+
+def _login(client, phone: str) -> str:
+    client.post("/api/v1/auth/sms/send", json={"phone": phone})
+    r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
+    assert r.status_code == 200, r.text
+    return r.json()["access_token"]
+
+
+def _auth(token: str) -> dict[str, str]:
+    return {"Authorization": f"Bearer {token}"}
+
+
+def _coin_balance(client, token: str) -> int:
+    r = client.get("/api/v1/wallet/account", headers=_auth(token))
+    assert r.status_code == 200, r.text
+    return r.json()["coin_balance"]
+
+
+def _my_code(client, token: str) -> str:
+    r = client.get("/api/v1/invite/me", headers=_auth(token))
+    assert r.status_code == 200, r.text
+    return r.json()["invite_code"]
+
+
+def test_invite_me_returns_stable_code(client) -> None:
+    """/invite/me 返回邀请码 + 分享链接;同一用户多次取码稳定。"""
+    token = _login(client, "13800002001")
+    r = client.get("/api/v1/invite/me", headers=_auth(token))
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["invite_code"]                      # 非空
+    assert f"ref={body['invite_code']}" in body["share_url"]
+    assert body["invited_count"] == 0
+    assert body["coins_earned"] == 0
+    # 再取一次,码不变(懒生成 + 持久化)
+    assert _my_code(client, token) == body["invite_code"]
+
+
+def test_bind_flow_both_get_coins(client) -> None:
+    """B 用 A 的码绑定 → 双方各得 1 万金币;A 战绩 +1。"""
+    a = _login(client, "13800002002")
+    b = _login(client, "13800002003")
+    a_code = _my_code(client, a)
+
+    assert _coin_balance(client, a) == 0
+    assert _coin_balance(client, b) == 0
+
+    r = client.post("/api/v1/invite/bind", json={"invite_code": a_code}, headers=_auth(b))
+    assert r.status_code == 200, r.text
+    res = r.json()
+    assert res["status"] == "success"
+    assert res["coins_awarded"] == INVITE_INVITEE_COINS
+
+    # 双方金币到账
+    assert _coin_balance(client, b) == INVITE_INVITEE_COINS
+    assert _coin_balance(client, a) == INVITE_INVITER_COINS
+
+    # A 的战绩:已邀 1 人,累计获得 = 邀请人那份
+    r = client.get("/api/v1/invite/me", headers=_auth(a))
+    assert r.json()["invited_count"] == 1
+    assert r.json()["coins_earned"] == INVITE_INVITER_COINS
+
+
+def test_bind_idempotent_no_double_reward(client) -> None:
+    """同一被邀请人二次绑定 → already_bound,不重复发奖。"""
+    a = _login(client, "13800002004")
+    b = _login(client, "13800002005")
+    a_code = _my_code(client, a)
+
+    r1 = client.post("/api/v1/invite/bind", json={"invite_code": a_code}, headers=_auth(b))
+    assert r1.json()["status"] == "success"
+    bal_b = _coin_balance(client, b)
+    bal_a = _coin_balance(client, a)
+
+    # 再绑一次(甚至换个码也不行,已被绑过)
+    c = _login(client, "13800002006")
+    c_code = _my_code(client, c)
+    r2 = client.post("/api/v1/invite/bind", json={"invite_code": c_code}, headers=_auth(b))
+    assert r2.json()["status"] == "already_bound"
+    assert r2.json()["coins_awarded"] == 0
+
+    # 余额没变(没二次发奖),C 也没拿到邀请奖励
+    assert _coin_balance(client, b) == bal_b
+    assert _coin_balance(client, a) == bal_a
+    assert _coin_balance(client, c) == 0
+
+
+def test_self_invite_blocked(client) -> None:
+    """填自己的邀请码 → self_invite,不发奖。"""
+    a = _login(client, "13800002007")
+    a_code = _my_code(client, a)
+    r = client.post("/api/v1/invite/bind", json={"invite_code": a_code}, headers=_auth(a))
+    assert r.status_code == 200, r.text
+    assert r.json()["status"] == "self_invite"
+    assert _coin_balance(client, a) == 0
+
+
+def test_invalid_code(client) -> None:
+    """无效邀请码 → invalid_code,不发奖。"""
+    b = _login(client, "13800002008")
+    r = client.post("/api/v1/invite/bind", json={"invite_code": "ZZZZZZ"}, headers=_auth(b))
+    assert r.status_code == 200, r.text
+    assert r.json()["status"] == "invalid_code"
+    assert _coin_balance(client, b) == 0
+
+
+def test_manual_channel_recorded(client) -> None:
+    """channel=manual 也能绑定(手动填码兜底路径)。"""
+    a = _login(client, "13800002009")
+    b = _login(client, "13800002010")
+    a_code = _my_code(client, a)
+    r = client.post(
+        "/api/v1/invite/bind",
+        json={"invite_code": a_code, "channel": "manual"},
+        headers=_auth(b),
+    )
+    assert r.status_code == 200, r.text
+    assert r.json()["status"] == "success"
+
+
+def test_invite_requires_auth(client) -> None:
+    """不带 token → 401。"""
+    assert client.get("/api/v1/invite/me").status_code == 401
+    assert client.post("/api/v1/invite/bind", json={"invite_code": "ABCDEF"}).status_code == 401
+
+
+def test_old_user_not_eligible(client) -> None:
+    """新用户闸:被邀请人注册超过窗口 → not_eligible,双方都不发奖。"""
+    a = _login(client, "13800002030")
+    a_code = _my_code(client, a)
+    b = _login(client, "13800002031")
+    # 把 b 的注册时间改成超出窗口 → 变"老用户"
+    with SessionLocal() as db:
+        u = get_user_by_phone(db, "13800002031")
+        assert u is not None
+        u.created_at = datetime.now(timezone.utc) - timedelta(hours=INVITE_NEW_USER_WINDOW_HOURS + 1)
+        db.commit()
+
+    r = client.post("/api/v1/invite/bind", json={"invite_code": a_code}, headers=_auth(b))
+    assert r.status_code == 200, r.text
+    assert r.json()["status"] == "not_eligible"
+    assert r.json()["coins_awarded"] == 0
+    assert _coin_balance(client, b) == 0
+    assert _coin_balance(client, a) == 0

tests/test_welfare.py

@@ -4,15 +4,20 @@
 """
 from __future__ import annotations
 
+import json
+
+from app.core.config import settings
 from app.core.rewards import (
     COIN_PER_CENT,
     COIN_PER_YUAN,
     MIN_EXCHANGE_COIN,
+    SIGNIN_BOOST_COIN,
     SIGNIN_REWARDS,
     TASK_ENABLE_NOTIFICATION,
     TASK_REWARDS,
 )
 from app.db.session import SessionLocal
+from app.integrations import pangle
 from app.repositories import wallet as crud_wallet
 from app.repositories.user import get_user_by_phone
 
@@ -29,6 +34,12 @@ def _auth(token: str) -> dict[str, str]:
     return {"Authorization": f"Bearer {token}"}
 
 
+def _signed_ad_callback(user_id: int, trans_id: str, **params: str) -> dict[str, str]:
+    body = {"user_id": str(user_id), "trans_id": trans_id, **params}
+    body["sign"] = pangle.build_sign(trans_id, settings.PANGLE_REWARD_SECRET)
+    return body
+
+
 def test_account_auto_created_empty(client) -> None:
     """首次访问账户:自动建空账户,余额全 0。"""
     token = _login(client, "13800001001")
@@ -91,8 +102,9 @@ def test_signin_flow(client) -> None:
 
 
 def test_signin_boost_flow(client) -> None:
-    """签到后看广告膨胀 → 补发一笔等额签到金币,每天只能膨胀一次。"""
-    token = _login(client, "13800001011")
+    """签到后看广告膨胀 → S2S 固定补发 2000 金币,每天只能膨胀一次。"""
+    phone = "13800001011"
+    token = _login(client, phone)
 
     r = client.post("/api/v1/signin/boost", json={}, headers=_auth(token))
     assert r.status_code == 409
@@ -101,14 +113,30 @@ def test_signin_boost_flow(client) -> None:
     assert r.status_code == 200, r.text
     first_coin = r.json()["coin_awarded"]
 
-    r = client.post("/api/v1/signin/boost", json={"ad_ref_id": "ad-session-1"}, headers=_auth(token))
+    with SessionLocal() as db:
+        user = get_user_by_phone(db, phone)
+        assert user is not None
+        uid = user.id
+
+    extra = json.dumps({"reward_scene": "signin_boost", "ad_session_id": "signin-session-1"})
+    r = client.get(
+        "/api/v1/ad/pangle-callback",
+        params=_signed_ad_callback(uid, "signin-boost-trans-1", extra=extra, ecpm="200"),
+    )
+    assert r.status_code == 200, r.text
+    assert r.json() == {"is_verify": True, "reason": 0}
+
+    r = client.post("/api/v1/signin/boost", json={"ad_ref_id": "signin-boost-trans-1"}, headers=_auth(token))
     assert r.status_code == 200, r.text
     body = r.json()
-    assert body["coin_awarded"] == first_coin
-    assert body["coin_balance"] == first_coin * 2
+    assert body["coin_awarded"] == SIGNIN_BOOST_COIN
+    assert body["coin_balance"] == first_coin + SIGNIN_BOOST_COIN
 
-    r = client.post("/api/v1/signin/boost", json={"ad_ref_id": "ad-session-1"}, headers=_auth(token))
-    assert r.status_code == 409
+    r = client.get(
+        "/api/v1/ad/pangle-callback",
+        params=_signed_ad_callback(uid, "signin-boost-trans-2", extra=extra, ecpm="200"),
+    )
+    assert r.status_code == 200, r.text
 
     r = client.get("/api/v1/wallet/coin-transactions", headers=_auth(token))
     types = [t["biz_type"] for t in r.json()["items"]]