chore(deploy): 穿山甲收益拉取 systemd 定时器(每天10:30)+ 运维手册

为 #92 的 scripts/sync_pangle_revenue.py 补 systemd 部署单元,让穿山甲 GroMore T+1 后台收益每天自动拉取入库,免人工敲命令。纯部署/文档文件,不改运行时代码。

- service:oneshot,跑 `--days 3` 回补近 3 天(幂等 upsert,扛偶发漏跑与穿山甲历史订正);10min 硬超时;加固块对齐主服务。
- timer:每天 10:30 触发(穿山甲约 10:00 出数留余量),Persistent 宕机后补跑。
- md:运维手册——上线前置(.env 填 PANGLE_REPORT_* 凭证、子账号需授权否则接口 118)、部署/健康检查/参数/注意事项(join key 用 ad_unit_id 非 code_id 等)。
- 风格对齐现有 deploy/daily-exchange.* 与 deploy/meituan-etl.*。

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

alembic/versions/761ef181ce7c_merge_jd_cps_order_fields_and_coupon_.py

@@ -1,26 +0,0 @@
-"""merge jd_cps_order_fields and coupon_session_origin_package heads
-
-Revision ID: 761ef181ce7c
-Revises: coupon_session_origin_package, jd_cps_order_fields
-Create Date: 2026-07-01 13:52:16.068808
-
-"""
-from typing import Sequence, Union
-
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision: str = '761ef181ce7c'
-down_revision: Union[str, Sequence[str], None] = ('coupon_session_origin_package', 'jd_cps_order_fields')
-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_session_origin_package.py

@@ -1,32 +0,0 @@
-"""coupon_session 加 origin_package 列(发起来源 App 包名 → admin「发起平台」)
-
-null=App 内(傻瓜比价首页)发起,非空=从美团/淘宝/京东弹券发起。与 trace_url 同理单独成迁移,
-已建表环境靠它补列、全新环境顺序应用,不重复加列。
-
-Revision ID: coupon_session_origin_package
-Revises: coupon_session_trace_url
-Create Date: 2026-06-30 00:00:00.000000
-
-"""
-
-from collections.abc import Sequence
-
-import sqlalchemy as sa
-
-from alembic import op
-
-# revision identifiers, used by Alembic.
-revision: str = "coupon_session_origin_package"
-down_revision: str | Sequence[str] | None = "coupon_session_trace_url"
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
-
-
-def upgrade() -> None:
-    with op.batch_alter_table("coupon_session", schema=None) as batch_op:
-        batch_op.add_column(sa.Column("origin_package", sa.String(length=64), nullable=True))
-
-
-def downgrade() -> None:
-    with op.batch_alter_table("coupon_session", schema=None) as batch_op:
-        batch_op.drop_column("origin_package")

alembic/versions/coupon_session_table.py

@@ -1,81 +0,0 @@
-"""coupon session table(领券任务全程流水 → admin「领券数据」看板数据源)
-
-一次领券一行(trace_id 唯一):客户端 POST /api/v1/coupon/session 两段上报 —— 发起建行
-(status=started)、收尾(completed/failed/abandoned)按 trace_id 更新同一行。记全程耗时
-elapsed_ms + 各平台耗时 platform_elapsed + 机型/ROM,供 admin 算发起/完成数、耗时分位、机型维度。
-
-Revision ID: coupon_session_table
-Revises: feedback_submit_env
-Create Date: 2026-06-30 00:00:00.000000
-
-"""
-
-from collections.abc import Sequence
-
-import sqlalchemy as sa
-from sqlalchemy.dialects import postgresql
-
-from alembic import op
-
-# revision identifiers, used by Alembic.
-revision: str = "coupon_session_table"
-down_revision: str | Sequence[str] | None = "feedback_submit_env"
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
-
-# PG 用 JSONB,SQLite(本地/测试)退化为通用 JSON(同 model 的 _JSON variant)。
-_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
-
-
-def upgrade() -> None:
-    op.create_table(
-        "coupon_session",
-        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("status", sa.String(length=16), nullable=False),
-        sa.Column("app_env", sa.String(length=16), nullable=True),
-        sa.Column("platforms", _JSON, nullable=True),
-        sa.Column("device_model", sa.String(length=128), nullable=True),
-        sa.Column("rom", sa.String(length=64), nullable=True),
-        sa.Column("started_at", sa.DateTime(timezone=True), nullable=False),
-        sa.Column("started_date", sa.Date(), nullable=False),
-        sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
-        sa.Column("elapsed_ms", sa.Integer(), nullable=True),
-        sa.Column("platform_elapsed", _JSON, nullable=True),
-        sa.Column("claimed_count", sa.Integer(), nullable=True),
-        sa.Column(
-            "created_at",
-            sa.DateTime(timezone=True),
-            server_default=sa.text("(CURRENT_TIMESTAMP)"),
-            nullable=False,
-        ),
-        sa.Column(
-            "updated_at",
-            sa.DateTime(timezone=True),
-            server_default=sa.text("(CURRENT_TIMESTAMP)"),
-            nullable=False,
-        ),
-        sa.PrimaryKeyConstraint("id"),
-        sa.UniqueConstraint("trace_id", name="uq_coupon_session_trace"),
-    )
-    with op.batch_alter_table("coupon_session", schema=None) as batch_op:
-        batch_op.create_index(
-            batch_op.f("ix_coupon_session_user_id"), ["user_id"], unique=False
-        )
-        batch_op.create_index(
-            batch_op.f("ix_coupon_session_app_env"), ["app_env"], unique=False
-        )
-        # admin 主聚合/筛选:按上海自然日 + 环境。
-        batch_op.create_index(
-            "ix_coupon_session_date_env", ["started_date", "app_env"], unique=False
-        )
-
-
-def downgrade() -> None:
-    with op.batch_alter_table("coupon_session", schema=None) as batch_op:
-        batch_op.drop_index("ix_coupon_session_date_env")
-        batch_op.drop_index(batch_op.f("ix_coupon_session_app_env"))
-        batch_op.drop_index(batch_op.f("ix_coupon_session_user_id"))
-    op.drop_table("coupon_session")

alembic/versions/coupon_session_trace_url.py

@@ -1,32 +0,0 @@
-"""coupon_session 加 trace_url 列(pricebot done 帧公网调试链接)
-
-建表迁移 coupon_session_table 落地后才追加本列,故单独成一个迁移:已建表的环境(本地/已 upgrade 过)
-靠它补列,全新环境则「建表(无 trace_url)→ 本迁移加列」,两条路一致、不重复加列。
-
-Revision ID: coupon_session_trace_url
-Revises: coupon_session_table
-Create Date: 2026-06-30 00:00:00.000000
-
-"""
-
-from collections.abc import Sequence
-
-import sqlalchemy as sa
-
-from alembic import op
-
-# revision identifiers, used by Alembic.
-revision: str = "coupon_session_trace_url"
-down_revision: str | Sequence[str] | None = "coupon_session_table"
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
-
-
-def upgrade() -> None:
-    with op.batch_alter_table("coupon_session", schema=None) as batch_op:
-        batch_op.add_column(sa.Column("trace_url", sa.String(length=512), nullable=True))
-
-
-def downgrade() -> None:
-    with op.batch_alter_table("coupon_session", schema=None) as batch_op:
-        batch_op.drop_column("trace_url")

app/admin/main.py

