perf(admin): aggregate record duration percentiles in PostgreSQL

.env.example

@@ -113,13 +113,6 @@ JD_UNION_APP_SECRET=
 JD_UNION_SITE_ID=
 JD_UNION_AUTH_KEY=
 
-# 美团 + 京东订单每天北京时间 05:00 自动对账；按更新时间回拉近 3 天，重叠防漏单并刷新状态。
-# 手动对账按钮不受该开关影响。通常保持开启；临时停自动任务时设为 false。
-CPS_AUTO_RECONCILE_ENABLED=true
-CPS_AUTO_RECONCILE_RUN_HOUR=5
-CPS_AUTO_RECONCILE_LOOKBACK_DAYS=3
-CPS_AUTO_RECONCILE_CHECK_INTERVAL_SEC=60
-
 # ===== Pricebot 上游 (领券/比价业务透传目标) =====
 # 客户端调本服务的 /api/v1/coupon/step 等,我们透传到 pricebot-backend。
 # 本地开发用 localhost:8000。生产部署改成内网地址(如 http://pricebot.internal:8000)。

alembic/versions/8e04cc13a211_merge_notification_comparison_user_idx_.py

@@ -1,26 +0,0 @@
-"""merge notification/comparison_user_idx/monitoring_audit heads
-
-Revision ID: 8e04cc13a211
-Revises: comparison_user_created_idx, monitoring_audit_rbac, notification_table
-Create Date: 2026-07-23 15:37:26.967540
-
-"""
-from typing import Sequence, Union
-
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision: str = '8e04cc13a211'
-down_revision: Union[str, Sequence[str], None] = ('comparison_user_created_idx', 'monitoring_audit_rbac', 'notification_table')
-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/coupon_claim_event.py

@@ -1,93 +0,0 @@
-"""add per-session coupon claim event table
-
-Revision ID: coupon_claim_event
-Revises: 8e04cc13a211
-Create Date: 2026-07-23
-"""
-
-from collections.abc import Sequence
-
-import sqlalchemy as sa
-from sqlalchemy.dialects import postgresql
-
-from alembic import op
-
-revision: str = "coupon_claim_event"
-down_revision: str | Sequence[str] | None = "8e04cc13a211"
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
-
-_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
-
-
-def upgrade() -> None:
-    op.create_table(
-        "coupon_claim_event",
-        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
-        sa.Column("trace_id", sa.String(length=64), 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("app_env", sa.String(length=16), nullable=True),
-        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("reason", sa.String(length=255), nullable=True),
-        sa.Column("extra", _JSON, nullable=True),
-        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
-        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
-        sa.PrimaryKeyConstraint("id"),
-        sa.UniqueConstraint(
-            "trace_id", "coupon_id",
-            name="uq_coupon_claim_event_trace_coupon",
-        ),
-    )
-    op.create_index(
-        "ix_coupon_claim_event_date_env",
-        "coupon_claim_event",
-        ["claim_date", "app_env"],
-        unique=False,
-    )
-    op.create_index(
-        op.f("ix_coupon_claim_event_app_env"),
-        "coupon_claim_event",
-        ["app_env"],
-        unique=False,
-    )
-    op.create_index(
-        op.f("ix_coupon_claim_event_trace_id"),
-        "coupon_claim_event",
-        ["trace_id"],
-        unique=False,
-    )
-    op.create_index(
-        op.f("ix_coupon_claim_event_user_id"),
-        "coupon_claim_event",
-        ["user_id"],
-        unique=False,
-    )
-
-    # 旧表只能回填当前仍保留的 trace；历史上已被每日去重覆盖的关联无法恢复。
-    op.execute(
-        """
-        INSERT INTO coupon_claim_event (
-            trace_id, device_id, user_id, coupon_id, claim_date, status, app_env,
-            vendor, coupon_name, claimed_count, reason, extra, created_at, updated_at
-        )
-        SELECT
-            trace_id, device_id, user_id, coupon_id, claim_date, status, app_env,
-            vendor, coupon_name, claimed_count, reason, extra, created_at, updated_at
-        FROM coupon_claim_record
-        WHERE trace_id IS NOT NULL
-        """
-    )
-
-
-def downgrade() -> None:
-    op.drop_index(op.f("ix_coupon_claim_event_user_id"), table_name="coupon_claim_event")
-    op.drop_index(op.f("ix_coupon_claim_event_trace_id"), table_name="coupon_claim_event")
-    op.drop_index(op.f("ix_coupon_claim_event_app_env"), table_name="coupon_claim_event")
-    op.drop_index("ix_coupon_claim_event_date_env", table_name="coupon_claim_event")
-    op.drop_table("coupon_claim_event")

alembic/versions/d8dd2106e438_merge_guide_video_seq_uq_and_coupon_.py

@@ -1,26 +0,0 @@
-"""merge guide_video seq_uq and coupon_claim_event heads
-
-Revision ID: d8dd2106e438
-Revises: coupon_claim_event, guide_video_user_seq_uq
-Create Date: 2026-07-24 11:52:18.290731
-
-"""
-from typing import Sequence, Union
-
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision: str = 'd8dd2106e438'
-down_revision: Union[str, Sequence[str], None] = ('coupon_claim_event', 'guide_video_user_seq_uq')
-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/d9c03cc3ea07_merge_guide_video_and_main_alembic_heads.py

@@ -1,26 +0,0 @@
-"""merge guide_video and main alembic heads
-
-Revision ID: d9c03cc3ea07
-Revises: 8e04cc13a211, guide_video_play_table
-Create Date: 2026-07-23 22:57:40.998161
-
-"""
-from typing import Sequence, Union
-
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision: str = 'd9c03cc3ea07'
-down_revision: Union[str, Sequence[str], None] = ('8e04cc13a211', 'guide_video_play_table')
-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/guide_video_play_table.py

@@ -1,50 +0,0 @@
-"""新手引导视频播放记录表(领券浮层前 N 次替代广告)
-
-见 app/models/guide_video.py:按账号计次(开播即计数)、play_token 幂等发币。
-配置(开关 / 视频地址 / 次数 / 金币)复用既有 app_config 表,无需建表。
-
-Revision ID: guide_video_play_table
-Revises: meituan_coupon_feed_indexes
-Create Date: 2026-07-23 12:00:00.000000
-
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-
-# revision identifiers, used by Alembic.
-revision: str = "guide_video_play_table"
-down_revision: Union[str, Sequence[str], None] = "meituan_coupon_feed_indexes"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.create_table(
-        "guide_video_play",
-        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
-        sa.Column("user_id", sa.Integer(), nullable=False),
-        sa.Column("play_token", sa.String(length=64), nullable=False),
-        sa.Column("scene", sa.String(length=16), nullable=False, server_default="coupon"),
-        sa.Column("seq", sa.Integer(), nullable=False, server_default="1"),
-        sa.Column("video_url", sa.String(length=512), nullable=True),
-        sa.Column("coin", sa.Integer(), nullable=False, server_default="0"),
-        sa.Column("status", sa.String(length=16), nullable=False, server_default="playing"),
-        sa.Column("completed", sa.Integer(), nullable=False, server_default="0"),
-        sa.Column(
-            "started_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
-        ),
-        sa.Column("granted_at", sa.DateTime(timezone=True), nullable=True),
-        sa.ForeignKeyConstraint(["user_id"], ["user.id"]),
-        sa.PrimaryKeyConstraint("id"),
-        sa.UniqueConstraint("play_token", name="uq_guide_video_play_token"),
-    )
-    op.create_index("ix_guide_video_play_user_id", "guide_video_play", ["user_id"])
-    op.create_index("ix_guide_video_play_started_at", "guide_video_play", ["started_at"])
-
-
-def downgrade() -> None:
-    op.drop_index("ix_guide_video_play_started_at", table_name="guide_video_play")
-    op.drop_index("ix_guide_video_play_user_id", table_name="guide_video_play")
-    op.drop_table("guide_video_play")

alembic/versions/guide_video_play_user_seq_unique.py

@@ -1,40 +0,0 @@
-"""guide_video_play 加 (user_id, seq) 唯一约束:堵住并发 /start 绕过次数上限
-
-start_play 是无锁 check-then-insert(读 COUNT(*) 算 seq=used+1 再插一行),N 个并发
-/start 会都读到同一个 used、算出同一个 seq、各插一行拿到各自的 play_token,于是 3 次
-上限被绕过、每个 token 都能换 120 金币。加唯一键后并发同 seq 必撞,start_play 捕获
-IntegrityError 降级为 should_play=false(客户端照旧放广告)。
-
-用 unique index 而不是 batch_alter_table 加 UniqueConstraint:SQLite 加约束要整表重建,
-而 CREATE UNIQUE INDEX 两边都原生支持,回滚也干净。
-
-注:若库里已有并发产生的重复 (user_id, seq),建索引会失败 —— 本功能尚未上线,表通常是空的;
-真撞上了先按 seq 去重(留 id 最小的一行,多发的金币按 scripts/reset_guide_video.py 的口径退)。
-
-Revision ID: guide_video_user_seq_uq
-Revises: d9c03cc3ea07
-Create Date: 2026-07-24 10:00:00.000000
-
-"""
-from typing import Sequence, Union
-
-from alembic import op
-
-# revision identifiers, used by Alembic.
-revision: str = "guide_video_user_seq_uq"
-down_revision: Union[str, Sequence[str], None] = "d9c03cc3ea07"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.create_index(
-        "uq_guide_video_play_user_seq",
-        "guide_video_play",
-        ["user_id", "seq"],
-        unique=True,
-    )
-
-
-def downgrade() -> None:
-    op.drop_index("uq_guide_video_play_user_seq", table_name="guide_video_play")

alembic/versions/meituan_coupon_feed_indexes.py

@@ -1,52 +0,0 @@
-"""meituan_coupon 首页 feed 分页复合索引(销量最高 / 智能推荐)
-
-「销量最高」「智能推荐」两个 tab 都是
-    WHERE city_id = ? [+ 过滤]  →  DISTINCT ON (dedup_key) ORDER BY dedup_key, <排序键> DESC
-的形状。列顺序对齐后 Postgres 可以顺着索引流式去重,免掉「每翻一页就把该城全部券重排一遍」,
-这是首页下滑到底越来越慢的根因之一(另一半在 app 层:见 api/v1/meituan.py 的 _paged_dedup_ids)。
-
-⚠️ 本文件同时是一个 **merge 迁移**:主干此前有 3 个并行 head
-(comparison_user_created_idx / monitoring_audit_rbac / notification_table),
-`alembic upgrade head` 会因 multiple heads 报错。这里一并收敛回单 head。
-
-Revision ID: meituan_coupon_feed_indexes
-Revises: comparison_user_created_idx, monitoring_audit_rbac, notification_table
-Create Date: 2026-07-23 10:00:00.000000
-
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-
-# revision identifiers, used by Alembic.
-revision: str = "meituan_coupon_feed_indexes"
-down_revision: Union[str, Sequence[str], None] = (
-    "comparison_user_created_idx",
-    "monitoring_audit_rbac",
-    "notification_table",
-)
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    # 销量最高:WHERE city_id=? AND sale_volume_num IS NOT NULL
-    #           ORDER BY dedup_key, sale_volume_num DESC, commission_percent DESC
-    op.create_index(
-        "ix_meituan_coupon_city_dedup_sales",
-        "meituan_coupon",
-        ["city_id", "dedup_key", sa.text("sale_volume_num DESC"), sa.text("commission_percent DESC")],
-    )
-    # 智能推荐:WHERE city_id=? AND commission_percent>=3.0
-    #           ORDER BY dedup_key, commission_percent DESC
-    op.create_index(
-        "ix_meituan_coupon_city_dedup_comm",
-        "meituan_coupon",
-        ["city_id", "dedup_key", sa.text("commission_percent DESC")],
-    )
-
-
-def downgrade() -> None:
-    op.drop_index("ix_meituan_coupon_city_dedup_comm", table_name="meituan_coupon")
-    op.drop_index("ix_meituan_coupon_city_dedup_sales", table_name="meituan_coupon")

app/admin/main.py

@@ -30,7 +30,6 @@ from app.admin.routers.analytics_health import router as analytics_health_router
 from app.admin.routers.event_logs import router as event_logs_router
 from app.admin.routers.feedback import router as feedback_router
 from app.admin.routers.feedback_qr import router as feedback_qr_router
-from app.admin.routers.guide_video import router as guide_video_router
 from app.admin.routers.huawei_review import router as huawei_review_router
 from app.admin.routers.onboarding import router as onboarding_router
 from app.admin.routers.ops_marquee_seed import router as ops_marquee_seed_router
@@ -41,7 +40,6 @@ from app.admin.routers.wallet import router as wallet_router
 from app.admin.routers.withdraw import router as withdraw_router
 from app.core.config import settings
 from app.core.logging import setup_logging
-from app.integrations import meituan as mt_meituan
 
 setup_logging(debug=settings.APP_DEBUG)
 logger = logging.getLogger("shagua.admin")
@@ -55,8 +53,6 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
         settings.DATABASE_URL.split("://", 1)[0],
     )
     yield
-    # CPS 后台页会打美团(routers/cps.py),那条共享 client 若被建过要在这里关掉连接池
-    mt_meituan.close_client()
     logger.info("admin app shutting down")
 
 
@@ -105,7 +101,6 @@ admin_app.include_router(feedback_router)
 admin_app.include_router(event_logs_router)
 admin_app.include_router(analytics_health_router)
 admin_app.include_router(feedback_qr_router)
-admin_app.include_router(guide_video_router)
 admin_app.include_router(admins_router)
 admin_app.include_router(roles_router)
 admin_app.include_router(audit_router)

app/admin/repositories/coupon_data.py

@@ -16,7 +16,7 @@ from sqlalchemy import case, func, or_, select
 from sqlalchemy.orm import Session
 
 from app.core import rewards
-from app.models.coupon_state import CouponClaimEvent, CouponClaimRecord, CouponSession
+from app.models.coupon_state import CouponClaimRecord, CouponSession
 from app.models.user import User
 from app.repositories import ad_ecpm as crud_ecpm
 from app.repositories.coupon_state import DEFAULT_PLATFORMS, coupon_id_to_platform
