后台审核新增批量处理接口 (#164)

## 改动
- 新增低价审核与用户反馈的批量通过、批量拒绝接口
- 单条仍保持独立事务、审计、发奖和通知；单项失败不影响同批其它记录
- 批量响应返回每条记录的成功状态或失败原因，供前端保留失败项重试
- 反馈审核补充行锁，降低并发重复发奖风险

## 验证
- `ruff check`（相关路由、Schema、测试）
- `pytest tests/test_admin_write.py -q`：22 passed

---------

Co-authored-by: unknown <798648091@qq.com>
Reviewed-on: #164
Co-authored-by: linkeyu <linkeyu@wonderable.ai>
Co-committed-by: linkeyu <linkeyu@wonderable.ai>

.env.example

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

@@ -0,0 +1,26 @@
+"""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

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

app/admin/repositories/coupon_data.py

@@ -1,7 +1,7 @@
 """admin「领券数据」看板聚合:发起/完成数、耗时均值与分位、按天/小时趋势、逐条明细。
 
-数据源 coupon_session(一次领券一行,客户端 /api/v1/coupon/session 两段上报)。量级不大,全量拉
+数据源 coupon_session(一次领券一行,客户端 /api/v1/coupon/session 两段上报)。生产 PostgreSQL
-区间数据后 Python 聚合(分位 SQLite 无 percentile,统一 Python 算,PG 上也一致)。
+使用 percentile_cont 聚合耗时分位；SQLite 本地/测试环境回退读取耗时单列计算。
 - 发起数 = 区间内全部 session(含 started/completed/failed/abandoned),= 流失统计的基数。
 - 完成数 / 耗时均值 / 分位 = 仅 status==completed 子集(成功跑完才有可比的"领券耗时")。
 - summary/daily/hourly/total 在全量上算,不受分页;items 为排序后当前页。
@@ -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 CouponClaimRecord, CouponSession
+from app.models.coupon_state import CouponClaimEvent, 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
@@ -45,6 +45,75 @@ def _percentile(sorted_vals: list[int], q: float) -> int | None:
     return round(sorted_vals[lo] * (1 - frac) + sorted_vals[hi] * frac)
 
 
+def _round_duration_ms(value) -> int | None:
+    """将数据库聚合结果按既有 Python round 口径转为整数毫秒。"""
+    if value is None:
+        return None
+    return int(round(value))
+
+
+def _coupon_summary_aggregate_stmt(conditions: list):
+    """PostgreSQL 汇总卡聚合语句；计数、均值与四个分位一次返回。"""
+    completed = CouponSession.status == "completed"
+    completed_elapsed = completed & CouponSession.elapsed_ms.is_not(None)
+    return select(
+        func.count(CouponSession.id),
+        func.sum(case((completed, 1), else_=0)),
+        func.avg(CouponSession.elapsed_ms).filter(completed_elapsed),
+        *(
+            func.percentile_cont(q)
+            .within_group(CouponSession.elapsed_ms)
+            .filter(completed_elapsed)
+            for q in (0.05, 0.5, 0.95, 0.99)
+        ),
+    ).where(*conditions)
+
+
+def _coupon_summary_aggregates(db: Session, conditions: list) -> dict:
+    """汇总卡基础指标；生产 PG 全部在数据库内完成，SQLite 仅作测试回退。"""
+    if db.bind is not None and db.bind.dialect.name == "postgresql":
+        row = db.execute(_coupon_summary_aggregate_stmt(conditions)).one()
+        return {
+            "started_count": int(row[0] or 0),
+            "completed_count": int(row[1] or 0),
+            "avg_elapsed_ms": _round_duration_ms(row[2]),
+            "p5_ms": _round_duration_ms(row[3]),
+            "p50_ms": _round_duration_ms(row[4]),
+            "p95_ms": _round_duration_ms(row[5]),
+            "p99_ms": _round_duration_ms(row[6]),
+        }
+
+    counts = db.execute(
+        select(
+            func.count(CouponSession.id),
+            func.sum(case((CouponSession.status == "completed", 1), else_=0)),
+        ).where(*conditions)
+    ).one()
+    # SQLite 没有 percentile_cont；本地/测试只回退读取耗时单列，不加载完整记录。
+    completed_elapsed = list(
+        db.execute(
+            select(CouponSession.elapsed_ms)
+            .where(
+                *conditions,
+                CouponSession.status == "completed",
+                CouponSession.elapsed_ms.is_not(None),
+            )
+            .order_by(CouponSession.elapsed_ms)
+        ).scalars()
+    )
+    return {
+        "started_count": int(counts[0] or 0),
+        "completed_count": int(counts[1] or 0),
+        "avg_elapsed_ms": _round_duration_ms(
+            sum(completed_elapsed) / len(completed_elapsed)
+        ) if completed_elapsed else None,
+        "p5_ms": _percentile(completed_elapsed, 5),
+        "p50_ms": _percentile(completed_elapsed, 50),
+        "p95_ms": _percentile(completed_elapsed, 95),
+        "p99_ms": _percentile(completed_elapsed, 99),
+    }
+
+
 def _avg(vals: list[int]) -> int | None:
     return round(sum(vals) / len(vals)) if vals else None
 
@@ -124,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((CouponClaimRecord.status.in_(_SLOT_OK), 1), else_=0))
+    succeeded = func.sum(case((CouponClaimEvent.status.in_(_SLOT_OK), 1), else_=0))
     rows = db.execute(
         select(
-            CouponClaimRecord.trace_id,
+            CouponClaimEvent.trace_id,
             succeeded.label("succeeded"),
             func.count().label("tried"),
         )
         .where(
-            CouponClaimRecord.trace_id.in_(trace_ids),
+            CouponClaimEvent.trace_id.in_(trace_ids),
-            CouponClaimRecord.status.in_(_SLOT_TRIED),
+            CouponClaimEvent.status.in_(_SLOT_TRIED),
         )
-        .group_by(CouponClaimRecord.trace_id)
+        .group_by(CouponClaimEvent.trace_id)
     ).all()
     return {
         trace_id: {"succeeded": int(success_count or 0), "tried": int(tried or 0)}
@@ -148,13 +217,13 @@ def coupon_point_details(db: Session, *, trace_id: str) -> list[dict]:
     """按单个 trace 查询逐券结果；仅在后台用户点击分数时调用。"""
     rows = db.execute(
         select(
-            CouponClaimRecord.coupon_id,
+            CouponClaimEvent.coupon_id,
-            CouponClaimRecord.coupon_name,
+            CouponClaimEvent.coupon_name,
-            CouponClaimRecord.status,
+            CouponClaimEvent.status,
-            CouponClaimRecord.reason,
+            CouponClaimEvent.reason,
         )
-        .where(CouponClaimRecord.trace_id == trace_id)
+        .where(CouponClaimEvent.trace_id == trace_id)
-        .order_by(CouponClaimRecord.id)
+        .order_by(CouponClaimEvent.id)
     ).all()
     return [
         {
@@ -215,30 +284,22 @@ def coupon_data_report(
         if not user_ids:
             return _empty_result()
 
-    stmt = select(CouponSession).where(
+    conditions = [
         CouponSession.started_date >= d_from,
         CouponSession.started_date <= d_to,
-    )
+    ]
     if app_env is not None:
-        stmt = stmt.where(CouponSession.app_env == app_env)
+        conditions.append(CouponSession.app_env == app_env)
     if statuses:
-        stmt = stmt.where(CouponSession.status.in_(statuses))
+        conditions.append(CouponSession.status.in_(statuses))
     if user_ids is not None:
-        stmt = stmt.where(CouponSession.user_id.in_(user_ids))
+        conditions.append(CouponSession.user_id.in_(user_ids))
+    stmt = select(CouponSession).where(*conditions)
     rows = list(db.execute(stmt).scalars())
 
     # ── 汇总卡 ──
-    completed_elapsed = sorted(
-        r.elapsed_ms for r in rows if r.status == "completed" and r.elapsed_ms is not None
-    )
     summary = {
-        "started_count": len(rows),
+        **_coupon_summary_aggregates(db, conditions),
-        "completed_count": sum(1 for r in rows if r.status == "completed"),
-        "avg_elapsed_ms": _avg(completed_elapsed),
-        "p5_ms": _percentile(completed_elapsed, 5),
-        "p50_ms": _percentile(completed_elapsed, 50),
-        "p95_ms": _percentile(completed_elapsed, 95),
-        "p99_ms": _percentile(completed_elapsed, 99),
         **_success_rates(rows),
     }
 
@@ -323,7 +384,7 @@ def coupon_data_report(
         "summary": summary,
         "daily": daily,
         "hourly": hourly,
-        "total": len(rows),
+        "total": summary["started_count"],
         "items": items,
     }
 

app/admin/repositories/cps.py

@@ -5,6 +5,7 @@
 """
 from __future__ import annotations
 
+import logging
 from datetime import datetime, timedelta, timezone
 from decimal import Decimal, InvalidOperation
 from typing import Any
@@ -15,12 +16,14 @@ 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"}
@@ -378,18 +381,35 @@ 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 = 0
+    fetched = inserted = updated = pages = api_requests = 0
     page = 1
     while page <= max_pages:
-        resp = meituan.query_order(
+        api_requests += 1
-            sid=sid, start_time=start_time, end_time=end_time,
+        try:
-            query_time_type=query_time_type, page=page, limit=100,
+            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
         rows = ((resp.get("data") or {}).get("dataList")) or []
         if not rows:
             break
@@ -419,30 +439,58 @@ def reconcile_orders(
             break
         page += 1
     db.commit()
-    return {"fetched": fetched, "inserted": inserted, "updated": updated, "pages": pages}
+    return {
+        "fetched": fetched,
+        "inserted": inserted,
+        "updated": updated,
+        "pages": pages,
+        "api_requests": api_requests,
+    }
 
 
 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 = 0
+    fetched = inserted = updated = pages = api_requests = windows = 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:
-            resp = jd_union.query_order_rows(
+            api_requests += 1
-                start_time=cur,
+            try:
-                end_time=win_end,
+                resp = jd_union.query_order_rows(
-                query_time_type=query_time_type,
+                    start_time=cur,
-                page_index=page,
+                    end_time=win_end,
-                page_size=200,
+                    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
             rows = resp.get("rows") or []
             has_more = bool(resp.get("has_more"))
             if not rows:
@@ -469,7 +517,14 @@ def reconcile_jd_orders(
             page += 1
         cur = win_end
     db.commit()
-    return {"fetched": fetched, "inserted": inserted, "updated": updated, "pages": pages}
+    return {
+        "fetched": fetched,
+        "inserted": inserted,
+        "updated": updated,
+        "pages": pages,
+        "api_requests": api_requests,
+        "windows": windows,
+    }
 
 
 def list_orders(

app/admin/repositories/queries.py

@@ -6,6 +6,7 @@
 from __future__ import annotations
 
 from datetime import date, datetime, time, timedelta, timezone
+from decimal import ROUND_HALF_UP, Decimal
 from zoneinfo import ZoneInfo
 
 from sqlalchemy import Select, asc, case, desc, func, or_, select
@@ -297,6 +298,58 @@ def _comparison_percentile(sorted_values: list[int], q: float) -> int | None:
     return int(value + 0.5)
 
 
+def _round_duration_ms(value) -> int | None:
+    """将数据库聚合结果按既有口径四舍五入为整数毫秒。"""
+    if value is None:
+        return None
+    return int(Decimal(str(value)).quantize(Decimal("1"), rounding=ROUND_HALF_UP))
+
+
+def _comparison_duration_aggregate_stmt(conditions: list, status: str, quantiles: tuple[float, ...]):
+    """PostgreSQL 耗时聚合语句；每种状态只返回一行。"""
+    return select(
+        func.avg(ComparisonRecord.total_ms),
+        *(
+            func.percentile_cont(q).within_group(ComparisonRecord.total_ms)
+            for q in quantiles
+        ),
+    ).where(
+        *conditions,
+        ComparisonRecord.status == status,
+        ComparisonRecord.total_ms.is_not(None),
+    )
+
+
+def _comparison_duration_aggregates(
+    db: Session,
+    *,
+    conditions: list,
+    status: str,
+    quantiles: tuple[float, ...],
+) -> list[int | None]:
+    """返回平均值和各分位数；生产 PG 在数据库内聚合，SQLite 仅作测试回退。"""
+    if db.bind is not None and db.bind.dialect.name == "postgresql":
+        row = db.execute(
+            _comparison_duration_aggregate_stmt(conditions, status, quantiles)
+        ).one()
+        return [_round_duration_ms(value) for value in row]
+
+    # SQLite 没有 percentile_cont；本地/测试只回退读取耗时单列，不加载完整记录。
+    values = list(
+        db.execute(
+            select(ComparisonRecord.total_ms)
+            .where(
+                *conditions,
+                ComparisonRecord.status == status,
+                ComparisonRecord.total_ms.is_not(None),
+            )
+            .order_by(ComparisonRecord.total_ms)
+        ).scalars()
+    )
+    average = _round_duration_ms(sum(values) / len(values)) if values else None
+    return [average, *(_comparison_percentile(values, q) for q in quantiles)]
+
+
 def comparison_records_summary(
     db: Session,
     *,
@@ -332,20 +385,18 @@ def comparison_records_summary(
     success = int(row[2] or 0)
     lower_price = int(row[4] or 0)
     cancelled = int(row[5] or 0)
-    success_durations = sorted(db.execute(
+    success_duration_stats = _comparison_duration_aggregates(
-        select(ComparisonRecord.total_ms).where(
+        db,
-            *conditions,
+        conditions=conditions,
-            ComparisonRecord.status == "success",
+        status="success",
-            ComparisonRecord.total_ms.is_not(None),
+        quantiles=(0.05, 0.5, 0.95, 0.99),
-        )
+    )
-    ).scalars().all())
+    cancelled_duration_stats = _comparison_duration_aggregates(
-    cancelled_durations = sorted(db.execute(
+        db,
-        select(ComparisonRecord.total_ms).where(
+        conditions=conditions,
-            *conditions,
+        status="cancelled",
-            ComparisonRecord.status == "cancelled",
+        quantiles=(0.05, 0.5, 0.95),
-            ComparisonRecord.total_ms.is_not(None),
+    )
-        )
-    ).scalars().all())
     success_rate_denominator = started - cancelled
     return {
         "started": started,
@@ -354,19 +405,16 @@ def comparison_records_summary(
         "success_rate": success / success_rate_denominator if success_rate_denominator else None,
         "avg_token_cost": float(row[3]) if row[3] is not None else None,
         "lower_price_rate": lower_price / success if success else None,
-        "avg_duration_ms": (
+        "avg_duration_ms": success_duration_stats[0],
-            int(sum(success_durations) / len(success_durations) + 0.5)
+        "p5_duration_ms": success_duration_stats[1],
-            if success_durations else None
+        "p50_duration_ms": success_duration_stats[2],
-        ),
+        "p95_duration_ms": success_duration_stats[3],
-        "p5_duration_ms": _comparison_percentile(success_durations, 0.05),
+        "p99_duration_ms": success_duration_stats[4],
-        "p50_duration_ms": _comparison_percentile(success_durations, 0.5),
-        "p95_duration_ms": _comparison_percentile(success_durations, 0.95),
-        "p99_duration_ms": _comparison_percentile(success_durations, 0.99),
         "cancelled": cancelled,
         "cancelled_rate": cancelled / started if started else None,
-        "cancelled_p5_ms": _comparison_percentile(cancelled_durations, 0.05),
+        "cancelled_p5_ms": cancelled_duration_stats[1],
-        "cancelled_p50_ms": _comparison_percentile(cancelled_durations, 0.5),
+        "cancelled_p50_ms": cancelled_duration_stats[2],
-        "cancelled_p95_ms": _comparison_percentile(cancelled_durations, 0.95),
+        "cancelled_p95_ms": cancelled_duration_stats[3],
     }
 
 

app/admin/repositories/stats.py

@@ -30,18 +30,17 @@ 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")
+REWARD_VIDEO_BIZ_TYPES = ("reward_video", "ad_reward", "signin_boost")
 # 领券/比价奖励金币的真实来源是信息流广告发奖(ad_feed_reward_record,按 feed_scene 分场景);
 # coin_transaction 里只有扁平的 feed_ad_reward、biz_type 不分 coupon/comparison,故这俩桶历史从未
-# 被写入,仅留作未来兜底,实际金额在下方按 feed_scene 汇总 ad_feed_reward_record 得出。reward_video/
+# 被写入,仅留作未来兜底,实际金额在下方按 feed_scene 汇总 ad_feed_reward_record 得出。
-# ad_reward 是激励视频,单独成桶、不再混进领券奖励(历史误并会把激励视频金币双计进领券)。
+# reward_video/ad_reward 及历史 signin_boost 均归看视频桶，不再混进领券奖励或常规任务。
 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,6 +12,10 @@ 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,
@@ -33,6 +37,123 @@ 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,
@@ -73,6 +194,50 @@ 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,
@@ -93,53 +258,7 @@ def approve_feedback(
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
 ) -> FeedbackOut:
-    fb = db.get(Feedback, feedback_id)
+    return _approve_feedback(db, admin, feedback_id, payload, get_client_ip(request))
-    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="拒绝采纳反馈")
@@ -150,41 +269,4 @@ def reject_feedback(
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
 ) -> FeedbackOut:
-    fb = db.get(Feedback, feedback_id)
+    return _reject_feedback(db, admin, feedback_id, payload, get_client_ip(request))
-    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/price_report.py

@@ -17,6 +17,10 @@ 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,
@@ -34,6 +38,59 @@ 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,
@@ -60,6 +117,50 @@ 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,
@@ -67,27 +168,7 @@ def approve_price_report(
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
 ) -> OkResponse:
-    # 行锁(SELECT FOR UPDATE):并发/连点双请求会都读到 pending → 各发一次金币双倍发奖,
+    _approve_price_report(db, admin, report_id, get_client_ip(request))
-    # 锁住该行串行化,第二个请求拿锁后看到 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()
 
 
@@ -99,16 +180,5 @@ def reject_price_report(
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
 ) -> OkResponse:
-    rep = db.get(PriceReport, report_id, with_for_update=True)  # 行锁,同 approve(防并发重复审核)
+    _reject_price_report(db, admin, report_id, body.reason, get_client_ip(request))
-    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/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
+from pydantic import BaseModel, ConfigDict, Field, field_validator
 
 from app.core.rewards import FEEDBACK_REWARD_MAX_COINS
 
@@ -55,6 +55,54 @@ 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/price_report.py

@@ -56,6 +56,42 @@ 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,6 +20,8 @@ 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,
@@ -35,6 +37,41 @@ 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 环境,给 coupon_claim_record 打 app_env 标(每券成功率表按它过滤;设计 §13)。
+        # 取本次 session 环境，给每日资产和逐次事件同时打环境标。
         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()
 
-    # 领券结果沉淀(资产):中间帧 last_coupon_result + done 帧 coupon_results 幂等写库。
+    # 领券结果沉淀:每日资产 + 逐次事件；中间帧和 done 全量帧均幂等写库。
     # 当前只记录、不参与"要不要领"判断(MVP 先不去重)。写库失败不影响返回。
     if device_id:
         results = _extract_coupon_results(resp_json)

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"),
+        transfer_auth_enabled=bool(auth and auth.state == "active" and auth.authorization_id),
         tiers=[WithdrawTierOut(**t) for t in crud_wallet.withdraw_tier_states(db, user.id, source)],
     )
 
@@ -335,7 +335,8 @@ 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"
-    return TransferAuthStatusOut(state=state, enabled=(state == "active"))
+    enabled = bool(auth and auth.state == "active" and auth.authorization_id)
+    return TransferAuthStatusOut(state=state, enabled=enabled)
 
 
 @router.post(

app/core/config.py

@@ -188,6 +188,13 @@ 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 移动应用,用于微信支付);这是【已认证服务号】。

app/core/cps_reconcile_worker.py

@@ -0,0 +1,428 @@
+"""美团、京东 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,6 +91,55 @@ 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
 
 
@@ -126,7 +175,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 = RotatingFileHandler(
+    file_handler = SafeRotatingFileHandler(
         log_file, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8",
     )
     file_handler.setFormatter(JsonFormatter(service))

app/main.py

@@ -43,6 +43,10 @@ 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,
@@ -89,6 +93,7 @@ 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()
@@ -98,6 +103,7 @@ 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)

app/models/__init__.py

@@ -21,6 +21,7 @@ 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,

app/models/coupon_state.py

@@ -96,6 +96,40 @@ 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/repositories/comparison.py

@@ -5,7 +5,7 @@
 """
 from __future__ import annotations
 
-from datetime import datetime
+from datetime import datetime, timedelta
 
 from sqlalchemy import func, or_, select
 from sqlalchemy.orm import Session, defer
@@ -14,8 +14,19 @@ 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 透传。"""
@@ -243,6 +254,78 @@ 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,6 +14,7 @@ from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
 from app.models.coupon_state import (
+    CouponClaimEvent,
     CouponClaimRecord,
     CouponDailyCompletion,
     CouponPromptEngagement,
@@ -164,11 +165,12 @@ 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)。
-    (device, coupon_id, 今天) 唯一:重复上报同张券走更新(status 以最后一次为准)。
+    - CouponClaimRecord 按 (device, coupon_id, 今天) 幂等，供每日资产口径使用。
+    - CouponClaimEvent 按 (trace_id, coupon_id) 幂等，供 admin 逐场统计使用。
     """
     today = today_cn()
     written = 0
@@ -208,6 +210,41 @@ 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/schemas/compare_record.py

@@ -13,7 +13,6 @@ from datetime import datetime
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator
 
-
 # ===== 上报请求 =====
 
 class ComparisonItemIn(BaseModel):
@@ -198,6 +197,20 @@ 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):
     """「我的」页省钱战绩卡(比价口径)聚合。"""
 

scripts/seed_coupon_session_mock.py

@@ -0,0 +1,197 @@
+"""生成 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_mock_feedback.py

@@ -0,0 +1,277 @@
+"""一次性 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,6 +30,7 @@ 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 控制台输出中文/¥
@@ -84,10 +85,11 @@ def seed(db) -> list[PriceReport]:
     # 1) 建 3 个 mock 用户
     users: dict[str, User] = {}
     for i, (phone, nickname) in enumerate(
-        zip(MOCK_PHONES, ["省钱小王", "比价老李", "薅羊毛阿珍"])
+        zip(MOCK_PHONES, ["省钱小王", "比价老李", "薅羊毛阿珍"], strict=True)
     ):
         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 datetime, timedelta, timezone
+from datetime import UTC, datetime, timedelta
 
 from sqlalchemy import delete, select
 
@@ -31,6 +31,7 @@ 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"):
@@ -51,7 +52,7 @@ REMARK = {"exchange_in": "金币兑入", "withdraw": "提现扣款", "withdraw_r
 
 def _naive_utc_now() -> datetime:
     """与 func.now() 在 SQLite 的口径一致:naive UTC。前端按 UTC 解析再转北京时间。"""
-    return datetime.now(timezone.utc).replace(tzinfo=None)
+    return datetime.now(UTC).replace(tzinfo=None)
 
 
 def _mock_transfer_no() -> str:
@@ -96,6 +97,7 @@ 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

@@ -468,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,
@@ -514,3 +514,51 @@ 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,6 +14,7 @@ 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
@@ -82,6 +83,26 @@ 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:
@@ -397,6 +418,100 @@ 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

@@ -0,0 +1,120 @@
+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_comparison_admin_summary.py

@@ -4,12 +4,28 @@ from __future__ import annotations
 from datetime import UTC, date, datetime
 
 import pytest
+from sqlalchemy.dialects import postgresql
 
 from app.admin.repositories import queries
 from app.db.session import SessionLocal
 from app.models.comparison import ComparisonRecord
 
 
+def test_postgresql_duration_summary_uses_ordered_set_aggregates() -> None:
+    stmt = queries._comparison_duration_aggregate_stmt(
+        [], "success", (0.05, 0.5, 0.95, 0.99)
+    )
+    sql = str(
+        stmt.compile(
+            dialect=postgresql.dialect(),
+            compile_kwargs={"literal_binds": True},
+        )
+    )
+
+    assert sql.count("percentile_cont") == 4
+    assert "comparison_record.status = 'success'" in sql
+
+
 def test_summary_uses_only_success_durations_and_filters_beijing_date() -> None:
     db = SessionLocal()
     try:

tests/test_coupon_claim_event.py

@@ -0,0 +1,54 @@
+"""逐次单券事件不能被同设备同日的每日去重记录串场。"""
+
+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_platform_success.py

@@ -7,8 +7,12 @@ from __future__ import annotations
 from datetime import UTC, date, datetime
 
 from sqlalchemy import delete, func, select
+from sqlalchemy.dialects import postgresql
 
-from app.admin.repositories.coupon_data import coupon_data_report
+from app.admin.repositories.coupon_data import (
+    _coupon_summary_aggregate_stmt,
+    coupon_data_report,
+)
 from app.db.session import SessionLocal
 from app.models.coupon_state import CouponSession
 from app.repositories.coupon_state import (
@@ -19,7 +23,12 @@ from app.repositories.coupon_state import (
 
 
 def _agg_session(
-    trace: str, platforms, platform_success, *, status: str = "completed"
+    trace: str,
+    platforms,
+    platform_success,
+    *,
+    status: str = "completed",
+    elapsed_ms: int | None = None,
 ) -> CouponSession:
     """构造一条聚合测试用 session(started_date 固定 2020-01-02、app_env=prod,不 commit)。"""
     return CouponSession(
@@ -31,9 +40,53 @@ def _agg_session(
         platform_success=platform_success,
         started_at=datetime(2020, 1, 2, tzinfo=UTC),
         started_date=date(2020, 1, 2),
+        elapsed_ms=elapsed_ms,
     )
 
 
+def test_postgresql_coupon_summary_uses_ordered_set_aggregates() -> None:
+    sql = str(
+        _coupon_summary_aggregate_stmt([]).compile(
+            dialect=postgresql.dialect(),
+            compile_kwargs={"literal_binds": True},
+        )
+    )
+
+    assert sql.count("percentile_cont") == 4
+    assert "FILTER (WHERE coupon_session.status = 'completed'" in sql
+
+
+def test_coupon_duration_summary_uses_only_completed_rows() -> None:
+    db = SessionLocal()
+    try:
+        db.add_all([
+            _agg_session("duration-a", [], [], elapsed_ms=1000),
+            _agg_session("duration-b", [], [], elapsed_ms=3000),
+            _agg_session(
+                "duration-failed", [], [], status="failed", elapsed_ms=100_000
+            ),
+        ])
+        db.flush()
+
+        summary = coupon_data_report(
+            db,
+            date_from="2020-01-02",
+            date_to="2020-01-02",
+            app_env="prod",
+        )["summary"]
+
+        assert summary["started_count"] == 3
+        assert summary["completed_count"] == 2
+        assert summary["avg_elapsed_ms"] == 2000
+        assert summary["p5_ms"] == 1100
+        assert summary["p50_ms"] == 2000
+        assert summary["p95_ms"] == 2900
+        assert summary["p99_ms"] == 2980
+    finally:
+        db.rollback()
+        db.close()
+
+
 def _make_session(db, trace_id: str, **kw) -> CouponSession:
     row = CouponSession(
         trace_id=trace_id,

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 CouponClaimRecord, CouponSession
+from app.models.coupon_state import CouponClaimEvent, 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([
-            CouponClaimRecord(
+            CouponClaimEvent(
+                trace_id=trace,
                 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(CouponClaimRecord(
+        db.add(CouponClaimEvent(
+            trace_id=trace,
             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([
-            CouponClaimRecord(
+            CouponClaimEvent(
+                trace_id=trace,
                 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(CouponClaimRecord(
+        db.add(CouponClaimEvent(
+            trace_id=trace,
             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(CouponClaimRecord).where(CouponClaimRecord.trace_id == trace))
+        db.execute(delete(CouponClaimEvent).where(CouponClaimEvent.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 CouponClaimRecord, CouponSession
+from app.models.coupon_state import CouponClaimEvent, CouponClaimRecord, CouponSession
 from app.repositories.coupon_state import record_claims, session_app_env
 
 
@@ -49,6 +49,7 @@ 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

@@ -0,0 +1,197 @@
+"""美团、京东 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_logging_rotation.py

@@ -0,0 +1,73 @@
+"""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
+from app.models.wallet import CoinAccount, WithdrawOrder, WechatTransferAuthorization
 from app.repositories import wallet as crud_wallet
 
 
@@ -426,3 +426,62 @@ 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