@@ -21,7 +21,6 @@ from app.admin.routers.audit import router as audit_router
 from app.admin.routers.auth import router as auth_router
 from app.admin.routers.comparison import router as comparison_router
 from app.admin.routers.config import router as config_router
-from app.admin.routers.coupon_data import router as coupon_data_router
 from app.admin.routers.cps import router as cps_router
 from app.admin.routers.dashboard import router as dashboard_router
 from app.admin.routers.device_liveness import router as device_liveness_router
@@ -102,7 +101,6 @@ admin_app.include_router(audit_router)
 admin_app.include_router(config_router)
 admin_app.include_router(comparison_router)
 admin_app.include_router(cps_router)
-admin_app.include_router(coupon_data_router)
 admin_app.include_router(ad_audit_router)
 admin_app.include_router(ad_config_router)
 admin_app.include_router(ad_revenue_router)

app/admin/repositories/ad_audit.py

@@ -136,16 +136,12 @@ def _feed_scene_matches(rec: AdFeedRewardRecord, scene: str | None) -> bool:
     """该信息流记录是否落入请求的展示筛选 scene。
     - scene=="feed":ad_type in ("feed", NULL)(旧数据 NULL 视为 feed,向后兼容)
     - scene=="draw":ad_type=="draw"
-    - scene=="feed_all":所有信息流(feed/draw/NULL 都要)——业务已全切 Draw 信息流,收益报表把「Draw 信息流」
-      当作整个信息流口径(含历史误标 feed/NULL),用它避免筛选漏历史。
     - scene 为 None:不筛(两类都要)。
     """
     if scene == "feed":
         return rec.ad_type in (None, "feed")
     if scene == "draw":
         return rec.ad_type == "draw"
-    if scene == "feed_all":
-        return True
     return True
 
 