@@ -193,18 +193,18 @@ def _point_scores_by_trace(db: Session, trace_ids: list[str]) -> dict[str, dict[
     """聚合查询批量返回逐场点位分数，不加载逐券明细。"""
     if not trace_ids:
         return {}
-    succeeded = func.sum(case((CouponClaimEvent.status.in_(_SLOT_OK), 1), else_=0))
+    succeeded = func.sum(case((CouponClaimRecord.status.in_(_SLOT_OK), 1), else_=0))
     rows = db.execute(
         select(
-            CouponClaimEvent.trace_id,
+            CouponClaimRecord.trace_id,
             succeeded.label("succeeded"),
             func.count().label("tried"),
         )
         .where(
-            CouponClaimEvent.trace_id.in_(trace_ids),
-            CouponClaimEvent.status.in_(_SLOT_TRIED),
+            CouponClaimRecord.trace_id.in_(trace_ids),
+            CouponClaimRecord.status.in_(_SLOT_TRIED),
         )
-        .group_by(CouponClaimEvent.trace_id)
+        .group_by(CouponClaimRecord.trace_id)
     ).all()
     return {
         trace_id: {"succeeded": int(success_count or 0), "tried": int(tried or 0)}
@@ -217,13 +217,13 @@ def coupon_point_details(db: Session, *, trace_id: str) -> list[dict]:
     """按单个 trace 查询逐券结果；仅在后台用户点击分数时调用。"""
     rows = db.execute(
         select(
-            CouponClaimEvent.coupon_id,
-            CouponClaimEvent.coupon_name,
-            CouponClaimEvent.status,
-            CouponClaimEvent.reason,
+            CouponClaimRecord.coupon_id,
+            CouponClaimRecord.coupon_name,
+            CouponClaimRecord.status,
+            CouponClaimRecord.reason,
         )
-        .where(CouponClaimEvent.trace_id == trace_id)
-        .order_by(CouponClaimEvent.id)
+        .where(CouponClaimRecord.trace_id == trace_id)
+        .order_by(CouponClaimRecord.id)
     ).all()
     return [
         {

app/admin/repositories/cps.py

@@ -5,7 +5,6 @@
 """
 from __future__ import annotations
 
-import logging
 from datetime import datetime, timedelta, timezone
 from decimal import Decimal, InvalidOperation
 from typing import Any
@@ -16,14 +15,12 @@ from sqlalchemy.orm import Session
 
 from app.admin.repositories.queries import _as_utc, offset_paginate
 from app.integrations import jd_union, meituan
+from app.repositories import cps_link as cps_link_repo
 from app.models.cps_activity import CpsActivity
 from app.models.cps_group import CpsGroup
 from app.models.cps_link import CpsClick, CpsLink
 from app.models.cps_order import CpsOrder
 from app.models.cps_wx_user import CpsWxUser
-from app.repositories import cps_link as cps_link_repo
-
-logger = logging.getLogger("shagua.cps_reconcile")
 
 # 美团订单状态:取消(4)/风控(5)不计佣金;结算(6)为佣金真正到账
 _INVALID_STATUS = {"4", "5"}
@@ -381,35 +378,18 @@ def effective_commission_cents(order: CpsOrder) -> int:
 def reconcile_orders(
     db: Session, *, start_time: int, end_time: int,
     query_time_type: int = 1, sid: str | None = None, max_pages: int = 200,
-    audit_context: dict[str, Any] | None = None,
 ) -> dict:
     """调美团 query_order 分页拉单 → 按 order_id upsert。返回 {fetched, inserted, updated, pages}。
 
     订单状态会随时间变(付款→完成→结算/退款),重复拉同一单则更新。max_pages 防异常死循环。
     """
-    fetched = inserted = updated = pages = api_requests = 0
+    fetched = inserted = updated = pages = 0
     page = 1
     while page <= max_pages:
-        api_requests += 1
-        try:
-            resp = meituan.query_order(
-                sid=sid, start_time=start_time, end_time=end_time,
-                query_time_type=query_time_type, page=page, limit=100,
-            )
-        except Exception:  # noqa: BLE001 - 记录失败页后保持原异常类型继续抛出
-            if audit_context is not None:
-                logger.exception(
-                    "CPS reconcile upstream request failed platform=meituan page=%s",
-                    page,
-                    extra={
-                        **audit_context,
-                        "event": "cps_reconcile.request_failed",
-                        "platform": "meituan",
-                        "failed_page": page,
-                        "api_request_number": api_requests,
-                    },
-                )
-            raise
+        resp = meituan.query_order(
+            sid=sid, start_time=start_time, end_time=end_time,
+            query_time_type=query_time_type, page=page, limit=100,
+        )
         rows = ((resp.get("data") or {}).get("dataList")) or []
         if not rows:
             break
@@ -439,58 +419,30 @@ def reconcile_orders(
             break
         page += 1
     db.commit()
-    return {
-        "fetched": fetched,
-        "inserted": inserted,
-        "updated": updated,
-        "pages": pages,
-        "api_requests": api_requests,
-    }
+    return {"fetched": fetched, "inserted": inserted, "updated": updated, "pages": pages}
 
 
 def reconcile_jd_orders(
     db: Session, *, start_time: datetime, end_time: datetime,
     query_time_type: int = 3, max_pages: int = 100,
-    audit_context: dict[str, Any] | None = None,
 ) -> dict:
     """调京东 order.row.query 拉单 → 按订单行 upsert。
 
     京东单次查询窗口最多 1 小时,这里按北京自然时间切窗并逐页拉取。
     """
-    fetched = inserted = updated = pages = api_requests = windows = 0
+    fetched = inserted = updated = pages = 0
     cur = start_time
     while cur < end_time:
         win_end = min(cur + timedelta(hours=1), end_time)
-        windows += 1
         page = 1
         while page <= max_pages:
-            api_requests += 1
-            try:
-                resp = jd_union.query_order_rows(
-                    start_time=cur,
-                    end_time=win_end,
-                    query_time_type=query_time_type,
-                    page_index=page,
-                    page_size=200,
-                )
-            except Exception:  # noqa: BLE001 - 记录失败窗口后保持原异常类型继续抛出
-                if audit_context is not None:
-                    logger.exception(
-                        "CPS reconcile upstream request failed platform=jd window=%s..%s page=%s",
-                        cur.isoformat(),
-                        win_end.isoformat(),
-                        page,
-                        extra={
-                            **audit_context,
-                            "event": "cps_reconcile.request_failed",
-                            "platform": "jd",
-                            "failed_window_start": cur.isoformat(),
-                            "failed_window_end": win_end.isoformat(),
-                            "failed_page": page,
-                            "api_request_number": api_requests,
-                        },
-                    )
-                raise
+            resp = jd_union.query_order_rows(
+                start_time=cur,
+                end_time=win_end,
+                query_time_type=query_time_type,
+                page_index=page,
+                page_size=200,
+            )
             rows = resp.get("rows") or []
             has_more = bool(resp.get("has_more"))
             if not rows:
@@ -517,14 +469,7 @@ def reconcile_jd_orders(
             page += 1
         cur = win_end
     db.commit()
-    return {
-        "fetched": fetched,
-        "inserted": inserted,
-        "updated": updated,
-        "pages": pages,
-        "api_requests": api_requests,
-        "windows": windows,
-    }
+    return {"fetched": fetched, "inserted": inserted, "updated": updated, "pages": pages}
 
 
 def list_orders(

app/admin/repositories/queries.py

@@ -1171,30 +1171,24 @@ def user_reward_stats(
     acc = db.get(CoinAccount, user_id)  # 现金余额:当前快照,不随窗口
     cash_balance = acc.cash_balance_cents if acc else 0
 
-    # 只投影本统计实际使用的列。避免滚动发布或旧本地库尚未补齐无关新列时，
-    # SQLAlchemy 因 select(ORM) 自动展开整表字段而让提现详情整体 500。
-    rv = db.execute(
-        select(AdRewardRecord.ecpm_raw, AdRewardRecord.coin).where(
+    rv = list(db.execute(
+        select(AdRewardRecord).where(
             AdRewardRecord.user_id == user_id,
             AdRewardRecord.reward_scene == "reward_video",
             AdRewardRecord.status == "granted",
             *_window_conds(AdRewardRecord.created_at, date_from, date_to),
         )
-    ).all()
+    ).scalars())
     rv_ecpms = [rewards.parse_ecpm_fen(r.ecpm_raw) for r in rv if r.ecpm_raw]
     rv_coins = sum(r.coin for r in rv)
 
-    feed = db.execute(
-        select(
-            AdFeedRewardRecord.unit_count,
-            AdFeedRewardRecord.ecpm_raw,
-            AdFeedRewardRecord.coin,
-        ).where(
+    feed = list(db.execute(
+        select(AdFeedRewardRecord).where(
             AdFeedRewardRecord.user_id == user_id,
             AdFeedRewardRecord.status == "granted",
             *_window_conds(AdFeedRewardRecord.created_at, date_from, date_to),
         )
-    ).all()
+    ).scalars())
     feed_ecpms = [rewards.parse_ecpm_fen(f.ecpm_raw) for f in feed if f.ecpm_raw]
     feed_coins = sum(f.coin for f in feed)
 
@@ -1252,14 +1246,8 @@ def user_coin_records(
     signin_from = date_from + timedelta(hours=8) if date_from is not None else None
     signin_to = date_to + timedelta(hours=8) if date_to is not None else None
 
-    # 三类来源都只取页面需要的列，避免无关 ORM 新列造成旧库查询失败。
     for rec in db.execute(
-        select(
-            AdRewardRecord.reward_scene,
-            AdRewardRecord.created_at,
-            AdRewardRecord.ecpm_raw,
-            AdRewardRecord.coin,
-        )
+        select(AdRewardRecord)
         .where(
             AdRewardRecord.user_id == user_id,
             AdRewardRecord.status == "granted",
@@ -1267,7 +1255,7 @@ def user_coin_records(
         )
         .order_by(AdRewardRecord.created_at.desc())
         .limit(fetch)
-    ).all():
+    ).scalars():
         is_video = rec.reward_scene == "reward_video"
         rows.append({
             "source": rec.reward_scene,
@@ -1278,12 +1266,7 @@ def user_coin_records(
         })
 
     for rec in db.execute(
-        select(
-            AdFeedRewardRecord.feed_scene,
-            AdFeedRewardRecord.created_at,
-            AdFeedRewardRecord.ecpm_raw,
-            AdFeedRewardRecord.coin,
-        )
+        select(AdFeedRewardRecord)
         .where(
             AdFeedRewardRecord.user_id == user_id,
             AdFeedRewardRecord.status == "granted",
@@ -1291,7 +1274,7 @@ def user_coin_records(
         )
         .order_by(AdFeedRewardRecord.created_at.desc())
         .limit(fetch)
-    ).all():
+    ).scalars():
         rows.append({
             "source": "feed",
             "source_label": _FEED_SCENE_LABEL.get(rec.feed_scene, "信息流广告"),
@@ -1301,7 +1284,7 @@ def user_coin_records(
         })
 
     for rec in db.execute(
-        select(CoinTransaction.created_at, CoinTransaction.amount)
+        select(CoinTransaction)
         .where(
             CoinTransaction.user_id == user_id,
             CoinTransaction.biz_type == "signin",
@@ -1309,7 +1292,7 @@ def user_coin_records(
         )
         .order_by(CoinTransaction.created_at.desc())
         .limit(fetch)
-    ).all():
+    ).scalars():
         rows.append({
             "source": "signin",
             "source_label": "签到",

app/admin/repositories/stats.py

@@ -30,17 +30,18 @@ from app.models.user import User
 from app.models.wallet import CoinTransaction, WithdrawOrder
 
 _BEIJING = timezone(timedelta(hours=8))
-REWARD_VIDEO_BIZ_TYPES = ("reward_video", "ad_reward", "signin_boost")
+REWARD_VIDEO_BIZ_TYPES = ("reward_video", "ad_reward")
 # 领券/比价奖励金币的真实来源是信息流广告发奖(ad_feed_reward_record,按 feed_scene 分场景);
 # coin_transaction 里只有扁平的 feed_ad_reward、biz_type 不分 coupon/comparison,故这俩桶历史从未
-# 被写入,仅留作未来兜底,实际金额在下方按 feed_scene 汇总 ad_feed_reward_record 得出。
-# reward_video/ad_reward 及历史 signin_boost 均归看视频桶，不再混进领券奖励或常规任务。
+# 被写入,仅留作未来兜底,实际金额在下方按 feed_scene 汇总 ad_feed_reward_record 得出。reward_video/
+# ad_reward 是激励视频,单独成桶、不再混进领券奖励(历史误并会把激励视频金币双计进领券)。
 COUPON_REWARD_BIZ_TYPES = ("coupon", "coupon_reward")
 COMPARISON_REWARD_BIZ_TYPES = ("comparison", "compare_reward", "comparison_reward")
 # 常规任务必须按明确来源相加；不能从全部正向流水反减排除项，否则新增广告/运营
 # biz_type 时会在排除清单更新前自动混入该桶。task_ 前缀在查询处单独覆盖现有及未来任务。
 REGULAR_TASK_EXACT_BIZ_TYPES = (
     "signin",
+    "signin_boost",
     "price_report_reward",
     "feedback_reward",
 )

app/admin/routers/feedback.py

@@ -12,10 +12,6 @@ from app.admin.repositories import mutations, queries
 from app.admin.schemas.common import CursorPage, OkResponse
 from app.admin.schemas.feedback import (
     FeedbackApproveRequest,
-    FeedbackBulkApproveRequest,
-    FeedbackBulkItemResult,
-    FeedbackBulkRejectRequest,
-    FeedbackBulkResult,
     FeedbackOut,
     FeedbackRejectRequest,
     FeedbackSummary,
@@ -37,123 +33,6 @@ def _ensure_pending(fb: Feedback) -> None:
         raise HTTPException(status_code=400, detail="反馈已审核")
 
 
-def _approve_feedback(
-    db: AdminDb,
-    admin: AdminUser,
-    feedback_id: int,
-    payload: FeedbackApproveRequest | FeedbackBulkApproveRequest,
-    ip: str,
-    *,
-    bulk: bool = False,
-) -> FeedbackOut:
-    fb = db.get(Feedback, feedback_id, with_for_update=True)
-    if fb is None:
-        raise HTTPException(status_code=404, detail="反馈不存在")
-    _ensure_pending(fb)
-
-    before = fb.status
-    mutations.review_feedback(
-        db,
-        fb,
-        status="adopted",
-        reward_coins=payload.reward_coins,
-        review_note=payload.note,
-        admin_reply=payload.reply,
-        reviewed_by_admin_id=admin.id,
-        commit=False,
-    )
-    wallet_repo.grant_coins(
-        db,
-        fb.user_id,
-        payload.reward_coins,
-        biz_type="feedback_reward",
-        ref_id=str(fb.id),
-        remark="意见反馈被采纳",
-    )
-    detail = {
-        "before": before,
-        "after": "adopted",
-        "reward_coins": payload.reward_coins,
-        "note": payload.note,
-        "reply": payload.reply,
-    }
-    if bulk:
-        detail["bulk"] = True
-    write_audit(
-        db,
-        admin,
-        action="feedback.approve",
-        target_type="feedback",
-        target_id=feedback_id,
-        detail=detail,
-        ip=ip,
-        commit=False,
-    )
-    db.commit()
-    db.refresh(fb)
-    out = FeedbackOut.model_validate(fb)
-    notification_events.notify_feedback_reward(db, fb)
-    return out
-
-
-def _reject_feedback(
-    db: AdminDb,
-    admin: AdminUser,
-    feedback_id: int,
-    payload: FeedbackRejectRequest | FeedbackBulkRejectRequest,
-    ip: str,
-    *,
-    bulk: bool = False,
-) -> FeedbackOut:
-    fb = db.get(Feedback, feedback_id, with_for_update=True)
-    if fb is None:
-        raise HTTPException(status_code=404, detail="反馈不存在")
-    _ensure_pending(fb)
-
-    before = fb.status
-    mutations.review_feedback(
-        db,
-        fb,
-        status="rejected",
-        reject_reason=payload.reason,
-        review_note=payload.note,
-        admin_reply=payload.reply,
-        reviewed_by_admin_id=admin.id,
-        commit=False,
-    )
-    detail = {
-        "before": before,
-        "after": "rejected",
-        "reason": payload.reason,
-        "note": payload.note,
-        "reply": payload.reply,
-    }
-    if bulk:
-        detail["bulk"] = True
-    write_audit(
-        db,
-        admin,
-        action="feedback.reject",
-        target_type="feedback",
-        target_id=feedback_id,
-        detail=detail,
-        ip=ip,
-        commit=False,
-    )
-    db.commit()
-    db.refresh(fb)
-    out = FeedbackOut.model_validate(fb)
-    notification_events.notify_feedback_reply(db, fb)
-    return out
-
-
-def _bulk_result(items: list[FeedbackBulkItemResult]) -> FeedbackBulkResult:
-    success = sum(1 for item in items if item.ok)
-    return FeedbackBulkResult(
-        total=len(items), success=success, failed=len(items) - success, items=items,
-    )
-
-
 @router.get("", response_model=CursorPage[FeedbackOut], summary="反馈工单列表")
 def list_feedbacks(
     db: AdminDb,
@@ -194,50 +73,6 @@ def feedback_summary(db: AdminDb) -> FeedbackSummary:
     return FeedbackSummary.model_validate(queries.feedback_summary(db))
 
 
-@router.post("/bulk/approve", response_model=FeedbackBulkResult, summary="批量采纳反馈并发金币")
-def bulk_approve_feedbacks(
-    body: FeedbackBulkApproveRequest,
-    request: Request,
-    admin: Annotated[AdminUser, Depends(require_role("operator"))],
-    db: AdminDb,
-) -> FeedbackBulkResult:
-    results: list[FeedbackBulkItemResult] = []
-    ip = get_client_ip(request)
-    for feedback_id in body.ids:
-        try:
-            out = _approve_feedback(db, admin, feedback_id, body, ip, bulk=True)
-            results.append(FeedbackBulkItemResult(id=feedback_id, ok=True, status=out.status))
-        except HTTPException as exc:
-            db.rollback()
-            results.append(FeedbackBulkItemResult(id=feedback_id, ok=False, error=str(exc.detail)))
-        except Exception:  # noqa: BLE001 - 单笔失败不打断整批
-            db.rollback()
-            results.append(FeedbackBulkItemResult(id=feedback_id, ok=False, error="系统异常"))
-    return _bulk_result(results)
-
-
-@router.post("/bulk/reject", response_model=FeedbackBulkResult, summary="批量拒绝采纳反馈")
-def bulk_reject_feedbacks(
-    body: FeedbackBulkRejectRequest,
-    request: Request,
-    admin: Annotated[AdminUser, Depends(require_role("operator"))],
-    db: AdminDb,
-) -> FeedbackBulkResult:
-    results: list[FeedbackBulkItemResult] = []
-    ip = get_client_ip(request)
-    for feedback_id in body.ids:
-        try:
-            out = _reject_feedback(db, admin, feedback_id, body, ip, bulk=True)
-            results.append(FeedbackBulkItemResult(id=feedback_id, ok=True, status=out.status))
-        except HTTPException as exc:
-            db.rollback()
-            results.append(FeedbackBulkItemResult(id=feedback_id, ok=False, error=str(exc.detail)))
-        except Exception:  # noqa: BLE001 - 单笔失败不打断整批
-            db.rollback()
-            results.append(FeedbackBulkItemResult(id=feedback_id, ok=False, error="系统异常"))
-    return _bulk_result(results)
-
-
 @router.post("/{feedback_id}/handle", response_model=OkResponse, summary="标记反馈已处理")
 def handle_feedback(
     feedback_id: int,
@@ -258,7 +93,53 @@ def approve_feedback(
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
 ) -> FeedbackOut:
-    return _approve_feedback(db, admin, feedback_id, payload, get_client_ip(request))
+    fb = db.get(Feedback, feedback_id)
+    if fb is None:
+        raise HTTPException(status_code=404, detail="反馈不存在")
+    _ensure_pending(fb)
+
+    before = fb.status
+    mutations.review_feedback(
+        db,
+        fb,
+        status="adopted",
+        reward_coins=payload.reward_coins,
+        review_note=payload.note,
+        admin_reply=payload.reply,
+        reviewed_by_admin_id=admin.id,
+        commit=False,
+    )
+    wallet_repo.grant_coins(
+        db,
+        fb.user_id,
+        payload.reward_coins,
+        biz_type="feedback_reward",
+        ref_id=str(fb.id),
+        remark="意见反馈被采纳",
+    )
+    write_audit(
+        db,
+        admin,
+        action="feedback.approve",
+        target_type="feedback",
+        target_id=feedback_id,
+        detail={
+            "before": before,
+            "after": "adopted",
+            "reward_coins": payload.reward_coins,
+            "note": payload.note,
+            "reply": payload.reply,
+        },
+        ip=get_client_ip(request),
+        commit=False,
+    )
+    db.commit()
+    db.refresh(fb)
+    out = FeedbackOut.model_validate(fb)
+    # PRD #10 反馈奖励:采纳发金币后通知用户(站内 + push,必带官方留言)。
+    # 业务已 commit,通知失败只 log 不影响审核结果。
+    notification_events.notify_feedback_reward(db, fb)
+    return out
 
 
 @router.post("/{feedback_id}/reject", response_model=FeedbackOut, summary="拒绝采纳反馈")
@@ -269,4 +150,41 @@ def reject_feedback(
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
 ) -> FeedbackOut:
-    return _reject_feedback(db, admin, feedback_id, payload, get_client_ip(request))
+    fb = db.get(Feedback, feedback_id)
+    if fb is None:
+        raise HTTPException(status_code=404, detail="反馈不存在")
+    _ensure_pending(fb)
+
+    before = fb.status
+    mutations.review_feedback(
+        db,
+        fb,
+        status="rejected",
+        reject_reason=payload.reason,
+        review_note=payload.note,
+        admin_reply=payload.reply,
+        reviewed_by_admin_id=admin.id,
+        commit=False,
+    )
+    write_audit(
+        db,
+        admin,
+        action="feedback.reject",
+        target_type="feedback",
+        target_id=feedback_id,
+        detail={
+            "before": before,
+            "after": "rejected",
+            "reason": payload.reason,
+            "note": payload.note,
+            "reply": payload.reply,
+        },
+        ip=get_client_ip(request),
+        commit=False,
+    )
+    db.commit()
+    db.refresh(fb)
+    out = FeedbackOut.model_validate(fb)
+    # PRD #9 官方回复:未采纳也回复了用户(原因/留言用户端可见),通知去反馈历史页查看。
+    notification_events.notify_feedback_reply(db, fb)
+    return out

app/admin/routers/guide_video.py

@@ -1,101 +0,0 @@
-"""admin 新手引导视频配置:读 / 改开关次数金币 / 上传视频 / 删视频(带审计)。
-
-整份配置存通用 app_config 表(见 app/repositories/guide_video.py),App 领券等候浮层
-每次展示前调 POST /api/v1/guide-video/start 同步。权限:operator 可改(运营维护),
-super 恒可;读为只读(任意已登录 admin)。
-
-⚠️ 视频上限 100MB(settings.GUIDE_VIDEO_MAX_BYTES),已在 admin nginx 为本接口单独放宽
-client_max_body_size,见 shaguabijia-admin-web/deploy/nginx/admin.shaguabijia.com.conf。
-"""
-from __future__ import annotations
-
-from typing import Annotated
-
-from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile
-
-from app.admin.audit import write_audit
-from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
-from app.admin.schemas.guide_video import GuideVideoConfigOut, GuideVideoConfigUpdate
-from app.core import media
-from app.models.admin import AdminUser
-from app.repositories import guide_video
-
-router = APIRouter(
-    prefix="/admin/api/guide-video",
-    tags=["admin-guide-video"],
-    dependencies=[Depends(get_current_admin)],
-)
-
-
-def _out(db: AdminDb) -> GuideVideoConfigOut:
-    """配置 + 播放统计合成响应(四个写接口都以最新状态返回,前端一次同步到位)。"""
-    return GuideVideoConfigOut(**guide_video.get_config(db), **guide_video.play_stats(db))
-
-
-@router.get("", response_model=GuideVideoConfigOut, summary="新手引导视频配置(领券浮层)")
-def get_config(db: AdminDb) -> GuideVideoConfigOut:
-    return _out(db)
-
-
-@router.patch("", response_model=GuideVideoConfigOut, summary="改开关/次数/金币(带审计)")
-def update_config(
-    body: GuideVideoConfigUpdate,
-    request: Request,
-    admin: Annotated[AdminUser, Depends(require_role("operator"))],
-    db: AdminDb,
-) -> GuideVideoConfigOut:
-    before, after = guide_video.update_config(
-        db,
-        enabled=body.enabled,
-        max_plays=body.max_plays,
-        reward_coin=body.reward_coin,
-        admin_id=admin.id,
-        commit=False,
-    )
-    write_audit(
-        db, admin, action="guide_video.update", target_type="guide_video", target_id=None,
-        detail={"before": before, "after": after}, ip=get_client_ip(request), commit=False,
-    )
-    db.commit()
-    return _out(db)
-
-
-@router.post("/video", response_model=GuideVideoConfigOut, summary="上传新手引导视频(MP4,带审计)")
-async def upload_video(
-    request: Request,
-    admin: Annotated[AdminUser, Depends(require_role("operator"))],
-    db: AdminDb,
-    file: UploadFile = File(...),
-) -> GuideVideoConfigOut:
-    data = await file.read()
-    try:
-        url = media.save_guide_video(data)
-    except media.MediaError as e:
-        raise HTTPException(status_code=400, detail=str(e)) from e
-    before, after = guide_video.set_video(db, url, admin_id=admin.id, commit=False)
-    write_audit(
-        db, admin, action="guide_video.set_video", target_type="guide_video", target_id=None,
-        detail={"before": before.get("video_url"), "after": url, "bytes": len(data)},
-        ip=get_client_ip(request), commit=False,
-    )
-    db.commit()
-    # 提交成功后再删旧片,避免新片没落库就把旧片丢了
-    media.delete_guide_video(before.get("video_url"))
-    return _out(db)
-
-
-@router.delete("/video", response_model=GuideVideoConfigOut, summary="移除新手引导视频(带审计)")
-def delete_video(
-    request: Request,
-    admin: Annotated[AdminUser, Depends(require_role("operator"))],
-    db: AdminDb,
-) -> GuideVideoConfigOut:
-    """移除后 /guide-video/start 一律返回 should_play=false,领券浮层回到「只放广告」。"""
-    before, after = guide_video.set_video(db, None, admin_id=admin.id, commit=False)
-    write_audit(
-        db, admin, action="guide_video.delete_video", target_type="guide_video", target_id=None,
-        detail={"before": before.get("video_url")}, ip=get_client_ip(request), commit=False,
-    )
-    db.commit()
-    media.delete_guide_video(before.get("video_url"))
-    return _out(db)

app/admin/routers/price_report.py

@@ -17,10 +17,6 @@ from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_ro
 from app.admin.repositories import mutations, queries
 from app.admin.schemas.common import CursorPage, OkResponse
 from app.admin.schemas.price_report import (
-    PriceReportBulkItemResult,
-    PriceReportBulkRejectRequest,
-    PriceReportBulkRequest,
-    PriceReportBulkResult,
     PriceReportOut,
     PriceReportRejectRequest,
     PriceReportSummary,
@@ -38,59 +34,6 @@ router = APIRouter(
 )
 
 
-def _approve_price_report(
-    db: AdminDb, admin: AdminUser, report_id: int, ip: str, *, bulk: bool = False
-) -> PriceReport:
-    rep = db.get(PriceReport, report_id, with_for_update=True)
-    if rep is None:
-        raise HTTPException(status_code=404, detail="上报记录不存在")
-    if rep.status != "pending":
-        raise HTTPException(status_code=400, detail=f"该上报已审核过(当前 {rep.status}),不可重复操作")
-    coins = PRICE_REPORT_REWARD_COINS
-    mutations.review_price_report(db, rep, status="approved", reward_coins=coins, commit=False)
-    wallet_repo.grant_coins(
-        db, rep.user_id, coins,
-        biz_type="price_report_reward", ref_id=str(rep.id), remark="上报更低价审核通过",
-    )
-    detail = {"reward_coins": coins, "user_id": rep.user_id}
-    if bulk:
-        detail["bulk"] = True
-    write_audit(
-        db, admin, action="price_report.approve", target_type="price_report", target_id=report_id,
-        detail=detail, ip=ip, commit=False,
-    )
-    db.commit()
-    notification_events.notify_report_approved(db, rep)
-    return rep
-
-
-def _reject_price_report(
-    db: AdminDb, admin: AdminUser, report_id: int, reason: str, ip: str, *, bulk: bool = False
-) -> PriceReport:
-    rep = db.get(PriceReport, report_id, with_for_update=True)
-    if rep is None:
-        raise HTTPException(status_code=404, detail="上报记录不存在")
-    if rep.status != "pending":
-        raise HTTPException(status_code=400, detail=f"该上报已审核过(当前 {rep.status}),不可重复操作")
-    mutations.review_price_report(db, rep, status="rejected", reject_reason=reason, commit=False)
-    detail = {"reason": reason, "user_id": rep.user_id}
-    if bulk:
-        detail["bulk"] = True
-    write_audit(
-        db, admin, action="price_report.reject", target_type="price_report", target_id=report_id,
-        detail=detail, ip=ip, commit=False,
-    )
-    db.commit()
-    return rep
-
-
-def _bulk_result(items: list[PriceReportBulkItemResult]) -> PriceReportBulkResult:
-    success = sum(1 for item in items if item.ok)
-    return PriceReportBulkResult(
-        total=len(items), success=success, failed=len(items) - success, items=items,
-    )
-
-
 @router.get("", response_model=CursorPage[PriceReportOut], summary="上报更低价列表(筛选+分页)")
 def list_price_reports(
     db: AdminDb,
@@ -117,50 +60,6 @@ def price_report_summary(db: AdminDb) -> PriceReportSummary:
     return PriceReportSummary.model_validate(queries.price_report_summary(db))
 
 
-@router.post("/bulk/approve", response_model=PriceReportBulkResult, summary="批量通过上报(发固定金币)")
-def bulk_approve_price_reports(
-    body: PriceReportBulkRequest,
-    request: Request,
-    admin: Annotated[AdminUser, Depends(require_role("operator"))],
-    db: AdminDb,
-) -> PriceReportBulkResult:
-    results: list[PriceReportBulkItemResult] = []
-    ip = get_client_ip(request)
-    for report_id in body.ids:
-        try:
-            rep = _approve_price_report(db, admin, report_id, ip, bulk=True)
-            results.append(PriceReportBulkItemResult(id=report_id, ok=True, status=rep.status))
-        except HTTPException as exc:
-            db.rollback()
-            results.append(PriceReportBulkItemResult(id=report_id, ok=False, error=str(exc.detail)))
-        except Exception:  # noqa: BLE001 - 单笔失败不打断整批
-            db.rollback()
-            results.append(PriceReportBulkItemResult(id=report_id, ok=False, error="系统异常"))
-    return _bulk_result(results)
-
-
-@router.post("/bulk/reject", response_model=PriceReportBulkResult, summary="批量拒绝上报")
-def bulk_reject_price_reports(
-    body: PriceReportBulkRejectRequest,
-    request: Request,
-    admin: Annotated[AdminUser, Depends(require_role("operator"))],
-    db: AdminDb,
-) -> PriceReportBulkResult:
-    results: list[PriceReportBulkItemResult] = []
-    ip = get_client_ip(request)
-    for report_id in body.ids:
-        try:
-            rep = _reject_price_report(db, admin, report_id, body.reason, ip, bulk=True)
-            results.append(PriceReportBulkItemResult(id=report_id, ok=True, status=rep.status))
-        except HTTPException as exc:
-            db.rollback()
-            results.append(PriceReportBulkItemResult(id=report_id, ok=False, error=str(exc.detail)))
-        except Exception:  # noqa: BLE001 - 单笔失败不打断整批
-            db.rollback()
-            results.append(PriceReportBulkItemResult(id=report_id, ok=False, error="系统异常"))
-    return _bulk_result(results)
-
-
 @router.post("/{report_id}/approve", response_model=OkResponse, summary="通过上报(发固定金币)")
 def approve_price_report(
     report_id: int,
@@ -168,7 +67,27 @@ def approve_price_report(
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
 ) -> OkResponse:
-    _approve_price_report(db, admin, report_id, get_client_ip(request))
+    # 行锁(SELECT FOR UPDATE):并发/连点双请求会都读到 pending → 各发一次金币双倍发奖,
+    # 锁住该行串行化,第二个请求拿锁后看到 approved → 走 400。SQLite 下 FOR UPDATE 为 no-op。
+    rep = db.get(PriceReport, report_id, with_for_update=True)
+    if rep is None:
+        raise HTTPException(status_code=404, detail="上报记录不存在")
+    if rep.status != "pending":
+        raise HTTPException(status_code=400, detail=f"该上报已审核过(当前 {rep.status}),不可重复操作")
+    coins = PRICE_REPORT_REWARD_COINS
+    # 改状态 + 发金币 + 审计同一事务(commit=False),最后一起 commit:改了就有痕、发了就留账
+    mutations.review_price_report(db, rep, status="approved", reward_coins=coins, commit=False)
+    wallet_repo.grant_coins(
+        db, rep.user_id, coins,
+        biz_type="price_report_reward", ref_id=str(rep.id), remark="上报更低价审核通过",
+    )
+    write_audit(
+        db, admin, action="price_report.approve", target_type="price_report", target_id=report_id,
+        detail={"reward_coins": coins, "user_id": rep.user_id}, ip=get_client_ip(request), commit=False,
+    )
+    db.commit()
+    # PRD #11 爆料审核通过:发金币后通知用户(站内 + push)。业务已 commit,通知失败只 log。
+    notification_events.notify_report_approved(db, rep)
     return OkResponse()
 
 
@@ -180,5 +99,16 @@ def reject_price_report(
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
 ) -> OkResponse:
-    _reject_price_report(db, admin, report_id, body.reason, get_client_ip(request))
+    rep = db.get(PriceReport, report_id, with_for_update=True)  # 行锁,同 approve(防并发重复审核)
+    if rep is None:
+        raise HTTPException(status_code=404, detail="上报记录不存在")
+    if rep.status != "pending":
+        raise HTTPException(status_code=400, detail=f"该上报已审核过(当前 {rep.status}),不可重复操作")
+    reason = body.reason.strip()
+    mutations.review_price_report(db, rep, status="rejected", reject_reason=reason, commit=False)
+    write_audit(
+        db, admin, action="price_report.reject", target_type="price_report", target_id=report_id,
+        detail={"reason": reason, "user_id": rep.user_id}, ip=get_client_ip(request), commit=False,
+    )
+    db.commit()
     return OkResponse()

app/admin/routers/withdraw.py

@@ -129,11 +129,13 @@ def withdraw_health_check(db: AdminDb) -> WxpayHealthCheckOut:
         issues.append("免确认授权回调地址未配置")
 
     # 实际是否自动对账 = env 部署总闸(worker 起没起)AND 运营后台 DB 开关(本轮跑不跑)。
-    # 自动查单属于非阻断运维能力:状态继续返回给调用方,但关闭时不计入微信提现配置 issues,
-    # 避免把“没有自动扫单”误报成“无法打款”。
     worker_running = settings.WITHDRAW_AUTO_RECONCILE_ENABLED
     daily_on = bool(app_config.get_value(db, "withdraw_auto_reconcile_enabled"))
     auto_reconcile_enabled = worker_running and daily_on
+    if not worker_running:
+        issues.append("自动对账 worker 未启动(部署侧 env WITHDRAW_AUTO_RECONCILE_ENABLED=false)")
+    elif not daily_on:
+        issues.append("自动对账运营开关已关闭(系统配置页可开)")
 
     return WxpayHealthCheckOut(
         ok=not issues,

app/admin/schemas/coupon_data.py

@@ -79,10 +79,10 @@ class CouponDataRow(BaseModel):
     started_at: datetime = Field(..., description="发起时刻(明细「时间」列)")
     claimed_count: int | None = None
     point_success_count: int | None = Field(
-        None, description="本次成功单券数(success+already_claimed);无逐券事件为空"
+        None, description="本次成功券点位数(success+already_claimed);无逐券埋点为空"
     )
     point_total_count: int | None = Field(
-        None, description="本次尝试单券数(success+already_claimed+failed,不含 skipped);无逐券事件为空"
+        None, description="本次尝试券点位数(success+already_claimed+failed,不含 skipped);无逐券埋点为空"
     )
     trace_url: str | None = Field(None, description="pricebot 公网 trace 链接(仅 completed 有);admin 渲染可点链接,无则显示可复制 trace_id")
     ad_revenue_yuan: float = Field(

app/admin/schemas/feedback.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 
 from datetime import datetime
 
-from pydantic import BaseModel, ConfigDict, Field, field_validator
+from pydantic import BaseModel, ConfigDict, Field
 
 from app.core.rewards import FEEDBACK_REWARD_MAX_COINS
 
@@ -55,54 +55,6 @@ class FeedbackRejectRequest(BaseModel):
     reply: str | None = Field(default=None, max_length=256, description="给用户的回复留言,用户端可见")
 
 
-class FeedbackBulkRequest(BaseModel):
-    ids: list[int] = Field(min_length=1, max_length=50, description="待审核反馈 ID 列表")
-
-    @field_validator("ids")
-    @classmethod
-    def _ids_must_be_unique(cls, ids: list[int]) -> list[int]:
-        if len(ids) != len(set(ids)):
-            raise ValueError("反馈 ID 不能重复")
-        return ids
-
-
-class FeedbackBulkApproveRequest(FeedbackBulkRequest):
-    reward_coins: int = Field(
-        ge=1,
-        le=FEEDBACK_REWARD_MAX_COINS,
-        description="每条采纳反馈发放的金币数",
-    )
-    note: str | None = Field(default=None, max_length=256, description="采纳要点/审核备注(内部)")
-    reply: str | None = Field(default=None, max_length=256, description="给用户的回复留言,用户端可见")
-
-
-class FeedbackBulkRejectRequest(FeedbackBulkRequest):
-    reason: str = Field(min_length=1, max_length=256, description="批量未采纳原因,用户端可见")
-    note: str | None = Field(default=None, max_length=256, description="运营内部审核备注")
-    reply: str | None = Field(default=None, max_length=256, description="给用户的回复留言,用户端可见")
-
-    @field_validator("reason")
-    @classmethod
-    def _reason_not_blank(cls, value: str) -> str:
-        if not value.strip():
-            raise ValueError("未采纳原因不能为空")
-        return value.strip()
-
-
-class FeedbackBulkItemResult(BaseModel):
-    id: int
-    ok: bool
-    status: str | None = None
-    error: str | None = None
-
-
-class FeedbackBulkResult(BaseModel):
-    total: int
-    success: int
-    failed: int
-    items: list[FeedbackBulkItemResult]
-
-
 class FeedbackSummary(BaseModel):
     """审核台顶部各状态计数(pending 含历史 new 态)。"""
 

app/admin/schemas/guide_video.py

@@ -1,25 +0,0 @@
-"""admin 新手引导视频配置 schemas(开关 / 视频地址 / 前几次 / 每次金币)。"""
-from __future__ import annotations
-
-from pydantic import BaseModel, Field
-
-from app.repositories.guide_video import MAX_PLAYS_LIMIT, REWARD_COIN_LIMIT
-
-
-class GuideVideoConfigOut(BaseModel):
-    enabled: bool
-    video_url: str | None = None   # 相对地址 /media/guide_video/xxx.mp4;未配片 = None
-    max_plays: int
-    reward_coin: int
-    updated_at: str | None = None
-    # 只读统计,后台展示用:已有多少次播放、其中已发币多少次。
-    total_plays: int = 0
-    granted_plays: int = 0
-
-
-class GuideVideoConfigUpdate(BaseModel):
-    """部分更新:只改传入(非 None)字段。视频文件走 /video 上传接口。"""
-
-    enabled: bool | None = None
-    max_plays: int | None = Field(default=None, ge=0, le=MAX_PLAYS_LIMIT)
-    reward_coin: int | None = Field(default=None, ge=0, le=REWARD_COIN_LIMIT)

app/admin/schemas/price_report.py

@@ -56,42 +56,6 @@ class PriceReportRejectRequest(BaseModel):
         return v.strip()
 
 
-class PriceReportBulkRequest(BaseModel):
-    ids: list[int] = Field(min_length=1, max_length=50, description="待审核上报 ID 列表")
-
-    @field_validator("ids")
-    @classmethod
-    def _ids_must_be_unique(cls, ids: list[int]) -> list[int]:
-        if len(ids) != len(set(ids)):
-            raise ValueError("上报 ID 不能重复")
-        return ids
-
-
-class PriceReportBulkRejectRequest(PriceReportBulkRequest):
-    reason: str = Field(min_length=1, max_length=256, description="批量拒绝理由,用户端记录页会看到")
-
-    @field_validator("reason")
-    @classmethod
-    def _reason_not_blank(cls, value: str) -> str:
-        if not value.strip():
-            raise ValueError("拒绝理由不能为空")
-        return value.strip()
-
-
-class PriceReportBulkItemResult(BaseModel):
-    id: int
-    ok: bool
-    status: str | None = None
-    error: str | None = None
-
-
-class PriceReportBulkResult(BaseModel):
-    total: int
-    success: int
-    failed: int
-    items: list[PriceReportBulkItemResult]
-
-
 class PriceReportSummary(BaseModel):
     """审核台顶部各状态计数。"""
 

app/api/v1/compare_record.py

@@ -20,8 +20,6 @@ from app.db.session import SessionLocal
 from app.models.comparison import ComparisonRecord
 from app.repositories import comparison as crud_compare
 from app.schemas.compare_record import (
-    CompareStartReserveIn,
-    CompareStartReserveOut,
     CompareStatsOut,
     ComparisonRecordCreatedOut,
     ComparisonRecordDetailOut,
@@ -37,41 +35,6 @@ logger = logging.getLogger("shagua.compare_record")
 router = APIRouter(prefix="/api/v1/compare", tags=["compare-record"])
 
 
-@router.post(
-    "/start",
-    response_model=CompareStartReserveOut,
-    summary="预占一次当日比价发起次数（每人每天最多100次）",
-)
-def reserve_compare_start(
-    payload: CompareStartReserveIn,
-    user: CurrentUser,
-    db: DbSession,
-) -> CompareStartReserveOut:
-    try:
-        _, used = crud_compare.reserve_daily_start(
-            db,
-            user_id=user.id,
-            trace_id=payload.trace_id,
-            business_type=payload.business_type,
-            device_id=payload.device_id,
-        )
-    except crud_compare.DailyCompareStartLimitExceeded:
-        raise HTTPException(
-            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
-            detail="今日已比价超过100次，请明天再试",
-        ) from None
-    except crud_compare.ComparisonTraceOwnershipError:
-        raise HTTPException(
-            status_code=status.HTTP_409_CONFLICT,
-            detail="比价任务标识冲突，请重新发起",
-        ) from None
-    return CompareStartReserveOut(
-        limit=crud_compare.DAILY_COMPARE_START_LIMIT,
-        used=used,
-        remaining=max(crud_compare.DAILY_COMPARE_START_LIMIT - used, 0),
-    )
-
-
 @router.post(
     "/record",
     response_model=ComparisonRecordCreatedOut,

app/api/v1/coupon.py

@@ -81,7 +81,7 @@ def _record_claims_blocking(
     device_id: str, user_id: int | None, trace_id: str | None, results: list[dict]
 ) -> None:
     with SessionLocal() as db:
-        # 取本次 session 环境，给每日资产和逐次事件同时打环境标。
+        # 取本次 session 环境,给 coupon_claim_record 打 app_env 标(每券成功率表按它过滤;设计 §13)。
         app_env = coupon_repo.session_app_env(db, trace_id)
         coupon_repo.record_claims(db, device_id, user_id, trace_id, results, app_env=app_env)
         # 顺带把本帧「成功平台」并入 coupon_session.platform_success(admin 领券数据 ②整单/③点位成功率;
@@ -176,7 +176,7 @@ async def coupon_step(
 
     resp_json = resp.json()
 
-    # 领券结果沉淀:每日资产 + 逐次事件；中间帧和 done 全量帧均幂等写库。
+    # 领券结果沉淀(资产):中间帧 last_coupon_result + done 帧 coupon_results 幂等写库。
     # 当前只记录、不参与"要不要领"判断(MVP 先不去重)。写库失败不影响返回。
     if device_id:
         results = _extract_coupon_results(resp_json)

app/api/v1/guide_video.py

@@ -1,64 +0,0 @@
-"""新手引导视频(领券等候浮层前 N 次替代广告)。
-
-路由前缀 `/api/v1/guide-video`(均需 Bearer):
-  POST /start    这次浮层放引导视频还是放广告?命中则**当场计次**并下发 play_token
-  POST /reward   播完 / 中途关闭都调,按 play_token 幂等发固定金币
-
-发币额度以**服务端配置**为准(运营后台可改),客户端只报"播完/关闭",报不了金额,
-所以被破解也刷不到超额金币;次数上限由 guide_video_play 行数(按账号)硬卡。
-"""
-from __future__ import annotations
-
-import logging
-
-from fastapi import APIRouter, Depends
-
-from app.api.deps import CurrentUser, DbSession
-from app.core.ratelimit import rate_limit
-from app.repositories import guide_video as crud_guide
-from app.schemas.guide_video import (
-    GuideVideoRewardIn,
-    GuideVideoRewardOut,
-    GuideVideoStartIn,
-    GuideVideoStartOut,
-)
-
-logger = logging.getLogger("shagua.guide_video")
-
-router = APIRouter(prefix="/api/v1/guide-video", tags=["guide-video"])
-
-
-@router.post(
-    "/start",
-    response_model=GuideVideoStartOut,
-    summary="领券浮层是否放新手引导视频(命中即计次)",
-    dependencies=[Depends(rate_limit(60, 60, "guide-video-start"))],
-)
-def start(payload: GuideVideoStartIn, user: CurrentUser, db: DbSession) -> GuideVideoStartOut:
-    """开播即计数:返回 should_play=True 时服务端已写下这一次,客户端必须真的播。
-
-    没配视频 / 开关关 / 次数用完 → should_play=False,客户端照旧走广告链路(行为不变)。
-    """
-    result = crud_guide.start_play(db, user.id, scene=payload.scene or "coupon")
-    logger.info(
-        "guide video start user_id=%d scene=%s should_play=%s seq=%d remaining=%d",
-        user.id, payload.scene, result["should_play"], result["seq"], result["remaining"],
-    )
-    return GuideVideoStartOut(**result)
-
-
-@router.post(
-    "/reward",
-    response_model=GuideVideoRewardOut,
-    summary="引导视频发金币(播完/中途关闭都发,play_token 幂等)",
-    dependencies=[Depends(rate_limit(60, 60, "guide-video-reward"))],
-)
-def reward(payload: GuideVideoRewardIn, user: CurrentUser, db: DbSession) -> GuideVideoRewardOut:
-    result = crud_guide.grant_play(
-        db, user.id, play_token=payload.play_token, completed=payload.completed
-    )
-    logger.info(
-        "guide video reward user_id=%d token=%s completed=%s granted=%s coin=%d",
-        user.id, payload.play_token[:12], payload.completed, result["granted"], result["coin"],
-    )
-    return GuideVideoRewardOut(**result)

app/api/v1/meituan.py

@@ -5,12 +5,11 @@
 from __future__ import annotations
 
 import logging
-from concurrent.futures import ThreadPoolExecutor
-from typing import TYPE_CHECKING, Any
+from concurrent.futures import ThreadPoolExecutor, as_completed
 
 from fastapi import APIRouter, Depends, HTTPException
 from sqlalchemy import nullslast, select
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import Session, aliased
 
 from app.core.config import settings
 from app.db.session import get_db
@@ -26,14 +25,8 @@ from app.schemas.meituan import (
     ReferralLinkResponse,
     TopSalesRequest,
 )
-from app.utils import mt_search_cursor
 from app.utils.meituan_city import get_meituan_city
 
-if TYPE_CHECKING:  # 仅供类型标注(本模块已开 from __future__ import annotations)
-    from collections.abc import Callable
-
-    from sqlalchemy import ColumnElement
-
 logger = logging.getLogger("shagua.meituan")
 
 
@@ -116,86 +109,6 @@ def _commission_pct(card: CouponCard) -> float:
         return 0.0
 
 
-# ────────────── 离线库分页(智能推荐 / 销量最高 共用) ──────────────
-# 去重+排序阶段**只投影这几列**:够 DISTINCT ON 分组、够排序、够回表定位,且全是定长小字段。
-# ⚠️ 关键性能点:`raw` 是整条美团原始返回(JSONB,每行数 KB)。原实现用 select(MeituanCoupon)
-# 做子查询,等于把整城几千行连 raw 一起塞进两次排序(DISTINCT ON 一次 + 分页一次),
-# 体量轻松超过 work_mem → Postgres 落盘做外部归并排序,而且**每翻一页都要重来一遍**。
-# 拆成「先在小列上排出本页 id,再按 id 回表取 raw」后,排序数据量降到原来的百分之几,
-# JSONB 只解析当前页 ~20 行。
-_DEDUP_COLS = (
-    MeituanCoupon.id,
-    MeituanCoupon.dedup_key,
-    MeituanCoupon.sale_volume_num,
-    MeituanCoupon.commission_percent,
-)
-
-
-def _paged_dedup_ids(
-    db: Session,
-    *,
-    conds: list[ColumnElement[bool]],
-    dedup_order: list[ColumnElement],
-    page_order: Callable[[Any], list[ColumnElement]],
-    page: int,
-    page_size: int,
-) -> tuple[list[int], bool]:
-    """DISTINCT ON(dedup_key) 跨源去重 → 整体排序 → 分页,返回 (本页 id 列表, 是否还有下一页)。
-
-    - `dedup_order`:同一个 dedup_key 的多条里留哪条(如销量最高/佣金最高)。
-    - `page_order`:接收去重子查询的列集合(`sub.c`),返回去重后的整体排序。
-    多取 1 条用于判断 has_next。
-    """
-    deduped = (
-        select(*_DEDUP_COLS)
-        .where(*conds)
-        .distinct(MeituanCoupon.dedup_key)
-        .order_by(MeituanCoupon.dedup_key, *dedup_order)
-        .subquery()
-    )
-    ids = db.execute(
-        select(deduped.c.id)
-        .order_by(*page_order(deduped.c))
-        .offset((page - 1) * page_size)
-        .limit(page_size + 1)
-    ).scalars().all()
-    return list(ids[:page_size]), len(ids) > page_size
-
-
-def _load_raws(db: Session, ids: list[int]) -> list[dict]:
-    """按给定 id 顺序取 raw(只回表本页 ~20 行)。缺行(被 ETL 清掉)静默跳过。"""
-    if not ids:
-        return []
-    raw_by_id = {
-        row_id: raw
-        for row_id, raw in db.execute(
-            select(MeituanCoupon.id, MeituanCoupon.raw).where(MeituanCoupon.id.in_(ids))
-        ).all()
-    }
-    return [raw_by_id[i] for i in ids if i in raw_by_id]
-
-
-def _cards_from_raws(raws: list[dict], *, hide_distance: bool) -> list[CouponCard]:
-    """raw → CouponCard;解析失败的单条跳过,不整页失败。
-
-    hide_distance:离线库里的距离是相对「城市默认点」算的,对用户无意义且误导 —— 智能推荐 /
-    销量最高两个 tab 一律置空,前端「距离 店名」那行只剩店名、自动顶到最左。
-    """
-    cards: list[CouponCard] = []
-    for raw in raws:
-        try:
-            card = CouponCard.from_raw(raw or {})
-        except Exception:  # noqa: BLE001
-            continue
-        if not card.product_view_sign:
-            continue
-        if hide_distance:
-            card.distance_text = None
-            card.distance_meters = None
-        cards.append(card)
-    return cards
-
-
 @router.post("/feed", response_model=FeedResponse, summary="混合feed(外卖+到店交叉);tab=rec智能推荐/distance距离最近")
 def feed(req: FeedRequest, db: Session = Depends(get_db)) -> FeedResponse:
     lon, lat = req.longitude, req.latitude
@@ -220,21 +133,17 @@ def feed(req: FeedRequest, db: Session = Depends(get_db)) -> FeedResponse:
             return [], True
 
     # 距离最近:搜索召回(外卖搜"外卖" + 到店搜"美食",都 sortField=6 离我最近)一页页拉。
-    # 搜索翻页必须用 searchId(pageNo 翻不动),而接口是无状态的(客户端只传页码)—— 原实现因此
-    # 每次都从第 1 页顺序重放到第 N 页,取第 N 页要向美团发 N 次请求,越往下滑越慢。
-    # 现在把沿途 searchId 记进 [mt_search_cursor],稳态下每翻一页恒定 1 次请求;两路仍并行。
-    # 按你位置实时算距离(库里没存 POI 经纬度,只能实时)。
+    # 搜索翻页必须用 searchId(pageNo 翻不动),所以每个 feed 页顺序翻到第 N 页;两路并行、page 1 最快。
+    # 无状态、不改 APP(传页码即可);按你位置实时算距离(库里没存 POI 经纬度,只能实时)。
     if tab == "distance":
         lon_i, lat_i = int(lon * 1_000_000), int(lat * 1_000_000)
 
-        def _replay(
-            platform: int, biz_line: int | None, keyword: str,
-            key: mt_search_cursor.RouteKey, start: int, sid: str | None, n: int,
-        ) -> tuple[list[dict], bool, bool]:
-            """从第 start 页(用 sid 取)顺序翻到第 n 页。start==1 时 sid 应为 None(走 pageNo=1)。"""
+        def _search_page_n(platform: int, biz_line: int | None, keyword: str, n: int) -> tuple[list[dict], bool, bool]:
+            """顺序翻到第 n 页(搜索须 searchId 续页),返回(第 n 页 items, 是否还有下一页, 是否调用失败)。"""
+            sid: str | None = None
             data: list[dict] = []
             has_next = False
-            for pg in range(start, n + 1):
+            for pg in range(1, n + 1):
                 body: dict = {
                     "platform": platform, "searchText": keyword, "sortField": 6,
                     "longitude": lon_i, "latitude": lat_i, "pageSize": 20,
@@ -252,27 +161,10 @@ def feed(req: FeedRequest, db: Session = Depends(get_db)) -> FeedResponse:
                 data = r.get("data") or []
                 sid = r.get("searchId")
                 has_next = bool(r.get("hasNext")) and bool(data)
-                # 记下「下一页要用哪个 searchId」;没有下一页就别记,免得存进死游标。
-                if sid and has_next:
-                    mt_search_cursor.remember(key, pg + 1, sid)
                 if not data or (not has_next and pg < n):
                     return [], False, False   # 没那么多页了(非错误)
             return data, has_next, False
 
-        def _search_page_n(platform: int, biz_line: int | None, keyword: str, n: int) -> tuple[list[dict], bool, bool]:
-            """取第 n 页,返回(第 n 页 items, 是否还有下一页, 是否调用失败)。
-
-            优先用缓存游标一发直达;缓存未命中/过期才从最近的已知页往后重放,并把沿途游标补进缓存。
-            """
-            key = mt_search_cursor.route_key(lat, lon, platform, keyword)
-            start, sid = mt_search_cursor.lookup(key, n)
-            data, has_next, failed = _replay(platform, biz_line, keyword, key, start, sid, n)
-            # 用缓存游标却打不通,多半是上游 searchId 过期:作废整条路线,回到第 1 页重放一次。
-            if failed and start > 1:
-                mt_search_cursor.drop(key)
-                data, has_next, failed = _replay(platform, biz_line, keyword, key, 1, None, n)
-            return data, has_next, failed
-
         with ThreadPoolExecutor(max_workers=2) as pool:
             f_wm = pool.submit(_search_page_n, 1, None, "外卖", req.page)
             f_dd = pool.submit(_search_page_n, 2, 1, "美食", req.page)
@@ -302,25 +194,39 @@ def feed(req: FeedRequest, db: Session = Depends(get_db)) -> FeedResponse:
             return FeedResponse(items=[], has_next=False, page=req.page, status="degraded")
         PAGE = 20
         try:
-            ids, has_next = _paged_dedup_ids(
-                db,
-                conds=[
-                    MeituanCoupon.commission_percent >= 3.0,
-                    MeituanCoupon.city_id == city_id,
-                ],
-                # 同一去重键留佣金最高那条
-                dedup_order=[MeituanCoupon.commission_percent.desc()],
-                # 销量高的优先(无销量档排后),同档佣金高优先,id 兜底稳定分页
-                page_order=lambda c: [
-                    nullslast(c.sale_volume_num.desc()), c.commission_percent.desc(), c.id,
-                ],
-                page=req.page, page_size=PAGE,
+            base = select(MeituanCoupon).where(
+                MeituanCoupon.commission_percent >= 3.0,
+                MeituanCoupon.city_id == city_id,
             )
-            raws = _load_raws(db, ids)
+            deduped = base.distinct(MeituanCoupon.dedup_key).order_by(
+                MeituanCoupon.dedup_key,
+                MeituanCoupon.commission_percent.desc(),
+            ).subquery()
+            m = aliased(MeituanCoupon, deduped)
+            start = (req.page - 1) * PAGE
+            rows = db.execute(
+                select(m)
+                # 销量高的优先(无销量档排后),同档佣金高优先,id 兜底稳定分页
+                .order_by(nullslast(m.sale_volume_num.desc()), m.commission_percent.desc(), m.id)
+                .offset(start)
+                .limit(PAGE + 1)
+            ).scalars().all()
         except Exception:  # noqa: BLE001
             logger.exception("[feed] rec 库查询失败,降级返空")
             return FeedResponse(items=[], has_next=False, page=req.page, status="degraded")
-        cards = _cards_from_raws(raws, hide_distance=True)
+        has_next = len(rows) > PAGE
+        cards: list[CouponCard] = []
+        for row in rows[:PAGE]:
+            try:
+                card = CouponCard.from_raw(row.raw or {})
+            except Exception:  # noqa: BLE001
+                continue
+            if card.product_view_sign:
+                # 智能推荐不显示距离:库里的距离是相对城市默认点的(对用户无意义、且误导)。
+                # 置空后前端"距离 店名"那行只剩店名、自动顶到最左(店名移到原距离的位置)。
+                card.distance_text = None
+                card.distance_meters = None
+                cards.append(card)
         if not cards and req.page == 1:
             # 命中城市却 0 券:该城确无 ≥3% 券,或 ETL 灌的 city_id 与 city_dict 口径不一致。
             logger.info("[feed] rec city_id=%s 命中 0 券(该城确无券?或 ETL/city_dict 的 city_id 口径不一致)", city_id)
@@ -376,36 +282,51 @@ def top_sales(req: TopSalesRequest, db: Session = Depends(get_db)) -> CouponList
     if not city_id:
         return CouponListResponse(items=[], has_next=False, search_id=None, status="degraded")
 
-    # 去重 + 排序 + 分页全在 SQL 做,每页只回表并解析当前页 ~20 条(见 _paged_dedup_ids 的性能说明)。
+    # 去重 + 排序 + 分页全在 SQL 做,每页只取并解析当前页 ~20 条。
+    # (之前实现每翻一页都全表拉取 + 全量 from_raw 解析,翻页慢 → 客户端滑动卡顿/翻不动。)
     # 库为空(prod 刚部署 / ETL 未跑完)时返空 + status=empty,不崩;库查询异常降级 degraded。
-    conds = [
-        MeituanCoupon.sale_volume_num.isnot(None),
-        MeituanCoupon.city_id == city_id,
-    ]
-    if req.platform is not None:
-        conds.append(MeituanCoupon.platform == req.platform)
     try:
-        ids, has_next = _paged_dedup_ids(
-            db,
-            conds=conds,
-            # 每个去重键(品牌|名|价)只留销量最高那条(同销量再按佣金)
-            dedup_order=[
-                MeituanCoupon.sale_volume_num.desc(),
-                MeituanCoupon.commission_percent.desc(),
-            ],
-            # 加 id 作稳定 tiebreaker:同销量同佣金的并列项排序确定,避免跨页重复/漏项
-            page_order=lambda c: [
-                c.sale_volume_num.desc(), c.commission_percent.desc(), c.id,
-            ],
-            page=req.page, page_size=req.page_size,
+        # 1) DISTINCT ON (dedup_key):每个去重键(品牌|名|价)只留销量最高那条(同销量再按佣金)
+        base = select(MeituanCoupon).where(
+            MeituanCoupon.sale_volume_num.isnot(None),
+            MeituanCoupon.city_id == city_id,
         )
-        raws = _load_raws(db, ids)
+        if req.platform is not None:
+            base = base.where(MeituanCoupon.platform == req.platform)
+        deduped = base.distinct(MeituanCoupon.dedup_key).order_by(
+            MeituanCoupon.dedup_key,
+            MeituanCoupon.sale_volume_num.desc(),
+            MeituanCoupon.commission_percent.desc(),
+        ).subquery()
+
+        # 2) 对去重结果按销量降序分页;多取 1 条判断 has_next,只对本页做 from_raw
+        m = aliased(MeituanCoupon, deduped)
+        start = (req.page - 1) * req.page_size
+        rows = db.execute(
+            select(m)
+            # 加 id 作稳定 tiebreaker:同销量同佣金的并列项排序确定,避免跨页重复/漏项
+            .order_by(m.sale_volume_num.desc(), m.commission_percent.desc(), m.id)
+            .offset(start)
+            .limit(req.page_size + 1)
+        ).scalars().all()
     except Exception:  # noqa: BLE001
         logger.exception("[top-sales] 库查询失败,降级返空")
         return CouponListResponse(items=[], has_next=False, search_id=None, status="degraded")
 
-    # 不显示距离:库里的距离是相对城市默认点的(对用户无意义、且误导),与推荐流口径一致。
-    cards = _cards_from_raws(raws, hide_distance=True)
+    has_next = len(rows) > req.page_size
+    cards: list[CouponCard] = []
+    for row in rows[:req.page_size]:
+        try:
+            card = CouponCard.from_raw(row.raw or {})
+        except Exception:  # noqa: BLE001
+            continue
+        if card.product_view_sign:
+            # 不显示距离:库里的距离是相对城市默认点的(对用户无意义、且误导)。
+            # 置空后前端"距离 店名"那行只剩店名、自动顶到最左(店名移到原距离的位置)。
+            # 逻辑与推荐流保持一致
+            card.distance_text = None
+            card.distance_meters = None
+            cards.append(card)
     if not cards and req.page == 1:
         # 命中城市却 0 券:可能该城确无券,也可能 ETL 灌的 city_id 与 city_dict 口径不一致(静默降级的隐患)。
         logger.info("[top-sales] city_id=%s 命中 0 券(该城确无券?或 ETL/city_dict 的 city_id 口径不一致)", city_id)

app/api/v1/wallet.py

@@ -192,7 +192,7 @@ def withdraw_info(
         wechat_bound=bool(u and u.wechat_openid),
         wechat_nickname=u.wechat_nickname if u else None,
         wechat_avatar_url=u.wechat_avatar_url if u else None,
-        transfer_auth_enabled=bool(auth and auth.state == "active" and auth.authorization_id),
+        transfer_auth_enabled=bool(auth and auth.state == "active"),
         tiers=[WithdrawTierOut(**t) for t in crud_wallet.withdraw_tier_states(db, user.id, source)],
     )
 
@@ -335,8 +335,7 @@ def open_transfer_auth(user: CurrentUser, db: DbSession) -> TransferAuthResultOu
 def transfer_auth_status(user: CurrentUser, db: DbSession) -> TransferAuthStatusOut:
     auth = crud_wallet.sync_transfer_auth(db, user.id)
     state = auth.state if auth else "none"
-    enabled = bool(auth and auth.state == "active" and auth.authorization_id)
-    return TransferAuthStatusOut(state=state, enabled=enabled)
+    return TransferAuthStatusOut(state=state, enabled=(state == "active"))
 
 
 @router.post(

app/core/config.py

@@ -188,13 +188,6 @@ class Settings(BaseSettings):
         """京东联盟订单查询凭证齐全。"""
         return bool(self.JD_UNION_APP_KEY and self.JD_UNION_APP_SECRET)
 
-    # 美团 + 京东 CPS 订单自动对账：进程内 worker 每天北京时间 05:00 后跑一轮。
-    # 按更新时间回拉近 N 天（重叠窗口防漏单并刷新状态），order_id 幂等更新；手动接口不受影响。
-    CPS_AUTO_RECONCILE_ENABLED: bool = True
-    CPS_AUTO_RECONCILE_RUN_HOUR: int = 5
-    CPS_AUTO_RECONCILE_LOOKBACK_DAYS: int = 3
-    CPS_AUTO_RECONCILE_CHECK_INTERVAL_SEC: int = 60
-
     # ===== 微信服务号(网页授权) =====
     # CPS 落地页在微信内拿用户 openid(base 静默)/昵称头像(userinfo),做用户级群统计。
     # ⚠️ 区别于 WECHAT_APP_ID(那是 App 移动应用,用于微信支付);这是【已认证服务号】。
@@ -384,9 +377,6 @@ class Settings(BaseSettings):
     MEDIA_ROOT: str = "./data/media"
     MEDIA_URL_PREFIX: str = "/media"
     AVATAR_MAX_BYTES: int = 5 * 1024 * 1024  # 头像最大 5MB
-    # 运营后台上传的新手引导视频上限。视频比图片大一个量级,单独一档;
-    # ⚠️ 改大时同步放宽网关 client_max_body_size(实测 QA 4MiB / prod 32MiB),否则 nginx 先挡下。
-    GUIDE_VIDEO_MAX_BYTES: int = 100 * 1024 * 1024  # 引导视频最大 100MB
 
     # ===== 邀请好友 =====
     # 分享落地页(二维码 / 分享链接指向这里;扫码 → 落地页 → 引导浏览器下载 APK)。

app/core/cps_reconcile_worker.py

@@ -1,428 +0,0 @@
-"""美团、京东 CPS 订单每日自动对账任务。
-
-每天北京时间 `CPS_AUTO_RECONCILE_RUN_HOUR`（默认 05:00）后执行一次，按更新时间
-回拉最近若干天订单并复用 admin CPS 仓储层的幂等 upsert。手动对账接口保持独立、不受影响。
-"""
-from __future__ import annotations
-
-import asyncio
-import contextlib
-import logging
-import os
-import socket
-import time
-from collections.abc import Callable, Iterator
-from datetime import date, datetime, timedelta
-from pathlib import Path
-from uuid import uuid4
-
-from app.admin.repositories import cps as cps_repo
-from app.core.config import settings
-from app.core.rewards import CN_TZ
-from app.db.session import SessionLocal, engine
-
-logger = logging.getLogger("shagua.cps_reconcile")
-_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "cps_reconcile.lock"
-
-
-def _cn_now() -> datetime:
-    return datetime.now(CN_TZ)
-
-
-def _new_run_id(now: datetime) -> str:
-    return f"{now:%Y%m%d-%H%M%S}-{uuid4().hex[:8]}"
-
-
-def _next_run_at(now: datetime, run_hour: int) -> datetime:
-    scheduled = now.replace(hour=run_hour, minute=0, second=0, microsecond=0)
-    return scheduled if now < scheduled else scheduled + timedelta(days=1)
-
-
-def _trigger_for_run(worker_started_at: datetime, now: datetime, run_hour: int) -> str:
-    if worker_started_at.date() == now.date() and worker_started_at.hour >= run_hour:
-        return "startup_catchup"
-    return "scheduled"
-
-
-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]:
-    """同机多进程保护：同一时间只允许一个 CPS 自动对账 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 _empty_result() -> dict:
-    return {
-        "fetched": 0,
-        "inserted": 0,
-        "updated": 0,
-        "pages": 0,
-        "api_requests": 0,
-    }
-
-
-def _platform_log_fields(platform: str, result: dict) -> dict:
-    fields = {
-        "platform": platform,
-        "platform_status": result["status"],
-        "duration_ms": result["duration_ms"],
-    }
-    for key in ("fetched", "inserted", "updated", "pages", "api_requests", "windows"):
-        if key in result:
-            fields[key] = result[key]
-    return fields
-
-
-def _run_platform(
-    *,
-    platform: str,
-    query_time_type: int,
-    common: dict,
-    reconcile: Callable[[], dict],
-) -> tuple[dict, str | None]:
-    started = time.perf_counter()
-    logger.info(
-        "CPS auto reconcile platform started run_id=%s platform=%s",
-        common["run_id"],
-        platform,
-        extra={
-            **common,
-            "event": "cps_reconcile.platform_started",
-            "platform": platform,
-            "query_time_type": query_time_type,
-        },
-    )
-    try:
-        raw_result = reconcile()
-    except Exception as exc:  # noqa: BLE001 - 单平台失败不能阻断另一平台
-        result = {
-            "status": "failed",
-            "duration_ms": round((time.perf_counter() - started) * 1000, 3),
-            "error_type": type(exc).__name__,
-            "error_summary": str(exc)[:500],
-        }
-        logger.exception(
-            "CPS auto reconcile platform failed run_id=%s platform=%s error_type=%s",
-            common["run_id"],
-            platform,
-            type(exc).__name__,
-            extra={
-                **common,
-                "event": "cps_reconcile.platform_failed",
-                **_platform_log_fields(platform, result),
-                "error_type": type(exc).__name__,
-                "error_summary": str(exc)[:500],
-            },
-        )
-        return result, str(exc)
-
-    result = {
-        **raw_result,
-        "status": "success",
-        "duration_ms": round((time.perf_counter() - started) * 1000, 3),
-    }
-    logger.info(
-        "CPS auto reconcile platform completed run_id=%s platform=%s fetched=%s inserted=%s updated=%s",
-        common["run_id"],
-        platform,
-        result.get("fetched", 0),
-        result.get("inserted", 0),
-        result.get("updated", 0),
-        extra={
-            **common,
-            "event": "cps_reconcile.platform_completed",
-            **_platform_log_fields(platform, result),
-        },
-    )
-    return result, None
-
-
-def _skip_platform(platform: str, common: dict) -> dict:
-    result = {
-        **_empty_result(),
-        "status": "skipped",
-        "skipped": "not_configured",
-        "duration_ms": 0.0,
-    }
-    logger.warning(
-        "CPS auto reconcile platform skipped run_id=%s platform=%s reason=not_configured",
-        common["run_id"],
-        platform,
-        extra={
-            **common,
-            "event": "cps_reconcile.platform_skipped",
-            **_platform_log_fields(platform, result),
-            "skip_reason": "not_configured",
-        },
-    )
-    return result
-
-
-def _reconcile_once(
-    now: datetime | None = None,
-    *,
-    run_id: str | None = None,
-    trigger: str = "scheduled",
-) -> dict:
-    """独立拉取美团和京东；单平台异常只记日志，不影响另一平台。"""
-    end = (now or _cn_now()).astimezone(CN_TZ)
-    run_id = run_id or _new_run_id(end)
-    lookback_days = max(1, int(settings.CPS_AUTO_RECONCILE_LOOKBACK_DAYS))
-    start = end - timedelta(days=lookback_days)
-    task_started = time.perf_counter()
-    common = {
-        "run_id": run_id,
-        "trigger": trigger,
-        "scheduled_date": end.date().isoformat(),
-        "app_env": settings.APP_ENV,
-        "db_dialect": engine.dialect.name,
-        "hostname": socket.gethostname(),
-        "pid": os.getpid(),
-        "window_start": start.isoformat(),
-        "window_end": end.isoformat(),
-        "lookback_days": lookback_days,
-    }
-    result = {
-        "run_id": run_id,
-        "trigger": trigger,
-        "window_start": start.isoformat(),
-        "window_end": end.isoformat(),
-        "meituan": None,
-        "jd": None,
-        "errors": {},
-    }
-    logger.info(
-        "CPS auto reconcile started run_id=%s trigger=%s window=%s..%s",
-        run_id,
-        trigger,
-        result["window_start"],
-        result["window_end"],
-        extra={**common, "event": "cps_reconcile.started"},
-    )
-
-    if settings.mt_cps_configured:
-        def reconcile_meituan() -> dict:
-            with SessionLocal() as db:
-                return cps_repo.reconcile_orders(
-                    db,
-                    start_time=int(start.timestamp()),
-                    end_time=int(end.timestamp()),
-                    query_time_type=2,
-                    audit_context=common,
-                )
-
-        result["meituan"], error = _run_platform(
-            platform="meituan",
-            query_time_type=2,
-            common=common,
-            reconcile=reconcile_meituan,
-        )
-        if error is not None:
-            result["errors"]["meituan"] = error
-    else:
-        result["meituan"] = _skip_platform("meituan", common)
-
-    _touch_lock()
-    if settings.jd_union_configured:
-        def reconcile_jd() -> dict:
-            with SessionLocal() as db:
-                return cps_repo.reconcile_jd_orders(
-                    db,
-                    start_time=start,
-                    end_time=end,
-                    query_time_type=3,
-                    audit_context=common,
-                )
-
-        result["jd"], error = _run_platform(
-            platform="jd",
-            query_time_type=3,
-            common=common,
-            reconcile=reconcile_jd,
-        )
-        if error is not None:
-            result["errors"]["jd"] = error
-    else:
-        result["jd"] = _skip_platform("jd", common)
-
-    successes = sum(
-        platform_result.get("status") == "success"
-        for platform_result in (result["meituan"], result["jd"])
-    )
-    if result["errors"]:
-        task_status = "partial_success" if successes else "failed"
-    else:
-        task_status = "success"
-    completed_at = _cn_now()
-    duration_ms = round((time.perf_counter() - task_started) * 1000, 3)
-    next_run_at = _next_run_at(
-        completed_at,
-        min(max(int(settings.CPS_AUTO_RECONCILE_RUN_HOUR), 0), 23),
-    ).isoformat()
-    result.update({
-        "status": task_status,
-        "duration_ms": duration_ms,
-        "manual_retry_required": bool(result["errors"]),
-        "next_run_at": next_run_at,
-    })
-    completion_fields = {
-        **common,
-        "event": "cps_reconcile.completed",
-        "task_status": task_status,
-        "duration_ms": duration_ms,
-        "manual_retry_required": result["manual_retry_required"],
-        "failed_platforms": sorted(result["errors"]),
-        "next_run_at": next_run_at,
-        "meituan_result": result["meituan"],
-        "jd_result": result["jd"],
-    }
-    log_method = logger.info if task_status == "success" else logger.warning
-    log_method(
-        "CPS auto reconcile completed run_id=%s status=%s duration_ms=%s failed_platforms=%s next_run_at=%s",
-        run_id,
-        task_status,
-        duration_ms,
-        ",".join(sorted(result["errors"])) or "-",
-        next_run_at,
-        extra=completion_fields,
-    )
-
-    return result
-
-
-def _should_run(last_run: date | None, now: datetime, run_hour: int) -> bool:
-    return last_run != now.date() and now.hour >= run_hour
-
-
-async def _run_loop() -> None:
-    interval = max(30, int(settings.CPS_AUTO_RECONCILE_CHECK_INTERVAL_SEC))
-    run_hour = min(max(int(settings.CPS_AUTO_RECONCILE_RUN_HOUR), 0), 23)
-    lock_stale_after = max(interval * 3, 1800)
-    with _single_instance_lock(lock_stale_after) as lock_acquired:
-        if not lock_acquired:
-            logger.warning(
-                "CPS auto reconcile worker skipped: another worker owns lock",
-                extra={
-                    "event": "cps_reconcile.worker_skipped",
-                    "skip_reason": "lock_not_acquired",
-                    "pid": os.getpid(),
-                },
-            )
-            return
-        await _run_locked_loop(interval, run_hour)
-
-
-async def _run_locked_loop(interval: int, run_hour: int) -> None:
-    worker_started_at = _cn_now()
-    logger.info(
-        "CPS auto reconcile worker started run_hour=%s interval=%ss lookback_days=%s",
-        run_hour,
-        interval,
-        settings.CPS_AUTO_RECONCILE_LOOKBACK_DAYS,
-        extra={
-            "event": "cps_reconcile.worker_started",
-            "pid": os.getpid(),
-            "hostname": socket.gethostname(),
-            "app_env": settings.APP_ENV,
-            "db_dialect": engine.dialect.name,
-            "run_hour": run_hour,
-            "interval_sec": interval,
-            "lookback_days": settings.CPS_AUTO_RECONCILE_LOOKBACK_DAYS,
-            "next_run_at": _next_run_at(worker_started_at, run_hour).isoformat(),
-        },
-    )
-    last_run: date | None = None
-    try:
-        while True:
-            run_id: str | None = None
-            try:
-                _touch_lock()
-                now = _cn_now()
-                if _should_run(last_run, now, run_hour):
-                    run_id = _new_run_id(now)
-                    trigger = _trigger_for_run(worker_started_at, now, run_hour)
-                    await asyncio.to_thread(
-                        _reconcile_once,
-                        now,
-                        run_id=run_id,
-                        trigger=trigger,
-                    )
-                    last_run = now.date()
-            except Exception:  # noqa: BLE001 - 后台任务不能因单次异常退出
-                logger.exception(
-                    "CPS auto reconcile unexpected worker error",
-                    extra={
-                        "event": "cps_reconcile.unexpected_failed",
-                        "run_id": run_id or "unassigned",
-                        "pid": os.getpid(),
-                    },
-                )
-            await asyncio.sleep(interval)
-    except asyncio.CancelledError:
-        logger.info(
-            "CPS auto reconcile worker stopped",
-            extra={"event": "cps_reconcile.worker_stopped", "pid": os.getpid()},
-        )
-        raise
-
-
-def start_cps_reconcile_worker() -> asyncio.Task | None:
-    if not settings.CPS_AUTO_RECONCILE_ENABLED:
-        logger.info(
-            "CPS auto reconcile disabled",
-            extra={
-                "event": "cps_reconcile.worker_skipped",
-                "skip_reason": "disabled",
-            },
-        )
-        return None
-    if not settings.mt_cps_configured and not settings.jd_union_configured:
-        logger.warning(
-            "CPS auto reconcile not started: Meituan and JD credentials are missing",
-            extra={
-                "event": "cps_reconcile.worker_skipped",
-                "skip_reason": "all_platform_credentials_missing",
-            },
-        )
-        return None
-    return asyncio.create_task(_run_loop(), name="cps-auto-reconcile")
-
-
-async def stop_cps_reconcile_worker(task: asyncio.Task | None) -> None:
-    if task is None:
-        return
-    task.cancel()
-    with contextlib.suppress(asyncio.CancelledError):
-        await task

app/core/logging.py

@@ -22,13 +22,13 @@ from __future__ import annotations
 import json
 import logging
 import os
-import shutil
 import sys
 from contextvars import ContextVar
 from datetime import datetime
 from logging.handlers import RotatingFileHandler
 from pathlib import Path
 
+
 # 请求级 trace_id:入口(如 compare.py 透传壳)set 之后, 本请求上下文(含 run_in_threadpool
 # 拷贝出去的线程)内所有日志自动带上。默认空串 = 非请求上下文(启动/后台 worker)。
 trace_id_ctx: ContextVar[str] = ContextVar("trace_id", default="")
@@ -91,55 +91,6 @@ class TextFormatter(logging.Formatter):
         return f"{base} trace={tid}" if tid else base
 
 
-class SafeRotatingFileHandler(RotatingFileHandler):
-    """Windows 下不会被外部句柄卡死的 RotatingFileHandler。
-
-    stdlib 轮转靠 rename 活动文件(app-server.log → .1);Windows 只要有别的句柄(IDE 索引、
-    app.admin.main 第二进程、残留 --reload worker、杀软扫描)开着它, rename 就 WinError 32,
-    轮转永久卡死——文件停在 maxBytes、之后每条日志被丢。这里 Windows 改用 copytruncate:把活动
-    文件拷进备份、再通过自己的句柄原地清空, 从不 rename 活动文件, 故外部句柄开着也能转。
-    POSIX(生产 Linux)rename 打开中的文件本就合法, 保留 stdlib 的原子轮转不变。
-
-    代价:copytruncate 在“拷贝→清空”极窄窗口内并发写可能丢几行(仅跨进程;同进程 emit 有
-    handler 锁串行, 无此问题)。对本地开发日志可接受。
-    """
-
-    def doRollover(self) -> None:
-        if os.name != "nt":
-            super().doRollover()
-            return
-        if self.stream is None:
-            self.stream = self._open()
-        else:
-            self.stream.flush()
-        try:
-            self._copytruncate_backups()
-        except OSError:
-            # 备份腾挪是尽力而为:任一备份被占用也绝不能挡住下面的清空, 否则活动文件继续涨、
-            # 轮转又卡死——那就白改了。
-            pass
-        # 通过自己独占的句柄原地清空:不涉及 rename, 外部只读句柄不受影响。
-        self.stream.seek(0)
-        self.stream.truncate()
-        self.stream.flush()
-
-    def _copytruncate_backups(self) -> None:
-        """把 .N-1→.N 逐级腾挪, 再把活动文件拷到 .1(不动活动文件本身)。"""
-        if self.backupCount <= 0:
-            return
-        for i in range(self.backupCount - 1, 0, -1):
-            sfn = self.rotation_filename(f"{self.baseFilename}.{i}")
-            dfn = self.rotation_filename(f"{self.baseFilename}.{i + 1}")
-            if os.path.exists(sfn):
-                if os.path.exists(dfn):
-                    os.remove(dfn)
-                os.replace(sfn, dfn)
-        dfn = self.rotation_filename(f"{self.baseFilename}.1")
-        if os.path.exists(dfn):
-            os.remove(dfn)
-        shutil.copyfile(self.baseFilename, dfn)
-
-
 _CONFIGURED = False
 
 
@@ -175,7 +126,7 @@ def setup_logging(debug: bool = False) -> None:
         Path(os.getenv("LOG_DIR", "logs")) / "app-server.log"
     )
     Path(log_file).parent.mkdir(parents=True, exist_ok=True)
-    file_handler = SafeRotatingFileHandler(
+    file_handler = RotatingFileHandler(
         log_file, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8",
     )
     file_handler.setFormatter(JsonFormatter(service))

app/core/media.py

@@ -77,35 +77,6 @@ def save_feedback_qr(data: bytes) -> str:
     return _save_named("feedback_qr", "qr", data)
 
 
-def _sniff_video_ext(data: bytes) -> str | None:
-    """按魔数判定视频类型,返回扩展名;非支持类型返回 None。
-
-    只认 MP4 家族(ISO BMFF):`....ftyp` 在偏移 4。Android ExoPlayer 与浏览器 <video>
-    都稳吃 H.264/AAC 的 mp4;放开 mkv/avi 只会让端上放不出来,不如在入口就挡掉。
-    """
-    if len(data) >= 12 and data[4:8] == b"ftyp":
-        return ".mp4"
-    return None
-
-
-def save_guide_video(data: bytes) -> str:
-    """保存新手引导视频(运营后台上传的运营素材),返回相对 URL(`/media/guide_video/<file>`)。
-
-    与图片分开一套校验:体积上限走 [settings.GUIDE_VIDEO_MAX_BYTES],类型只认 MP4。
-    """
-    if not data:
-        raise MediaError("空文件")
-    limit = settings.GUIDE_VIDEO_MAX_BYTES
-    if len(data) > limit:
-        raise MediaError(f"视频过大(上限 {limit // (1024 * 1024)}MB)")
-    if _sniff_video_ext(data) is None:
-        raise MediaError("仅支持 MP4 视频(H.264 编码)")
-
-    fname = f"guide_{secrets.token_hex(8)}.mp4"
-    (_media_dir("guide_video") / fname).write_bytes(data)
-    return f"{settings.MEDIA_URL_PREFIX}/guide_video/{fname}"
-
-
 def save_cps_image(admin_id: int, data: bytes) -> str:
     """保存 CPS 活动落地页图,返回相对 URL(`/media/cps/<file>`)。admin_id 入文件名便于追溯。"""
     return _save_image("cps", admin_id, data)
@@ -145,8 +116,3 @@ def delete_avatar(url: str | None) -> None:
 def delete_feedback_qr(url: str | None) -> None:
     """删除本服务托管的旧反馈页二维码文件;外部 URL 或空值不处理。"""
     _delete_managed("feedback_qr", url)
-
-
-def delete_guide_video(url: str | None) -> None:
-    """删除本服务托管的旧新手引导视频文件;外部 URL 或空值不处理。"""
-    _delete_managed("guide_video", url)

app/integrations/meituan.py

@@ -11,7 +11,6 @@ import hashlib
 import hmac
 import json
 import logging
-import threading
 import time
 from typing import Any
 
@@ -26,44 +25,6 @@ class MeituanCpsError(Exception):
     """美团 CPS 接口调用失败。"""
 
 
-# ────────────────────── 共享 httpx.Client(连接池 + keep-alive) ──────────────────────
-# 原实现每次 _call 都 `with httpx.Client(...)` 新建再关掉,等于每发一次请求就:
-#   ① 重建一套 SSL 上下文(加载 certifi CA,实测几百 ms) ② 重做一次 TLS 握手 ③ 用完即弃连接。
-# 首页 feed 一次翻页要向美团发好几次请求(外卖/到店两路,「距离最近」还要续页),这份固定开销被成倍放大,
-# 是「滑到底部加载很慢」的一大块。改成进程内单例:握手一次、后续走 keep-alive 复用。
-# httpx.Client 本身线程安全,可被 feed 的 ThreadPoolExecutor 两条抓取线程共用。
-_client: httpx.Client | None = None
-_client_lock = threading.Lock()
-
-
-def get_client() -> httpx.Client:
-    """取美团 CPS 共享 client(幂等,懒建)。lifespan 启动时预热,把建 SSL 的成本摊到启动。"""
-    global _client
-    if _client is None:
-        with _client_lock:
-            if _client is None:
-                # 走 MT_CPS_PROXY(本机开发直连会 SSL EOF,必须走代理;线上留空=直连)。
-                # trust_env=False:不读进程环境的 HTTP_PROXY,只认配置,避免被错误/失效代理误导。
-                _client = httpx.Client(
-                    proxy=settings.MT_CPS_PROXY or None,
-                    trust_env=False,
-                    timeout=settings.MT_CPS_TIMEOUT_SEC,
-                    # 池子留足:feed 每个请求会起 2 条抓取线程,并发用户多时别在池上排队
-                    # (排满会等到 MT_CPS_TIMEOUT_SEC 抛 PoolTimeout,表现成"又慢又降级")。
-                    limits=httpx.Limits(max_keepalive_connections=16, max_connections=32),
-                )
-    return _client
-
-
-def close_client() -> None:
-    """lifespan 关停时调,优雅关连接池。"""
-    global _client
-    with _client_lock:
-        if _client is not None:
-            _client.close()
-            _client = None
-
-
 def _content_md5(body: bytes) -> str:
     return base64.b64encode(hashlib.md5(body).digest()).decode()
 
@@ -101,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 = get_client().post(url, content=body, headers=headers)
+        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/main.py

@@ -29,7 +29,6 @@ from app.api.v1.coupon import router as coupon_router
 from app.api.v1.cps_redirect import router as cps_redirect_router
 from app.api.v1.device import router as device_router
 from app.api.v1.feedback import router as feedback_router
-from app.api.v1.guide_video import router as guide_video_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.notifications import router as notifications_router
@@ -44,10 +43,6 @@ from app.api.v1.user import router as user_router
 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.cps_reconcile_worker import (
-    start_cps_reconcile_worker,
-    stop_cps_reconcile_worker,
-)
 from app.core.daily_exchange_worker import (
     start_daily_exchange_worker,
     stop_daily_exchange_worker,
@@ -71,7 +66,6 @@ from app.core.withdraw_reconcile_worker import (
     start_withdraw_reconcile_worker,
     stop_withdraw_reconcile_worker,
 )
-from app.integrations import meituan as mt_meituan
 
 setup_logging(debug=settings.APP_DEBUG)
 logger = logging.getLogger("shagua.main")
@@ -88,9 +82,6 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
         settings.DATABASE_URL.split("://", 1)[0],
     )
     get_pricebot_client()  # 预热透传 client:把建 SSL 上下文的一次性成本付在启动,首个领券请求即热
-    if settings.mt_cps_configured:
-        # 同理预热美团 CPS client(TLS 上下文 + 连接池建一次,后续 keep-alive 复用)
-        mt_meituan.get_client()
     try:
         # 预热离线地理库:首次加载 ~2.5M 行 CSV + 建 KDTree,摊到启动、不砸首个按城市过滤的请求
         from app.utils import geo
@@ -98,7 +89,6 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
     except Exception:  # noqa: BLE001
         logger.exception("reverse_geocoder 预热失败(城市反查将在首个请求时懒加载)")
     reconcile_task = start_withdraw_reconcile_worker()
-    cps_reconcile_task = start_cps_reconcile_worker()
     heartbeat_task = start_heartbeat_monitor()
     daily_exchange_task = start_daily_exchange_worker()
     observe_task = start_observe_worker()
@@ -108,12 +98,10 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
     finally:
         await stop_heartbeat_monitor(heartbeat_task)
         await stop_withdraw_reconcile_worker(reconcile_task)
-        await stop_cps_reconcile_worker(cps_reconcile_task)
         await stop_daily_exchange_worker(daily_exchange_task)
         await stop_observe_worker(observe_task)
         await stop_inactivity_reset_worker(inactivity_task)
         await aclose_pricebot_client()
-        mt_meituan.close_client()
         logger.info("shutting down")
 
 
@@ -160,8 +148,6 @@ app.include_router(signin_router)
 app.include_router(tasks_router)
 app.include_router(savings_router)
 app.include_router(ad_router)
-# 新手引导视频(领券等候浮层前 N 次替代广告;运营后台传片,见 repositories/guide_video.py)
-app.include_router(guide_video_router)
 app.include_router(order_router)
 app.include_router(report_router)
 # 消息通知中心(PRD;数据落库 notification 表,见 repositories/notification.py)

app/models/__init__.py

@@ -21,14 +21,12 @@ from app.models.cps_wx_user import CpsWxUser  # noqa: F401
 from app.models.comparison_milestone import ComparisonMilestoneClaim  # noqa: F401
 from app.models.device import DeviceLiveness  # noqa: F401
 from app.models.coupon_state import (  # noqa: F401
-    CouponClaimEvent,
     CouponClaimRecord,
     CouponDailyCompletion,
     CouponPromptEngagement,
     CouponSession,
 )
 from app.models.feedback import Feedback  # noqa: F401
-from app.models.guide_video import GuideVideoPlay  # noqa: F401
 from app.models.inactivity import (  # noqa: F401
     InactivityNotificationLog,
     InactivityResetLog,

app/models/coupon_state.py

@@ -96,40 +96,6 @@ class CouponClaimRecord(Base):
         )
 
 
-class CouponClaimEvent(Base):
-    """一次领券任务中的单券结果，按 ``(trace_id, coupon_id)`` 幂等。"""
-
-    __tablename__ = "coupon_claim_event"
-    __table_args__ = (
-        UniqueConstraint(
-            "trace_id", "coupon_id",
-            name="uq_coupon_claim_event_trace_coupon",
-        ),
-        Index("ix_coupon_claim_event_date_env", "claim_date", "app_env"),
-    )
-
-    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
-    trace_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
-    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)
-    claim_date: Mapped[date] = mapped_column(Date, nullable=False)
-    status: Mapped[str] = mapped_column(String(24), nullable=False)
-    app_env: Mapped[str | None] = mapped_column(String(16), index=True, nullable=True)
-    vendor: Mapped[str | None] = mapped_column(String(48), nullable=True)
-    coupon_name: Mapped[str | None] = mapped_column(String(128), nullable=True)
-    claimed_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
-    reason: Mapped[str | None] = mapped_column(String(255), nullable=True)
-    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,
-    )
-
-
 class CouponDailyCompletion(Base):
     """按 (device, 自然日) 记"今天是否已跑完整轮领券(到 done 帧)"——首页置灰源。
 

app/models/guide_video.py

@@ -1,73 +0,0 @@
-"""新手引导视频播放记录(领券浮层前 N 次用它替代广告)。
-
-产品规则(2026-07 拍板):新用户点「一键自动领取」后的等候浮层,**前 3 次**不放广告,
-改放运营后台上传的引导视频;每次固定 120 金币,中途关闭也算看完照发。
-
-口径:
-- **计次按账号**(user_id),与设备无关 —— 换设备不重新送 3 次。
-- **开播即计数**:客户端每次要展示浮层时调 `/api/v1/guide-video/start`,服务端当场
-  写一行(status='playing')并返回 play_token;`COUNT(*)` 即已用次数。用户中途 kill
-  App 也算用掉一次(产品选定口径,防反复进出刷金币)。
-- **发币幂等**靠 play_token 定位 + `status='playing'` 条件更新:并发两次上报只有一次
-  改到行(另一次 rowcount=0),所以只发一次币。光有 play_token 唯一键挡不住 —— 发币走的是
-  UPDATE,不 INSERT,撞不到任何唯一键。
-- **次数上限**靠 (user_id, seq) 唯一键兜底,防并发 /start 绕过 COUNT 判定(见下)。
-
-与广告收益(ad_feed_reward_record)彻底分离:引导视频不是广告,不该进广告收益报表。
-"""
-from __future__ import annotations
-
-from datetime import datetime
-
-from sqlalchemy import DateTime, ForeignKey, Index, Integer, String, UniqueConstraint, func
-from sqlalchemy.orm import Mapped, mapped_column
-
-from app.db.base import Base
-
-
-class GuideVideoPlay(Base):
-    """一次引导视频播放一行。开播时建(status='playing'),发币后置 'granted'。"""
-
-    __tablename__ = "guide_video_play"
-    __table_args__ = (
-        # 客户端幂等键:同一次播放重复上报奖励只发一次。
-        UniqueConstraint("play_token", name="uq_guide_video_play_token"),
-        # 次数上限的**硬约束**:start_play 是无锁 check-then-insert(读 COUNT 算 seq 再插),
-        # N 个并发 /start 会都读到同一个已用次数、算出同一个 seq,不拦就能各拿一个 token、
-        # 各发一次金币,3 次上限形同虚设(改包即可无限刷)。seq 唯一 → 并发同 seq 必撞,
-        # start_play 捕获 IntegrityError 降级成"这次不放视频"。
-        # 用 unique Index 而非 UniqueConstraint:与迁移里的 create_index 对齐(SQLite 加约束
-        # 要整表重建),autogenerate 才不会每次报一条假 diff。
-        Index("uq_guide_video_play_user_seq", "user_id", "seq", unique=True),
-    )
-
-    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
-    user_id: Mapped[int] = mapped_column(
-        Integer, ForeignKey("user.id"), index=True, nullable=False
-    )
-    # 服务端生成下发给客户端的幂等键(uuid hex)。
-    play_token: Mapped[str] = mapped_column(String(64), nullable=False)
-    # 触发场景:目前只有 coupon(领券等候浮层);留字段以便日后比价等场景复用。
-    scene: Mapped[str] = mapped_column(String(16), nullable=False, default="coupon")
-    # 本账号第几次(1-based),= 建行时已有行数 + 1。日常判定仍以 COUNT 为准,但 (user_id, seq)
-    # 唯一键让并发 /start 只能成一个 —— 见 __table_args__。
-    seq: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
-    # 当次下发的视频地址(运营换片后能回溯用户当时看的是哪支)。
-    video_url: Mapped[str | None] = mapped_column(String(512), nullable=True)
-    # 实发金币;未发时 0。
-    coin: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
-    # playing(已开播未发币) / granted(已发币)。
-    status: Mapped[str] = mapped_column(String(16), nullable=False, default="playing")
-    # 客户端上报时是否播完(true=自然播完 / false=中途关闭)。仅留痕:两者都发币。
-    completed: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
-
-    started_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
-    )
-    granted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
-
-    def __repr__(self) -> str:  # pragma: no cover
-        return (
-            f"<GuideVideoPlay user={self.user_id} seq={self.seq} "
-            f"{self.status} coin={self.coin}>"
-        )

app/models/meituan_coupon.py

@@ -23,7 +23,7 @@ from __future__ import annotations
 
 from datetime import datetime
 
-from sqlalchemy import JSON, DateTime, Float, Index, Integer, String, UniqueConstraint, func
+from sqlalchemy import JSON, DateTime, Float, Integer, String, UniqueConstraint, func
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.orm import Mapped, mapped_column
 
@@ -95,23 +95,3 @@ class MeituanCoupon(Base):
             f"<MeituanCoupon id={self.id} source={self.source} "
             f"name={self.name!r} sale={self.sale_volume} comm={self.commission_percent}>"
         )
-
-
-# 首页「销量最高 / 智能推荐」两个 tab 的分页索引(见 api/v1/meituan.py 的 _paged_dedup_ids)。
-# 两条 tab 都是 `WHERE city_id=? [+过滤] ORDER BY dedup_key, <排序键> DESC` 的 DISTINCT ON 去重,
-# 列顺序对齐后 Postgres 可以顺着索引流式去重,免掉整城数据的排序 —— 否则每翻一页都要把该城
-# 全部券重排一遍(下滑到底越来越慢的根因之一)。
-# 定义放在类外:__table_args__ 里拿不到还没建好的类属性,写不了 .desc()。
-Index(
-    "ix_meituan_coupon_city_dedup_sales",
-    MeituanCoupon.city_id,
-    MeituanCoupon.dedup_key,
-    MeituanCoupon.sale_volume_num.desc(),
-    MeituanCoupon.commission_percent.desc(),
-)
-Index(
-    "ix_meituan_coupon_city_dedup_comm",
-    MeituanCoupon.city_id,
-    MeituanCoupon.dedup_key,
-    MeituanCoupon.commission_percent.desc(),
-)

app/repositories/comparison.py

@@ -5,7 +5,7 @@
 """
 from __future__ import annotations
 
-from datetime import datetime, timedelta
+from datetime import datetime
 
 from sqlalchemy import func, or_, select
 from sqlalchemy.orm import Session, defer
@@ -14,19 +14,8 @@ from app.core.rewards import CN_TZ
 from app.models.ad_feed_reward import AdFeedRewardRecord
 from app.models.comparison import ComparisonRecord
 from app.models.savings import SavingsRecord
-from app.models.user import User
 from app.schemas.compare_record import ComparisonRecordIn
 
-DAILY_COMPARE_START_LIMIT = 100
-
-
-class DailyCompareStartLimitExceeded(Exception):
-    """The authenticated user has consumed today's comparison-start quota."""
-
-
-class ComparisonTraceOwnershipError(Exception):
-    """A trace id already belongs to a different authenticated user."""
-
 
 def _yuan_to_cents(yuan: float | None) -> int | None:
     """元(float)→ 分(int)。None 透传。"""
@@ -254,78 +243,6 @@ def _get_by_trace(db: Session, trace_id: str) -> ComparisonRecord | None:
     ).scalar_one_or_none()
 
 
-def reserve_daily_start(
-    db: Session,
-    *,
-    user_id: int,
-    trace_id: str,
-    business_type: str = "food",
-    device_id: str | None = None,
-    now: datetime | None = None,
-) -> tuple[ComparisonRecord, int]:
-    """Atomically reserve one of a user's 100 Beijing-day comparison starts.
-
-    ``trace_id`` makes client retries idempotent. Locking the user row serializes
-    concurrent starts for one account, so parallel requests cannot both consume
-    the final available slot. The reservation is the existing ``running``
-    comparison row; later result reporting updates that same row.
-    """
-    db.execute(select(User.id).where(User.id == user_id).with_for_update()).scalar_one()
-
-    existing = _get_by_trace(db, trace_id)
-    if existing is not None:
-        if existing.user_id not in (None, user_id):
-            raise ComparisonTraceOwnershipError
-        if existing.user_id is None:
-            existing.user_id = user_id
-            if existing.device_id is None and device_id:
-                existing.device_id = device_id
-            db.commit()
-            db.refresh(existing)
-
-        existing_at = existing.created_at
-        if existing_at.tzinfo is not None:
-            existing_at = existing_at.astimezone(CN_TZ).replace(tzinfo=None)
-        day_start = existing_at.replace(hour=0, minute=0, second=0, microsecond=0)
-        day_end = day_start + timedelta(days=1)
-        used = db.scalar(
-            select(func.count(ComparisonRecord.id)).where(
-                ComparisonRecord.user_id == user_id,
-                ComparisonRecord.created_at >= day_start,
-                ComparisonRecord.created_at < day_end,
-            )
-        ) or 0
-        return existing, int(used)
-
-    current = now or datetime.now(CN_TZ)
-    if current.tzinfo is not None:
-        current = current.astimezone(CN_TZ).replace(tzinfo=None)
-    day_start = current.replace(hour=0, minute=0, second=0, microsecond=0)
-    day_end = day_start + timedelta(days=1)
-    used = db.scalar(
-        select(func.count(ComparisonRecord.id)).where(
-            ComparisonRecord.user_id == user_id,
-            ComparisonRecord.created_at >= day_start,
-            ComparisonRecord.created_at < day_end,
-        )
-    ) or 0
-    if used >= DAILY_COMPARE_START_LIMIT:
-        raise DailyCompareStartLimitExceeded
-
-    rec = ComparisonRecord(
-        trace_id=trace_id,
-        user_id=user_id,
-        business_type=business_type or "food",
-        device_id=device_id,
-        status="running",
-        created_at=current,
-    )
-    db.add(rec)
-    db.commit()
-    db.refresh(rec)
-    return rec, int(used) + 1
-
-
 def harvest_running(
     db: Session,
     *,

app/repositories/coupon_state.py

@@ -14,7 +14,6 @@ from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
 from app.models.coupon_state import (
-    CouponClaimEvent,
     CouponClaimRecord,
     CouponDailyCompletion,
     CouponPromptEngagement,
@@ -165,12 +164,11 @@ def record_claims(
     results: list[dict],
     app_env: str | None = None,
 ) -> int:
-    """一批券领取结果同时写入每日资产表和逐次事件表。
+    """一批券领取结果幂等写入,返回写入(新增 + 更新)条数。
 
     results 单项取自 pricebot 的 last_coupon_result / done.coupon_results,识别字段:
     coupon_id(必需)/ status(必需)/ name / vendor / reason /(display_count)。
-    - CouponClaimRecord 按 (device, coupon_id, 今天) 幂等，供每日资产口径使用。
-    - CouponClaimEvent 按 (trace_id, coupon_id) 幂等，供 admin 逐场统计使用。
+    (device, coupon_id, 今天) 唯一:重复上报同张券走更新(status 以最后一次为准)。
     """
     today = today_cn()
     written = 0
@@ -210,41 +208,6 @@ def record_claims(
                 claimed_count=count, trace_id=trace_id, reason=r.get("reason"),
                 extra=r,
             ))
-        if trace_id:
-            event = db.execute(
-                select(CouponClaimEvent).where(
-                    CouponClaimEvent.trace_id == trace_id,
-                    CouponClaimEvent.coupon_id == coupon_id,
-                )
-            ).scalar_one_or_none()
-            if event is not None:
-                event.device_id = device_id
-                event.status = status
-                event.reason = r.get("reason")
-                event.vendor = r.get("vendor")
-                event.coupon_name = r.get("name")
-                event.extra = r
-                if user_id is not None:
-                    event.user_id = user_id
-                if count is not None:
-                    event.claimed_count = count
-                if app_env is not None:
-                    event.app_env = app_env
-            else:
-                db.add(CouponClaimEvent(
-                    trace_id=trace_id,
-                    device_id=device_id,
-                    user_id=user_id,
-                    coupon_id=coupon_id,
-                    claim_date=today,
-                    status=status,
-                    app_env=app_env,
-                    vendor=r.get("vendor"),
-                    coupon_name=r.get("name"),
-                    claimed_count=count,
-                    reason=r.get("reason"),
-                    extra=r,
-                ))
         written += 1
     if written == 0:
         return 0

app/repositories/guide_video.py

@@ -1,290 +0,0 @@
-"""新手引导视频:运营配置读写 + 播放计次 + 发币。
-
-**配置**(开关 / 视频地址 / 前几次 / 每次金币)整体作为一个 JSON 存进通用 app_config 表
-(key=coupon_guide_video),写法完全对齐 feedback_qr —— 不进 CONFIG_DEFS,所以不会污染
-系统配置页的通用列表,由本模块独占维护。
-
-**计次**按账号(user_id)、**开播即计数**:客户端每次要展示领券等候浮层时调
-`/api/v1/guide-video/start`,命中则当场写一行 guide_video_play(status='playing')。
-已用次数 = 该账号的行数,达到 max_plays 后不再下发,客户端改放广告(原逻辑)。
-COUNT 判定本身无锁,真正卡住次数上限的是 (user_id, seq) 唯一键:并发 /start 只能成一个。
-
-**发币**幂等键是 play_token,落地方式是 `status='playing' → 'granted'` 的**条件更新**:
-同一次播放重复上报只入账一次(网络重试 / 关闭与播完同时触发都靠它挡住)。
-中途关闭也照发 —— 产品拍板「中途关闭也算看完」。
-
-两处都是直接铸币的路径,改动前先看 `start_play` / `grant_play` 上的并发注释。
-"""
-from __future__ import annotations
-
-import uuid
-from datetime import datetime
-from typing import Any
-
-from sqlalchemy import func, select, update
-from sqlalchemy.exc import IntegrityError
-from sqlalchemy.orm import Session
-
-from app.core import rewards
-from app.models.app_config import AppConfig
-from app.models.guide_video import GuideVideoPlay
-from app.repositories import wallet as crud_wallet
-
-_KEY = "coupon_guide_video"
-
-#: 金币流水 biz_type。客户端收益明细按它显示「新手引导视频奖励」。
-BIZ_TYPE = "guide_video"
-
-# 默认值 = 「运营还没配」时的行为:video_url 为空 → 一律不下发引导视频,浮层维持现状(放广告)。
-# 所以本功能上线后**不配视频就等于没上线**,不会影响存量用户。
-_DEFAULTS: dict[str, Any] = {
-    "enabled": True,
-    "video_url": None,   # None/空 = 未配片 → 不下发,浮层照旧放广告
-    "max_plays": 3,      # 每个账号前 N 次浮层放引导视频
-    "reward_coin": 120,  # 每次固定金币
-}
-
-_FIELDS = tuple(_DEFAULTS.keys())
-
-# 后台可配范围的护栏:防手滑把次数/金币填成天文数字(配置直接决定发币)。
-MAX_PLAYS_LIMIT = 50
-REWARD_COIN_LIMIT = 10_000
-
-
-# ===== 配置 =====
-
-
-def _merge(raw: Any) -> dict[str, Any]:
-    """DB 里(可能不全的)dict 叠加到默认上,得到完整配置(4 个字段,无 updated_at)。"""
-    out = dict(_DEFAULTS)
-    if isinstance(raw, dict):
-        for k in _FIELDS:
-            v = raw.get(k)
-            if v is not None:
-                out[k] = v
-    return out
-
-
-def get_config(db: Session) -> dict[str, Any]:
-    """完整配置 + updated_at(admin 读 / 业务读共用)。"""
-    row = db.get(AppConfig, _KEY)
-    cfg = _merge(row.value if row is not None else None)
-    cfg["updated_at"] = row.updated_at.isoformat() if row is not None and row.updated_at else None
-    return cfg
-
-
-def _write(db: Session, value: dict[str, Any], *, admin_id: int, commit: bool) -> dict[str, Any]:
-    """整体覆写该行(value 须为完整字段 dict),返回合并后的完整配置(含 updated_at)。"""
-    row = db.get(AppConfig, _KEY)
-    if row is None:
-        row = AppConfig(key=_KEY, value=value, updated_by_admin_id=admin_id)
-        db.add(row)
-    else:
-        row.value = value  # 整体重新赋值,SQLAlchemy 才侦测得到变更
-        row.updated_by_admin_id = admin_id
-    if commit:
-        db.commit()
-        db.refresh(row)
-    else:
-        db.flush()
-    out = _merge(row.value)
-    out["updated_at"] = row.updated_at.isoformat() if row.updated_at else None
-    return out
-
-
-def update_config(
-    db: Session,
-    *,
-    enabled: bool | None = None,
-    max_plays: int | None = None,
-    reward_coin: int | None = None,
-    admin_id: int,
-    commit: bool = True,
-) -> tuple[dict[str, Any], dict[str, Any]]:
-    """改开关 / 次数 / 金币(只改传了的字段;视频走 set_video)。返回 (before, after) 供审计。"""
-    row = db.get(AppConfig, _KEY)
-    before = _merge(row.value if row is not None else None)
-    new_value = {k: before[k] for k in _FIELDS}
-    if enabled is not None:
-        new_value["enabled"] = enabled
-    if max_plays is not None:
-        new_value["max_plays"] = max(0, min(int(max_plays), MAX_PLAYS_LIMIT))
-    if reward_coin is not None:
-        new_value["reward_coin"] = max(0, min(int(reward_coin), REWARD_COIN_LIMIT))
-    after = _write(db, new_value, admin_id=admin_id, commit=commit)
-    return before, after
-
-
-def set_video(
-    db: Session, video_url: str | None, *, admin_id: int, commit: bool = True
-) -> tuple[dict[str, Any], dict[str, Any]]:
-    """设置/清空引导视频地址。返回 (before, after);before['video_url'] 供调用方删旧文件。"""
-    row = db.get(AppConfig, _KEY)
-    before = _merge(row.value if row is not None else None)
-    new_value = {k: before[k] for k in _FIELDS}
-    new_value["video_url"] = video_url
-    after = _write(db, new_value, admin_id=admin_id, commit=commit)
-    return before, after
-
-
-# ===== 播放计次 =====
-
-
-def used_plays(db: Session, user_id: int) -> int:
-    """该账号已用掉的引导视频次数(开播即算,含未发币的)。"""
-    return int(
-        db.execute(
-            select(func.count()).select_from(GuideVideoPlay).where(
-                GuideVideoPlay.user_id == user_id
-            )
-        ).scalar_one()
-    )
-
-
-def play_stats(db: Session) -> dict[str, int]:
-    """全站播放统计(admin 页展示):总播放次数 / 其中已发币次数。"""
-    total = int(
-        db.execute(select(func.count()).select_from(GuideVideoPlay)).scalar_one()
-    )
-    granted = int(
-        db.execute(
-            select(func.count()).select_from(GuideVideoPlay).where(
-                GuideVideoPlay.status == "granted"
-            )
-        ).scalar_one()
-    )
-    return {"total_plays": total, "granted_plays": granted}
-
-
-def start_play(
-    db: Session, user_id: int, *, scene: str = "coupon", commit: bool = True
-) -> dict[str, Any]:
-    """决定这次浮层是否放引导视频;命中则**当场计次**并返回 play_token。
-
-    返回 dict:
-      should_play  是否放引导视频(False → 客户端照旧放广告)
-      video_url    相对地址(/media/...);客户端自行拼 BASE_URL
-      play_token   发币幂等键(should_play=False 时为空串)
-      reward_coin  播完/中途关闭都发的固定金币
-      seq / remaining  第几次 / 发完这次还剩几次(仅展示与排查用)
-    """
-    cfg = get_config(db)
-    video_url = (cfg.get("video_url") or "").strip()
-    max_plays = int(cfg.get("max_plays") or 0)
-    reward_coin = int(cfg.get("reward_coin") or 0)
-    used = used_plays(db, user_id)
-
-    def _miss(used_now: int) -> dict[str, Any]:
-        return {
-            "should_play": False,
-            "video_url": None,
-            "play_token": "",
-            "reward_coin": reward_coin,
-            "seq": used_now,
-            "remaining": max(0, max_plays - used_now),
-        }
-
-    if not cfg.get("enabled") or not video_url or max_plays <= 0 or used >= max_plays:
-        return _miss(used)
-
-    seq = used + 1
-    play = GuideVideoPlay(
-        user_id=user_id,
-        play_token=uuid.uuid4().hex,
-        scene=scene,
-        seq=seq,
-        video_url=video_url,
-        coin=0,
-        status="playing",
-        completed=0,
-        started_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
-    )
-    db.add(play)
-    # 上面的 COUNT 判定是无锁 check-then-insert:并发 /start 会都算出同一个 seq。
-    # (user_id, seq) 唯一键让只有一个能落库,其余撞键 → 回滚后按"这次不放视频"降级,
-    # 客户端照旧走广告链路。没有它,并发就能绕过 max_plays 无限刷金币。
-    try:
-        if commit:
-            db.commit()
-        else:
-            db.flush()
-    except IntegrityError:
-        db.rollback()
-        return _miss(used_plays(db, user_id))
-    return {
-        "should_play": True,
-        "video_url": video_url,
-        "play_token": play.play_token,
-        "reward_coin": reward_coin,
-        "seq": seq,
-        "remaining": max(0, max_plays - seq),
-    }
-
-
-def _find_play(db: Session, user_id: int, token: str) -> GuideVideoPlay | None:
-    """按 (play_token, user_id) 取播放行 —— 带 user_id 是防拿别人的 token 来兑。"""
-    return db.execute(
-        select(GuideVideoPlay).where(
-            GuideVideoPlay.play_token == token,
-            GuideVideoPlay.user_id == user_id,
-        )
-    ).scalar_one_or_none()
-
-
-def grant_play(
-    db: Session, user_id: int, *, play_token: str, completed: bool
-) -> dict[str, Any]:
-    """按 play_token 发这次引导视频的金币(幂等)。播完 / 中途关闭都发。
-
-    返回 {granted, coin, status}:granted=True 表示**本次调用真的入账了**;
-    重复上报返回 granted=False + 已发金币(客户端据此不重复累加 toast 金额)。
-    """
-    token = (play_token or "").strip()
-    # 金币额度以**服务端配置**为准,不信客户端(客户端只上报"播完/关闭")。
-    coin = int(get_config(db).get("reward_coin") or 0)
-
-    # 幂等核心:把 status 放进 WHERE 做条件更新(compare-and-set),而不是"先读再判再写"。
-    # 「播完」与「✕ 关闭」抢跑、或客户端超时重试时,两个请求会都读到 status='playing',
-    # 无锁的话就都往下发币、都 commit,金币入账两次(不用恶意,重试就会中招)。改成条件更新后
-    # 并发里只有一条 rowcount=1,另一条拿 0 → 按已发返回,不二次铸币。
-    # (PG READ COMMITTED 下后到的 UPDATE 阻塞到对手提交,再按新版本重判 status;SQLite 写串行。)
-    #
-    # 别指望 IntegrityError 兜底:这里只 UPDATE 不 INSERT,撞不到 uq_guide_video_play_token;
-    # 而 biz_type='guide_video' 的金币流水也不在 ux_coin_transaction_task_ref 的谓词
-    # (biz_type LIKE 'task%')覆盖范围内 —— 两个唯一键在这条路径上都是不生效的。
-    won = db.execute(
-        update(GuideVideoPlay)
-        .where(
-            GuideVideoPlay.play_token == token,
-            GuideVideoPlay.user_id == user_id,
-            GuideVideoPlay.status == "playing",
-        )
-        .values(
-            status="granted",
-            coin=coin,
-            completed=1 if completed else 0,
-            granted_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
-        )
-        .execution_options(synchronize_session=False)
-    ).rowcount
-
-    if not won:
-        # 没抢到:token 不存在 / 不是本人的 / 已被另一次上报发过。回滚拿干净快照再区分两者
-        # (对手此时必然已提交 —— 我们就是被它挡下的,所以读得到它写的 coin)。
-        db.rollback()
-        play = _find_play(db, user_id, token)
-        if play is None:
-            return {"granted": False, "coin": 0, "status": "not_found"}
-        return {"granted": False, "coin": play.coin, "status": "already_granted"}
-
-    if coin > 0:
-        crud_wallet.grant_coins(
-            db,
-            user_id,
-            coin,
-            biz_type=BIZ_TYPE,
-            ref_id=token,
-            remark="新手引导视频奖励",
-        )
-    db.commit()
-    return {"granted": True, "coin": coin, "status": "granted"}

app/schemas/compare_record.py

@@ -13,6 +13,7 @@ from datetime import datetime
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator
 
+
 # ===== 上报请求 =====
 
 class ComparisonItemIn(BaseModel):
@@ -197,20 +198,6 @@ class ComparisonRecordCreatedOut(BaseModel):
     id: int = Field(..., description="写入(或已存在)的记录 id")
 
 
-class CompareStartReserveIn(BaseModel):
-    """Reserve one authenticated comparison start before the agent begins."""
-
-    trace_id: str = Field(..., min_length=1, max_length=64)
-    business_type: str = Field(default="food", min_length=1, max_length=16)
-    device_id: str | None = Field(default=None, max_length=64)
-
-
-class CompareStartReserveOut(BaseModel):
-    limit: int
-    used: int
-    remaining: int
-
-
 class CompareStatsOut(BaseModel):
     """「我的」页省钱战绩卡(比价口径)聚合。"""
 

app/schemas/guide_video.py

@@ -1,36 +0,0 @@
-"""新手引导视频(领券等候浮层前 N 次替代广告)的客户端请求/响应契约。"""
-from __future__ import annotations
-
-from pydantic import BaseModel, Field
-
-
-class GuideVideoStartIn(BaseModel):
-    """开播询问。scene 目前只有 coupon(领券浮层);预留给日后比价等场景。"""
-
-    scene: str = Field(default="coupon", max_length=16)
-
-
-class GuideVideoStartOut(BaseModel):
-    """should_play=False 时客户端照旧走广告链路,其余字段无意义。"""
-
-    should_play: bool
-    video_url: str | None = None   # 相对地址 /media/...;客户端自行拼 BASE_URL
-    play_token: str = ""           # 发奖幂等键
-    reward_coin: int = 0           # 播完/中途关闭都发的固定金币
-    seq: int = 0                   # 本账号第几次
-    remaining: int = 0             # 发完这次还剩几次
-
-
-class GuideVideoRewardIn(BaseModel):
-    """播完或中途关闭都调这个;completed 只做留痕,两者都发币。"""
-
-    play_token: str = Field(min_length=1, max_length=64)
-    completed: bool = False
-
-
-class GuideVideoRewardOut(BaseModel):
-    """granted=True 表示本次调用真的入账(重复上报为 False,coin 是已发金额)。"""
-
-    granted: bool
-    coin: int
-    status: str

app/utils/mt_search_cursor.py

@@ -1,97 +0,0 @@
-"""美团搜索翻页游标(searchId)缓存 —— 「距离最近」tab 翻页提速。
-
-**问题**:美团 query_coupon 的搜索结果只能靠 searchId 续页(pageNo 翻不动),而我们的 HTTP 接口
-是无状态的、客户端只传页码。原实现因此每次都从第 1 页顺序重放到第 N 页 —— 取第 N 页要向美团发
-N 次请求,用户越往下滑越慢(第 5 页 ≈ 第 1 页的 5 倍耗时,且每次调用都可能撞 402 限流)。
-
-**做法**:把「翻到第 n 页所需的 searchId」按 (量化坐标, platform, 关键词) 记下来,
-下次请求第 n 页直接一发命中;未命中才从缓存里最深的那一页往后重放,并把沿途游标补进缓存。
-稳态下每翻一页恒定 1 次请求。
-
-**坐标量化**到 2 位小数(~1km),与 `utils/meituan_city` 同口径:原始 GPS 每次抖动到小数点后 5~6 位,
-不量化几乎不命中缓存;而同一路搜索的排序原点相差 1km 以内,对券列表的实际顺序无感知差别。
-
-**TTL** 取 10 分钟:searchId 属于上游会话态,放太久会翻到过期游标(调用失败 → 调用方回退重放)。
-缓存是**进程内**的,多 worker 各存一份、不共享,这没问题:命中率只影响快慢,不影响正确性。
-"""
-from __future__ import annotations
-
-import threading
-import time
-
-# 缓存键:(量化纬度, 量化经度, platform, 搜索词)
-RouteKey = tuple[float, float, int, str]
-
-_TTL_SEC = 10 * 60
-_MAX_ROUTES = 512          # 最多缓存多少条搜索路线(每条 = 一个坐标×平台×关键词)
-_MAX_PAGES_PER_ROUTE = 60  # 单条路线最多记多少页游标,防某个坐标被无限翻页撑爆
-
-# {route_key: {page_no: (search_id, 写入时刻)}};page_no 表示「用这个 searchId 能取到第几页」。
-# 第 1 页不需要 searchId(直接 pageNo=1),故永远不入表。
-_cursors: dict[RouteKey, dict[int, tuple[str, float]]] = {}
-_lock = threading.Lock()
-
-
-def route_key(latitude: float, longitude: float, platform: int, keyword: str) -> RouteKey:
-    """构造缓存键(坐标量化到 ~1km)。"""
-    return (round(latitude, 2), round(longitude, 2), platform, keyword)
-
-
-def lookup(key: RouteKey, page: int) -> tuple[int, str | None]:
-    """找到能最省调用地拿到第 `page` 页的起点。
-
-    返回 (起始页, 该页要用的 search_id):
-      - (page, "xxx")  缓存命中 → 调用方一发直达第 page 页
-      - (m, "xxx")     命中更浅的第 m 页(1 < m < page)→ 从第 m 页重放到第 page 页
-      - (1, None)      全未命中 → 从第 1 页(pageNo=1)重放
-    """
-    if page <= 1:
-        return 1, None
-    now = time.time()
-    with _lock:
-        pages = _cursors.get(key)
-        if not pages:
-            return 1, None
-        for p in range(page, 1, -1):
-            hit = pages.get(p)
-            if hit is None:
-                continue
-            search_id, wrote_at = hit
-            if now - wrote_at > _TTL_SEC:
-                pages.pop(p, None)   # 过期即清,继续往浅了找
-                continue
-            return p, search_id
-    return 1, None
-
-
-def remember(key: RouteKey, page: int, search_id: str) -> None:
-    """记下「用 search_id 可取到第 page 页」。page<=1 无意义(第 1 页不用游标)。"""
-    if page <= 1 or not search_id:
-        return
-    now = time.time()
-    with _lock:
-        pages = _cursors.get(key)
-        if pages is None:
-            if len(_cursors) >= _MAX_ROUTES:
-                _evict_oldest_route_locked()
-            pages = _cursors[key] = {}
-        pages[page] = (search_id, now)
-        if len(pages) > _MAX_PAGES_PER_ROUTE:
-            # 越浅的页越容易被重新走到,优先丢最深的那些(它们下次多半也过期了)
-            for p in sorted(pages, reverse=True)[: len(pages) - _MAX_PAGES_PER_ROUTE]:
-                pages.pop(p, None)
-
-
-def drop(key: RouteKey) -> None:
-    """整条路线作废。用在「拿缓存游标去请求却失败」时(多半是上游 searchId 过期)。"""
-    with _lock:
-        _cursors.pop(key, None)
-
-
-def _evict_oldest_route_locked() -> None:
-    """淘汰最久没更新过的一条路线(调用方须已持锁)。"""
-    oldest_key = min(
-        _cursors,
-        key=lambda k: max((w for _, w in _cursors[k].values()), default=0.0),
-    )
-    _cursors.pop(oldest_key, None)

docs/api/meituan/meituan-feed.md

@@ -30,8 +30,7 @@
 ## 各 tab 行为
 - **`rec` 智能推荐**:走【离线库 `meituan_coupon`】筛佣金率 ≥ 3%,`DISTINCT ON(dedup_key)` 去重后按销量降序分页。**纯库查询、不打美团、不依赖 MT 凭证**。库为空(prod 刚部署 / ETL 未跑完)→ `empty`;库异常 → `degraded`。**不显示距离**(库里距离相对城市默认点,对用户无意义)。
   - 为何不实时:实测同城热销榜中位佣金 ~0.8%,筛佣金≥3% 后每页剩 0–1 条,既撞 402 又填不满,故从库出;库空时也**不回退实时**(回退同样填不满)。
-- **`distance` 距离最近**:外卖搜「外卖」+ 到店搜「美食」(均 `sortField=6`),两路并行翻页,实时按你坐标算距离、由近及远。两路**都失败**且无结果 → `degraded`;否则有结果 `ok` / 无结果 `empty`。
-  - 翻页成本:美团搜索只能靠 `searchId` 续页,本接口又是无状态的(客户端只传页码)。服务端把沿途 `searchId` 按「量化坐标(~1km)+平台+关键词」缓存 10 分钟(`utils/mt_search_cursor`),**稳态下每翻一页恒定 1 次上游请求**(改前是「取第 N 页 = 发 N 次」);缓存冷/过期才从最近的已知页往后重放。缓存是进程内的,多 worker 不共享 —— 只影响快慢,不影响结果。
+- **`distance` 距离最近**:外卖搜「外卖」+ 到店搜「美食」(均 `sortField=6`),两路并行顺序翻页,实时按你坐标算距离、由近及远。两路**都失败**且无结果 → `degraded`;否则有结果 `ok` / 无结果 `empty`。
 - **默认(空 tab)**:逐轮分页的混合 feed(2 外卖 + 1 到店交叉,写死 3 页:爆款 / 今日必推 / 精选+限时),第 4 页返空。两路榜单都失败 → `degraded`。
 
 ## 错误码

docs/api/meituan/meituan-top-sales.md

@@ -17,7 +17,6 @@
 
 ## 说明
 - 从 `meituan_coupon` 取 `sale_volume_num` 非空 **且 `city_id` = 反查城市** 的券(#116,同城销量榜),`DISTINCT ON(dedup_key)` 跨源去重(每个「品牌|名|价」只留销量最高一条,同销量再按佣金),按销量降序分页;每页只对当前 ~20 条做 `from_raw` 解析(翻页快,不全表拉取)。
-- 分页查询分两步(与 `rec` 共用 `_paged_dedup_ids`):**① 只在 `id + 排序键` 这几个小列上去重/排序/分页,② 再按 id 回表取本页 `raw`**。`raw` 是整条美团原始返回(JSONB,每行数 KB),让它参与排序会把整城数据推过 `work_mem`、落盘做外部归并,而且每翻一页都重来一遍 —— 这是此前「滑到底越来越慢」的主因之一。配套索引 `ix_meituan_coupon_city_dedup_sales` / `..._comm`(见 `alembic/versions/meituan_coupon_feed_indexes.py`)让 `DISTINCT ON` 顺着索引流式去重,免掉排序。
 - **不依赖 MT 凭证**(纯库查询)。库为空(prod 刚部署 / ETL 未跑完)→ `status=empty`;库查询异常 → `status=degraded`。均返 `200`、不抛 5xx。
 - **仅 PostgreSQL**(`DISTINCT ON` 为 PG 专用)。
 

docs/database/meituan_coupon.md

@@ -42,12 +42,8 @@
 ## 索引与约束
 - 唯一约束 `uq_meituan_coupon_source_sign`(`source`, `product_view_sign`)—— ETL upsert 的冲突目标。
 - 单列 index:`source`、`city_id`、`brand_name`、`sale_volume_num`、`commission_percent`、`dedup_key`、`last_seen`。
-- 复合 index(`meituan_coupon_feed_indexes` 迁移,2026-07-23 加):
-  - `ix_meituan_coupon_city_dedup_sales`(`city_id`, `dedup_key`, `sale_volume_num DESC`, `commission_percent DESC`)——「销量最高」。
-  - `ix_meituan_coupon_city_dedup_comm`(`city_id`, `dedup_key`, `commission_percent DESC`)——「智能推荐」。
-  - 列顺序对齐 `WHERE city_id=? … ORDER BY dedup_key, <排序键> DESC`,让 `DISTINCT ON` 顺着索引流式去重、免掉排序。
 - **字段核对(2026-06-10)**:模型 ↔ 迁移(`meituan_coupon_table`)↔ ETL 解析三处一致;解析截断宽度均 ≤ 列宽(name256 / brand128 / poi128 / head512 / sign128,`dedup_key` = md5 32 字符 ≤ 64)。
-- **索引评估修订(2026-07-23)**:此前判断「单城几千行 seq scan 即亚毫秒,复合索引留作放量时再加」——**漏算了 `raw`**。原查询用 `select(MeituanCoupon)` 做子查询,`raw`(JSONB,每行数 KB)被一起拖进 `DISTINCT ON` 与分页两轮排序,整城体量轻松推过 `work_mem` → 落盘外部归并,且**每翻一页重来一次**,是首页「滑到底越来越慢」的主因之一。现已双管齐下:接口侧改成「先在小列上排出本页 id,再按 id 回表取 `raw`」(`api/v1/meituan.py` 的 `_paged_dedup_ids`),库侧补上上面两条复合索引。
+- **索引评估**:`rec` / `sales` 查询的理想索引是复合 `(dedup_key, sale_volume_num DESC)` 与 `(dedup_key, commission_percent DESC)`(匹配 `DISTINCT ON`);但当前**单城几千行,seq scan 即亚毫秒**,现有单列索引已够用,复合索引留作多城 / 放量时再加,避免过早优化。
 
 ## 过期清理策略
 - ETL 每轮(每小时)末尾按 **`last_seen` 的 TTL** 清理:`DELETE WHERE last_seen < now() - prune_hours`(默认 24h,`--prune-hours` 可调,0=不清)。

scripts/reset_guide_video.py

@@ -1,216 +0,0 @@
-"""重置指定账号的「领券引导视频」已播次数,让领券等候浮层重新放引导视频,方便反复看这支片。
-
-原理:浮层放不放引导视频只由两件事决定(见 app/repositories/guide_video.py `start_play`):
-  1. 运营配置 app_config(key=coupon_guide_video):enabled / video_url / max_plays;
-  2. 该账号 guide_video_play 的**行数**(开播即计次)—— 行数 >= max_plays 就不再下发,
-     `/api/v1/guide-video/start` 返 should_play=false,客户端照旧放广告。
-所以「想再看一遍」= 删掉这个账号的 guide_video_play 行。配置本脚本**只读不改**(要换片子 /
-开关去运营后台「配置 - 领券引导视频」;次数 3 / 金币 120 是服务端默认值,后台不开放调整),
-这样不会把别的账号的线上行为一起动了。
-
-默认连**当时发的金币一起退**(每次 reward_coin,现配置 120):这些流水 biz_type='guide_video',
-不退的话每重置一轮余额就白涨一轮,收益明细里还会堆出一串「新手引导视频奖励」。想留着用 --keep-coins。
-例外:金币已被兑换成现金、余额兜不住时**整笔跳过不退** —— coin_balance 必须恒等于流水总和,
-硬退会退成负数,夹到 0 又会吃掉别处赚的金币,两种做法都会让账对不上(同 reset_signin_today.py)。
-
-用法(在项目根、已 pip install -e . 的环境里跑):
-  python scripts/reset_guide_video.py                  # 默认测试号 11111111111
-  python scripts/reset_guide_video.py 13800138000      # 指定手机号
-  python scripts/reset_guide_video.py --user-id 5      # 直接指定 user_id
-  python scripts/reset_guide_video.py --dry-run        # 预览(照常执行再回滚),不落库
-  python scripts/reset_guide_video.py --keep-coins     # 只清次数,已发金币不退
-
-走 SessionLocal 连 DATABASE_URL(SQLite / Postgres 都行),因此**只允许 APP_ENV=dev 改库**
-(--dry-run 只读,任何环境都能跑)。
-"""
-from __future__ import annotations
-
-import argparse
-import sys
-from pathlib import Path
-
-from sqlalchemy import select
-
-from app.core.config import settings
-from app.db.session import SessionLocal, engine
-from app.models.guide_video import GuideVideoPlay
-from app.models.user import User
-from app.models.wallet import CoinAccount, CoinTransaction
-from app.repositories import guide_video as crud_guide
-
-# Windows 控制台默认 GBK,强制 UTF-8 否则中文输出乱码。stderr 也要设:
-# SystemExit(如"用户不存在")的中文提示走的是 stderr。
-for _stream in (sys.stdout, sys.stderr):
-    if hasattr(_stream, "reconfigure"):
-        _stream.reconfigure(encoding="utf-8")
-
-# dev 下 engine 是 echo=True(APP_DEBUG),几十行 SQL 会把前后对比刷没。echo 走 SQLAlchemy 自己的
-# InstanceLogger,不吃 logging.setLevel,只能改 engine.echo。
-engine.echo = False
-
-DEFAULT_PHONE = "11111111111"
-
-
-def resolve_user(db, phone: str, user_id: int | None) -> User:
-    if user_id is not None:
-        user = db.get(User, user_id)
-        if user is None:
-            raise SystemExit(f"user_id={user_id} 不存在")
-        return user
-    user = db.execute(select(User).where(User.phone == phone)).scalar_one_or_none()
-    if user is None:
-        raise SystemExit(f"手机号 {phone} 没有对应用户(注意 phone 才是登录账号,username 是展示 ID)")
-    return user
-
-
-def load_plays(db, user_id: int) -> list[GuideVideoPlay]:
-    return list(db.execute(
-        select(GuideVideoPlay)
-        .where(GuideVideoPlay.user_id == user_id)
-        .order_by(GuideVideoPlay.id)
-    ).scalars().all())
-
-
-def print_config(db) -> dict:
-    """打印运营配置 + 片子是否真在盘上(配了地址但文件丢了,客户端会黑屏/加载失败)。"""
-    cfg = crud_guide.get_config(db)
-    video_url = (cfg.get("video_url") or "").strip()
-    print("--- 运营配置(app_config: coupon_guide_video,本脚本不改) ---")
-    print(f"  enabled={cfg['enabled']}  max_plays={cfg['max_plays']}  reward_coin={cfg['reward_coin']}")
-    print(f"  video_url={video_url or '(未配片)'}")
-    if video_url.startswith(settings.MEDIA_URL_PREFIX + "/"):
-        rel = video_url[len(settings.MEDIA_URL_PREFIX) + 1:]
-        f = Path(settings.MEDIA_ROOT) / rel
-        if f.is_file():
-            print(f"  文件: {f} ({f.stat().st_size / 1024 / 1024:.1f} MB) ✓")
-        else:
-            print(f"  ⚠️ 文件不存在: {f} —— 客户端会加载失败,请去运营后台重新上传")
-    return cfg
-
-
-def print_state(db, user: User, cfg: dict, label: str) -> None:
-    plays = load_plays(db, user.id)
-    max_plays = int(cfg.get("max_plays") or 0)
-    used = len(plays)
-    print(f"--- {label} ---")
-    print(f"  guide_video_play: {used} 行(已用次数,开播即计),上限 {max_plays}")
-    for p in plays:
-        print(f"    #{p.id} 第{p.seq}次 {p.status} +{p.coin}金币 "
-              f"completed={p.completed} 开播于 {p.started_at} token={p.play_token[:12]}…")
-
-    # 用仓储层原样复算一遍"下次会不会放",而不是脚本里自己判规则 —— 这就是客户端会拿到的结果
-    enabled = bool(cfg.get("enabled"))
-    has_video = bool((cfg.get("video_url") or "").strip())
-    will_play = enabled and has_video and max_plays > 0 and used < max_plays
-    reason = (
-        "会放引导视频" if will_play
-        else "开关关着" if not enabled
-        else "没配片" if not has_video
-        else "max_plays=0" if max_plays <= 0
-        else f"次数已用完({used}/{max_plays})"
-    )
-    print(f"  [下次点一键领取] should_play={will_play} —— {reason}")
-
-    acc = db.get(CoinAccount, user.id)
-    if acc is None:
-        print("  coin_account: (无)")
-    else:
-        print(f"  coin_account: coin={acc.coin_balance} earned={acc.total_coin_earned}")
-
-
-def refund_plays(db, user_id: int, tokens: list[str]) -> None:
-    """退回这些播放发的金币:删流水 + 扣余额。余额兜不住的整笔跳过(见模块 docstring)。"""
-    if not tokens:
-        return
-    rows = list(db.execute(
-        select(CoinTransaction).where(
-            CoinTransaction.user_id == user_id,
-            CoinTransaction.biz_type == crud_guide.BIZ_TYPE,
-            CoinTransaction.ref_id.in_(tokens),
-        )
-    ).scalars().all())
-    if not rows:
-        print("  没有可退的金币流水(可能是 reward_coin=0,或本来就没发过)")
-        return
-    acc = db.get(CoinAccount, user_id)
-    if acc is None:
-        return
-
-    # 从最近一笔往回退,退到余额兜不住为止 —— 保证"最新发的那笔"一定被退掉。
-    rows.sort(key=lambda r: r.id, reverse=True)
-    refundable: list[CoinTransaction] = []
-    total = 0
-    for r in rows:
-        if total + r.amount > acc.coin_balance:
-            break
-        refundable.append(r)
-        total += r.amount
-
-    for r in refundable:
-        db.delete(r)
-    if total:
-        acc.coin_balance -= total
-        acc.total_coin_earned = max(0, acc.total_coin_earned - total)
-        print(f"  已退回 {total} 金币({len(refundable)}/{len(rows)} 笔引导视频奖励)")
-
-    stuck = len(rows) - len(refundable)
-    if stuck:
-        print(f"  ⚠️ 还有 {stuck} 笔退不掉(金币已被兑换/花掉,余额 {acc.coin_balance} 兜不住),原样保留 —— "
-              "硬退会让余额和流水总和对不上。收益明细里会多留几条,不影响再看视频。")
-
-
-def main() -> None:
-    parser = argparse.ArgumentParser(description="重置账号的领券引导视频次数,让浮层重新放引导视频")
-    parser.add_argument("phone", nargs="?", default=DEFAULT_PHONE,
-                        help=f"手机号(默认 {DEFAULT_PHONE})")
-    parser.add_argument("--user-id", type=int, default=None, help="直接按 user_id 定位,优先于 phone")
-    parser.add_argument("--keep-coins", action="store_true",
-                        help="不退已发金币(余额会越测越高,收益明细堆重复流水)")
-    parser.add_argument("--dry-run", action="store_true", help="预览,最后回滚不落库")
-    args = parser.parse_args()
-
-    if not args.dry_run and settings.APP_ENV != "dev":
-        raise SystemExit(f"APP_ENV={settings.APP_ENV},拒绝改库(只有 dev 能改;--dry-run 可任意环境)")
-
-    db = SessionLocal()
-    try:
-        user = resolve_user(db, args.phone, args.user_id)
-        print(f"DB: {settings.DATABASE_URL}  APP_ENV: {settings.APP_ENV}")
-        print(f"用户: id={user.id} phone={user.phone} username={user.username}  keep_coins: {args.keep_coins}")
-        cfg = print_config(db)
-        print_state(db, user, cfg, "before")
-
-        plays = load_plays(db, user.id)
-        if not plays:
-            print("该账号没有任何播放记录,次数本来就是满的,无需重置。")
-            db.rollback()
-            return
-
-        tokens = [p.play_token for p in plays]
-        for p in plays:
-            db.delete(p)
-        db.flush()
-
-        if args.keep_coins:
-            print("(--keep-coins:保留已发金币,流水和余额不动)")
-        else:
-            refund_plays(db, user.id, tokens)
-            # 注:更早流水的 balance_after 是当时的快照,不回改 —— 收益明细里历史行的余额列
-            # 会与现余额对不上,dev 测试库无妨。
-
-        print_state(db, user, cfg, "after")
-
-        if args.dry_run:
-            db.rollback()
-            print(f"(dry-run:以上 after 为预览,已回滚,库没动;真跑会删 {len(plays)} 条播放记录)")
-            return
-        db.commit()
-        print(f"完成:删掉 {len(plays)} 条播放记录,{user.phone} 下次点「一键自动领取」浮层会重新放引导视频。")
-        print("提醒:客户端是在浮层建窗时调 /guide-video/start 的,不用重装、不用重登,直接再点一次一键领取即可;"
-              "但「广告显示」开关关掉时整块浮层不建窗(连引导视频也不放),测之前先确认它是开的。")
-    finally:
-        db.close()
-
-
-if __name__ == "__main__":
-    main()

scripts/seed_coupon_session_mock.py

@@ -1,197 +0,0 @@
-"""生成 admin「领券记录」本地联调数据。
-
-用法:
-    python scripts/seed_coupon_session_mock.py
-
-脚本只清理 ``mock-coupon-repeat-*`` 前缀的数据并重新生成。打开后台「领券记录」，
-日期选今天；分别切换 prod/dev，可验证同一设备同一天多次领券仍各自显示正确分数。
-"""
-
-from __future__ import annotations
-
-import sys
-from datetime import UTC, datetime
-from pathlib import Path
-from zoneinfo import ZoneInfo
-
-from sqlalchemy import delete, select
-
-ROOT = Path(__file__).resolve().parents[1]
-if str(ROOT) not in sys.path:
-    sys.path.insert(0, str(ROOT))
-
-from app.db.session import SessionLocal, engine  # noqa: E402
-from app.models.coupon_state import (  # noqa: E402
-    CouponClaimEvent,
-    CouponClaimRecord,
-    CouponSession,
-)
-from app.models.user import User  # noqa: E402
-from app.repositories.coupon_state import record_claims, today_cn  # noqa: E402
-
-PREFIX = "mock-coupon-repeat-"
-CN_TZ = ZoneInfo("Asia/Shanghai")
-DEVICE_REPEAT = f"{PREFIX}device"
-DEVICE_CONTROL = f"{PREFIX}control-device"
-PHONE = "19900009001"
-USERNAME = "80000009001"
-PLATFORM_ELAPSED = {
-    "meituan-waimai": 46_800,
-    "taobao-shanguang": 19_500,
-    "jd-waimai": 24_000,
-}
-
-FIRST_RESULTS = [
-    {"coupon_id": "mt_mock_1", "name": "美团模拟券1", "vendor": "meituan", "status": "success"},
-    {"coupon_id": "mt_mock_2", "name": "美团模拟券2", "vendor": "meituan", "status": "success"},
-    {"coupon_id": "mt_mock_3", "name": "美团模拟券3", "vendor": "meituan", "status": "success"},
-    {"coupon_id": "mt_mock_4", "name": "美团模拟券4", "vendor": "meituan", "status": "failed", "reason": "模拟失败"},
-    {"coupon_id": "tb_mock_1", "name": "淘宝模拟券1", "vendor": "taobao", "status": "success"},
-    {"coupon_id": "tb_mock_2", "name": "淘宝模拟券2", "vendor": "taobao", "status": "success"},
-    {"coupon_id": "tb_mock_3", "name": "淘宝模拟券3", "vendor": "taobao", "status": "success"},
-    {"coupon_id": "jd_mock_1", "name": "京东模拟券", "vendor": "jingdong", "status": "success"},
-]
-
-SECOND_RESULTS = [
-    {"coupon_id": "mt_mock_1", "name": "美团模拟券1", "vendor": "meituan", "status": "skipped", "reason": "模拟跳过，不计分母"},
-    {"coupon_id": "mt_mock_2", "name": "美团模拟券2", "vendor": "meituan", "status": "already_claimed"},
-    {"coupon_id": "mt_mock_3", "name": "美团模拟券3", "vendor": "meituan", "status": "success"},
-    {"coupon_id": "mt_mock_4", "name": "美团模拟券4", "vendor": "meituan", "status": "success"},
-    {"coupon_id": "tb_mock_1", "name": "淘宝模拟券1", "vendor": "taobao", "status": "success"},
-    {"coupon_id": "tb_mock_2", "name": "淘宝模拟券2", "vendor": "taobao", "status": "success"},
-    {"coupon_id": "tb_mock_3", "name": "淘宝模拟券3", "vendor": "taobao", "status": "already_claimed"},
-    {"coupon_id": "jd_mock_1", "name": "京东模拟券", "vendor": "jingdong", "status": "success"},
-]
-
-
-def _started_at(hour: int, minute: int) -> datetime:
-    local = datetime.combine(today_cn(), datetime.min.time()).replace(
-        hour=hour, minute=minute, tzinfo=CN_TZ
-    )
-    return local.astimezone(UTC)
-
-
-def _session(
-    *,
-    trace_id: str,
-    device_id: str,
-    user_id: int,
-    app_env: str,
-    hour: int,
-    minute: int,
-    status: str = "completed",
-    elapsed_ms: int | None = 91_900,
-    platform_elapsed: dict[str, int] | None = None,
-) -> CouponSession:
-    started_at = _started_at(hour, minute)
-    return CouponSession(
-        trace_id=trace_id,
-        device_id=device_id,
-        user_id=user_id,
-        status=status,
-        app_env=app_env,
-        platforms=[],
-        origin_package=None,
-        device_model="Mock Phone",
-        rom="MockOS 1",
-        started_at=started_at,
-        started_date=today_cn(),
-        finished_at=started_at if status != "started" else None,
-        elapsed_ms=elapsed_ms,
-        platform_elapsed=platform_elapsed,
-        platform_success=(
-            ["meituan-waimai", "taobao-shanguang", "jd-waimai"]
-            if status == "completed" else None
-        ),
-        claimed_count=7 if status == "completed" else 0,
-    )
-
-
-def main() -> None:
-    # 本地旧库 Alembic 版本链可能未同步；仅为联调补建新事件表，正式环境仍走 migration。
-    CouponClaimEvent.__table__.create(bind=engine, checkfirst=True)
-    with SessionLocal() as db:
-        db.execute(delete(CouponClaimEvent).where(
-            CouponClaimEvent.trace_id.startswith(PREFIX)
-        ))
-        db.execute(delete(CouponClaimRecord).where(
-            CouponClaimRecord.device_id.startswith(PREFIX)
-        ))
-        db.execute(delete(CouponSession).where(
-            CouponSession.trace_id.startswith(PREFIX)
-        ))
-        user = db.execute(select(User).where(User.phone == PHONE)).scalar_one_or_none()
-        if user is None:
-            user = User(
-                phone=PHONE,
-                username=USERNAME,
-                register_channel="sms",
-                nickname="领券重复测试",
-            )
-            db.add(user)
-            db.flush()
-
-        first_trace = f"{PREFIX}dev-first"
-        second_trace = f"{PREFIX}prod-second"
-        abandoned_trace = f"{PREFIX}prod-abandoned"
-        control_trace = f"{PREFIX}prod-control"
-        db.add_all([
-            _session(
-                trace_id=first_trace,
-                device_id=DEVICE_REPEAT,
-                user_id=user.id,
-                app_env="dev",
-                hour=10,
-                minute=0,
-                platform_elapsed=PLATFORM_ELAPSED,
-            ),
-            _session(
-                trace_id=second_trace,
-                device_id=DEVICE_REPEAT,
-                user_id=user.id,
-                app_env="prod",
-                hour=15,
-                minute=0,
-                platform_elapsed=PLATFORM_ELAPSED,
-            ),
-            _session(
-                trace_id=abandoned_trace,
-                device_id=DEVICE_REPEAT,
-                user_id=user.id,
-                app_env="prod",
-                hour=16,
-                minute=0,
-                status="abandoned",
-                elapsed_ms=21_500,
-                platform_elapsed={"meituan-waimai": 20_500},
-            ),
-            _session(
-                trace_id=control_trace,
-                device_id=DEVICE_CONTROL,
-                user_id=user.id,
-                app_env="prod",
-                hour=17,
-                minute=0,
-                platform_elapsed=PLATFORM_ELAPSED,
-            ),
-        ])
-        db.commit()
-
-        record_claims(
-            db, DEVICE_REPEAT, user.id, first_trace, FIRST_RESULTS, app_env="dev"
-        )
-        record_claims(
-            db, DEVICE_REPEAT, user.id, second_trace, SECOND_RESULTS, app_env="prod"
-        )
-        record_claims(
-            db, DEVICE_CONTROL, user.id, control_trace, FIRST_RESULTS, app_env="prod"
-        )
-
-    print(f"已生成 {today_cn()} 的领券 mock 数据。")
-    print("筛选用户 19900009001。")
-    print("prod 应有：7/7（100.0%）、-、7/8（87.5%）三条。")
-    print("dev 应有：7/8（87.5%）一条。")
-
-
-if __name__ == "__main__":
-    main()

scripts/seed_meituan_coupon_mock.py

@@ -1,222 +0,0 @@
-"""本地 mock:往 meituan_coupon 灌一批假券,让首页「智能推荐 / 销量最高」在无美团凭证时也有数据。
-
-为什么需要:这两个 tab 不实时打美团,只查 `meituan_coupon` 表,数据靠 scripts/pull_meituan_coupons.py
-定时抓。本地既没 MT_CPS 凭证、也没线上导出的 TSV 时,表是空的 → 接口返 status=empty,页面永远空白,
-分页 / 左右滑动 / 触底加载全都没法验。本脚本纯造数,不联网、不需要任何凭证。
-
-覆盖不到的:「距离最近」tab 必须实时打美团搜索接口(库里没存 POI 经纬度),假数据救不了;
-点「抢」换推广链接也会失败(productViewSign 是假的)。要验这两条只能配 MT_CPS_APP_KEY/SECRET。
-
-图片:生成纯色 PNG 落到 data/media/mock_coupon/,由后端 /media 静态服务出图。
-文件名**带 FEED_THUMB_PARAM 后缀**是故意的 —— schemas/meituan.py 的 feed_image_url 会给每个
-headUrl 无条件拼上该后缀(美团 CDN 的缩放参数),本地静态服务不认参数、只会当成文件名的一部分,
-所以磁盘上的文件必须叫 `xxx.png@375w_375h_1e_1c.webp` 才取得到。
-
-city_id 必须与端上定位反查出的一致:rec/top-sales 都按 city_id 过滤,而 city_id 由经纬度经
-app/utils/meituan_city.py 反查。默认灌北京,测试机的模拟定位也要设成北京,否则照样是空。
-
-幂等:重跑先按 MOCK_SIGN_PREFIX 清掉旧 mock 行再重建,不碰真实抓取的数据。
-
-    python -m scripts.seed_meituan_coupon_mock                  # 默认北京 400 条
-    python -m scripts.seed_meituan_coupon_mock --count 800
-    python -m scripts.seed_meituan_coupon_mock --city-id 2QSF6IG3KMDXWO5VP7FXHMMKXA   # 上海
-    python -m scripts.seed_meituan_coupon_mock --clean-only     # 只清 mock 数据
-"""
-from __future__ import annotations
-
-import argparse
-import hashlib
-import random
-import struct
-import sys
-import zlib
-from pathlib import Path
-
-from sqlalchemy import delete
-
-from app.core.config import settings
-from app.db.session import SessionLocal
-from app.models.meituan_coupon import MeituanCoupon
-from app.schemas.meituan import FEED_THUMB_PARAM
-
-if hasattr(sys.stdout, "reconfigure"):
-    sys.stdout.reconfigure(encoding="utf-8")  # Windows 控制台输出中文
-
-# mock 行的 productViewSign 前缀:清理时按此删,保证不误伤真实抓取的券。
-MOCK_SIGN_PREFIX = "MOCK-"
-
-# 北京。其他城市的 id 见 app/integrations/data/meituan_cities.json。
-DEFAULT_CITY_ID = "WKV2HMXUEK634WP64CUCUQGM64"
-
-_IMG_DIR = Path(settings.MEDIA_ROOT) / "mock_coupon"
-_IMG_COUNT = 16
-
-_BRANDS = [
-    "蜜雪冰城", "瑞幸咖啡", "肯德基", "麦当劳", "华莱士", "塔斯汀",
-    "沪上阿姨", "古茗", "茶百道", "霸王茶姬", "正新鸡排", "杨国福麻辣烫",
-    "绝味鸭脖", "喜茶", "奈雪的茶", "汉堡王", "德克士", "必胜客",
-    "老乡鸡", "南城香",
-]
-_WAIMAI_ITEMS = [
-    "单人套餐", "双人超值餐", "3选1套餐", "招牌奶茶券", "全场通用券",
-    "汉堡可乐套餐", "炸鸡拼盘", "麻辣烫单人餐", "早餐组合", "夜宵拼盘",
-]
-_DAODIAN_ITEMS = [
-    "10元代金券", "20元代金券", "50元代金券", "双人自助餐", "四人聚餐套餐",
-    "下午茶双人套餐", "火锅双人餐", "烤肉四人餐",
-]
-# (saleVolume 文案, 排序用下界) —— 与 pull 脚本的 _sale_volume_num 解析口径一致
-_SALE_VOLUMES = [
-    ("月售99+", 99), ("月售199+", 199), ("月售500+", 500), ("月售999+", 999),
-    ("热销2000+", 2000), ("热销5000+", 5000), ("热销1w+", 10000), ("热销3w+", 30000),
-]
-_PRICE_LABELS = [None, "15天低价", "30天低价", "近期低价"]
-
-
-def _solid_png(width: int, height: int, rgb: tuple[int, int, int]) -> bytes:
-    """生成一张纯色 PNG(truecolor RGB)的字节,无需 Pillow。"""
-    def _chunk(typ: bytes, data: bytes) -> bytes:
-        body = typ + data
-        return struct.pack(">I", len(data)) + body + struct.pack(">I", zlib.crc32(body) & 0xFFFFFFFF)
-
-    ihdr = struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0)  # 8bit/通道, color type 2 = RGB
-    row = b"\x00" + bytes(rgb) * width  # 每行前缀 filter byte 0
-    idat = zlib.compress(row * height, 9)
-    return b"\x89PNG\r\n\x1a\n" + _chunk(b"IHDR", ihdr) + _chunk(b"IDAT", idat) + _chunk(b"IEND", b"")
-
-
-def _hsv_rgb(i: int, total: int) -> tuple[int, int, int]:
-    """色相均匀铺开,出一组肉眼可区分的高饱和色(免得整屏一个色、看不出卡片边界)。"""
-    h = 6.0 * i / total
-    f = h - int(h)
-    v, p, q, t = 235, 90, int(235 - 145 * f), int(90 + 145 * f)
-    return [(v, t, p), (q, v, p), (p, v, t), (p, q, v), (t, p, v), (v, p, q)][int(h) % 6]
-
-
-def _write_images() -> list[str]:
-    """写 mock 头图,返回可直接塞进 raw.headUrl 的 URL 列表(不含缩放后缀)。"""
-    _IMG_DIR.mkdir(parents=True, exist_ok=True)
-    urls: list[str] = []
-    for i in range(_IMG_COUNT):
-        name = f"mock_coupon_{i:02d}.png"
-        # 落盘名带后缀,原因见模块 docstring
-        (_IMG_DIR / f"{name}{FEED_THUMB_PARAM}").write_bytes(_solid_png(375, 375, _hsv_rgb(i, _IMG_COUNT)))
-        urls.append(name)
-    return urls
-
-
-def _clean(db, city_id: str | None) -> int:
-    """删掉本脚本造的 mock 行(可限定城市),返回删除条数。"""
-    stmt = delete(MeituanCoupon).where(MeituanCoupon.product_view_sign.like(f"{MOCK_SIGN_PREFIX}%"))
-    if city_id:
-        stmt = stmt.where(MeituanCoupon.city_id == city_id)
-    n = db.execute(stmt).rowcount or 0
-    db.commit()
-    for f in _IMG_DIR.glob(f"mock_coupon_*.png{FEED_THUMB_PARAM}"):
-        f.unlink()
-    return n
-
-
-def _build_row(i: int, rng: random.Random, city_id: str, img_urls: list[str], base_url: str) -> MeituanCoupon:
-    to_store = rng.random() < 0.35   # 三成半到店,其余外卖
-    platform = 2 if to_store else 1
-    biz_line = rng.choice([1, 2]) if to_store else None
-    source = "store_supply" if to_store else rng.choice(["search_waimai", "search_meishi"])
-
-    brand = rng.choice(_BRANDS)
-    item = rng.choice(_DAODIAN_ITEMS if to_store else _WAIMAI_ITEMS)
-    # 名字带序号:保证 dedup_key(brand|name|price) 唯一,不会被 DISTINCT ON 折叠掉,分页才铺得开
-    name = f"{brand}{item}#{i:04d}"
-
-    sell_cents = rng.randrange(590, 8900, 10)
-    orig_cents = int(sell_cents * rng.uniform(1.35, 2.6))
-    sell, orig = f"{sell_cents / 100:.2f}", f"{orig_cents / 100:.2f}"
-
-    # 六成券佣金 ≥3%(智能推荐的门槛),其余低佣金:两个 tab 的结果集才有明显区别
-    pct = round(rng.uniform(3.0, 8.5), 1) if rng.random() < 0.6 else round(rng.uniform(0.3, 2.9), 1)
-    comm_cents = int(sell_cents * pct / 100)
-
-    sv_text, sv_num = rng.choice(_SALE_VOLUMES)
-    head_url = f"{base_url}{settings.MEDIA_URL_PREFIX}/mock_coupon/{rng.choice(img_urls)}"
-    poi_name = f"{brand}(mock{rng.randrange(1, 60):02d}店)"
-    dist_km = round(rng.uniform(0.2, 8.0), 2)   # 外卖侧单位是 km,到店是 m(见 CouponCard.from_raw)
-    sign = f"{MOCK_SIGN_PREFIX}{i:06d}"
-
-    raw = {
-        "couponPackDetail": {
-            "productViewSign": sign,
-            "skuViewId": f"{sign}-SKU",
-            "platform": platform,
-            "bizLine": biz_line,
-            "name": name,
-            "headUrl": head_url,
-            "sellPrice": sell,
-            "originalPrice": orig,
-            "saleVolume": sv_text,
-            "couponNum": rng.choice([1, 1, 1, 3, 5]),
-            "productLabel": {
-                "pricePowerLabel": {"historyPriceLabel": rng.choice(_PRICE_LABELS)},
-                "productRankLabel": f"2小时北京销量榜第{rng.randrange(1, 30)}名" if rng.random() < 0.25 else None,
-                "dianPingRankLabel": f"{rng.uniform(3.8, 4.9):.1f}分" if rng.random() < 0.5 else None,
-            },
-        },
-        "brandInfo": {"brandName": brand, "brandLogoUrl": head_url},
-        # commissionPercent 是「百分比 ×100」(140 → 1.4%),与 pull 脚本的换算保持一致
-        "commissionInfo": {"commissionPercent": int(pct * 100), "commission": f"{comm_cents / 100:.2f}"},
-        "deliverablePoiInfo": {"poiName": poi_name, "deliveryDistance": dist_km if platform == 1 else dist_km * 1000},
-        "availablePoiInfo": {"availablePoiNum": rng.randrange(1, 200)},
-        "couponValidTimeInfo": {"couponValidDay": rng.choice([7, 15, 30, 90])},
-    }
-
-    return MeituanCoupon(
-        source=source, platform=platform, biz_line=biz_line, city_id=city_id,
-        product_view_sign=sign, sku_view_id=f"{sign}-SKU",
-        name=name, brand_name=brand,
-        sell_price_cents=sell_cents, original_price_cents=orig_cents,
-        head_url=head_url, image_size=None, image_type="image/png",
-        sale_volume=sv_text, sale_volume_num=sv_num,
-        commission_percent=pct, commission_amount_cents=comm_cents,
-        poi_name=poi_name, available_poi_num=raw["availablePoiInfo"]["availablePoiNum"],
-        delivery_distance_m=dist_km * 1000,
-        dedup_key=hashlib.md5(f"{brand}|{name}|{sell_cents}".encode()).hexdigest(),
-        raw=raw,
-    )
-
-
-def main() -> None:
-    ap = argparse.ArgumentParser(description="往 meituan_coupon 灌 mock 券(本地无凭证时用)")
-    ap.add_argument("--city-id", default=DEFAULT_CITY_ID, help=f"美团 cityId,默认北京 {DEFAULT_CITY_ID}")
-    ap.add_argument("--count", type=int, default=400, help="造多少条(默认 400,约六成过智能推荐的佣金门槛)")
-    ap.add_argument("--base-url", default="http://127.0.0.1:8770",
-                    help="头图用的后端地址,须为测试机可达(默认 http://127.0.0.1:8770,对齐 local.properties)")
-    ap.add_argument("--clean-only", action="store_true", help="只清 mock 数据,不重建")
-    ap.add_argument("--seed", type=int, default=20260723, help="随机种子(固定则每次造出同一批)")
-    args = ap.parse_args()
-
-    db = SessionLocal()
-    try:
-        removed = _clean(db, None if args.clean_only else args.city_id)
-        print(f"清理旧 mock: {removed} 条")
-        if args.clean_only:
-            return
-
-        rng = random.Random(args.seed)
-        img_urls = _write_images()
-        print(f"头图: {_IMG_COUNT} 张 -> {_IMG_DIR}")
-
-        rows = [_build_row(i, rng, args.city_id, img_urls, args.base_url.rstrip("/"))
-                for i in range(args.count)]
-        db.add_all(rows)
-        db.commit()
-
-        rec = sum(1 for r in rows if (r.commission_percent or 0) >= 3.0)
-        print(f"入库: {len(rows)} 条  city_id={args.city_id}")
-        print(f"  智能推荐(佣金≥3%): {rec} 条 ≈ {(rec + 19) // 20} 页")
-        print(f"  销量最高(有销量):   {len(rows)} 条 ≈ {(len(rows) + 19) // 20} 页")
-        print("\n测试机的模拟定位记得设成对应城市,否则 city_id 对不上仍然是空。")
-    finally:
-        db.close()
-
-
-if __name__ == "__main__":
-    main()

scripts/seed_mock_feedback.py

@@ -1,277 +0,0 @@
-"""一次性 mock:造几条不同状态的用户反馈,供运营后台「反馈工单」页联调验收。
-
-覆盖:
-  - 三个状态 tab(待审核 pending / 已采纳 adopted / 未采纳 rejected),重点铺「待审核」;
-  - 两种反馈类型 source(普通反馈 profile /「我的」页入口、比价反馈 comparison / 比价结果页入口),
-    比价反馈带「问题场景」scene(找错商品/优惠不对/比价太慢…);
-  - 提交端环境快照(app_version / device_model / rom_name / android_version)——新端反馈才有,
-    另留 1~2 条历史反馈(env 全 NULL、contact 有值)测「旧数据」展示;
-  - 截图:在 data/media/feedback/ 生成真实可加载的纯色 PNG(手写字节,无需 Pillow),
-    让审核抽屉的图能真加载出来(与 seed_mock_price_reports 同法)。
-  - 已采纳条带 reward_coins + admin_reply + review_note;未采纳条带 reject_reason + admin_reply。
-
-幂等:每次运行先按固定 mock 手机号清掉旧 mock 用户/反馈 + 删 mock 截图再重建。仅清理用 --clean-only。
-
-  .venv\\Scripts\\python.exe scripts\\seed_mock_feedback.py
-  .venv\\Scripts\\python.exe scripts\\seed_mock_feedback.py --clean-only
-
-看图:前端「反馈工单」页(http://localhost:3001 → 反馈)。图经 NEXT_PUBLIC_MEDIA_BASE
-(本地 = http://localhost:8770)由 App 后端 /media 加载——改过 .env.local 后需重启 next dev,
-且 App 后端(:8770)要在跑。
-"""
-from __future__ import annotations
-
-import argparse
-import struct
-import sys
-import zlib
-from datetime import UTC, datetime, timedelta
-from pathlib import Path
-
-from sqlalchemy import delete, select
-
-from app.core.config import settings
-from app.db.session import SessionLocal
-from app.models.feedback import Feedback
-from app.models.user import User
-from app.repositories.user import _gen_unique_username
-
-# Windows GBK 控制台下也能正常打印中文/¥(避免 UnicodeEncodeError)
-if hasattr(sys.stdout, "reconfigure"):
-    sys.stdout.reconfigure(encoding="utf-8")
-
-# 固定 mock 手机号:脚本只动这些号,便于幂等重建 / 清理。
-# 刻意与 seed_mock_withdraws / seed_mock_price_reports 的号段错开,互不干扰。
-MOCK_PHONES = [
-    "13255550001",
-    "13255550002",
-    "13255550003",
-    "13255550004",
-    "13255550005",
-]
-
-_FEEDBACK_DIR = Path(settings.MEDIA_ROOT) / "feedback"
-_MOCK_IMG_GLOB = "mock_fb_*.png"  # 本脚本生成的图前缀,清理时按此删
-
-
-def _naive_utc_now() -> datetime:
-    """与 func.now() 在 SQLite 的口径一致:naive UTC。反馈 created_at 走 server_default=func.now(),
-    这里显式造数据也用 naive UTC,和真实反馈行同源,前端展示口径一致。"""
-    return datetime.now(UTC).replace(tzinfo=None)
-
-
-def _solid_png(width: int, height: int, rgb: tuple[int, int, int]) -> bytes:
-    """生成一张纯色 PNG(truecolor RGB)的字节,无需 Pillow。颜色块即可肉眼判断「图加载出来了」。"""
-    def _chunk(typ: bytes, data: bytes) -> bytes:
-        body = typ + data
-        return struct.pack(">I", len(data)) + body + struct.pack(">I", zlib.crc32(body) & 0xFFFFFFFF)
-
-    ihdr = struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0)  # 8bit/通道, color type 2 = RGB
-    row = b"\x00" + bytes(rgb) * width  # 每行前缀 filter byte 0
-    idat = zlib.compress(row * height, 9)
-    return b"\x89PNG\r\n\x1a\n" + _chunk(b"IHDR", ihdr) + _chunk(b"IDAT", idat) + _chunk(b"IEND", b"")
-
-
-def _write_mock_image(name: str, rgb: tuple[int, int, int]) -> str:
-    """写一张 mock 截图到 media/feedback/,返回其相对 URL(/media/feedback/<name>)。"""
-    _FEEDBACK_DIR.mkdir(parents=True, exist_ok=True)
-    (_FEEDBACK_DIR / name).write_bytes(_solid_png(320, 320, rgb))
-    return f"{settings.MEDIA_URL_PREFIX}/feedback/{name}"
-
-
-def clean(db) -> int:
-    uids = list(db.execute(select(User.id).where(User.phone.in_(MOCK_PHONES))).scalars())
-    if uids:
-        db.execute(delete(Feedback).where(Feedback.user_id.in_(uids)))
-        db.execute(delete(User).where(User.id.in_(uids)))
-        db.commit()
-    # 删 mock 截图文件
-    if _FEEDBACK_DIR.exists():
-        for f in _FEEDBACK_DIR.glob(_MOCK_IMG_GLOB):
-            f.unlink(missing_ok=True)
-    return len(uids)
-
-
-def seed(db) -> list[Feedback]:
-    now = _naive_utc_now()
-
-    def ago(**kw) -> datetime:
-        return now - timedelta(**kw)
-
-    # 1) 建 5 个 mock 用户(U5 昵称留空测 "-" 展示)
-    users_spec = [
-        ("13255550001", "反馈小达人"),
-        ("13255550002", "比价挑刺王"),
-        ("13255550003", "热心用户阿明"),
-        ("13255550004", "老用户张姐"),
-        ("13255550005", None),
-    ]
-    users: dict[str, User] = {}
-    for phone, nickname in users_spec:
-        u = User(
-            phone=phone,
-            username=_gen_unique_username(db),  # 列 NOT NULL + unique,仿真实注册生成
-            nickname=nickname,
-            register_channel="sms",
-            status="active",
-            created_at=ago(days=15),
-            last_login_at=ago(hours=1),
-        )
-        db.add(u)
-        users[phone] = u
-    db.flush()  # 拿 user.id
-
-    # 2) 生成 mock 截图(不同颜色块,便于肉眼区分「都加载出来了」)
-    palette = [
-        (24, 144, 255),   # 蓝
-        (82, 196, 26),    # 绿
-        (250, 173, 20),   # 橙
-        (245, 34, 45),    # 红
-    ]
-    imgs = [_write_mock_image(f"mock_fb_{i}.png", palette[i]) for i in range(len(palette))]
-
-    # 3) 造反馈记录
-    def fb(
-        phone: str,
-        content: str,
-        *,
-        source: str = "profile",
-        scene: str | None = None,
-        images: list[str] | None = None,
-        contact: str = "",
-        status: str = "pending",
-        reject_reason: str | None = None,
-        reward_coins: int | None = None,
-        review_note: str | None = None,
-        admin_reply: str | None = None,
-        app_version: str | None = None,
-        device_model: str | None = None,
-        rom_name: str | None = None,
-        android_version: str | None = None,
-        created: datetime,
-    ) -> Feedback:
-        return Feedback(
-            user_id=users[phone].id,
-            content=content,
-            contact=contact,  # 列 NOT NULL:新端存空串,历史数据有值
-            source=source,
-            scene=scene,
-            images=images,
-            status=status,
-            reject_reason=reject_reason,
-            reward_coins=reward_coins,
-            review_note=review_note,
-            admin_reply=admin_reply,
-            app_version=app_version,
-            device_model=device_model,
-            rom_name=rom_name,
-            android_version=android_version,
-            reviewed_at=(created + timedelta(hours=2)) if status != "pending" else None,
-            created_at=created,
-        )
-
-    feedbacks = [
-        # ===== 待审核 pending(默认 tab,重点铺量)=====
-        # 普通反馈 · 新端(带环境快照)· 无图
-        fb("13255550001", "签到金币到账有时候会延迟一两分钟,能不能做成实时到账?",
-           source="profile",
-           app_version="2.3.1", device_model="PJA110", rom_name="ColorOS", android_version="14",
-           created=ago(minutes=6)),
-        # 比价反馈 · scene=优惠不对 · 新端 · 2 图
-        fb("13255550002", "这家店京东外卖的到手价比你们算出来的最低价还低,截图为证,麻烦核实。",
-           source="comparison", scene="优惠不对", images=[imgs[0], imgs[1]],
-           app_version="2.3.1", device_model="M2012K11AC", rom_name="MIUI", android_version="13",
-           created=ago(minutes=22)),
-        # 比价反馈 · scene=找错商品 · 新端 · 无图
-        fb("13255550002", "比价结果里的商品跟我搜的不是同一个规格,数量对不上。",
-           source="comparison", scene="找错商品",
-           app_version="2.3.0", device_model="V2309A", rom_name="OriginOS", android_version="14",
-           created=ago(hours=1)),
-        # 普通反馈 · 新端 · 1 图(表扬 + 小问题)
-        fb("13255550003", "提现秒到账,好评!顺手反馈个小 bug:金币记录页偶尔白屏,要退出去重进。",
-           source="profile", images=[imgs[2]],
-           app_version="2.3.1", device_model="23078RKD5C", rom_name="MIUI", android_version="14",
-           created=ago(hours=3)),
-        # 比价反馈 · scene=比价太慢 · 历史数据(env 全 NULL、contact 有值)
-        fb("13255550004", "比价转圈太久了,经常要等十几秒才出结果,体验不太好。",
-           source="comparison", scene="比价太慢", contact="微信 zhangjie_66",
-           created=ago(days=1, hours=2)),
-        # 普通反馈 · 历史数据(env 全 NULL、contact 有值)· 无昵称用户
-        fb("13255550005", "希望能增加支付宝提现,微信零钱用不太习惯。",
-           source="profile", contact="QQ 100200300",
-           created=ago(days=1, hours=8)),
-
-        # ===== 已采纳 adopted(发金币 + 回复)=====
-        fb("13255550003", "建议在比价结果页加个「一键复制口令」,分享给家人更方便。",
-           source="profile", images=[imgs[3]],
-           status="adopted", reward_coins=2000,
-           review_note="有效产品建议,已排期到 2.4.0", admin_reply="感谢反馈!该功能已在规划中,金币奖励已发放~",
-           app_version="2.2.8", device_model="PJA110", rom_name="ColorOS", android_version="13",
-           created=ago(days=2)),
-
-        # ===== 未采纳 rejected(带原因 + 回复)=====
-        fb("13255550002", "你们算的价格不准,我看到的更便宜。",
-           source="comparison", scene="价格不准",
-           status="rejected", reject_reason="截图价格为限时活动价且已过期,不满足「长期可复现更低价」条件,暂不采纳。",
-           admin_reply="感谢参与,本次未通过,欢迎继续上报有效更低价~",
-           app_version="2.3.0", device_model="M2012K11AC", rom_name="MIUI", android_version="13",
-           created=ago(days=3)),
-    ]
-    db.add_all(feedbacks)
-    db.commit()
-    for f in feedbacks:
-        db.refresh(f)
-    return feedbacks
-
-
-def main() -> None:
-    parser = argparse.ArgumentParser(description="造用户反馈 mock 数据(运营后台反馈工单页联调用)")
-    parser.add_argument("--clean-only", action="store_true", help="只清理 mock 数据,不重建")
-    args = parser.parse_args()
-
-    db = SessionLocal()
-    try:
-        removed = clean(db)
-        if removed:
-            print(f"🧹 已清理旧 mock:{removed} 个用户及其反馈 + mock 截图")
-        if args.clean_only:
-            print("✅ 仅清理,已完成。")
-            return
-
-        feedbacks = seed(db)
-
-        status_label = {"pending": "待审核", "adopted": "已采纳", "rejected": "未采纳"}
-        source_label = {"profile": "普通反馈", "comparison": "比价反馈"}
-        by_status: dict[str, list[Feedback]] = {}
-        for f in feedbacks:
-            by_status.setdefault(f.status, []).append(f)
-
-        print(f"\n✅ 已生成 {len(feedbacks)} 条反馈(截图落 {_FEEDBACK_DIR}),分布:")
-        for st in ("pending", "adopted", "rejected"):
-            lst = by_status.get(st, [])
-            print(f"   {status_label[st]:<4} {len(lst)} 条")
-
-        print("\n   明细(#id | 状态 | 类型/场景 | 图 | 内容):")
-        uid2phone = dict(
-            db.execute(select(User.id, User.phone).where(User.phone.in_(MOCK_PHONES))).all()
-        )
-        for f in feedbacks:
-            src = source_label.get(f.source, f.source)
-            scene = f"·{f.scene}" if f.scene else ""
-            nimg = len(f.images or [])
-            snippet = f.content[:20] + ("…" if len(f.content) > 20 else "")
-            print(
-                f"   #{f.id} [{status_label.get(f.status, f.status)}] "
-                f"{src}{scene}  {nimg}图  {uid2phone.get(f.user_id, '?')}  {snippet}"
-            )
-        print(
-            "\n👉 打开 http://localhost:3001 → 反馈 查看(默认「待审核」tab)。"
-            "\n   图加载不出来时排查:① 是否重启过 next dev(读 .env.local 的 NEXT_PUBLIC_MEDIA_BASE)"
-            "  ② App 后端(:8770)是否在跑(它托管 /media)。"
-        )
-    finally:
-        db.close()
-
-
-if __name__ == "__main__":
-    main()

scripts/seed_mock_price_reports.py

@@ -30,7 +30,6 @@ from app.core.config import settings
 from app.db.session import SessionLocal
 from app.models.price_report import PriceReport
 from app.models.user import User
-from app.repositories.user import _gen_unique_username
 
 if hasattr(sys.stdout, "reconfigure"):
     sys.stdout.reconfigure(encoding="utf-8")  # Windows 控制台输出中文/¥
@@ -85,11 +84,10 @@ def seed(db) -> list[PriceReport]:
     # 1) 建 3 个 mock 用户
     users: dict[str, User] = {}
     for i, (phone, nickname) in enumerate(
-        zip(MOCK_PHONES, ["省钱小王", "比价老李", "薅羊毛阿珍"], strict=True)
+        zip(MOCK_PHONES, ["省钱小王", "比价老李", "薅羊毛阿珍"])
     ):
         u = User(
             phone=phone,
-            username=_gen_unique_username(db),  # 列 NOT NULL + unique,仿真实注册生成
             nickname=nickname,
             register_channel="sms",
             status="active",

scripts/seed_mock_withdraws.py

@@ -21,7 +21,7 @@ import argparse
 import sys
 import uuid
 from collections import defaultdict
-from datetime import UTC, datetime, timedelta
+from datetime import datetime, timedelta, timezone
 
 from sqlalchemy import delete, select
 
@@ -31,7 +31,6 @@ from app.models.ad_feed_reward import AdFeedRewardRecord
 from app.models.ad_reward import AdRewardRecord
 from app.models.user import User
 from app.models.wallet import CashTransaction, CoinAccount, CoinTransaction, WithdrawOrder
-from app.repositories.user import _gen_unique_username
 
 # Windows GBK 控制台下也能正常打印中文(避免 UnicodeEncodeError)
 if hasattr(sys.stdout, "reconfigure"):
@@ -52,7 +51,7 @@ REMARK = {"exchange_in": "金币兑入", "withdraw": "提现扣款", "withdraw_r
 
 def _naive_utc_now() -> datetime:
     """与 func.now() 在 SQLite 的口径一致:naive UTC。前端按 UTC 解析再转北京时间。"""
-    return datetime.now(UTC).replace(tzinfo=None)
+    return datetime.now(timezone.utc).replace(tzinfo=None)
 
 
 def _mock_transfer_no() -> str:
@@ -97,7 +96,6 @@ def seed(db) -> list[WithdrawOrder]:
     for phone, nickname in users_spec:
         u = User(
             phone=phone,
-            username=_gen_unique_username(db),  # 列 NOT NULL + unique,仿真实注册生成
             nickname=nickname,
             register_channel="sms",
             status="active",

tests/test_admin_read.py

@@ -5,11 +5,10 @@ from datetime import datetime
 
 import pytest
 from fastapi.testclient import TestClient
-from sqlalchemy import event
 
 from app.admin.main import admin_app
 from app.admin.repositories import admin_user as admin_repo
-from app.db.session import SessionLocal, engine
+from app.db.session import SessionLocal
 from app.models.comparison import ComparisonRecord
 from app.models.feedback import Feedback
 from app.models.wallet import CashTransaction, WithdrawOrder
@@ -133,35 +132,6 @@ def test_user_list_and_detail(admin_client: TestClient, admin_token: str) -> Non
     assert admin_client.get("/admin/api/users/999999", headers=_auth(admin_token)).status_code == 404
 
 
-def test_user_reward_detail_does_not_select_unrelated_new_ad_columns(
-    admin_client: TestClient, admin_token: str
-) -> None:
-    """旧库缺少无关新列时，提现详情的统计和金币记录仍应可读。"""
-    uid = _seed_user_with_data("13800000022")
-
-    def reject_full_ad_reward_projection(
-        _conn, _cursor, statement: str, _parameters, _context, _executemany
-    ) -> None:
-        if "ad_reward_record.boost_round_id" in statement:
-            raise AssertionError("提现详情不应查询未使用的 boost_round_id")
-
-    event.listen(engine, "before_cursor_execute", reject_full_ad_reward_projection)
-    try:
-        stats = admin_client.get(
-            f"/admin/api/users/{uid}/reward-stats", headers=_auth(admin_token)
-        )
-        records = admin_client.get(
-            f"/admin/api/users/{uid}/coin-records",
-            params={"limit": 10, "cursor": 0},
-            headers=_auth(admin_token),
-        )
-    finally:
-        event.remove(engine, "before_cursor_execute", reject_full_ad_reward_projection)
-
-    assert stats.status_code == 200, stats.text
-    assert records.status_code == 200, records.text
-
-
 def test_user_filter_by_status(admin_client: TestClient, admin_token: str) -> None:
     _seed_user_with_data("13800000003")
     r = admin_client.get("/admin/api/users", params={"status": "active"}, headers=_auth(admin_token))
@@ -498,13 +468,13 @@ def test_period_regular_task_coin_uses_explicit_allowlist(
     d = "2021-06-18"
     included = {
         "signin": 100,
+        "signin_boost": 200,
         "task_enable_notification": 300,
         "task_other": 400,
         "price_report_reward": 500,
         "feedback_reward": 600,
     }
     excluded = {
-        "signin_boost": 200,
         "feed_ad_reward_coupon": 700,
         "feed_ad_reward_comparison": 800,
         "feed_ad_reward": 900,
@@ -544,51 +514,3 @@ def test_period_regular_task_coin_uses_explicit_allowlist(
     coins = response.json()["period"]["coins"]
     assert coins["regular_task_coin_total"] == sum(included.values())
     assert coins["task_coin_total"] == 700
-    assert coins["reward_video_coin_total"] == 1200
-
-
-def test_period_signin_boost_moves_to_reward_video_without_double_count(
-    admin_client: TestClient, admin_token: str
-) -> None:
-    """历史签到膨胀归看视频桶，不再进常规任务；本期发放总额不变且不重复计算。"""
-    from datetime import datetime
-
-    from app.models.wallet import CoinTransaction
-
-    d = "2021-06-19"
-    rows = [
-        ("signin_boost", 200),
-        ("reward_video", 100),
-        ("signin", 50),
-    ]
-    db = SessionLocal()
-    try:
-        uid = user_repo.upsert_user_for_login(
-            db, phone="13800008805", register_channel="sms"
-        ).id
-        balance = 0
-        for index, (biz_type, amount) in enumerate(rows, start=1):
-            balance += amount
-            db.add(CoinTransaction(
-                user_id=uid,
-                amount=amount,
-                balance_after=balance,
-                biz_type=biz_type,
-                ref_id=f"signin-boost-route-{index}",
-                created_at=datetime(2021, 6, 19, 12, 0, index),
-            ))
-        db.commit()
-    finally:
-        db.close()
-
-    response = admin_client.get(
-        "/admin/api/stats/overview",
-        params={"date_from": d, "date_to": d},
-        headers=_auth(admin_token),
-    )
-    assert response.status_code == 200, response.text
-    coins = response.json()["period"]["coins"]
-    assert coins["granted_total"] == 350
-    assert coins["reward_video_coin_total"] == 300
-    assert coins["regular_task_coin_total"] == 50
-    assert coins["signin_boost_coin_total"] == 200

tests/test_admin_write.py

@@ -14,7 +14,6 @@ from app.core.security import hash_password
 from app.db.session import SessionLocal
 from app.models.admin import AdminAuditLog
 from app.models.feedback import Feedback
-from app.models.price_report import PriceReport
 from app.models.user import User
 from app.models.wallet import CoinAccount, CoinTransaction, WithdrawOrder
 from app.repositories import user as user_repo
@@ -83,26 +82,6 @@ def _seed_feedback(phone: str) -> int:
         db.close()
 
 
-def _seed_price_report(phone: str) -> int:
-    uid = _seed_user(phone)
-    db = SessionLocal()
-    try:
-        report = PriceReport(
-            user_id=uid,
-            store_name="测试门店",
-            reported_platform_id="eleme",
-            reported_platform_name="饿了么",
-            reported_price_cents=2990,
-            images=[],
-            status="pending",
-        )
-        db.add(report)
-        db.commit()
-        return report.id
-    finally:
-        db.close()
-
-
 # ===== 调金币 =====
 
 def test_grant_coins_writes_txn_and_audit(admin_client: TestClient, finance_token: str) -> None:
@@ -418,100 +397,6 @@ def test_feedback_review_stores_admin_reply(
     assert r.json()["admin_reply"] == "已收到,后续跟进"
 
 
-def test_bulk_approve_feedbacks_returns_per_item_results(
-    admin_client: TestClient, operator_token: str
-) -> None:
-    first_id = _seed_feedback("13900000031")
-    second_id = _seed_feedback("13900000032")
-    r = admin_client.post(
-        "/admin/api/feedbacks/bulk/approve",
-        json={"ids": [first_id, second_id, 999999], "reward_coins": 600, "note": "批量采纳"},
-        headers=_auth(operator_token),
-    )
-    assert r.status_code == 200, r.text
-    payload = r.json()
-    assert payload["total"] == 3 and payload["success"] == 2 and payload["failed"] == 1
-    assert payload["items"][-1] == {"id": 999999, "ok": False, "status": None, "error": "反馈不存在"}
-    db = SessionLocal()
-    try:
-        for feedback_id in (first_id, second_id):
-            feedback = db.get(Feedback, feedback_id)
-            assert feedback is not None and feedback.status == "adopted"
-            assert db.get(CoinAccount, feedback.user_id).coin_balance == 600
-            log = db.execute(
-                select(AdminAuditLog).where(
-                    AdminAuditLog.action == "feedback.approve",
-                    AdminAuditLog.target_id == str(feedback_id),
-                )
-            ).scalar_one()
-            assert log.detail["bulk"] is True
-    finally:
-        db.close()
-
-
-def test_bulk_approve_price_reports_returns_per_item_results(
-    admin_client: TestClient, operator_token: str
-) -> None:
-    first_id = _seed_price_report("13900000041")
-    second_id = _seed_price_report("13900000042")
-    r = admin_client.post(
-        "/admin/api/price-reports/bulk/approve",
-        json={"ids": [first_id, second_id, 999999]},
-        headers=_auth(operator_token),
-    )
-    assert r.status_code == 200, r.text
-    payload = r.json()
-    assert payload["total"] == 3 and payload["success"] == 2 and payload["failed"] == 1
-    assert payload["items"][-1] == {"id": 999999, "ok": False, "status": None, "error": "上报记录不存在"}
-    db = SessionLocal()
-    try:
-        for report_id in (first_id, second_id):
-            report = db.get(PriceReport, report_id)
-            assert report is not None and report.status == "approved"
-            assert report.reward_coins == 1000
-            assert db.get(CoinAccount, report.user_id).coin_balance == 1000
-            log = db.execute(
-                select(AdminAuditLog).where(
-                    AdminAuditLog.action == "price_report.approve",
-                    AdminAuditLog.target_id == str(report_id),
-                )
-            ).scalar_one()
-            assert log.detail["bulk"] is True
-    finally:
-        db.close()
-
-
-def test_bulk_reject_review_requests_apply_shared_reason(
-    admin_client: TestClient, operator_token: str
-) -> None:
-    feedback_id = _seed_feedback("13900000051")
-    report_id = _seed_price_report("13900000052")
-    feedback_response = admin_client.post(
-        "/admin/api/feedbacks/bulk/reject",
-        json={"ids": [feedback_id], "reason": "信息不足", "reply": "请补充完整截图"},
-        headers=_auth(operator_token),
-    )
-    report_response = admin_client.post(
-        "/admin/api/price-reports/bulk/reject",
-        json={"ids": [report_id], "reason": "截图无法核实"},
-        headers=_auth(operator_token),
-    )
-    assert feedback_response.status_code == 200, feedback_response.text
-    assert report_response.status_code == 200, report_response.text
-    assert feedback_response.json()["success"] == 1
-    assert report_response.json()["success"] == 1
-    db = SessionLocal()
-    try:
-        feedback = db.get(Feedback, feedback_id)
-        report = db.get(PriceReport, report_id)
-        assert feedback is not None and feedback.status == "rejected"
-        assert feedback.reject_reason == "信息不足" and feedback.admin_reply == "请补充完整截图"
-        assert report is not None and report.status == "rejected"
-        assert report.reject_reason == "截图无法核实"
-    finally:
-        db.close()
-
-
 # ===== admin 账号管理(super_admin) =====
 
 def test_create_and_update_admin(admin_client: TestClient, super_token: str) -> None:

tests/test_compare_daily_limit.py

@@ -1,120 +0,0 @@
-from __future__ import annotations
-
-import time
-from datetime import datetime, timedelta
-
-from sqlalchemy import func, select
-
-from app.core.rewards import CN_TZ
-from app.core.security import decode_token
-from app.db.session import SessionLocal
-from app.models.comparison import ComparisonRecord
-
-
-def _login(client) -> tuple[str, int]:
-    phone = f"137{int(time.time() * 1000) % 100000000:08d}"
-    sent = client.post("/api/v1/auth/sms/send", json={"phone": phone})
-    assert sent.status_code == 200, sent.text
-    logged_in = client.post(
-        "/api/v1/auth/sms/login",
-        json={"phone": phone, "code": "123456"},
-    )
-    assert logged_in.status_code == 200, logged_in.text
-    token = logged_in.json()["access_token"]
-    return token, int(decode_token(token, expected_type="access")["sub"])
-
-
-def _headers(token: str) -> dict[str, str]:
-    return {"Authorization": f"Bearer {token}"}
-
-
-def test_compare_start_requires_login(client) -> None:
-    response = client.post(
-        "/api/v1/compare/start",
-        json={"trace_id": "quota-no-auth", "business_type": "food"},
-    )
-    assert response.status_code == 401
-
-
-def test_compare_start_is_idempotent_by_trace_id(client) -> None:
-    token, user_id = _login(client)
-    payload = {
-        "trace_id": f"quota-idempotent-{user_id}",
-        "business_type": "ecom",
-        "device_id": "quota-device",
-    }
-
-    first = client.post("/api/v1/compare/start", json=payload, headers=_headers(token))
-    retry = client.post("/api/v1/compare/start", json=payload, headers=_headers(token))
-
-    assert first.status_code == 200, first.text
-    assert first.json() == {"limit": 100, "used": 1, "remaining": 99}
-    assert retry.status_code == 200, retry.text
-    assert retry.json() == first.json()
-    with SessionLocal() as db:
-        count = db.scalar(
-            select(func.count(ComparisonRecord.id)).where(
-                ComparisonRecord.trace_id == payload["trace_id"]
-            )
-        )
-        record = db.execute(
-            select(ComparisonRecord).where(
-                ComparisonRecord.trace_id == payload["trace_id"]
-            )
-        ).scalar_one()
-    assert count == 1
-    assert record.user_id == user_id
-    assert record.status == "running"
-    assert record.business_type == "ecom"
-    assert record.device_id == "quota-device"
-
-
-def test_compare_start_rejects_101st_beijing_day_attempt(client) -> None:
-    token, user_id = _login(client)
-    now = datetime.now(CN_TZ).replace(tzinfo=None)
-    with SessionLocal() as db:
-        db.add_all(
-            [
-                ComparisonRecord(
-                    user_id=user_id,
-                    trace_id=f"quota-full-{user_id}-{index}",
-                    status="failed",
-                    created_at=now,
-                )
-                for index in range(99)
-            ]
-        )
-        db.add(
-            ComparisonRecord(
-                user_id=user_id,
-                trace_id=f"quota-yesterday-{user_id}",
-                status="success",
-                created_at=now - timedelta(days=1),
-            )
-        )
-        db.commit()
-
-    final_allowed_trace = f"quota-final-allowed-{user_id}"
-    allowed = client.post(
-        "/api/v1/compare/start",
-        json={"trace_id": final_allowed_trace, "business_type": "food"},
-        headers=_headers(token),
-    )
-    assert allowed.status_code == 200, allowed.text
-    assert allowed.json() == {"limit": 100, "used": 100, "remaining": 0}
-
-    rejected_trace = f"quota-rejected-{user_id}"
-    response = client.post(
-        "/api/v1/compare/start",
-        json={"trace_id": rejected_trace, "business_type": "food"},
-        headers=_headers(token),
-    )
-
-    assert response.status_code == 429
-    assert response.json()["detail"] == "今日已比价超过100次，请明天再试"
-    with SessionLocal() as db:
-        assert db.scalar(
-            select(func.count(ComparisonRecord.id)).where(
-                ComparisonRecord.trace_id == rejected_trace
-            )
-        ) == 0

tests/test_coupon_claim_event.py

@@ -1,54 +0,0 @@
-"""逐次单券事件不能被同设备同日的每日去重记录串场。"""
-
-from sqlalchemy import delete, select
-
-from app.admin.repositories.coupon_data import _point_scores_by_trace, coupon_point_details
-from app.db.session import SessionLocal
-from app.models.coupon_state import CouponClaimEvent, CouponClaimRecord
-from app.repositories.coupon_state import record_claims
-
-
-def test_same_device_same_day_keeps_scores_for_each_trace() -> None:
-    db = SessionLocal()
-    device = "event-repeat-device"
-    first_trace = "event-repeat-first"
-    second_trace = "event-repeat-second"
-    first_results = [
-        {"coupon_id": "mt-repeat", "name": "美团测试券", "status": "success"},
-        {"coupon_id": "tb-repeat", "name": "淘宝测试券", "status": "failed"},
-    ]
-    second_results = [
-        {"coupon_id": "mt-repeat", "name": "美团测试券", "status": "already_claimed"},
-        {"coupon_id": "tb-repeat", "name": "淘宝测试券", "status": "success"},
-    ]
-    try:
-        record_claims(
-            db, device, None, first_trace, first_results, app_env="dev"
-        )
-        record_claims(
-            db, device, None, second_trace, second_results, app_env="prod"
-        )
-
-        assets = db.execute(
-            select(CouponClaimRecord).where(CouponClaimRecord.device_id == device)
-        ).scalars().all()
-        assert len(assets) == 2
-        assert {row.trace_id for row in assets} == {first_trace}
-
-        events = db.execute(
-            select(CouponClaimEvent).where(CouponClaimEvent.device_id == device)
-        ).scalars().all()
-        assert len(events) == 4
-        assert {row.trace_id for row in events} == {first_trace, second_trace}
-
-        scores = _point_scores_by_trace(db, [first_trace, second_trace])
-        assert scores[first_trace] == {"succeeded": 1, "tried": 2}
-        assert scores[second_trace] == {"succeeded": 2, "tried": 2}
-        assert [row["status"] for row in coupon_point_details(
-            db, trace_id=second_trace
-        )] == ["already_claimed", "success"]
-    finally:
-        db.execute(delete(CouponClaimEvent).where(CouponClaimEvent.device_id == device))
-        db.execute(delete(CouponClaimRecord).where(CouponClaimRecord.device_id == device))
-        db.commit()
-        db.close()

tests/test_coupon_point_score.py

@@ -13,7 +13,7 @@ from app.admin.repositories.coupon_data import (
 )
 from app.admin.security import create_admin_token
 from app.db.session import SessionLocal
-from app.models.coupon_state import CouponClaimEvent, CouponSession
+from app.models.coupon_state import CouponClaimRecord, CouponSession
 
 
 def test_point_scores_by_trace() -> None:
@@ -22,14 +22,14 @@ def test_point_scores_by_trace() -> None:
     trace = "point-score-trace"
     try:
         db.add_all([
-            CouponClaimEvent(
-                trace_id=trace,
+            CouponClaimRecord(
                 device_id="score-device",
                 coupon_id=f"mt-score-{status}",
                 claim_date=date(2020, 1, 2),
                 status=status,
                 coupon_name=f"测试点位-{status}",
                 reason="测试失败" if status == "failed" else None,
+                trace_id=trace,
             )
             for status in ("success", "already_claimed", "failed", "skipped")
         ])
@@ -53,12 +53,12 @@ def test_skipped_detail_does_not_create_a_score() -> None:
     db = SessionLocal()
     trace = "point-score-skipped"
     try:
-        db.add(CouponClaimEvent(
-            trace_id=trace,
+        db.add(CouponClaimRecord(
             device_id="score-device-skipped",
             coupon_id="mt-score-skipped-only",
             claim_date=date(2020, 1, 2),
             status="skipped",
+            trace_id=trace,
         ))
         db.flush()
 
@@ -87,12 +87,12 @@ def test_coupon_data_report_returns_scores_without_embedding_details() -> None:
             started_date=report_date,
         ))
         db.add_all([
-            CouponClaimEvent(
-                trace_id=trace,
+            CouponClaimRecord(
                 device_id="score-report-device",
                 coupon_id=f"mt-report-{status}",
                 claim_date=report_date,
                 status=status,
+                trace_id=trace,
             )
             for status in ("success", "failed")
         ])
@@ -128,14 +128,14 @@ def test_coupon_point_details_endpoint() -> None:
                 role="super_admin",
             )
         token, _expires_at = create_admin_token(admin_id=admin.id, role=admin.role)
-        db.add(CouponClaimEvent(
-            trace_id=trace,
+        db.add(CouponClaimRecord(
             device_id="point-details-endpoint-device",
             coupon_id="mt-point-details-endpoint",
             coupon_name="接口测试券",
             claim_date=date(2020, 1, 5),
             status="failed",
             reason="接口测试失败",
+            trace_id=trace,
         ))
         db.commit()
 
@@ -156,6 +156,6 @@ def test_coupon_point_details_endpoint() -> None:
         }
     finally:
         db.rollback()
-        db.execute(delete(CouponClaimEvent).where(CouponClaimEvent.trace_id == trace))
+        db.execute(delete(CouponClaimRecord).where(CouponClaimRecord.trace_id == trace))
         db.commit()
         db.close()

tests/test_coupon_slots.py

@@ -7,7 +7,7 @@ from sqlalchemy import delete, select
 
 from app.admin.repositories.coupon_data import coupon_slot_report
 from app.db.session import SessionLocal
-from app.models.coupon_state import CouponClaimEvent, CouponClaimRecord, CouponSession
+from app.models.coupon_state import CouponClaimRecord, CouponSession
 from app.repositories.coupon_state import record_claims, session_app_env
 
 
@@ -49,7 +49,6 @@ def test_record_claims_stamps_app_env() -> None:
         assert row.app_env == "prod"
         assert row.status == "already_claimed"
     finally:
-        db.execute(delete(CouponClaimEvent).where(CouponClaimEvent.trace_id == "t-stamp"))
         db.execute(delete(CouponClaimRecord).where(CouponClaimRecord.device_id == dev))
         db.commit()
         db.close()

tests/test_cps_reconcile_worker.py

@@ -1,197 +0,0 @@
-"""美团、京东 CPS 每日自动对账 worker。"""
-from __future__ import annotations
-
-import logging
-from datetime import date, datetime, timedelta
-
-import pytest
-
-from app.core import cps_reconcile_worker as worker
-from app.core.config import settings
-from app.core.rewards import CN_TZ
-from app.db.session import SessionLocal
-from app.integrations.meituan import MeituanCpsError
-
-
-def _configure_platforms(monkeypatch) -> None:
-    monkeypatch.setattr(settings, "MT_CPS_APP_KEY", "mt-key")
-    monkeypatch.setattr(settings, "MT_CPS_APP_SECRET", "mt-secret")
-    monkeypatch.setattr(settings, "JD_UNION_APP_KEY", "jd-key")
-    monkeypatch.setattr(settings, "JD_UNION_APP_SECRET", "jd-secret")
-    monkeypatch.setattr(settings, "CPS_AUTO_RECONCILE_LOOKBACK_DAYS", 3)
-
-
-def test_reconcile_once_pulls_meituan_and_jd_by_update_time(monkeypatch) -> None:
-    _configure_platforms(monkeypatch)
-    calls: dict[str, dict] = {}
-
-    def fake_meituan(db, **kwargs):
-        calls["meituan"] = kwargs
-        return {
-            "fetched": 2,
-            "inserted": 1,
-            "updated": 1,
-            "pages": 1,
-            "api_requests": 1,
-        }
-
-    def fake_jd(db, **kwargs):
-        calls["jd"] = kwargs
-        return {
-            "fetched": 3,
-            "inserted": 2,
-            "updated": 1,
-            "pages": 2,
-            "api_requests": 72,
-            "windows": 72,
-        }
-
-    monkeypatch.setattr(worker.cps_repo, "reconcile_orders", fake_meituan)
-    monkeypatch.setattr(worker.cps_repo, "reconcile_jd_orders", fake_jd)
-    monkeypatch.setattr(worker, "_touch_lock", lambda: None)
-    now = datetime(2026, 7, 22, 5, 0, tzinfo=CN_TZ)
-
-    result = worker._reconcile_once(now)
-
-    assert result["errors"] == {}
-    assert result["meituan"]["fetched"] == 2
-    assert result["jd"]["fetched"] == 3
-    assert calls["meituan"]["query_time_type"] == 2
-    assert calls["jd"]["query_time_type"] == 3
-    assert calls["meituan"]["end_time"] == int(now.timestamp())
-    assert calls["meituan"]["start_time"] == int((now - timedelta(days=3)).timestamp())
-    assert calls["jd"]["start_time"] == now - timedelta(days=3)
-    assert calls["jd"]["end_time"] == now
-    assert calls["meituan"]["audit_context"]["run_id"] == result["run_id"]
-    assert calls["jd"]["audit_context"]["run_id"] == result["run_id"]
-    assert result["status"] == "success"
-    assert result["manual_retry_required"] is False
-
-
-def test_meituan_failure_does_not_block_jd(monkeypatch) -> None:
-    _configure_platforms(monkeypatch)
-    jd_called = False
-
-    def fail_meituan(db, **kwargs):
-        raise MeituanCpsError("temporary failure")
-
-    def fake_jd(db, **kwargs):
-        nonlocal jd_called
-        jd_called = True
-        return {"fetched": 1, "inserted": 1, "updated": 0, "pages": 1}
-
-    monkeypatch.setattr(worker.cps_repo, "reconcile_orders", fail_meituan)
-    monkeypatch.setattr(worker.cps_repo, "reconcile_jd_orders", fake_jd)
-    monkeypatch.setattr(worker, "_touch_lock", lambda: None)
-
-    result = worker._reconcile_once(datetime(2026, 7, 22, 5, 0, tzinfo=CN_TZ))
-
-    assert result["errors"]["meituan"] == "temporary failure"
-    assert jd_called is True
-    assert result["jd"]["fetched"] == 1
-    assert result["meituan"]["status"] == "failed"
-    assert result["status"] == "partial_success"
-    assert result["manual_retry_required"] is True
-
-
-def test_should_run_once_after_five_am() -> None:
-    before = datetime(2026, 7, 22, 4, 59, tzinfo=CN_TZ)
-    at_five = datetime(2026, 7, 22, 5, 0, tzinfo=CN_TZ)
-
-    assert worker._should_run(None, before, 5) is False
-    assert worker._should_run(None, at_five, 5) is True
-    assert worker._should_run(date(2026, 7, 22), at_five, 5) is False
-
-
-def test_trigger_and_next_run_are_beijing_time() -> None:
-    before_five = datetime(2026, 7, 22, 4, 30, tzinfo=CN_TZ)
-    after_five = datetime(2026, 7, 22, 6, 0, tzinfo=CN_TZ)
-
-    assert worker._trigger_for_run(before_five, after_five, 5) == "scheduled"
-    assert worker._trigger_for_run(after_five, after_five, 5) == "startup_catchup"
-    assert worker._next_run_at(before_five, 5) == datetime(2026, 7, 22, 5, 0, tzinfo=CN_TZ)
-    assert worker._next_run_at(after_five, 5) == datetime(2026, 7, 23, 5, 0, tzinfo=CN_TZ)
-
-
-def test_reconcile_emits_structured_audit_logs(monkeypatch, caplog) -> None:
-    _configure_platforms(monkeypatch)
-    monkeypatch.setattr(
-        worker.cps_repo,
-        "reconcile_orders",
-        lambda db, **kwargs: {
-            "fetched": 2,
-            "inserted": 1,
-            "updated": 1,
-            "pages": 1,
-            "api_requests": 1,
-        },
-    )
-    monkeypatch.setattr(
-        worker.cps_repo,
-        "reconcile_jd_orders",
-        lambda db, **kwargs: {
-            "fetched": 3,
-            "inserted": 2,
-            "updated": 1,
-            "pages": 1,
-            "api_requests": 72,
-            "windows": 72,
-        },
-    )
-    monkeypatch.setattr(worker, "_touch_lock", lambda: None)
-    monkeypatch.setattr(
-        worker,
-        "_cn_now",
-        lambda: datetime(2026, 7, 22, 5, 1, tzinfo=CN_TZ),
-    )
-
-    with caplog.at_level(logging.INFO, logger=worker.logger.name):
-        result = worker._reconcile_once(
-            datetime(2026, 7, 22, 5, 0, tzinfo=CN_TZ),
-            run_id="audit-run-1",
-            trigger="scheduled",
-        )
-
-    records = {record.event: record for record in caplog.records if hasattr(record, "event")}
-    assert records["cps_reconcile.started"].run_id == "audit-run-1"
-    assert records["cps_reconcile.started"].lookback_days == 3
-    assert records["cps_reconcile.started"].scheduled_date == "2026-07-22"
-    assert records["cps_reconcile.started"].hostname
-    assert records["cps_reconcile.completed"].task_status == "success"
-    assert records["cps_reconcile.completed"].manual_retry_required is False
-    assert records["cps_reconcile.completed"].meituan_result["fetched"] == 2
-    assert records["cps_reconcile.completed"].jd_result["api_requests"] == 72
-    assert result["next_run_at"] == "2026-07-23T05:00:00+08:00"
-
-
-def test_jd_request_failure_logs_exact_window(monkeypatch, caplog) -> None:
-    start = datetime(2026, 7, 20, 5, 0, tzinfo=CN_TZ)
-
-    def fail_query(**kwargs):
-        raise RuntimeError("upstream timeout")
-
-    monkeypatch.setattr(worker.cps_repo.jd_union, "query_order_rows", fail_query)
-    with SessionLocal() as db, caplog.at_level(logging.ERROR, logger=worker.logger.name):
-        with pytest.raises(RuntimeError, match="upstream timeout"):
-            worker.cps_repo.reconcile_jd_orders(
-                db,
-                start_time=start,
-                end_time=start + timedelta(hours=2),
-                audit_context={"run_id": "audit-run-2", "trigger": "scheduled"},
-            )
-
-    record = next(
-        item
-        for item in caplog.records
-        if getattr(item, "event", "") == "cps_reconcile.request_failed"
-    )
-    assert record.run_id == "audit-run-2"
-    assert record.platform == "jd"
-    assert record.failed_window_start == "2026-07-20T05:00:00+08:00"
-    assert record.failed_window_end == "2026-07-20T06:00:00+08:00"
-    assert record.failed_page == 1
-
-
-def test_start_worker_respects_auto_switch(monkeypatch) -> None:
-    monkeypatch.setattr(settings, "CPS_AUTO_RECONCILE_ENABLED", False)
-    assert worker.start_cps_reconcile_worker() is None

tests/test_guide_video.py

@@ -1,263 +0,0 @@
-"""新手引导视频:次数上限 + 发币幂等。
-
-这条链路直接铸币,两处并发缺陷曾经都是真漏洞,所以本文件的重点不是走通 happy path,
-而是**并发失败分支**:
-  - `/start` 并发算出同一个 seq → (user_id, seq) 唯一键挡下,降级为"这次不放视频";
-  - `/reward` 并发抢跑 → status 条件更新只让一个 rowcount=1,输的那个不得入账。
-
-真并发在 SQLite 测试库里复现不了(读事务会直接把写方锁死,而不是让它读到旧快照),
-所以用"把对手已经落库的状态先摆好、再让被测方按旧读数往下走"来模拟,断言的是同一件事:
-输的一方必须空手而归,且账不能乱。
-"""
-from __future__ import annotations
-
-import pytest
-from sqlalchemy import func, select
-
-from app.db.session import SessionLocal
-from app.models.guide_video import GuideVideoPlay
-from app.models.wallet import CoinAccount, CoinTransaction
-from app.repositories import guide_video as crud_guide
-from app.repositories.user import get_user_by_phone
-
-VIDEO_URL = "/media/guide_video/pytest_guide.mp4"
-
-
-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 _user_id(phone: str) -> int:
-    with SessionLocal() as db:
-        user = get_user_by_phone(db, phone)
-        assert user is not None
-        return user.id
-
-
-def _coin_balance(user_id: int) -> int:
-    with SessionLocal() as db:
-        acc = db.get(CoinAccount, user_id)
-        return acc.coin_balance if acc else 0
-
-
-def _guide_txns(user_id: int) -> list[CoinTransaction]:
-    """该用户的引导视频金币流水(按 id 升序)。"""
-    with SessionLocal() as db:
-        return list(db.execute(
-            select(CoinTransaction)
-            .where(
-                CoinTransaction.user_id == user_id,
-                CoinTransaction.biz_type == crud_guide.BIZ_TYPE,
-            )
-            .order_by(CoinTransaction.id)
-        ).scalars().all())
-
-
-def _assert_ledger_balanced(user_id: int) -> None:
-    """coin_balance 必须恒等于流水总和 —— 双发/丢更新都会先在这里露馅。"""
-    with SessionLocal() as db:
-        total = db.execute(
-            select(func.coalesce(func.sum(CoinTransaction.amount), 0)).where(
-                CoinTransaction.user_id == user_id
-            )
-        ).scalar_one()
-        acc = db.get(CoinAccount, user_id)
-        assert acc is not None
-        assert acc.coin_balance == total, f"余额 {acc.coin_balance} != 流水总和 {total}"
-
-
-@pytest.fixture()
-def guide_configured():
-    """给全局配置塞一支片子(默认 max_plays=3 / reward_coin=120),用完还原成未配片。"""
-    with SessionLocal() as db:
-        crud_guide.set_video(db, VIDEO_URL, admin_id=1)
-        cfg = crud_guide.get_config(db)
-    yield cfg
-    with SessionLocal() as db:
-        crud_guide.set_video(db, None, admin_id=1)
-
-
-# ===== 基本闭环 =====
-
-
-def test_start_miss_when_no_video_configured(client) -> None:
-    """没配片 → should_play=False,客户端照旧放广告(本功能不配视频就等于没上线)。"""
-    token = _login(client, "13920000001")
-    r = client.post("/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(token))
-    assert r.status_code == 200, r.text
-    body = r.json()
-    assert body["should_play"] is False
-    assert body["play_token"] == ""
-
-
-def test_start_then_reward_grants_once(client, guide_configured) -> None:
-    """开播 → 上报 → 到账 120;重复上报不再加钱,只回已发金额。"""
-    cfg = guide_configured
-    phone = "13920000002"
-    token = _login(client, phone)
-    uid = _user_id(phone)
-
-    r = client.post("/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(token))
-    body = r.json()
-    assert body["should_play"] is True
-    assert body["video_url"] == VIDEO_URL
-    assert body["seq"] == 1
-    assert body["reward_coin"] == cfg["reward_coin"]
-    play_token = body["play_token"]
-    assert play_token
-
-    r = client.post(
-        "/api/v1/guide-video/reward",
-        json={"play_token": play_token, "completed": True},
-        headers=_auth(token),
-    )
-    assert r.status_code == 200, r.text
-    assert r.json() == {"granted": True, "coin": cfg["reward_coin"], "status": "granted"}
-    assert _coin_balance(uid) == cfg["reward_coin"]
-
-    # 重复上报(客户端超时重试 / 播完与 ✕ 都报了一次)
-    r = client.post(
-        "/api/v1/guide-video/reward",
-        json={"play_token": play_token, "completed": False},
-        headers=_auth(token),
-    )
-    assert r.json() == {"granted": False, "coin": cfg["reward_coin"], "status": "already_granted"}
-    assert _coin_balance(uid) == cfg["reward_coin"], "重复上报不得二次入账"
-    assert len(_guide_txns(uid)) == 1, "一个 play_token 只能有一条金币流水"
-    _assert_ledger_balanced(uid)
-
-
-def test_reward_rejects_unknown_and_others_token(client, guide_configured) -> None:
-    """乱填 token / 拿别人的 token 都发不出币(grant 按 user_id + token 双条件定位)。"""
-    victim = _login(client, "13920000003")
-    r = client.post("/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(victim))
-    stolen = r.json()["play_token"]
-
-    attacker_phone = "13920000004"
-    attacker = _login(client, attacker_phone)
-    attacker_uid = _user_id(attacker_phone)
-
-    r = client.post(
-        "/api/v1/guide-video/reward",
-        json={"play_token": stolen, "completed": True},
-        headers=_auth(attacker),
-    )
-    assert r.json() == {"granted": False, "coin": 0, "status": "not_found"}
-
-    r = client.post(
-        "/api/v1/guide-video/reward",
-        json={"play_token": "deadbeef" * 4, "completed": True},
-        headers=_auth(attacker),
-    )
-    assert r.json() == {"granted": False, "coin": 0, "status": "not_found"}
-    assert _coin_balance(attacker_uid) == 0
-
-
-def test_start_stops_at_max_plays(client, guide_configured) -> None:
-    """开播即计次:用满 max_plays 后不再下发,客户端回到广告链路。"""
-    max_plays = int(guide_configured["max_plays"])
-    token = _login(client, "13920000005")
-
-    for i in range(1, max_plays + 1):
-        body = client.post(
-            "/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(token)
-        ).json()
-        assert body["should_play"] is True, f"第 {i} 次应该还能放"
-        assert body["seq"] == i
-        assert body["remaining"] == max_plays - i
-
-    body = client.post(
-        "/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(token)
-    ).json()
-    assert body["should_play"] is False
-    assert body["remaining"] == 0
-
-
-# ===== 并发失败分支(回归) =====
-
-
-def test_start_loses_seq_race_degrades_to_ad(client, guide_configured, monkeypatch) -> None:
-    """并发 /start 抢到同一个 seq 时,输的一方降级成"不放视频",而不是多拿一个 token。
-
-    模拟:对手已经提交了 seq=1,而本次请求读到的还是旧计数(used=0)—— 这正是无锁
-    check-then-insert 的race window。没有 (user_id, seq) 唯一键的话,这里会插成第二行、
-    换回第二个 play_token,3 次上限就能被并发无限绕过(改包即可 N 倍刷币)。
-    """
-    phone = "13920000006"
-    token = _login(client, phone)
-    uid = _user_id(phone)
-
-    # 对手先落一行 seq=1
-    first = client.post(
-        "/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(token)
-    ).json()
-    assert first["should_play"] is True and first["seq"] == 1
-
-    # 本次请求读到的是过期计数 → 仍会算出 seq=1
-    monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id: 0)
-    with SessionLocal() as db:
-        result = crud_guide.start_play(db, uid)
-
-    assert result["should_play"] is False, "撞 seq 唯一键后必须降级,不能再发一个 token"
-    assert result["play_token"] == ""
-
-    monkeypatch.undo()
-    with SessionLocal() as db:
-        rows = db.execute(
-            select(func.count()).select_from(GuideVideoPlay).where(
-                GuideVideoPlay.user_id == uid
-            )
-        ).scalar_one()
-    assert rows == 1, "抢输的那次不该留下播放行"
-
-
-def test_reward_loses_race_does_not_double_mint(client, guide_configured) -> None:
-    """并发 /reward 抢输的一方不得入账 —— 条件更新(status 进 WHERE)的失败分支。
-
-    模拟的是真并发的**必要条件**:输的一方在对手提交之前就已经读到 status='playing',
-    之后带着这个旧认知继续往下走(「播完」与「✕ 关闭」抢跑、或超时重试都会造出这一幕)。
-    这里靠 Session 的 identity map 固化那次旧读数 —— 后续同一 Session 再查同一行,拿回的
-    还是这份旧快照,等价于 PG READ COMMITTED 下两个事务都读到 playing。
-
-    旧实现在这里会照发第二笔:它先读再判再写,而 `except IntegrityError` 兜底根本不可能
-    触发 —— 发币走 UPDATE 撞不到 uq_guide_video_play_token,biz_type='guide_video' 的流水
-    也不在 ux_coin_transaction_task_ref 的谓词(biz_type LIKE 'task%')覆盖内。
-    """
-    coin = int(guide_configured["reward_coin"])
-    phone = "13920000007"
-    token = _login(client, phone)
-    uid = _user_id(phone)
-
-    play_token = client.post(
-        "/api/v1/guide-video/start", json={"scene": "coupon"}, headers=_auth(token)
-    ).json()["play_token"]
-
-    db_loser = SessionLocal()
-    try:
-        # 输的一方先读到 playing —— 并发下两个请求都会读到它
-        stale = db_loser.execute(
-            select(GuideVideoPlay).where(GuideVideoPlay.play_token == play_token)
-        ).scalar_one()
-        assert stale.status == "playing"
-
-        # 对手抢先发币并提交
-        with SessionLocal() as db_winner:
-            won = crud_guide.grant_play(db_winner, uid, play_token=play_token, completed=True)
-        assert won == {"granted": True, "coin": coin, "status": "granted"}
-
-        # 输的一方带着旧认知继续:必须空手而归
-        lost = crud_guide.grant_play(db_loser, uid, play_token=play_token, completed=False)
-        assert lost == {"granted": False, "coin": coin, "status": "already_granted"}
-    finally:
-        db_loser.close()
-
-    assert _coin_balance(uid) == coin, "一次播放只能发一次币"
-    assert len(_guide_txns(uid)) == 1, "一个 play_token 只能有一条金币流水"
-    _assert_ledger_balanced(uid)

tests/test_logging_rotation.py

@@ -1,73 +0,0 @@
-"""SafeRotatingFileHandler:Windows 轮转不被外部句柄卡死(WinError 32)。
-
-stdlib RotatingFileHandler 靠 rename 活动文件轮转;Windows 上只要有第二个句柄(admin
-第二进程、残留 --reload worker、IDE 索引、杀软)开着它, rename 就 WinError 32、轮转永久
-卡死。Safe 版在 Windows 改走 copytruncate(拷贝→就地清空, 从不 rename 活动文件)。
-
-这些用例用 monkeypatch 把 os.name 强制成 "nt", 使 copytruncate 分支在任何 OS 的 CI 上都被
-覆盖;在真实 Windows 上则天然命中。
-"""
-from __future__ import annotations
-
-import logging
-from pathlib import Path
-
-from app.core.logging import SafeRotatingFileHandler
-
-
-def _emit(handler: logging.Handler, msg: str) -> None:
-    handler.emit(logging.LogRecord("t", logging.INFO, __file__, 0, msg, (), None))
-
-
-def test_rollover_survives_second_open_handle(tmp_path: Path, monkeypatch) -> None:
-    """第二个句柄开着活动文件时轮转:不抛异常, 且确实转了(生成 .1、活动文件就地清空)。"""
-    monkeypatch.setattr("app.core.logging.os.name", "nt")
-
-    log_file = tmp_path / "app-server.log"
-    # maxBytes 放大, 避免 emit 期间自动轮转干扰;本用例手动触发 doRollover。
-    handler = SafeRotatingFileHandler(
-        str(log_file), maxBytes=10**9, backupCount=3, encoding="utf-8",
-    )
-    handler.setFormatter(logging.Formatter("%(message)s"))
-    try:
-        for i in range(20):
-            _emit(handler, f"line-{i:03d}")
-        handler.flush()
-        before = log_file.stat().st_size
-        assert before > 0
-
-        # 正是 Windows 上 rename 失败的条件:另一个句柄开着活动文件。
-        with open(log_file, "a", encoding="utf-8"):
-            handler.doRollover()  # 不应抛 PermissionError / WinError 32
-
-        backup = tmp_path / "app-server.log.1"
-        assert backup.exists()
-        assert backup.stat().st_size == before   # 轮转前内容完整进了备份
-        assert log_file.stat().st_size == 0      # 活动文件就地清空(不是 rename)
-
-        # 句柄没被 rename 破坏, 仍能继续写。
-        _emit(handler, "after-rollover")
-        handler.flush()
-        assert log_file.stat().st_size > 0
-    finally:
-        handler.close()
-
-
-def test_backups_shift_and_capped(tmp_path: Path, monkeypatch) -> None:
-    """多次轮转:.1/.2 逐级腾挪, 超过 backupCount 的丢弃(不出现 .3)。"""
-    monkeypatch.setattr("app.core.logging.os.name", "nt")
-
-    log_file = tmp_path / "app-server.log"
-    handler = SafeRotatingFileHandler(
-        str(log_file), maxBytes=10**9, backupCount=2, encoding="utf-8",
-    )
-    handler.setFormatter(logging.Formatter("%(message)s"))
-    try:
-        for _ in range(4):
-            _emit(handler, "x" * 50)
-            handler.doRollover()
-        assert (tmp_path / "app-server.log.1").exists()
-        assert (tmp_path / "app-server.log.2").exists()
-        assert not (tmp_path / "app-server.log.3").exists()
-    finally:
-        handler.close()

tests/test_withdraw.py

@@ -9,7 +9,7 @@ from sqlalchemy import select
 
 from app.db.session import SessionLocal
 from app.models.user import User
-from app.models.wallet import CoinAccount, WithdrawOrder, WechatTransferAuthorization
+from app.models.wallet import CoinAccount, WithdrawOrder
 from app.repositories import wallet as crud_wallet
 
 
@@ -426,62 +426,3 @@ def test_withdraw_reject_refunds(client, monkeypatch) -> None:
     r = client.get("/api/v1/wallet/withdraw/status", params={"out_bill_no": bill}, headers=_auth(token))
     assert r.json()["status"] == "rejected"
     assert r.json()["fail_reason"] == "测试拒绝"
-
-
-# ===== §fix 免确认授权 enabled 判定收严:active+authorization_id 非空才算已授权 =====
-
-def _seed_transfer_auth(phone: str, state: str, authorization_id: str | None) -> None:
-    """直接在库里写/改该用户的免确认授权记录(绕过微信,用于判定测试)。"""
-    db = SessionLocal()
-    try:
-        user = db.execute(select(User).where(User.phone == phone)).scalar_one()
-        auth = db.get(WechatTransferAuthorization, user.id)
-        if auth is None:
-            auth = WechatTransferAuthorization(
-                user_id=user.id, openid=user.wechat_openid or "openid_test_abc",
-                out_authorization_no=f"oan_{user.id}",
-            )
-            db.add(auth)
-        auth.state = state
-        auth.authorization_id = authorization_id
-        db.commit()
-    finally:
-        db.close()
-
-
-def test_withdraw_info_auth_enabled_requires_authorization_id(client, monkeypatch) -> None:
-    _patch_userinfo(monkeypatch)
-    token = _login(client, "13800002051")
-    # 触发建号 + 绑定微信(withdraw-info 读 openid)
-    client.get("/api/v1/wallet/withdraw-info", headers=_auth(token))
-    client.post("/api/v1/wallet/bind-wechat", json={"code": "c1"}, headers=_auth(token))
-
-    # active 但无 authorization_id → 视为未授权
-    _seed_transfer_auth("13800002051", "active", None)
-    r = client.get("/api/v1/wallet/withdraw-info", headers=_auth(token))
-    assert r.status_code == 200, r.text
-    assert r.json()["transfer_auth_enabled"] is False
-
-    # active 且有 authorization_id → 已授权
-    _seed_transfer_auth("13800002051", "active", "wx_auth_123")
-    r = client.get("/api/v1/wallet/withdraw-info", headers=_auth(token))
-    assert r.status_code == 200, r.text
-    assert r.json()["transfer_auth_enabled"] is True
-
-
-def test_transfer_auth_status_requires_authorization_id(client, monkeypatch) -> None:
-    _patch_userinfo(monkeypatch)
-    token = _login(client, "13800002052")
-    client.get("/api/v1/wallet/withdraw-info", headers=_auth(token))
-    client.post("/api/v1/wallet/bind-wechat", json={"code": "c2"}, headers=_auth(token))
-
-    _seed_transfer_auth("13800002052", "active", None)
-    r = client.get("/api/v1/wallet/transfer-auth/status", headers=_auth(token))
-    assert r.status_code == 200, r.text
-    assert r.json()["state"] == "active"
-    assert r.json()["enabled"] is False
-
-    _seed_transfer_auth("13800002052", "active", "wx_auth_456")
-    r = client.get("/api/v1/wallet/transfer-auth/status", headers=_auth(token))
-    assert r.status_code == 200, r.text
-    assert r.json()["enabled"] is True