@@ -188,7 +184,6 @@ def _feed_rows(
                 "record_id": rec.id,
                 "user_id": rec.user_id,
                 "ad_session_id": rec.ad_session_id,
-                "trace_id": rec.trace_id,
                 "app_env": rec.app_env,
                 "our_code_id": rec.our_code_id,
                 "created_at": rec.created_at,
@@ -214,7 +209,6 @@ def _feed_rows(
                 "record_id": rec.id,
                 "user_id": rec.user_id,
                 "ad_session_id": rec.ad_session_id,
-                "trace_id": rec.trace_id,
                 "app_env": rec.app_env,
                 "our_code_id": rec.our_code_id,
                 "created_at": rec.created_at,
@@ -247,7 +241,7 @@ def audit_rows(
     rows: list[dict] = []
     if scene in (None, "reward_video"):
         rows.extend(_reward_video_rows(db, date=date, user_id=user_id))
-    if scene in (None, "feed", "draw", "feed_all"):
+    if scene in (None, "feed", "draw"):
         rows.extend(_feed_rows(db, date=date, user_id=user_id, scene=scene))
     return rows
 

app/admin/repositories/ad_revenue.py

@@ -3,11 +3,9 @@
 只读。每行 = 一次广告事件(不再按用户聚合):
 - **激励视频**:一次观看 = 1 条展示(ad_ecpm)+ 1 条发奖(ad_reward),按 ad_session_id 合并成一行,
   直接给出 eCPM / 收益 + 状态 / 应发 / 实发 / 一致;点开看该条金币复算因子。
-- **信息流(比价/领券)**:一次比价 / 一次领券 = 一条整场发奖(ad_feed_reward)一行,给出 eCPM /
+- **信息流**:轮播每条展示各一行(impressionId 各自独立);整场发奖(ad_feed_reward,client_event_id)
-  发奖金币 + 应发 / 实发 / 一致;点开看金币复算因子。⚠️ draw 的逐条展示(ad_ecpm,impressionId 各自
+  与逐条展示无法对应,单独成「纯发奖」行。
-  独立、与整场发奖无公共键、无法归到「哪一次」)**不再单独占行**(2026-07 按「一次比价/领券放一块」调整)——
+- 兜底:有展示无发奖(中途关 / 未达发奖)、有发奖无展示(未上报 eCPM)都各自成行。
-  其展示数 / eCPM / 预估收益仍进全量统计(合计 / 趋势 / 分类大盘 / 穿山甲对照),只是主表不逐条铺开。
-- 兜底:激励视频有展示无发奖(中途关 / 未达发奖)、有发奖无展示(未上报 eCPM)仍各自成行。
 
 展示与收益来自 ad_ecpm_record(收益 = eCPM元 ÷ 1000);应发 / 实发金币复用金币审计逐条复算
 (ad_audit.audit_rows,与正式发奖同一公式口径,不另写公式)。合计与对账在全量上统计,
@@ -60,6 +58,14 @@ def _date_range(date_from: str, date_to: str) -> list[str]:
 _AUDIT_SCENES = {"reward_video", "feed", "draw"}
 
 
+def _event_ad_type(row: dict) -> str:
+    """纯发奖事件行的 ad_type:信息流行用 audit 带回的真实 ad_type(feed/draw),回退 feed;
+    激励视频行恒 reward_video。不再用 scene 硬映射,避免把 draw 丢成 feed。"""
+    if row["scene"] == "reward_video":
+        return "reward_video"
+    return row.get("ad_type") or "feed"
+
+
 # 发奖复算明细字段(展开下钻看「金币怎么算出来的」)——从 audit 行原样取这些 key。
 _REWARD_DETAIL_KEYS = (
     "record_id", "created_at", "status", "ecpm", "ecpm_factor", "units",
@@ -102,14 +108,9 @@ def ad_revenue_report(
     #    同时保留全量列表,未被展示合并的成「纯发奖」事件。
     reward_by_session: dict[tuple[int, str], list[dict]] = {}
     all_reward_rows: list[dict] = []
-    # 报表 ad_type → audit scene:reward_video/feed 直传;**draw(前端「Draw 信息流」)映射成 feed_all**
+    # 报表 ad_type 直接当 audit scene 用(取值一致);未知/无效 ad_type 不取发奖行。draw 在此被
-    # ——业务已全切 Draw,把「Draw 信息流」当作整个信息流口径(含历史误标 feed/NULL),否则筛选会漏历史。
+    # 正确传成 scene="draw",audit 会按 ad_type 筛出 Draw 发奖,不再丢成 feed。
-    if ad_type == "draw":
+    audit_scene = ad_type if ad_type in _AUDIT_SCENES else None
-        audit_scene = "feed_all"
-    elif ad_type in _AUDIT_SCENES:
-        audit_scene = ad_type
-    else:
-        audit_scene = None
     if ad_type is None or audit_scene is not None:
         for d in _date_range(date_from, date_to):
             for row in ad_audit.audit_rows(db, date=d, user_id=user_id, scene=audit_scene):
@@ -139,10 +140,7 @@ def ad_revenue_report(
     )
     if user_id is not None:
         stmt = stmt.where(AdEcpmRecord.user_id == user_id)
-    if ad_type == "draw":
+    if ad_type is not None:
-        # draw = 所有信息流展示(业务已全 Draw,含历史误标 feed);展示行只进统计,不占主表行
-        stmt = stmt.where(AdEcpmRecord.ad_type.in_(["draw", "feed"]))
-    elif ad_type is not None:
         stmt = stmt.where(AdEcpmRecord.ad_type == ad_type)
     for rec in db.execute(stmt).scalars():
         rwd = _pop_reward(rec.user_id, rec.ad_session_id)
@@ -167,8 +165,6 @@ def ad_revenue_report(
             ),
             "adn": rec.adn,
             "slot_id": rec.slot_id,
-            "sub_rewards": [],
-            "sub_count": 1,
         }
         if rwd is not None:
             ev.update({
@@ -188,88 +184,33 @@ def ad_revenue_report(
             })
         events.append(ev)
 
-    # 3) 未被展示合并的发奖行 → 事件:
+    # 3) 未被展示合并的发奖行 → 「纯发奖」事件(信息流整场发奖 / 有发奖无展示)。
-    #    - 激励视频(reward_video):逐条成「纯发奖」事件(每次一个 ad_session_id;有发奖无展示等)。
+    #    收益恒 0(收益只算展示侧,避免与展示行重复计)。
-    #    - 信息流(feed/draw):同一次比价/领券的多条广告共享**整场 ad_session_id**(客户端整场复用),
-    #      按 (user_id, ad_session_id) 聚成**一次比价 / 一次领券**父事件;sub_rewards 为组内逐条明细,
-    #      应发/实发取组内合计;业务已全 Draw → 类型统一 "draw"。session 缺失(极少旧数据)各自单独成组。
-    feed_groups: dict[tuple[int, str], list[dict]] = {}
     for row in all_reward_rows:
         if row["record_id"] in used_reward_ids:
             continue
-        if row["scene"] == "reward_video":
-            events.append({
-                "event_key": f"rwd-{row['record_id']}",
-                "report_date": row["_report_date"],
-                "user_id": row["user_id"],
-                "ad_type": "reward_video",
-                "feed_scene": row.get("feed_scene"),
-                "app_env": row.get("app_env"),
-                "our_code_id": row.get("our_code_id"),
-                "created_at": row["created_at"],
-                "hour": _cn_hour(row["created_at"]) if by_hour else None,
-                "has_impression": False,
-                "impressions": 0,
-                "ecpm": row["ecpm"],
-                "revenue_yuan": 0.0,
-                "adn": None,
-                "slot_id": None,
-                "has_reward": True,
-                "status": row["status"],
-                "expected_coin": int(row["expected_coin"]),
-                "actual_coin": int(row["actual_coin"]),
-                "matched": bool(row["matched"]),
-                "reward_detail": _reward_detail(row),
-                "sub_rewards": [],
-                "sub_count": 1,
-            })
-        else:
-            # 聚合单位 = 一次完整比价/领券流程:优先用 trace_id(比价带 comparisonTraceId、领券带 sessionTraceId,
-            # 整个流程不变;即使中途点广告致浮层关闭重弹、ad_session_id 变了,trace_id 仍不变 → 全流程聚成一行)。
-            # 无 trace_id(历史领券未上报 / 旧数据)回退整场 ad_session_id;再无则 record_id 各自成组、不误并。
-            grp_key = row.get("trace_id") or row.get("ad_session_id") or f"_rid-{row['record_id']}"
-            feed_groups.setdefault((row["user_id"], grp_key), []).append(row)
-
-    # 信息流分组 → 「一次比价 / 一次领券」父事件(收益恒 0:收益只算展示侧,避免与展示行重复计)。
-    for (uid, grp_key), group in feed_groups.items():
-        group.sort(key=lambda r: (r["created_at"], r["record_id"]))
-        rep = group[-1]  # 代表条(最新一条):时间/场景/应用/代码位取它
-        expected_sum = sum(int(g["expected_coin"]) for g in group)
-        actual_sum = sum(int(g["actual_coin"]) for g in group)
-        # 父行 eCPM:组内各条 eCPM(分)均值(展示用,各条不同);无有效值则取代表条
-        ecpm_fens = [rewards.parse_ecpm_fen(g["ecpm"]) for g in group if g.get("ecpm")]
-        avg_ecpm = str(round(sum(ecpm_fens) / len(ecpm_fens))) if ecpm_fens else rep.get("ecpm")
-        # 主表逐行显示用:这次发奖广告的预估收益之和(发奖侧 eCPM 折算,钳顶同展示侧)。只放进
-        # row_revenue_yuan 给主表逐行展示,不进 revenue_yuan/合计/趋势——避免与展示侧 total 重复计。
-        row_revenue = round(sum(
-            min(rewards.parse_ecpm_yuan(g["ecpm"]), rewards.AD_ECPM_MAX_FEN / 100.0) / 1000.0
-            for g in group if g.get("ecpm")
-        ), 6)
         events.append({
-            "event_key": f"feedgrp-{uid}-{grp_key}",
+            "event_key": f"rwd-{row['record_id']}",
-            "report_date": rep["_report_date"],
+            "report_date": row["_report_date"],
-            "user_id": uid,
+            "user_id": row["user_id"],
-            "ad_type": "draw",  # 业务已全切 Draw 信息流,聚合行统一 draw
+            "ad_type": _event_ad_type(row),
-            "feed_scene": rep.get("feed_scene"),
+            "feed_scene": row.get("feed_scene"),
-            "app_env": rep.get("app_env"),
+            "app_env": row.get("app_env"),
-            "our_code_id": rep.get("our_code_id"),
+            "our_code_id": row.get("our_code_id"),
-            "created_at": rep["created_at"],
+            "created_at": row["created_at"],
-            "hour": _cn_hour(rep["created_at"]) if by_hour else None,
+            "hour": _cn_hour(row["created_at"]) if by_hour else None,
             "has_impression": False,
             "impressions": 0,
-            "ecpm": avg_ecpm,
+            "ecpm": row["ecpm"],
             "revenue_yuan": 0.0,
-            "row_revenue_yuan": row_revenue,
             "adn": None,
             "slot_id": None,
             "has_reward": True,
-            "status": rep["status"],  # 代表状态(逐条见展开)
+            "status": row["status"],
-            "expected_coin": expected_sum,
+            "expected_coin": int(row["expected_coin"]),
-            "actual_coin": actual_sum,
+            "actual_coin": int(row["actual_coin"]),
-            "matched": all(bool(g["matched"]) for g in group),
+            "matched": bool(row["matched"]),
-            "reward_detail": None,
+            "reward_detail": _reward_detail(row),
-            "sub_rewards": [_reward_detail(g) for g in group],
-            "sub_count": len(group),
         })
 
     # 「场景」作为全局筛选(与 user_id/ad_type 一致):同时作用于明细、合计与 daily/hourly 趋势。
@@ -390,20 +331,9 @@ def ad_revenue_report(
     is_today = date_from == date_to == rewards.cn_today().isoformat()
     dau = admin_stats.today_dau(db) if is_today else None
 
-    # 主表「逐行」= 单次广告行为(2026-07 按「一次比价/领券放一块」聚合):激励视频 = 一次观看一行(展示+发奖
-    # 按 ad_session_id 合并);一次比价 / 一次领券 = 该次整场多条广告按 ad_session_id 聚成一行(展开看逐条)。
-    # 信息流(draw/feed)的逐条展示(ad_ecpm,impressionId 各自独立、与整场发奖无公共键)不再单独占行
-    # ——其展示数 / eCPM / 预估收益已计入上面的全量统计(total_*、daily / hourly、type_stats、穿山甲对照),
-    # 只是主表不逐条铺开;逐条明细在父行展开里看(sub_rewards)。合计 / 趋势 / 分类大盘均基于全量 events,
-    # 不受此过滤影响;total / 分页只作用于主表行。
-    main_rows = [
-        e for e in events
-        if not (e["ad_type"] in ("draw", "feed") and e["has_impression"] and not e["has_reward"])
-    ]
-
     return {
-        "total": len(main_rows),
+        "total": len(events),
-        "truncated": len(main_rows) > offset + limit,
+        "truncated": len(events) > offset + limit,
         "total_impressions": total_impressions,
         "total_revenue_yuan": total_revenue_yuan,
         # 穿山甲后台收益合计(元):预估 revenue + 收益Api;非全量视图(带 user/类型/场景过滤)或无数据为 None。
@@ -417,5 +347,5 @@ def ad_revenue_report(
         "hourly": hourly,
         "type_stats": type_stats,
         "dau": dau,
-        "items": main_rows[offset:offset + limit],
+        "items": events[offset:offset + limit],
     }

app/admin/repositories/coupon_data.py

@@ -1,225 +0,0 @@
-"""admin「领券数据」看板聚合:发起/完成数、耗时均值与分位、按天/小时趋势、逐条明细。
-
-数据源 coupon_session(一次领券一行,客户端 /api/v1/coupon/session 两段上报)。量级不大,全量拉
-区间数据后 Python 聚合(分位 SQLite 无 percentile,统一 Python 算,PG 上也一致)。
-- 发起数 = 区间内全部 session(含 started/completed/failed/abandoned),= 流失统计的基数。
-- 完成数 / 耗时均值 / 分位 = 仅 status==completed 子集(成功跑完才有可比的"领券耗时")。
-- summary/daily/hourly/total 在全量上算,不受分页;items 为排序后当前页。
-"""
-from __future__ import annotations
-
-from datetime import UTC, date as _date, datetime
-
-from sqlalchemy import func, or_, select
-from sqlalchemy.orm import Session
-
-from app.core import rewards
-from app.models.coupon_state import CouponSession
-from app.models.user import User
-
-
-def _cn_hour(dt: datetime) -> int:
-    """started_at(UTC 口径)→ 北京时间小时(0–23)。naive 当 UTC(sqlite),tz-aware 直接换算(pg)。"""
-    if dt.tzinfo is None:
-        dt = dt.replace(tzinfo=UTC)
-    return dt.astimezone(rewards.CN_TZ).hour
-
-
-def _percentile(sorted_vals: list[int], q: float) -> int | None:
-    """线性插值分位(q=0..100,numpy 默认法)。sorted_vals 须已升序;空返回 None。"""
-    if not sorted_vals:
-        return None
-    if len(sorted_vals) == 1:
-        return sorted_vals[0]
-    idx = (len(sorted_vals) - 1) * q / 100.0
-    lo = int(idx)
-    hi = min(lo + 1, len(sorted_vals) - 1)
-    frac = idx - lo
-    return round(sorted_vals[lo] * (1 - frac) + sorted_vals[hi] * frac)
-
-
-def _avg(vals: list[int]) -> int | None:
-    return round(sum(vals) / len(vals)) if vals else None
-
-
-def _session_to_row(r, phone: str | None = None, nickname: str | None = None) -> dict:
-    """CouponSession ORM → 明细行 dict(主表「领券数据」与「用户全部领券」抽屉共用)。"""
-    return {
-        "id": r.id,
-        "trace_id": r.trace_id,
-        "user_id": r.user_id,
-        "user_phone": phone,
-        "user_nickname": nickname,
-        "status": r.status,
-        "platforms": r.platforms,
-        "origin_package": r.origin_package,
-        "elapsed_ms": r.elapsed_ms,
-        "platform_elapsed": r.platform_elapsed,
-        "device_model": r.device_model,
-        "rom": r.rom,
-        "app_env": r.app_env,
-        "started_at": r.started_at,
-        "claimed_count": r.claimed_count,
-        "trace_url": r.trace_url,
-    }
-
-
-def _empty_result() -> dict:
-    return {
-        "summary": {
-            "started_count": 0, "completed_count": 0, "avg_elapsed_ms": None,
-            "p5_ms": None, "p50_ms": None, "p95_ms": None, "p99_ms": None,
-        },
-        "daily": [],
-        "hourly": [],
-        "total": 0,
-        "items": [],
-    }
-
-
-def coupon_data_report(
-    db: Session,
-    *,
-    date_from: str,
-    date_to: str,
-    user: str | None = None,
-    app_env: str | None = None,
-    granularity: str = "day",
-    limit: int = 500,
-    offset: int = 0,
-    sort: str = "time",
-) -> dict:
-    """日期区间(北京自然日 started_date,闭区间)领券数据:汇总卡 + 趋势 + 逐条明细。
-
-    - user:手机号/昵称模糊搜(匹配不到任何用户 → 空结果)。
-    - app_env:prod/dev 精确;None=全部。
-    - sort:time=发起时刻倒序(默认) / elapsed=全程耗时倒序(None 末尾)。
-    """
-    by_hour = granularity == "hour"
-    d_from = _date.fromisoformat(date_from)
-    d_to = _date.fromisoformat(date_to)
-
-    # user 模糊 → 先定位匹配用户 id;匹配不到直接空结果(不全表扫)。
-    user_ids: set[int] | None = None
-    if user:
-        like = f"%{user}%"
-        user_ids = set(db.execute(
-            select(User.id).where(or_(User.phone.like(like), User.nickname.like(like)))
-        ).scalars().all())
-        if not user_ids:
-            return _empty_result()
-
-    stmt = select(CouponSession).where(
-        CouponSession.started_date >= d_from,
-        CouponSession.started_date <= d_to,
-    )
-    if app_env is not None:
-        stmt = stmt.where(CouponSession.app_env == app_env)
-    if user_ids is not None:
-        stmt = stmt.where(CouponSession.user_id.in_(user_ids))
-    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),
-        "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),
-    }
-
-    # ── 按天趋势(柱=发起/完成数,线=平均耗时)──
-    daily_map: dict[str, dict] = {}
-    for r in rows:
-        d = r.started_date.isoformat()
-        b = daily_map.get(d)
-        if b is None:
-            b = {"date": d, "started_count": 0, "completed_count": 0, "_elapsed": []}
-            daily_map[d] = b
-        b["started_count"] += 1
-        if r.status == "completed":
-            b["completed_count"] += 1
-            if r.elapsed_ms is not None:
-                b["_elapsed"].append(r.elapsed_ms)
-    daily = [
-        {
-            "date": b["date"],
-            "started_count": b["started_count"],
-            "completed_count": b["completed_count"],
-            "avg_elapsed_ms": _avg(b["_elapsed"]),
-        }
-        for b in sorted(daily_map.values(), key=lambda x: x["date"])
-    ]
-
-    # ── 按小时趋势(单日 hour 粒度)──
-    hourly: list[dict] = []
-    if by_hour:
-        hour_map: dict[int, dict] = {}
-        for r in rows:
-            h = _cn_hour(r.started_at)
-            b = hour_map.get(h)
-            if b is None:
-                b = {"hour": h, "started_count": 0, "completed_count": 0, "_elapsed": []}
-                hour_map[h] = b
-            b["started_count"] += 1
-            if r.status == "completed":
-                b["completed_count"] += 1
-                if r.elapsed_ms is not None:
-                    b["_elapsed"].append(r.elapsed_ms)
-        hourly = [
-            {
-                "hour": b["hour"],
-                "started_count": b["started_count"],
-                "completed_count": b["completed_count"],
-                "avg_elapsed_ms": _avg(b["_elapsed"]),
-            }
-            for b in sorted(hour_map.values(), key=lambda x: x["hour"])
-        ]
-
-    # ── 明细:排序 + 分页 + 补用户手机号/昵称(批量,防 N+1)──
-    if sort == "elapsed":
-        rows.sort(key=lambda r: (r.elapsed_ms is None, -(r.elapsed_ms or 0)))
-    else:  # time:发起时刻倒序
-        rows.sort(key=lambda r: r.started_at, reverse=True)
-    page = rows[offset:offset + limit]
-
-    uids = {r.user_id for r in page if r.user_id is not None}
-    user_map: dict[int, tuple[str | None, str | None]] = {}
-    if uids:
-        user_map = {
-            uid: (phone, nickname)
-            for uid, phone, nickname in db.execute(
-                select(User.id, User.phone, User.nickname).where(User.id.in_(uids))
-            ).all()
-        }
-    items = []
-    for r in page:
-        phone, nickname = user_map.get(r.user_id, (None, None)) if r.user_id is not None else (None, None)
-        items.append(_session_to_row(r, phone, nickname))
-
-    return {
-        "summary": summary,
-        "daily": daily,
-        "hourly": hourly,
-        "total": len(rows),
-        "items": items,
-    }
-
-
-def coupon_user_records(db: Session, *, user_id: int, limit: int = 100) -> dict:
-    """某用户全部领券记录(点手机号抽屉用):按发起时刻倒序、不限日期,total=该用户领券总次数。"""
-    rows = list(db.execute(
-        select(CouponSession)
-        .where(CouponSession.user_id == user_id)
-        .order_by(CouponSession.started_at.desc())
-        .limit(limit)
-    ).scalars())
-    total = db.execute(
-        select(func.count()).select_from(CouponSession).where(CouponSession.user_id == user_id)
-    ).scalar_one()
-    return {"items": [_session_to_row(r) for r in rows], "total": int(total)}

app/admin/repositories/queries.py

@@ -806,12 +806,6 @@ def get_user_overview(db: Session, user_id: int) -> dict | None:
     }
 
 
-def _as_utc_naive(value: datetime) -> datetime:
-    """窗口入参 → UTC naive(= _as_utc 去时区),与库里按 naive UTC 存取的 created_at 同口径比较。
-    历史遗留:_window_conds 一直引用本函数却未定义(自定义区间会 NameError),此处补上。"""
-    return _as_utc(value).replace(tzinfo=None)
-
-
 def _window_conds(col, date_from: datetime | None, date_to: datetime | None) -> list:
     """把 [date_from, date_to] 转成对 col(created_at)的过滤条件;都为 None = 全量(注册至今)。"""
     conds = []
@@ -901,13 +895,6 @@ def user_reward_stats(
     }
 
 
-def _cn_wall_to_utc(dt: datetime) -> datetime:
-    """coin_transaction 存的是北京 wall-clock(naive,见 wallet.grant_coins「存北京 wall-clock」),转成 UTC naive,
-    与广告表(func.now() UTC)统一 —— 让本函数按同一绝对时刻排序、且前端 apiTime(把无时区时间当 UTC 再 +8 展示)
-    口径一致;否则签到会比实际多显示 8 小时(北京时间又被 +8)。"""
-    return dt.replace(tzinfo=rewards.CN_TZ).astimezone(timezone.utc).replace(tzinfo=None)
-
-
 def user_coin_records(
     db: Session,
     user_id: int,
@@ -928,9 +915,6 @@ def user_coin_records(
     offset = max(cursor or 0, 0)
     fetch = offset + limit + 1
     rows: list[dict] = []
-    # coin_transaction 存北京 wall-clock(其余表存 UTC);签到窗口边界 +8h 对齐北京,过滤/计数才不偏移 8 小时
-    signin_from = date_from + timedelta(hours=8) if date_from is not None else None
-    signin_to = date_to + timedelta(hours=8) if date_to is not None else None
 
     for rec in db.execute(
         select(AdRewardRecord)
@@ -974,7 +958,7 @@ def user_coin_records(
         .where(
             CoinTransaction.user_id == user_id,
             CoinTransaction.biz_type == "signin",
-            *_window_conds(CoinTransaction.created_at, signin_from, signin_to),
+            *_window_conds(CoinTransaction.created_at, date_from, date_to),
         )
         .order_by(CoinTransaction.created_at.desc())
         .limit(fetch)
@@ -982,8 +966,7 @@ def user_coin_records(
         rows.append({
             "source": "signin",
             "source_label": "签到",
-            # 北京 wall-clock → UTC,与广告记录统一(前端 apiTime 会 +8 回北京展示,不然签到会多 8 小时)
+            "created_at": rec.created_at,
-            "created_at": _cn_wall_to_utc(rec.created_at),
             "ecpm": None,
             "coin": rec.amount,
         })
@@ -1009,7 +992,7 @@ def user_coin_records(
         + _count(
             CoinTransaction, CoinTransaction.user_id == user_id,
             CoinTransaction.biz_type == "signin",
-            *_window_conds(CoinTransaction.created_at, signin_from, signin_to),
+            *_window_conds(CoinTransaction.created_at, date_from, date_to),
         )
     )
     return rows[offset:offset + limit], (offset + limit if has_more else None), total

app/admin/routers/coupon_data.py

@@ -1,106 +0,0 @@
-"""admin「领券数据」看板:发起/完成数 + 领券耗时(均值 + P5/P50/P95/P99)+ 按天趋势 + 逐条明细。
-
-任意已登录 admin 可看(只读)。聚合逻辑在 app/admin/repositories/coupon_data.py。
-数据源 coupon_session(客户端 /api/v1/coupon/session 两段上报)。
-"""
-from __future__ import annotations
-
-from datetime import date as _date
-from typing import Annotated
-
-from fastapi import APIRouter, Depends, HTTPException, Query
-
-from app.admin.deps import AdminDb, get_current_admin
-from app.admin.repositories import coupon_data
-from app.admin.schemas.coupon_data import (
-    CouponDataDaily,
-    CouponDataHourly,
-    CouponDataOut,
-    CouponDataRow,
-    CouponDataSummary,
-    CouponUserRecordsOut,
-)
-from app.core.rewards import cn_today
-
-router = APIRouter(
-    prefix="/admin/api/coupon-data",
-    tags=["admin-coupon-data"],
-    dependencies=[Depends(get_current_admin)],
-)
-
-# 区间最大跨度(天);超出拒绝,避免一次拉过多天拖垮接口(对齐广告收益报表)。
-_MAX_RANGE_DAYS = 92
-
-
-def _parse_day(value: str | None, *, field: str, default: _date) -> _date:
-    if value is None:
-        return default
-    try:
-        return _date.fromisoformat(value)
-    except ValueError as e:
-        raise HTTPException(status_code=422, detail=f"{field} 需为 YYYY-MM-DD") from e
-
-
-@router.get(
-    "",
-    response_model=CouponDataOut,
-    summary="领券数据看板(发起/完成数 + 耗时分位 + 按天趋势 + 逐条明细)",
-)
-def get_coupon_data(
-    db: AdminDb,
-    date_from: Annotated[str | None, Query(description="起始日 北京 YYYY-MM-DD,默认今天")] = None,
-    date_to: Annotated[str | None, Query(description="结束日 北京 YYYY-MM-DD,闭区间,默认=date_from")] = None,
-    user: Annotated[str | None, Query(description="用户手机号/昵称模糊搜;不传=全部")] = None,
-    app_env: Annotated[str, Query(description="prod(默认) / dev / all(全部环境)")] = "prod",
-    granularity: Annotated[
-        str, Query(description="day=按天 / hour=按小时(北京);区间>1 天建议 day")
-    ] = "day",
-    limit: Annotated[int, Query(ge=1, le=1000, description="每页条数(分页大小)")] = 500,
-    offset: Annotated[int, Query(ge=0, description="分页偏移(已跳过条数)=(页码-1)×每页条数")] = 0,
-    sort: Annotated[
-        str, Query(description="排序:time=发起时间倒序(默认) / elapsed=耗时倒序")
-    ] = "time",
-) -> CouponDataOut:
-    today = cn_today()
-    d_from = _parse_day(date_from, field="date_from", default=today)
-    d_to = _parse_day(date_to, field="date_to", default=d_from)
-    if d_to < d_from:
-        raise HTTPException(status_code=422, detail="date_to 不能早于 date_from")
-    if (d_to - d_from).days + 1 > _MAX_RANGE_DAYS:
-        raise HTTPException(status_code=422, detail=f"区间最长 {_MAX_RANGE_DAYS} 天")
-
-    # 报表默认只看 prod(对齐广告报表防串台口径);app_env=all 时不过滤、看全部环境。
-    env = None if app_env == "all" else app_env
-    result = coupon_data.coupon_data_report(
-        db, date_from=d_from.isoformat(), date_to=d_to.isoformat(),
-        user=user, app_env=env, granularity=granularity,
-        limit=limit, offset=offset, sort=sort,
-    )
-    return CouponDataOut(
-        date_from=d_from.isoformat(),
-        date_to=d_to.isoformat(),
-        summary=CouponDataSummary(**result["summary"]),
-        daily=[CouponDataDaily(**d) for d in result["daily"]],
-        hourly=[CouponDataHourly(**h) for h in result["hourly"]],
-        total=result["total"],
-        items=[CouponDataRow(**r) for r in result["items"]],
-    )
-
-
-@router.get(
-    "/user-records",
-    response_model=CouponUserRecordsOut,
-    summary="某用户全部领券记录(点手机号抽屉:领券次数 + 记录列表)",
-)
-def get_user_coupon_records(
-    db: AdminDb,
-    user_id: Annotated[int, Query(description="用户 id")],
-    limit: Annotated[int, Query(ge=1, le=500, description="最多返回条数")] = 100,
-    sort_by: Annotated[str, Query(description="兼容 UserRecordsDrawer 参数;固定按发起时间倒序")] = "created_at",
-    sort_order: Annotated[str, Query(description="兼容参数,忽略")] = "desc",
-) -> CouponUserRecordsOut:
-    result = coupon_data.coupon_user_records(db, user_id=user_id, limit=limit)
-    return CouponUserRecordsOut(
-        items=[CouponDataRow(**r) for r in result["items"]],
-        total=result["total"],
-    )

app/admin/schemas/ad_revenue.py

@@ -94,11 +94,6 @@ class AdRevenueRow(BaseModel):
     impressions: int = Field(..., description="本行展示条数:有展示=1 / 纯发奖=0(供日汇总、趋势图复用)")
     ecpm: str | None = Field(None, description="eCPM 原始值(分/千次);展示行取展示值,纯发奖行取发奖采用值")
     revenue_yuan: float = Field(..., description="本次展示预估收益(元)= eCPM元 ÷ 1000;纯发奖行=0")
-    row_revenue_yuan: float | None = Field(
-        None,
-        description="主表逐行展示用的预估收益(元):一次比价/领券聚合行=该次发奖广告 eCPM 折算之和;"
-        "其它行为空(前端回退取 revenue_yuan)。不进合计/趋势,避免与展示侧重复计",
-    )
     adn: str | None = Field(None, description="实际填充 ADN 子渠道(pangle/gdt…);纯发奖行为空")
     slot_id: str | None = Field(None, description="底层 mediation rit(非我们配置的广告位 ID);纯发奖行为空")
     # ── 发奖侧 ──
@@ -111,15 +106,6 @@ class AdRevenueRow(BaseModel):
         None,
         description="发奖复算明细(eCPM/因子1/份数/LT/因子2/应发/实发/一致);点行展开下钻用,纯展示为空",
     )
-    sub_rewards: list[AdRevenueRecord] = Field(
-        default_factory=list,
-        description="一次比价/领券聚合行的组内逐条发奖明细(同一整场 ad_session_id 的多条广告);"
-        "点行展开渲染多行。激励视频/纯展示行为空(单条看 reward_detail)",
-    )
-    sub_count: int = Field(
-        1,
-        description="本行聚合的发奖条数:一次比价/领券=该次广告条数(≥1);激励视频/纯展示=1",
-    )
 
 
 class AdRevenueReportOut(BaseModel):

app/admin/schemas/coupon_data.py

@@ -1,83 +0,0 @@
-"""admin「领券数据」看板 schemas:汇总卡 + 按天/小时趋势 + 逐条领券明细。
-
-数据源 coupon_session(一次领券一行)。耗时单位 ms(前端按需折秒);均值/分位只统计 completed。
-"""
-from __future__ import annotations
-
-from datetime import datetime
-
-from pydantic import BaseModel, Field
-
-
-class CouponDataSummary(BaseModel):
-    """汇总卡:发起/完成数 + 耗时均值与分位(P5/P50/P95/P99,基于 completed 的 elapsed_ms)。"""
-
-    started_count: int = Field(..., description="发起数(区间内所有领券 session)")
-    completed_count: int = Field(..., description="完成数(status=completed)")
-    avg_elapsed_ms: int | None = Field(None, description="平均耗时(ms,仅 completed;无数据为空)")
-    p5_ms: int | None = Field(None, description="耗时 5 分位(ms)")
-    p50_ms: int | None = Field(None, description="耗时 50 分位(ms,中位数)")
-    p95_ms: int | None = Field(None, description="耗时 95 分位(ms)")
-    p99_ms: int | None = Field(None, description="耗时 99 分位(ms)")
-
-
-class CouponDataDaily(BaseModel):
-    """按天趋势(全量,不受分页影响):柱=发起/完成数,线=平均耗时。"""
-
-    date: str = Field(..., description="北京时间 YYYY-MM-DD")
-    started_count: int
-    completed_count: int
-    avg_elapsed_ms: int | None = Field(None, description="当天平均耗时(ms,仅 completed)")
-
-
-class CouponDataHourly(BaseModel):
-    """按北京小时(0–23)趋势(单日 granularity=hour 时非空)。"""
-
-    hour: int = Field(..., description="北京时间小时 0–23")
-    started_count: int
-    completed_count: int
-    avg_elapsed_ms: int | None = None
-
-
-class CouponDataRow(BaseModel):
-    """一条领券明细(一次领券任务)。"""
-
-    id: int = Field(..., description="coupon_session 主键(抽屉 rowKey 用)")
-    trace_id: str
-    user_id: int | None = None
-    user_phone: str | None = Field(None, description="手机号(admin 展示;匿名领券/查不到为空)")
-    user_nickname: str | None = Field(None, description="昵称")
-    status: str = Field(..., description="started / completed / failed / abandoned")
-    platforms: list[str] | None = Field(None, description="发起勾选平台")
-    origin_package: str | None = Field(None, description="发起来源 App 包名;null=App 内(傻瓜比价首页)发起")
-    elapsed_ms: int | None = Field(None, description="全程耗时(ms)")
-    platform_elapsed: dict[str, int] | None = Field(
-        None, description="各平台耗时 {meituan-waimai/taobao-shanguang/jd-waimai: ms}"
-    )
-    device_model: str | None = None
-    rom: str | None = None
-    app_env: str | None = None
-    started_at: datetime = Field(..., description="发起时刻(明细「时间」列)")
-    claimed_count: int | None = None
-    trace_url: str | None = Field(None, description="pricebot 公网 trace 链接(仅 completed 有);admin 渲染可点链接,无则显示可复制 trace_id")
-
-
-class CouponDataOut(BaseModel):
-    """领券数据看板响应:汇总卡 + 趋势 + 明细分页。"""
-
-    date_from: str
-    date_to: str
-    summary: CouponDataSummary
-    daily: list[CouponDataDaily] = Field(default_factory=list, description="按天趋势(全量)")
-    hourly: list[CouponDataHourly] = Field(
-        default_factory=list, description="按小时趋势(单日 hour 粒度时非空)"
-    )
-    total: int = Field(..., description="明细总条数(全量,不受分页)")
-    items: list[CouponDataRow] = Field(..., description="逐条领券明细(当前页)")
-
-
-class CouponUserRecordsOut(BaseModel):
-    """某用户全部领券记录(点手机号抽屉用):total=该用户领券总次数,items=记录列表(UserRecordsDrawer 渲染)。"""
-
-    items: list[CouponDataRow]
-    total: int

app/admin/schemas/user.py

@@ -17,7 +17,6 @@ class AdminUserListItem(BaseModel):
     status: str
     debug_trace_enabled: bool = False
     wechat_openid: str | None = None
-    wechat_nickname: str | None = None
     created_at: datetime
     last_login_at: datetime
 

app/api/v1/coupon.py

@@ -29,7 +29,6 @@ from app.schemas.coupon_state import (
     CouponPromptDismissIn,
     CouponPromptShouldShowOut,
     CouponPromptShownIn,
-    CouponSessionIn,
     CouponStatsOut,
 )
 
@@ -195,34 +194,6 @@ async def coupon_step(
     return resp_json
 
 
-@router.post("/session", summary="领券任务流水上报(admin 领券数据看板数据源)")
-def coupon_session(payload: CouponSessionIn, db: DbSession) -> dict[str, bool]:
-    """客户端两段上报一次领券流水(发起 started / 收尾 completed-failed-abandoned),按 trace_id upsert
-    到 coupon_session。不鉴权(同领券循环 MVP,按 device_id/trace_id);供 admin「领券数据」看板算
-    发起/完成数、耗时分位、机型维度。写库失败不应连累客户端(本就 fire-and-forget),吞掉返回 ok。"""
-    try:
-        coupon_repo.upsert_coupon_session(
-            db,
-            trace_id=payload.trace_id,
-            device_id=payload.device_id,
-            status=payload.status,
-            started_at_ms=payload.started_at_ms,
-            user_id=payload.user_id,
-            platforms=payload.platforms,
-            origin_package=payload.origin_package,
-            device_model=payload.device_model,
-            rom=payload.rom,
-            app_env=payload.app_env,
-            elapsed_ms=payload.elapsed_ms,
-            platform_elapsed=payload.platform_elapsed,
-            claimed_count=payload.claimed_count,
-            trace_url=payload.trace_url,
-        )
-    except Exception as e:  # noqa: BLE001
-        logger.warning("coupon session write failed: %s", e)
-    return {"ok": True}
-
-
 @router.post("/prompt/shown", summary="领券引导窗弹出即上报(按 App 记 shown)")
 def coupon_prompt_shown(payload: CouponPromptShownIn, db: DbSession) -> dict[str, bool]:
     """客户端弹出引导窗那刻调 → 记一条今日 engagement(shown),今天**这个 App** 不再自动弹。

app/models/__init__.py

@@ -19,7 +19,6 @@ from app.models.coupon_state import (  # noqa: F401
     CouponClaimRecord,
     CouponDailyCompletion,
     CouponPromptEngagement,
-    CouponSession,
 )
 from app.models.feedback import Feedback  # noqa: F401
 from app.models.invite import InviteRelation  # noqa: F401

app/models/coupon_state.py

@@ -181,79 +181,3 @@ class CouponPromptEngagement(Base):
             f"<CouponPromptEngagement device={self.device_id} "
             f"date={self.engage_date} type={self.engage_type}>"
         )
-
-
-class CouponSession(Base):
-    """一次领券任务的全程流水(admin「领券数据」看板数据源,2026-06-30)。
-
-    与 coupon_claim_record(按券一天一条去重)、coupon_daily_completion(按设备一天一条)都不同:
-    本表**一次领券一条**(trace_id 唯一),记从发起(started)到收尾(completed/failed/abandoned)的
-    全程耗时 + 各平台耗时 + 机型/ROM。客户端 POST /api/v1/coupon/session 两段上报:发起建行、
-    收尾按 trace_id 更新同一行。发起即落库 → admin 可算「发起数」与中途流失(started 无终态=未完成)。
-
-    口径:elapsed_ms 由客户端全程计时(点发起→收尾)、权威;started_at/finished_at 为时刻留痕。
-    started_date = started_at 的 Asia/Shanghai 自然日,供 admin 按天聚合 / 日期筛选(索引)。
-    """
-
-    __tablename__ = "coupon_session"
-    __table_args__ = (
-        # 一次领券一行:trace_id 幂等 upsert(发起建、收尾更新同一行)。
-        UniqueConstraint("trace_id", name="uq_coupon_session_trace"),
-        # admin 主聚合/筛选:按上海自然日 + 环境(报表默认只看 prod,避免测试数据串台)。
-        Index("ix_coupon_session_date_env", "started_date", "app_env"),
-    )
-
-    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
-
-    # 一次领券唯一 id(客户端 UUID,全程贯穿),upsert 键。
-    trace_id: Mapped[str] = mapped_column(String(64), nullable=False)
-    device_id: Mapped[str] = mapped_column(String(64), nullable=False)
-    # 登录态才带(admin join 用户表出手机号/昵称);匿名领券为空。
-    user_id: Mapped[int | None] = mapped_column(Integer, index=True, nullable=True)
-
-    # started / completed / failed / abandoned。started 无终态 = 中途流失(未完成)。
-    status: Mapped[str] = mapped_column(String(16), nullable=False)
-    # prod / dev(客户端 BuildConfig.DEBUG)。admin 报表默认只看 prod(对齐广告报表防串台口径)。
-    app_env: Mapped[str | None] = mapped_column(String(16), index=True, nullable=True)
-
-    # 发起勾选平台 ["meituan-waimai", ...](空=全领)。
-    platforms: Mapped[list | None] = mapped_column(_JSON, nullable=True)
-    # 发起来源外卖 App 包名;null=App 内(傻瓜比价首页)发起,非空=从美团/淘宝/京东弹券发起。
-    # admin「发起平台」列据此区分(空→傻瓜比价,包名→对应平台)。
-    origin_package: Mapped[str | None] = mapped_column(String(64), nullable=True)
-    # 机型(Build.MANUFACTURER + MODEL)与 ROM(OemDetector,如 "ColorOS 14")。明细「机型/ROM」列 + 维度筛选。
-    device_model: Mapped[str | None] = mapped_column(String(128), nullable=True)
-    rom: Mapped[str | None] = mapped_column(String(64), nullable=True)
-
-    # 发起时刻(客户端墙钟):明细「时间」列、趋势 X 轴。
-    started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
-    # 发起的 Asia/Shanghai 自然日:按天聚合 / 日期范围筛选(索引)。
-    started_date: Mapped[date] = mapped_column(Date, nullable=False)
-    # 收尾时刻(服务端 now);未收尾(流失)为空。
-    finished_at: Mapped[datetime | None] = mapped_column(
-        DateTime(timezone=True), nullable=True
-    )
-
-    # 全程耗时(ms,客户端点发起→收尾):平均 / 分位都基于它(只统计 completed)。
-    elapsed_ms: Mapped[int | None] = mapped_column(Integer, nullable=True)
-    # 各平台领券耗时 {"meituan-waimai":3200,...}(ms)。明细美团/淘宝/京东耗时列。
-    platform_elapsed: Mapped[dict | None] = mapped_column(_JSON, nullable=True)
-    # 领到总张数(收尾帧带)。
-    claimed_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
-    # pricebot done 帧回传的公网调试链接(price.shaguabijia.com/traces/{dir});含落盘时分秒、拼不出,只能存
-    # (同 ComparisonRecord.trace_url)。admin「领券数据」明细据此渲染可点 trace 链接;未到 done(failed/abandoned)为空。
-    trace_url: Mapped[str | None] = mapped_column(String(512), nullable=True)
-
-    created_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), server_default=func.now(), nullable=False
-    )
-    updated_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), server_default=func.now(), onupdate=func.now(),
-        nullable=False,
-    )
-
-    def __repr__(self) -> str:  # pragma: no cover
-        return (
-            f"<CouponSession trace={self.trace_id} status={self.status} "
-            f"elapsed_ms={self.elapsed_ms}>"
-        )

app/repositories/coupon_state.py

@@ -6,7 +6,7 @@
 from __future__ import annotations
 
 import logging
-from datetime import date, datetime, timezone
+from datetime import date, datetime
 from zoneinfo import ZoneInfo
 
 from sqlalchemy import delete, func, select
@@ -17,7 +17,6 @@ from app.models.coupon_state import (
     CouponClaimRecord,
     CouponDailyCompletion,
     CouponPromptEngagement,
-    CouponSession,
 )
 
 logger = logging.getLogger("shagua.coupon_state")
@@ -233,91 +232,3 @@ def sum_claimed_count(db: Session, user_id: int) -> int:
         )
     ).scalar_one()
     return int(total or 0)
-
-
-# ===== 领券任务流水(coupon_session,admin「领券数据」看板数据源)=====
-
-def upsert_coupon_session(
-    db: Session,
-    *,
-    trace_id: str,
-    device_id: str,
-    status: str,
-    started_at_ms: int,
-    user_id: int | None = None,
-    platforms: list[str] | None = None,
-    origin_package: str | None = None,
-    device_model: str | None = None,
-    rom: str | None = None,
-    app_env: str | None = None,
-    elapsed_ms: int | None = None,
-    platform_elapsed: dict | None = None,
-    claimed_count: int | None = None,
-    trace_url: str | None = None,
-) -> None:
-    """一条领券流水按 trace_id 幂等 upsert(发起 started 建行、收尾终态更新同一行)。
-
-    - 乱序/重复兜底:终态(completed/failed/abandoned)先到也建行;started 重复到不覆盖已有终态
-      (状态只前进,不降级)。
-    - started_at 由客户端墙钟毫秒转;started_date 取其 Asia/Shanghai 自然日(admin 按天聚合/筛选)。
-    - 终态帧补 finished_at=服务端 now;各字段非空才写(避免 started 帧的 None 抹掉收尾值,反之亦然)。
-    并发 IntegrityError 回滚忽略(本就幂等)。
-    """
-    started_at = datetime.fromtimestamp(started_at_ms / 1000, tz=timezone.utc)
-    started_date = started_at.astimezone(_CN_TZ).date()
-    is_terminal = status in ("completed", "failed", "abandoned")
-
-    row = db.execute(
-        select(CouponSession).where(CouponSession.trace_id == trace_id)
-    ).scalar_one_or_none()
-
-    if row is None:
-        db.add(CouponSession(
-            trace_id=trace_id,
-            device_id=device_id,
-            user_id=user_id,
-            status=status,
-            app_env=app_env,
-            platforms=platforms,
-            origin_package=origin_package,
-            device_model=device_model,
-            rom=rom,
-            started_at=started_at,
-            started_date=started_date,
-            finished_at=datetime.now(timezone.utc) if is_terminal else None,
-            elapsed_ms=elapsed_ms,
-            platform_elapsed=platform_elapsed,
-            claimed_count=claimed_count,
-            trace_url=trace_url,
-        ))
-    else:
-        # 状态只前进:started 帧重复到(如 START_STICKY 重启)不把已有终态降级回 started。
-        if not (status == "started" and row.status in ("completed", "failed", "abandoned")):
-            row.status = status
-        if is_terminal:
-            row.finished_at = datetime.now(timezone.utc)
-        if user_id is not None:
-            row.user_id = user_id
-        if platforms is not None:
-            row.platforms = platforms
-        if origin_package is not None:
-            row.origin_package = origin_package
-        if device_model is not None:
-            row.device_model = device_model
-        if rom is not None:
-            row.rom = rom
-        if app_env is not None:
-            row.app_env = app_env
-        if elapsed_ms is not None:
-            row.elapsed_ms = elapsed_ms
-        if platform_elapsed is not None:
-            row.platform_elapsed = platform_elapsed
-        if claimed_count is not None:
-            row.claimed_count = claimed_count
-        if trace_url is not None:
-            row.trace_url = trace_url
-    try:
-        db.commit()
-    except IntegrityError:
-        # 并发下另一请求刚插了同 trace_id → 唯一约束撞,回滚忽略(本就幂等)。
-        db.rollback()

app/schemas/coupon_state.py

@@ -49,28 +49,3 @@ class CouponStatsOut(BaseModel):
     """
 
     coupon_count: int
-
-
-class CouponSessionIn(BaseModel):
-    """客户端领券流水上报体(admin「领券数据」看板数据源,POST /api/v1/coupon/session)。
-
-    一次领券两段上报,按 trace_id upsert 到 coupon_session:
-    - 发起(status=started):带勾选平台 + 机型/ROM/app_env + started_at_ms(发起墙钟毫秒)。
-    - 收尾(completed/failed/abandoned):带 elapsed_ms(全程耗时)+ platform_elapsed(各平台耗时)+ claimed_count。
-    不鉴权(同领券循环 MVP,按 device_id/trace_id),user_id 登录态带上做留痕(可空)。
-    """
-
-    trace_id: str
-    device_id: str
-    status: str  # started / completed / failed / abandoned
-    started_at_ms: int  # 发起墙钟毫秒(客户端 System.currentTimeMillis)
-    user_id: int | None = None
-    platforms: list[str] | None = None
-    origin_package: str | None = None
-    device_model: str | None = None
-    rom: str | None = None
-    app_env: str | None = None
-    elapsed_ms: int | None = None
-    platform_elapsed: dict[str, int] | None = None
-    claimed_count: int | None = None
-    trace_url: str | None = None