feat(admin): 广告收益报表改为逐条广告事件 + 用户列显示手机号

报表主表从「按 用户×类型×应用×代码位 聚合」改成「逐条广告事件」(每次广告一行):
- 激励视频:展示(ad_ecpm)与发奖(ad_reward)按 ad_session_id 合并成一行,直接给出
  eCPM/收益 + 状态/应发/实发/一致;展开看该条金币复算因子
- 信息流:轮播每条展示各一行;整场发奖(client_event_id 与展示 impressionId 对不上)单独成行
- 纯展示行不计对账(matched 恒 true);有展示无发奖 / 有发奖无展示各自成行
- 每行补 user_phone(批量查 User.phone,完整不脱敏,与用户/钱包/比价记录页一致)
- 合计与对账在全量上统计、不受 limit 影响;event_key 作前端稳定 rowKey

ad_audit.audit_rows 顺带补返回 ad_session_id(供展示↔发奖按会话合并)。
真实库验证:逐条输出正确、合计交叉核对一致(展示条数=ecpm行数、实发=库实发)、schema 校验通过。

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

alembic/versions/meituan_coupon_image_meta.py

@@ -0,0 +1,34 @@
+"""meituan_coupon image size & type
+
+给 meituan_coupon 增加 image_size(字节)/ image_type(MIME)两列,采集时 HEAD 头图得到,
+供读取侧按图片大小决定是否拼缩放参数提速。两列均 nullable,不动存量行;下一轮全量 ETL
+upsert 自动回填,无需手动 backfill。
+
+Revision ID: meituan_coupon_image_meta
+Revises: feedback_review_fields
+Create Date: 2026-06-23 00:00:00.000000
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "meituan_coupon_image_meta"
+down_revision: str | Sequence[str] | None = "feedback_review_fields"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("meituan_coupon", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("image_size", sa.Integer(), nullable=True))
+        batch_op.add_column(sa.Column("image_type", sa.String(length=32), nullable=True))
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("meituan_coupon", schema=None) as batch_op:
+        batch_op.drop_column("image_type")
+        batch_op.drop_column("image_size")

app/admin/repositories/ad_audit.py

@@ -67,6 +67,7 @@ def _reward_video_rows(
                 "scene": "reward_video",
                 "record_id": rec.id,
                 "user_id": rec.user_id,
+                "ad_session_id": rec.ad_session_id,
                 "app_env": rec.app_env,
                 "our_code_id": rec.our_code_id,
                 "created_at": rec.created_at,
@@ -88,6 +89,7 @@ def _reward_video_rows(
                 "scene": "reward_video",
                 "record_id": rec.id,
                 "user_id": rec.user_id,
+                "ad_session_id": rec.ad_session_id,
                 "app_env": rec.app_env,
                 "our_code_id": rec.our_code_id,
                 "created_at": rec.created_at,
@@ -154,6 +156,7 @@ def _feed_rows(db: Session, *, date: str, user_id: int | None) -> list[dict]:
                 "scene": "feed",
                 "record_id": rec.id,
                 "user_id": rec.user_id,
+                "ad_session_id": rec.ad_session_id,
                 "app_env": rec.app_env,
                 "our_code_id": rec.our_code_id,
                 "created_at": rec.created_at,
@@ -174,6 +177,7 @@ def _feed_rows(db: Session, *, date: str, user_id: int | None) -> list[dict]:
                 "scene": "feed",
                 "record_id": rec.id,
                 "user_id": rec.user_id,
+                "ad_session_id": rec.ad_session_id,
                 "app_env": rec.app_env,
                 "our_code_id": rec.our_code_id,
                 "created_at": rec.created_at,

app/admin/repositories/ad_revenue.py

@@ -1,17 +1,18 @@
-"""admin 广告收益报表:按 用户 / 日期 / 广告类型 / 应用 / 代码位 聚合(单表含发奖对账)。
+"""admin 广告收益报表:**逐条广告事件**列表(每行一次广告,含展示 + 发奖对账)。
 
-只读。聚合键 = user_id × ad_type × app_env × our_code_id;每组一行同时给出:
-- 展示条数 + 收益:`ad_ecpm_record`(每行 = 客户端一次广告展示;收益 = Σ eCPM元 ÷ 1000)。
-  激励视频每次展示上报一行;信息流轮播每条展示各上报一行(每条独立 id,不复用会话)。
-- 应发金币 / 实发金币:复用金币审计的**逐条复算**(`ad_audit.audit_rows`,与正式发奖同一公式口径,
-  不另写公式),把每条发奖记录的 expected/actual 按同维度求和;`matched` = 组内**逐条**全部一致
-  (任一条不符该组即不符,不用「应发和==实发和」以免互相抵消掩盖错误)。**不改发奖逻辑**,只读复算。
+只读。每行 = 一次广告事件(不再按用户聚合):
+- **激励视频**:一次观看 = 1 条展示(ad_ecpm)+ 1 条发奖(ad_reward),按 ad_session_id 合并成一行,
+  直接给出 eCPM / 收益 + 状态 / 应发 / 实发 / 一致;点开看该条金币复算因子。
+- **信息流**:轮播每条展示各一行(impressionId 各自独立);整场发奖(ad_feed_reward,client_event_id)
+  与逐条展示无法对应,单独成「纯发奖」行。
+- 兜底:有展示无发奖(中途关 / 未达发奖)、有发奖无展示(未上报 eCPM)都各自成行。
 
-展示与发奖来自不同表,做并集:有展示无发奖(用户中途关 / 未达发奖)、有发奖无展示
-(未上报 eCPM)都各自成行。app_env/our_code_id 旧数据为 NULL → 归到「来源未知」组。
+展示与收益来自 ad_ecpm_record(收益 = eCPM元 ÷ 1000);应发 / 实发金币复用金币审计逐条复算
+(ad_audit.audit_rows,与正式发奖同一公式口径,不另写公式)。合计与对账在全量上统计,
+不受 limit(只截断 items)影响。
 
-⚠️ 局限:① 历史 Draw 发奖混在 ad_feed_reward_record 无类型标记,金币侧统一记 `feed`(迁移后 Draw
-不再产生新数据)。② 聚合级只能看出「某组应发≠实发」,定位到具体哪条仍需逐条审计接口(ad-coin-audit)。
+⚠️ 局限:① 历史 Draw 发奖混在 ad_feed_reward_record 无类型标记,金币侧统一记 feed。
+② 跨天 S2S 回调:同一次广告的展示与发奖偶尔落相邻日,各自按 report_date / reward_date 归日。
 """
 from __future__ import annotations
 
@@ -23,6 +24,7 @@ from sqlalchemy.orm import Session
 from app.admin.repositories import ad_audit
 from app.core import rewards
 from app.models.ad_ecpm import AdEcpmRecord
+from app.models.user import User
 
 
 def _cn_hour(dt: datetime) -> int:
@@ -32,17 +34,6 @@ def _cn_hour(dt: datetime) -> int:
     return dt.astimezone(rewards.CN_TZ).hour
 
 
-def _key(
-    report_date: str,
-    user_id: int,
-    ad_type: str,
-    app_env: str | None,
-    our_code_id: str | None,
-    hour: int | None,
-) -> tuple:
-    return (report_date, user_id, ad_type, app_env or None, our_code_id or None, hour)
-
-
 def _date_range(date_from: str, date_to: str) -> list[str]:
     """闭区间内逐日 'YYYY-MM-DD' 串(含首尾)。date_from > date_to 时返回空。"""
     d0 = _date.fromisoformat(date_from)
@@ -58,6 +49,18 @@ def _date_range(date_from: str, date_to: str) -> list[str]:
 # 审计行的 scene 与报表 ad_type 一一对应
 _SCENE_TO_AD_TYPE = {"reward_video": "reward_video", "feed": "feed"}
 
+# 发奖复算明细字段(展开下钻看「金币怎么算出来的」)——从 audit 行原样取这些 key。
+_REWARD_DETAIL_KEYS = (
+    "record_id", "created_at", "status", "ecpm", "ecpm_factor", "units",
+    "lt_index_start", "lt_index_end", "lt_factor_start", "lt_factor_end",
+    "expected_coin", "actual_coin", "matched",
+)
+
+
+def _reward_detail(row: dict) -> dict:
+    """从 audit 行抽出发奖复算明细(给前端展开行渲染因子1/因子2/份数/LT/应发实发)。"""
+    return {k: row[k] for k in _REWARD_DETAIL_KEYS}
+
 
 def ad_revenue_report(
     db: Session,
@@ -69,44 +72,42 @@ def ad_revenue_report(
     granularity: str = "day",
     limit: int = 500,
 ) -> dict:
-    """日期区间(北京时间,闭区间)广告收益聚合 + 发奖对账。单日时 date_from==date_to。
+    """日期区间(北京时间,闭区间)**逐条广告事件**列表 + 发奖对账。单日时 date_from==date_to。
 
-    聚合键含**日期**:report_date × user × ad_type × app_env × our_code_id(× 北京小时,granularity=hour)。
-    ad_type: None=全部 / reward_video / feed / draw。
-    granularity: "day"=按天 / "hour"=按小时(聚合键再加北京小时 0–23,每组一行)。
-    limit 只截断展示明细,total 与 total_* / daily 在全量上统计(不受 limit 影响),数字始终可信。
-
-    返回额外含 `daily`(按日期汇总的展示/收益/应发/实发,供前端按天趋势图;不受 limit 影响)。
-
-    注:按小时下,展示按 ecpm 记录的小时、金币按发奖记录的小时各自归桶——S2S 回调可能比展示晚
-    一会儿,故同一次广告的展示与金币偶尔落相邻小时(按天则一致)。
+    每个 item = 一次广告事件(展示与发奖按 ad_session_id 合并;信息流展示 / 发奖各自成行)。
+    ad_type: None=全部 / reward_video / feed / draw。granularity=hour 时每行带北京小时(由各自时间算)。
+    limit 只截断 items(事件明细),total 与 total_* / daily 在全量上统计,数字始终可信。
     """
     by_hour = granularity == "hour"
-    groups: dict[tuple, dict] = {}
 
-    def _grp(key: tuple) -> dict:
-        g = groups.get(key)
-        if g is None:
-            rdate, uid, atype, app_env, code_id, hour = key
-            g = {
-                "report_date": rdate,
-                "user_id": uid,
-                "ad_type": atype,
-                "app_env": app_env,
-                "our_code_id": code_id,
-                "hour": hour,
-                "impressions": 0,
-                "revenue_yuan": 0.0,
-                "expected_coin": 0,
-                "actual_coin": 0,
-                "adns": set(),
-                "impression_records": [],  # 该组逐条展示明细(展开下钻用)
-                "records": [],  # 该组逐条发奖复算明细(展开下钻用)
-            }
-            groups[key] = g
-        return g
+    # 1) 发奖行(逐日 audit 复算):建 (user_id, ad_session_id) → [行] 映射用于和展示合并;
+    #    同时保留全量列表,未被展示合并的成「纯发奖」事件。
+    reward_by_session: dict[tuple[int, str], list[dict]] = {}
+    all_reward_rows: list[dict] = []
+    audit_scene = _SCENE_TO_AD_TYPE.get(ad_type) if ad_type is not None else 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):
+                row["_report_date"] = d
+                all_reward_rows.append(row)
+                sid = row.get("ad_session_id")
+                if sid:
+                    reward_by_session.setdefault((row["user_id"], sid), []).append(row)
 
-    # 1) 展示条数 + 收益 ← ad_ecpm_record(report_date 闭区间;字符串 YYYY-MM-DD 字典序即日期序)
+    used_reward_ids: set[int] = set()
+    events: list[dict] = []
+
+    def _pop_reward(uid: int, sid: str | None) -> dict | None:
+        """取一条与 (uid, sid) 匹配且未被用过的发奖行(激励视频展示↔发奖按会话 1:1 合并)。"""
+        if not sid:
+            return None
+        for r in reward_by_session.get((uid, sid), ()):
+            if r["record_id"] not in used_reward_ids:
+                used_reward_ids.add(r["record_id"])
+                return r
+        return None
+
+    # 2) 展示记录(ad_ecpm):每条一个事件;能匹配到发奖则合并成「展示 + 发奖」一行。
     stmt = select(AdEcpmRecord).where(
         AdEcpmRecord.report_date >= date_from,
         AdEcpmRecord.report_date <= date_to,
@@ -116,124 +117,116 @@ def ad_revenue_report(
     if ad_type is not None:
         stmt = stmt.where(AdEcpmRecord.ad_type == ad_type)
     for rec in db.execute(stmt).scalars():
-        hour = _cn_hour(rec.created_at) if by_hour else None
-        g = _grp(_key(rec.report_date, rec.user_id, rec.ad_type, rec.app_env, rec.our_code_id, hour))
-        g["impressions"] += 1
-        # 单次展示收益(元) = eCPM元 ÷ 1000(每千次→单次);用与发奖同源的解析,口径一致。
-        rev = rewards.parse_ecpm_yuan(rec.ecpm_raw) / 1000.0
-        g["revenue_yuan"] += rev
-        if rec.adn:
-            g["adns"].add(rec.adn)
-        g["impression_records"].append({
-            "id": rec.id,
+        rwd = _pop_reward(rec.user_id, rec.ad_session_id)
+        ev = {
+            "event_key": f"imp-{rec.id}",
+            "report_date": rec.report_date,
+            "user_id": rec.user_id,
+            "ad_type": rec.ad_type,
+            "app_env": rec.app_env,
+            "our_code_id": rec.our_code_id,
             "created_at": rec.created_at,
+            "hour": _cn_hour(rec.created_at) if by_hour else None,
+            "has_impression": True,
+            "impressions": 1,
             "ecpm": rec.ecpm_raw,
-            "revenue_yuan": round(rev, 6),
+            # 单次展示收益(元)= eCPM元 ÷ 1000(每千次→单次);与发奖同源解析,口径一致。
+            "revenue_yuan": round(rewards.parse_ecpm_yuan(rec.ecpm_raw) / 1000.0, 6),
             "adn": rec.adn,
             "slot_id": rec.slot_id,
+        }
+        if rwd is not None:
+            ev.update({
+                "has_reward": True,
+                "status": rwd["status"],
+                "expected_coin": int(rwd["expected_coin"]),
+                "actual_coin": int(rwd["actual_coin"]),
+                "matched": bool(rwd["matched"]),
+                "reward_detail": _reward_detail(rwd),
+            })
+        else:
+            # 纯展示(信息流逐条展示、激励视频缺发奖记录):不计对账,matched=True。
+            ev.update({
+                "has_reward": False, "status": None,
+                "expected_coin": 0, "actual_coin": 0, "matched": True,
+                "reward_detail": None,
+            })
+        events.append(ev)
+
+    # 3) 未被展示合并的发奖行 → 「纯发奖」事件(信息流整场发奖 / 有发奖无展示)。
+    #    收益恒 0(收益只算展示侧,避免与展示行重复计)。
+    for row in all_reward_rows:
+        if row["record_id"] in used_reward_ids:
+            continue
+        events.append({
+            "event_key": f"rwd-{row['record_id']}",
+            "report_date": row["_report_date"],
+            "user_id": row["user_id"],
+            "ad_type": _SCENE_TO_AD_TYPE.get(row["scene"], row["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),
         })
 
-    # 2) 应发 / 实发金币 ← 复用金币审计逐条复算(同一公式口径),按同维度求和。
-    #    audit_rows 是单日的,区间逐日调用,每天的行归到当天 report_date(语义与单日报表完全一致)。
-    #    ad_type=draw 时审计无对应记录(scene 只有 reward_video/feed),金币侧自然为空。
-    audit_scene = _SCENE_TO_AD_TYPE.get(ad_type) if ad_type is not None else 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):
-                atype = _SCENE_TO_AD_TYPE.get(row["scene"], row["scene"])
-                hour = _cn_hour(row["created_at"]) if by_hour else None
-                g = _grp(_key(d, row["user_id"], atype, row.get("app_env"), row.get("our_code_id"), hour))
-                g["expected_coin"] += int(row["expected_coin"])
-                g["actual_coin"] += int(row["actual_coin"])
-                # 逐条明细(eCPM/因子1/份数/LT/因子2/应发/实发/一致)——前端展开该组时下钻展示。
-                g["records"].append({
-                    "record_id": row["record_id"],
-                    "created_at": row["created_at"],
-                    "status": row["status"],
-                    "ecpm": row["ecpm"],
-                    "ecpm_factor": row["ecpm_factor"],
-                    "units": row["units"],
-                    "lt_index_start": row["lt_index_start"],
-                    "lt_index_end": row["lt_index_end"],
-                    "lt_factor_start": row["lt_factor_start"],
-                    "lt_factor_end": row["lt_factor_end"],
-                    "expected_coin": row["expected_coin"],
-                    "actual_coin": row["actual_coin"],
-                    "matched": row["matched"],
-                })
+    events.sort(key=lambda e: (e["report_date"], e["user_id"], e["created_at"]))
 
-    rows = list(groups.values())
-    rows.sort(
-        key=lambda r: (
-            r["report_date"],
-            r["user_id"],
-            r["hour"] if r["hour"] is not None else -1,
-            r["ad_type"] or "",
-            r["our_code_id"] or "",
-        )
-    )
+    # 补手机号(admin 展示用,完整不脱敏,与用户 / 钱包 / 比价记录页一致):批量一次查,避免 N+1。
+    uids = {e["user_id"] for e in events}
+    phone_map: dict[int, str] = {}
+    if uids:
+        phone_map = {
+            uid: phone
+            for uid, phone in db.execute(
+                select(User.id, User.phone).where(User.id.in_(uids))
+            ).all()
+        }
+    for e in events:
+        e["user_phone"] = phone_map.get(e["user_id"])
 
-    total_impressions = sum(r["impressions"] for r in rows)
-    total_expected_coin = sum(r["expected_coin"] for r in rows)
-    total_actual_coin = sum(r["actual_coin"] for r in rows)
-    total_revenue_yuan = round(sum(r["revenue_yuan"] for r in rows), 6)
+    total_impressions = sum(e["impressions"] for e in events)
+    total_revenue_yuan = round(sum(e["revenue_yuan"] for e in events), 6)
+    total_expected_coin = sum(e["expected_coin"] for e in events)
+    total_actual_coin = sum(e["actual_coin"] for e in events)
+    mismatch_count = sum(1 for e in events if e["has_reward"] and not e["matched"])
 
     # 按日期汇总(全量,不受 limit):供前端按天趋势图。
     daily_map: dict[str, dict] = {}
-    for r in rows:
-        d = daily_map.get(r["report_date"])
+    for e in events:
+        d = daily_map.get(e["report_date"])
         if d is None:
-            d = {
-                "date": r["report_date"],
-                "impressions": 0,
-                "revenue_yuan": 0.0,
-                "expected_coin": 0,
-                "actual_coin": 0,
-            }
-            daily_map[r["report_date"]] = d
-        d["impressions"] += r["impressions"]
-        d["revenue_yuan"] += r["revenue_yuan"]
-        d["expected_coin"] += r["expected_coin"]
-        d["actual_coin"] += r["actual_coin"]
+            d = {"date": e["report_date"], "impressions": 0, "revenue_yuan": 0.0,
+                 "expected_coin": 0, "actual_coin": 0}
+            daily_map[e["report_date"]] = d
+        d["impressions"] += e["impressions"]
+        d["revenue_yuan"] += e["revenue_yuan"]
+        d["expected_coin"] += e["expected_coin"]
+        d["actual_coin"] += e["actual_coin"]
     daily = [
         {**d, "revenue_yuan": round(d["revenue_yuan"], 6)}
         for d in sorted(daily_map.values(), key=lambda x: x["date"])
     ]
 
-    items = [
-        {
-            "report_date": r["report_date"],
-            "user_id": r["user_id"],
-            "ad_type": r["ad_type"],
-            "app_env": r["app_env"],
-            "our_code_id": r["our_code_id"],
-            "hour": r["hour"],
-            "impressions": r["impressions"],
-            "revenue_yuan": round(r["revenue_yuan"], 6),
-            "expected_coin": r["expected_coin"],
-            "actual_coin": r["actual_coin"],
-            # 组内**逐条**全部一致才记一致——不能用「应发和==实发和」,否则一条多发+一条少发会互相
-            # 抵消、求和相等被误判为 ✓,掩盖真实发奖错误。纯展示无发奖记录的组 all([]) → True。
-            "matched": all(rec["matched"] for rec in r["records"]),
-            "adns": sorted(r["adns"]),
-            "impression_records": sorted(
-                r["impression_records"], key=lambda x: (x["created_at"], x["id"])
-            ),
-            "records": sorted(r["records"], key=lambda x: (x["created_at"], x["record_id"])),
-        }
-        for r in rows[:limit]
-    ]
-
     return {
-        "total": len(rows),
-        "truncated": len(rows) > limit,
+        "total": len(events),
+        "truncated": len(events) > limit,
         "total_impressions": total_impressions,
         "total_revenue_yuan": total_revenue_yuan,
         "total_expected_coin": total_expected_coin,
         "total_actual_coin": total_actual_coin,
-        "mismatch_count": sum(
-            1 for r in rows if not all(rec["matched"] for rec in r["records"])
-        ),
+        "mismatch_count": mismatch_count,
         "daily": daily,
-        "items": items,
+        "items": events[:limit],
     }

app/admin/schemas/ad_revenue.py

@@ -1,7 +1,7 @@
 """广告收益报表 schemas。
 
-按 用户 / 日期 / 广告类型 / 应用 / 代码位 聚合的只读报表:展示条数、收益(元)、金币、来源。
-字段 snake_case;收益按元(float),金币按整数。
+**逐条广告事件**只读报表:每行一次广告(激励视频展示+发奖按会话合并;信息流展示/发奖各自成行),
+含 展示条数、收益(元)、应发/实发金币、对账。字段 snake_case;收益按元(float),金币按整数。
 """
 from __future__ import annotations
 
@@ -50,27 +50,33 @@ class AdRevenueDaily(BaseModel):
 
 
 class AdRevenueRow(BaseModel):
-    """一个聚合组(report_date × user × ad_type × app_env × our_code_id)的汇总。"""
+    """一次广告事件(逐条一行):激励视频展示与发奖按 ad_session_id 合并;信息流展示 / 发奖各自成行。"""
 
-    report_date: str = Field(..., description="该组所属日期(北京时间 YYYY-MM-DD)")
+    event_key: str = Field(..., description="该事件稳定唯一键(imp-{ecpm_id} / rwd-{reward_id});前端 rowKey 用")
+    report_date: str = Field(..., description="该事件所属日期(北京时间 YYYY-MM-DD)")
     user_id: int
+    user_phone: str | None = Field(None, description="用户手机号(admin 展示用,完整;用户已删 / 查不到为空)")
     ad_type: str = Field(..., description="reward_video(激励视频) / feed(信息流) / draw(历史 Draw 信息流)")
     app_env: str | None = Field(None, description="我们的应用:prod(傻瓜比价正式) / test(测试应用);旧数据为空")
     our_code_id: str | None = Field(None, description="我们后台配置的代码位 ID(104xxx);旧数据为空")
     hour: int | None = Field(None, description="北京时间小时 0–23(granularity=hour 时有值;按天为 null)")
-    impressions: int = Field(..., description="展示条数(每条广告展示一条;轮播每条各计一次)")
-    revenue_yuan: float = Field(..., description="收益(元)= Σ(eCPM元 ÷ 1000);测试应用多为 0")
-    expected_coin: int = Field(..., description="应发金币(按公式复算,与金币审计同源)")
-    actual_coin: int = Field(..., description="实发金币(实际入账,按现发奖算法)")
-    matched: bool = Field(..., description="该组应发==实发(组内任一条不符则 false)")
-    adns: list[str] = Field(default_factory=list, description="实际填充的底层 ADN 子渠道集合(如 pangle/gdt)")
-    impression_records: list[AdRevenueImpression] = Field(
-        default_factory=list,
-        description="该组逐条展示明细(时间/eCPM/收益/adn);展开下钻用,无发奖也有(只要有展示)",
-    )
-    records: list[AdRevenueRecord] = Field(
-        default_factory=list,
-        description="该组逐条发奖复算明细(eCPM/因子1/份数/LT/因子2/应发/实发/一致);展开下钻用,纯展示无发奖记录的组为空",
+    created_at: datetime = Field(..., description="事件时间(有展示=展示时间,纯发奖=发奖时间)")
+    # ── 展示侧 ──
+    has_impression: bool = Field(..., description="是否有广告展示(信息流逐条展示=True,纯发奖行=False)")
+    impressions: int = Field(..., description="本行展示条数:有展示=1 / 纯发奖=0(供日汇总、趋势图复用)")
+    ecpm: str | None = Field(None, description="eCPM 原始值(分/千次);展示行取展示值,纯发奖行取发奖采用值")
+    revenue_yuan: float = Field(..., description="本次展示预估收益(元)= eCPM元 ÷ 1000;纯发奖行=0")
+    adn: str | None = Field(None, description="实际填充 ADN 子渠道(pangle/gdt…);纯发奖行为空")
+    slot_id: str | None = Field(None, description="底层 mediation rit(非我们配置的广告位 ID);纯发奖行为空")
+    # ── 发奖侧 ──
+    has_reward: bool = Field(..., description="是否有发奖记录(激励视频合并行 / 信息流整场发奖行=True;纯展示=False)")
+    status: str | None = Field(None, description="发奖状态 granted/closed_early/too_short/…;纯展示为空")
+    expected_coin: int = Field(..., description="应发金币(公式复算,与金币审计同源);纯展示=0")
+    actual_coin: int = Field(..., description="实发金币(实际入账);纯展示=0")
+    matched: bool = Field(..., description="本条应发==实发;纯展示恒 True(不计对账)")
+    reward_detail: AdRevenueRecord | None = Field(
+        None,
+        description="发奖复算明细(eCPM/因子1/份数/LT/因子2/应发/实发/一致);点行展开下钻用,纯展示为空",
     )
 
 
@@ -80,11 +86,11 @@ class AdRevenueReportOut(BaseModel):
     date_from: str = Field(..., description="报表起始日期(北京时间 YYYY-MM-DD)")
     date_to: str = Field(..., description="报表结束日期(北京时间 YYYY-MM-DD,闭区间;单日时与 date_from 相同)")
     daily: list[AdRevenueDaily] = Field(..., description="按日期汇总序列(全量,供按天趋势图)")
-    total: int = Field(..., description="聚合组总数(全量,不受 limit 影响)")
+    total: int = Field(..., description="广告事件总数(全量,不受 limit 影响)")
     truncated: bool = Field(..., description="明细是否被 limit 截断")
     total_impressions: int = Field(..., description="全量展示条数合计")
     total_revenue_yuan: float = Field(..., description="全量收益合计(元)")
     total_expected_coin: int = Field(..., description="全量应发金币合计")
     total_actual_coin: int = Field(..., description="全量实发金币合计")
-    mismatch_count: int = Field(..., description="应发≠实发的组数(=0 说明全部按公式发放)")
-    items: list[AdRevenueRow] = Field(..., description="聚合明细(按 用户→类型→代码位 排序)")
+    mismatch_count: int = Field(..., description="应发≠实发的发奖条数(=0 说明全部按公式发放)")
+    items: list[AdRevenueRow] = Field(..., description="逐条广告事件(按 日期→用户→时间 排序)")

app/models/meituan_coupon.py

@@ -55,6 +55,10 @@ class MeituanCoupon(Base):
     sell_price_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
     original_price_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
     head_url: Mapped[str | None] = mapped_column(String(512), nullable=True)
+    # 头图字节大小 / MIME 类型:采集时 HEAD head_url 得到(Content-Length / Content-Type)。
+    # 读取侧已统一压缩(不再按大小判定),这两列保留用于分析/监控(GIF 占比、体积分布等)。
+    image_size: Mapped[int | None] = mapped_column(Integer, nullable=True)         # 字节
+    image_type: Mapped[str | None] = mapped_column(String(32), nullable=True)      # 如 image/jpeg
 
     # ===== 销量(美团只给粗档位:热销1w+;num=排序用的下界数值,如 1w+ → 10000) =====
     sale_volume: Mapped[str | None] = mapped_column(String(32), nullable=True)

app/repositories/ops_marquee.py

@@ -6,13 +6,13 @@ feed 取最近的成功且省到钱的比价记录(脱敏用户名);不足 N 条
 种子是「生成规则」:用户名可空(空→随机合成脱敏名,避开同屏撞名)、金额是区间(每次随机取值)、
 feed 公平随机抽取(不看 sort_order),所有启用种子都有机会露出。
 
-脱敏名规则(全 feed 统一,按字符算、中英文皆适用):
-  - 有昵称 → 昵称脱敏:≥3 字「首+隐藏字数个星+末」(省钱小能手→省***手 / SaveKing→S******g),
-    2 字「首+星」(阿强→阿*),1 字「用户+该字」(喵→用户喵);
-  - 没昵称 → 合成一个假名(见下)再脱敏,少数露「用户+id 后 3 位」(用户********618)。
-真实条:设过昵称的按真实昵称脱敏;当前多数用户没设昵称,直接展示「用户****」会清一色,故给它们合成
-假名——后期真实昵称多了,真实条会直接用真实昵称,合成占比自然下降。种子 / 兜底同样合成、避开同屏撞名;
-种子留空 / 旧「用户****xxx」模板名一律重新合成(自愈历史种子,无需迁移)。
+脱敏名规则(对齐 PRD「用户标识打码规则」,按字符数算、中英文皆适用):
+  - 有昵称 → 昵称脱敏:n≥5「首+***+末」(AAA省钱小能手→A***手 / Micky_ 毛→M***毛),
+    n=4「首+**+末」(你好不好→你**好),n≤3「首+**」(小确呀→小** / 阿强→阿** / 喵→喵**);
+  - 没昵称(真实用户)→「用户」+5星+id 后 2 位(用户*****08),按 user_id 稳定。
+真实条按上述 PRD 规则脱敏。种子 / 兜底无真实用户(无昵称/无 id,套不上「用户*****+id后2位」),
+仍本地合成多样假名再走同一套昵称脱敏(避开同屏撞名),让横条不至清一色;种子留空 / 旧「用户****xxx」
+模板名一律重新合成(自愈历史种子,无需迁移)。
 假名合成:纯本地语料**组合生成**(中文姓池×名字池 拼真名 / 中文网络昵称 / 英文昵称 / 英文名带数字尾),
 组合空间上万、中文为主英文为辅,脱敏后像真实异质用户群。**真实条按 `user_id` 播种确定性合成**(同一用户
 每次展示恒定、不同用户各异),刷新/翻页不变脸;种子 / 兜底用运行时随机源出多样填充。不依赖外部库。
@@ -44,12 +44,12 @@ _REAL_ROWS_TTL_SECONDS = 30
 _REAL_ROWS_FETCH_CAP = 600  # 一次多取些,够 limit≤30 去重后取数;命中缓存后复用
 _real_rows_cache: dict = {"at": None, "rows": None}
 
-# ===== 用户标识脱敏 + 没真实昵称时的假名合成 =====
-# 脱敏规则(中英文皆适用,按字符算):有昵称→≥3 字「首+隐藏字数个星+末」(省钱小能手→省***手 /
-# SaveKing→S******g)、2 字「首+星」(阿强→阿*)、1 字「用户+该字」(喵→用户喵);没昵称→用户+8星+id 后 3 位。
-# 没真实昵称时(当前多数用户 / 种子 / 兜底)合成假名:本地姓/名字池组合出真名 + 网络昵称语料(中/英)
-# 混播再走同一套脱敏,像真实异质用户群。后期真实昵称多了,真实条直接用真实昵称,这里占比自然下降。
-_ANON_STARS = "*" * 8  # 无昵称匿名串固定 8 星(用户********xxx)
+# ===== 用户标识脱敏(对齐 PRD) + 种子无真实昵称时的假名合成 =====
+# 脱敏规则(按字符数,中英文皆适用):有昵称→n≥5「首+***+末」、n=4「首+**+末」、n≤3「首+**」;
+# 没昵称(真实用户)→「用户」+5星+id 后 2 位(用户*****08)。
+# 种子/兜底没有真实用户(无昵称/无 id),本地用姓/名字池组合真名 + 网络昵称语料(中/英)合成假名,
+# 再走同一套昵称脱敏,让横条像真实异质用户群、不清一色(PRD 只约束真实用户,种子是运营填充)。
+_ANON_STARS = "*" * 5  # 「用户」匿名串固定 5 星(用户*****xx,对齐 PRD)
 
 # 中文真名组合池:姓(百家姓常见 ~100)× 名字单字(~120)拼「姓 + 1~2 字」,组合空间上万。
 # 脱敏后只露「姓*」或「姓*末」,故无需真实姓名库——组合够多即看着各异(替代原 Faker zh_CN)。
@@ -86,19 +86,23 @@ _NICK_EN = [
 
 
 def _mask_nickname(nick: str) -> str:
-    """昵称脱敏(按字符,中英文皆可):≥3 字→首+(隐藏字数个)星+末(省钱小能手→省***手 / SaveKing→S******g);
-    2 字→首+星(阿强→阿*);1 字→「用户」+该字(喵→用户喵)。调用方需保证 nick 已 strip 且非空。"""
+    """昵称脱敏(按字符数,中英文皆可,对齐 PRD「用户标识打码规则」):
+    n≥5 → 首+***+末(AAA省钱小能手→A***手 / Micky_ 毛→M***毛);
+    n=4 → 首+**+末(你好不好→你**好 / Lady→L**y);
+    n≤3 → 首+**(小确呀→小** / 阿强→阿** / 喵→喵** / Lay→L**)。
+    调用方需保证 nick 已 strip 且非空。"""
     n = len(nick)
-    if n >= 3:
-        return nick[0] + "*" * (n - 2) + nick[-1]
-    if n == 2:
-        return nick[0] + "*"
-    return "用户" + nick
+    if n >= 5:
+        return nick[0] + "***" + nick[-1]
+    if n == 4:
+        return nick[0] + "**" + nick[-1]
+    return nick[0] + "**"
 
 
-def _anon_by_id(user_id: int) -> str:
-    """没昵称匿名:固定「用户」+ 8 星 + id 后 3 位(不足 3 位前补 0):用户********618。"""
-    return "用户" + _ANON_STARS + f"{user_id % 1000:03d}"
+def _mask_anon(user_id: int) -> str:
+    """没昵称(真实用户,对齐 PRD):「用户」+ 5 星 + id 后 2 位(不足 2 位前补 0):用户*****08。
+    注:user_id 为整数,「id 后 2 位」取末两位数字(PRD 示例「用户*****x8」的 x 为占位符)。"""
+    return "用户" + _ANON_STARS + f"{user_id % 100:02d}"
 
 
 def _synth_full_name(rng: random.Random) -> str:
@@ -118,27 +122,24 @@ def _synth_full_name(rng: random.Random) -> str:
 
 
 def _synth_masked_name(rng: random.Random) -> str:
-    """合成一条「已脱敏」假名(供没真实昵称的真实用户 / 种子 / 兜底用)。用传入 rng 决定一切随机。"""
+    """合成一条「已脱敏」假名(供种子 / 兜底用,真实用户已改走 PRD 规则)。用传入 rng 决定一切随机。"""
     full = _synth_full_name(rng).strip()
-    return _mask_nickname(full) if full else "用户" + _ANON_STARS + f"{rng.randint(0, 999):03d}"
+    return _mask_nickname(full) if full else "用户" + _ANON_STARS + f"{rng.randint(0, 99):02d}"
 
 
 def _mask_real(nickname: str | None, user_id: int) -> str:
-    """真实用户脱敏:设过昵称→昵称脱敏(中英文皆可);没昵称→**按 user_id 播种确定性合成假名**再脱敏
-    (同一用户每次展示恒定、刷新不变脸,不同用户各异),避免无昵称用户清一色「用户****xxx」;
-    少数(~15%)露「用户+真实 id 后 3 位」。当前多数用户没昵称,后期真实昵称多了占比下降。"""
+    """真实用户脱敏(对齐 PRD「用户标识打码规则」):设过昵称→昵称脱敏(中英文皆可);
+    没昵称→「用户」+5星+id 后 2 位(用户*****08),按 user_id 稳定、刷新不变脸。"""
     nick = (nickname or "").strip()
     if nick:
         return _mask_nickname(nick)
-    # 关键:按 user_id 播种本地 rng → 同一用户的合成名跨请求/刷新恒定(0.15 抽签也用它)。
-    seeded = random.Random(user_id)
-    return _anon_by_id(user_id) if seeded.random() < 0.15 else _synth_masked_name(seeded)
+    return _mask_anon(user_id)
 
 
 def _synth_name(rng: random.Random) -> str:
-    """合成一条脱敏名(种子留空 / 旧模板名 / 兜底用):绝大多数=假名脱敏,少数=用户+随机 3 位(用户********618)。"""
+    """合成一条脱敏名(种子留空 / 旧模板名 / 兜底用):绝大多数=假名脱敏,少数=用户+随机 2 位(用户*****08)。"""
     if rng.random() < 0.15:
-        return "用户" + _ANON_STARS + f"{rng.randint(0, 999):03d}"
+        return "用户" + _ANON_STARS + f"{rng.randint(0, 99):02d}"
     return _synth_masked_name(rng)
 
 
@@ -148,7 +149,7 @@ def _unique_name(used: set[str]) -> str:
         n = _synth_name(_rng)
         if n not in used:
             return n
-    return "用户" + _ANON_STARS + f"{_rng.randint(0, 999):03d}"  # 兜底:数字尾号天然好去重
+    return "用户" + _ANON_STARS + f"{_rng.randint(0, 99):02d}"  # 兜底:数字尾号天然好去重
 
 
 def _validate_amount(min_cents: int, max_cents: int) -> None:

app/schemas/meituan.py

@@ -13,6 +13,21 @@ from pydantic import BaseModel, Field
 FeedStatus = Literal["ok", "empty", "degraded"]
 
 
+# ───────────────── 头图传输优化（统一缩放 + 转 WebP） ─────────────────
+# 美团图片 CDN 支持在 URL 后拼 @<w>w_<h>h_1e_1c[.webp] 让服务端缩放/转码。feed 卡片只占
+# 124dp(≈372px@3x),而库里 head_url 是原图(实测中位 137KB、长尾到 17MB、GIF 均 4.8MB),
+# 是首页 feed 图片加载偏慢的来源。按产品要求,头图**统一**缩到 124dp + 转 WebP 再下发
+# (全部压缩、不设大小阈值),由美团 CDN 服务端缩放/转码,实测省 76–99% 体积。
+FEED_THUMB_PARAM = "@375w_375h_1e_1c.webp"          # 124dp@3x 缩略 + 转 WebP
+
+
+def feed_image_url(head_url: str) -> str:
+    """统一给头图 URL 拼缩放参数(缩到 124dp + 转 WebP),让美团 CDN 出小图;空 URL 原样返回。"""
+    if not head_url:
+        return head_url
+    return head_url.split("@")[0] + FEED_THUMB_PARAM
+
+
 # ───────────────── 券卡片（归一化后给客户端） ─────────────────
 
 class CouponCard(BaseModel):
@@ -94,7 +109,7 @@ class CouponCard(BaseModel):
             platform=platform,
             biz_line=biz_line,
             name=cpd.get("name") or "",
-            head_image_url=head_url,
+            head_image_url=feed_image_url(head_url),
             brand_name=brand.get("brandName"),
             brand_logo_url=brand.get("brandLogoUrl"),
             sell_price=sell,

deploy/backfill_image_meta.sh

@@ -0,0 +1,48 @@
+#!/usr/bin/env bash
+# 美团券头图大小/类型回填 —— 全量重跑一轮 ETL,把 meituan_coupon.image_size / image_type 灌满。
+# 上线本 PR 后在服务器手动跑一次,立即回填存量券(不必等定时器逐轮 upsert 自然补齐)。
+#
+# 用法(服务器, root):
+#   cd /opt/shaguabijia-app-server
+#   sudo bash deploy/backfill_image_meta.sh
+#   # 全量一轮 ~70min,建议在 tmux/screen 里跑,防 SSH 断开中断
+#
+# 前提:已拉到含本 PR 的 main、已跑 `alembic upgrade head`(脚本会自检列是否存在)、
+#       已 restart shaguabijia-app-server.service。服务器 MT_CPS_PROXY 留空(直连 CDN)。
+set -euo pipefail
+
+APP_DIR="${APP_DIR:-/opt/shaguabijia-app-server}"
+PY="$APP_DIR/.venv/bin/python"
+cd "$APP_DIR"
+mkdir -p data
+LOG="data/etl_backfill_$(date +%Y%m%d_%H%M%S).log"
+
+# 退出时务必恢复定时器(即使 ETL 中途失败,也不把 timer 留在停用态)
+restore_timer() { systemctl start meituan-etl.timer 2>/dev/null || true; }
+trap restore_timer EXIT
+
+echo "[0/3] 自检 image_size/image_type 列(缺列说明没迁移,先跑 alembic upgrade head)..."
+"$PY" -c "from sqlalchemy import inspect; from app.db.session import engine; \
+cols=[c['name'] for c in inspect(engine).get_columns('meituan_coupon')]; \
+import sys; (('image_size' in cols) and ('image_type' in cols)) or sys.exit('  ❌ 缺列,请先: sudo .venv/bin/python -m alembic upgrade head'); \
+print('  ✓ 列已就绪')"
+
+echo "[1/3] 停定时器,防与定时轮次并发占锁..."
+systemctl stop meituan-etl.timer 2>/dev/null || true
+
+echo "[2/3] 全量重跑 ETL(~70min;日志 $LOG)..."
+# 不走 systemctl:meituan-etl.service 有 20min 硬超时,全量一轮跑不完会被掐断。
+"$PY" -m scripts.pull_meituan_coupons --once --prune-hours 24 2>&1 | tee "$LOG"
+
+echo "[3/3] 验证填充率..."
+"$PY" - <<'PYEOF'
+from sqlalchemy import text
+from app.db.session import engine
+engine.echo = False
+with engine.connect() as c:
+    tot = c.execute(text("SELECT count(*) FROM meituan_coupon")).scalar()
+    enr = c.execute(text("SELECT count(*) FROM meituan_coupon WHERE image_size IS NOT NULL")).scalar()
+    gif = c.execute(text("SELECT count(*) FROM meituan_coupon WHERE image_type='image/gif'")).scalar()
+    print(f"  总 {tot} 行 | 有 image_size {enr} ({enr*100//max(tot,1)}%) | GIF {gif} 张")
+PYEOF
+echo "✅ 回填完成(定时器已恢复)。"

docs/README.md

@@ -16,7 +16,7 @@
 | [api/](./api/) | **API 接口文档**:"索引 + 一接口一文件"(组织方式见下)。想查"某个接口的协议"看这里。 |
 | [database/](./database/) | **数据库文档**:每张表一个文件(表结构/字段/索引/约定) + 两份迁移指南——[数据库迁移.md](./database/数据库迁移.md)(Alembic 操作:clone 后建表/日常升级/新增迁移/多 head 排查) + [postgres-migration.md](./database/postgres-migration.md)(SQLite → PostgreSQL 切换步骤,配套 `scripts/init_postgres.py`)。想"把库跑起来 / 改表结构 / 查某张表"看这里。 |
 | [integrations/](./integrations/) | **集成层实现文档**:穿山甲验签 / 微信支付 / 极光 / 短信 / 美团 CPS 等 SDK 集成的签名、加解密、协议细节与踩坑。想知道"接外部服务那块到底怎么实现"看这里。 |
-| [guides/](./guides/) | **开发 / 上线 / 功能 指南**:[待办与技术债.md](./guides/待办与技术债.md)(跨前后端 backlog,P1 鉴权/用户绑定、引擎移植待办等,"还欠什么、以后要补什么"看这份) + [看广告赚金币上线清单.md](./guides/看广告赚金币上线清单.md)(看广告发奖上线 checklist) + [邀请功能-实现原理与本地测试.md](./guides/邀请功能-实现原理与本地测试.md)(invite-mvp 实现原理 + 本地内网全链路测试)。 |
+| [guides/](./guides/) | **开发 / 上线 / 功能 指南**:[待办与技术债.md](./guides/待办与技术债.md)(跨前后端 backlog,P1 鉴权/用户绑定、引擎移植待办等,"还欠什么、以后要补什么"看这份) + [看广告赚金币上线清单.md](./guides/看广告赚金币上线清单.md)(看广告发奖上线 checklist) + [邀请功能-实现原理与本地测试.md](./guides/邀请功能-实现原理与本地测试.md)(invite-mvp 实现原理 + 本地内网全链路测试) + [CPS发券分发与微信授权.md](./guides/CPS发券分发与微信授权.md)(**CPS 发券系统交接文档**:设群/建活动/生成落地页短链/美团对账/统计 + 微信网页授权拿 openid 做用户级统计;含数据流/表/平台差异/端点/配置/代码地图/排障/技术债)。 |
 
 ## api/ 目录是怎么组织的(传送门式)
 

docs/api/README.md

@@ -173,7 +173,7 @@
 | `platform` | int | `1`=外卖/到家, `2`=到店 |
 | `biz_line` | int \| null | 到店子类:1到餐 2到综 3酒店 4门票 |
 | `name` | string | 商品名 |
-| `head_image_url` | string | 头图 |
+| `head_image_url` | string | 头图;后端已**统一**拼缩放参数 `@375w_375h_1e_1c.webp`(美团 CDN 服务端缩到 124dp + 转 WebP,实测省 76–99%),客户端直接用 |
 | `brand_name` | string \| null | 品牌名 |
 | `brand_logo_url` | string \| null | 品牌 logo |
 | `sell_price` | string | 现价 |

docs/guides/CPS发券分发与微信授权.md

@@ -0,0 +1,194 @@
+# CPS 发券分发 + 微信授权 系统说明(交接文档)
+
+> 给接手人。读完应能:看懂系统、带运营走完"设群→建活动→生成链接→看统计"、接微信授权、排障、知道边界。
+> 代码引用一律用「文件:符号」(不写行号,重构后符号比行号稳)。
+
+---
+
+## 一、这是什么 / 业务目标
+
+傻瓜比价的 **CPS 群发分发**系统:运营在 admin 后台建「群」、建「活动」(美团/淘宝/京东的券)、生成带渠道追踪的**落地页短链 `/c/{code}`** 发到各微信群;用户点链接 → 落地页 → 跳转下单或复制淘口令领券 → 后端记点击、(美团)拉订单对账、(微信内)拿 openid 做用户级统计。
+
+**目标**:把券发出去 + 把「点击 → 下单 → 佣金」漏斗、「哪个群/哪个微信用户领了券」统计出来,指导投放。
+
+---
+
+## 二、两套入口 / 两个域名(先建立全局)
+
+| | 域名 | 反代 | 代码 | 鉴权 |
+|---|---|---|---|---|
+| 运营后台 | admin-web.shaguabijia.com | admin 子应用 :8771 | `admin/routers/cps.py`,前缀 `/admin/api/cps` | admin JWT(operator/finance) |
+| 用户落地页 | **coupon.shaguabijia.com** | app-server :8770 | `api/v1/cps_redirect.py`,**无前缀** | **公网无鉴权**(群里谁都能点) |
+
+> 两个域名都全反代到 app-server/admin。落地页域名 `coupon.shaguabijia.com` 由 `settings.CPS_REDIRECT_BASE` 配置,生成短链时拼前缀。
+
+---
+
+## 三、数据流
+
+```
+[admin] 建群(cps_group) → 建活动(cps_activity) → 生成链接(cps_link, 得 /c/{code})
+                                                          │ 复制发到微信群
+                                                          ▼
+[用户点 /c/{code}] cps_landing:
+   ├─ (微信内+开关开+无cookie) 302 跳微信授权 → 回调换 openid → 存 cps_wx_user + 种 cookie → 跳回
+   ├─ 记一条点击 cps_click(visit, 带 openid)
+   └─ 美团/京东:302 跳 target_url   |   淘宝:返回 H5 落地页(图+复制淘口令按钮)
+                                                          │ (淘宝)点复制 → POST /c/{code}/copy 记 copy
+                                                          ▼
+[美团] 运营在 admin 点「刷新对账」→ query_order 拉订单 → cps_order(按 sid 归群)
+                                                          ▼
+[admin 统计] 对账统计页 + 群详情(折线图 / 每日明细 / 群内微信用户)
+```
+
+---
+
+## 四、数据表(6 张,均在 `app/models/cps_*`)
+
+| 表 | 模型 | 职责 / 关键字段 |
+|---|---|---|
+| cps_group | `CpsGroup` | 群。`name` / `platforms`(多选 meituan/taobao/jd) / `sid`(美团渠道追踪位,仅美团群有;纯淘宝/京东为空) / `member_count` / `status`(active/archived) |
+| cps_activity | `CpsActivity` | 可推广活动池。`platform` / `name` / `act_id`(美团活动物料ID) / `product_view_sign`(美团商品券) / `payload`(淘宝整段淘口令 / 京东链接) / `image_url`(淘宝落地页主视觉图,绝对URL) |
+| cps_link | `CpsLink` | 群发短链。`code`(短码,`/c/{code}`) / `group_id` / `activity_id` / `sid` / `target_url`(美团短链/京东链接/淘宝淘口令) / `platform` |
+| cps_click | `CpsClick` | 点击事件。`link_id` / `group_id` / `sid` / `event_type`(**visit**=进页/被跳转,**copy**=淘宝点复制口令) / `ip` / `ua` / **`openid`**(微信授权拿到,非微信/未授权为空) |
+| cps_order | `CpsOrder` | 美团订单(query_order 拉回)。`order_id`(唯一,upsert) / `sid`(按它归群) / `pay_price_cents` / `commission_cents`(预估佣金) / `refund_profit_cents`(退款佣金) / `mt_status`(2付款3完成4取消5风控6结算) / `pay_time` |
+| cps_wx_user | `CpsWxUser` | 微信用户。`openid`(唯一) / `unionid` / `nickname` / `headimgurl` / `first_group_id`(首次授权来源群) |
+
+> 金额**全程存「分」**,前端 ÷100 显示「元」。时间统一 tz-aware UTC,展示按北京。
+
+---
+
+## 五、三平台差异(最关键的一张表)
+
+| | 美团 | 淘宝 | 京东 |
+|---|---|---|---|
+| 链接生成 | 调 `meituan.get_referral_link` 转链(**需群 sid**) | 直接用活动 `payload`(淘口令) | 直接用活动 `payload`(链接) |
+| 落地页行为 | 302 跳美团短链 | **H5 落地页**(主视觉图 + 复制淘口令按钮) | 302 跳京东链接 |
+| 对账 | ✓ `query_order` 拉订单/佣金 | ✗ 无开放 API | ✗ 无开放 API |
+| 能统计到 | 点击 + 下单 + 佣金 | 点击 + 复制次数 | 点击 |
+
+> 淘宝/京东订单类指标(有效单/成交额/佣金)在统计页显示「-」(无法对账)。只有美团群(有 sid)有订单数据。
+
+---
+
+## 六、运营操作流程(带新人就照这个)
+
+1. **建群**:CPS 分发 → 群管理 → 新建群。选这个群发哪些平台的券(可多选);含美团时可填 sid(留空自动 `g<id>`)。
+2. **建活动**:活动管理 → 新建活动。按平台填:美团填 `actId`(联盟「我要推广-活动推广」第一列) 或 商品券 `productViewSign`;淘宝填整段淘口令 + **必须上传落地页图**;京东填推广链接。
+3. **生成链接**:群管理 → 某群 → 「生成链接」→ 勾选要发的活动(只列该群平台范围内的)→ 每个活动出一条 `https://coupon.shaguabijia.com/c/{code}` → 复制发到群里。
+4. **对账(仅美团,finance 角色)**:对账统计 → 「刷新对账(拉美团订单)」→ 拉近 N 天订单入库。**非实时,要手动点**。
+5. **看统计**:对账统计页看各群点击/独立/复制/有效单/取消风控/成交额/预估佣金/结算佣金;**点群名进群详情** → 折线图(点击趋势,天/小时)+ 每日明细大表格(点击+订单按天,含退款/净佣金)+ 群内微信用户(头像/昵称/领券次数)。
+
+---
+
+## 七、微信授权(用户级统计)
+
+### 目的
+用户在微信内打开落地页时,拿到 `openid`(识别这个人) + 昵称头像,做「每个微信用户在每个群领没领券」的用户级统计。
+
+### 链路
+```
+/c/{code} (微信内 + 开关开 + 无 wx_openid cookie)
+  → 302 跳微信 base 授权(snsapi_base,静默不打断)
+  → 微信回调 /wx/oauth/cb?code=..&state=base:{code}
+  → wx_oauth.exchange_code 换 openid → cps_wx_user_repo.upsert 存用户 + 关联群 → 种 wx_openid cookie(30天)
+  → 302 跳回 /c/{code}(这次有 cookie,正常展示,记 visit 带 openid)
+[淘宝群] 用户点「复制口令」→ 若还没昵称头像 → 跳 snsapi_userinfo 授权(用户点击=交互触发,避免微信"快照页")
+  → 回调拉 userinfo 补 nickname/headimgurl → 跳回落地页自动复制
+```
+代码:`cps_redirect.py:cps_landing` / `cps_redirect.py:wx_oauth_cb` / `cps_redirect.py:_taobao_landing_html`(注入 userinfo 授权链接);微信调用封装 `integrations/wx_oauth.py`。
+
+### 三个硬前提
+1. **必须是已认证服务号**(订阅号、未认证都不支持网页授权 → 报 10005)。
+2. **网页授权域名** = `coupon.shaguabijia.com`,在服务号后台「设置与开发→公众号设置→功能设置→网页授权域名」配置 + 放校验文件。校验文件由本服务端点 serve(见 `cps_redirect.py` 里 `MP_verify_*.txt` 路由,换号/换文件时改这里返回的串)。
+3. **凭证 + 开关**:`.env` 配 `WX_MP_APPID` / `WX_MP_SECRET`(服务号的,**≠** 微信支付那个 `WECHAT_APP_ID`);总开关 `WX_MP_OAUTH_ENABLED=true`。开关默认关 → 落地页走原逻辑、不拿 openid(认证未就绪时用)。判定见 `config.py:Settings.wx_oauth_active`。
+
+### 群内微信用户怎么算(重要,踩过坑)
+`admin/repositories/cps.py:group_wx_users` 按「**该群有带 openid 的点击**」(cps_click.group_id + openid)算群内用户,再关联 cps_wx_user 拿画像。**不是**按 `cps_wx_user.first_group_id`——因为一个用户(cookie 已存)会去多个群,first_group_id 只记首次授权群,会漏算。
+- **美团/京东群只有 openid、没有昵称头像**:它们是 302 跳转、没有"点复制领券"那一步,触发不了 userinfo。只有**淘宝群**点复制口令时才补昵称头像。
+
+---
+
+## 八、端点清单
+
+### admin(`/admin/api/cps`,operator/finance 鉴权)— `admin/routers/cps.py`
+- 群:`GET /groups` `POST /groups` `PATCH /groups/{id}` `DELETE /groups/{id}`
+- 活动:`GET /activities` `POST /activities` `PATCH /activities/{id}` `DELETE /activities/{id}` `POST /upload-image` `GET /activity-images`
+- 生成链接:`POST /referral-links`(批量,按 activity_ids,美团转链)
+- 对账:`POST /orders/reconcile`(拉美团订单) `GET /orders`(明细)
+- 统计:`GET /stats`(按群对账统计) `GET /groups/{id}/timeseries`(折线) `GET /groups/{id}/daily`(每日明细) `GET /groups/{id}/wx-users`(群内微信用户)
+
+### 落地页(coupon 域,公网无鉴权)— `api/v1/cps_redirect.py`
+- `GET /c/{code}`(落地页:授权+记点击+跳转/H5)
+- `POST /c/{code}/copy`(淘宝点复制,记 copy)
+- `GET /wx/oauth/cb`(微信授权回调)
+- `GET /MP_verify_*.txt`(微信域名归属校验文件)
+
+> 删除/编辑都不级联删已生成的 cps_link——**已发出去的短链永远要能用**,落地页只查 link 本身,不依赖 group/activity 是否还在(删了淘宝活动,其落地页图回退默认图)。
+
+---
+
+## 九、配置项(`.env`,见 `app/core/config.py`)
+
+| 配置 | 说明 |
+|---|---|
+| `CPS_REDIRECT_BASE` | 落地页域名,如 `https://coupon.shaguabijia.com`。生成 `/c/{code}` 拼前缀;也是图/回调的绝对 URL 基址 |
+| `MT_CPS_APP_KEY` / `MT_CPS_APP_SECRET` | 美团联盟凭证(转链 + 对账)。缺则美团相关优雅降级 |
+| `MT_CPS_DEFAULT_SID` | 美团默认渠道位(群没 sid 时兜底) |
+| `WX_MP_APPID` / `WX_MP_SECRET` | **微信服务号**网页授权凭证(≠ `WECHAT_APP_ID` 那个 App 支付用的) |
+| `WX_MP_OAUTH_ENABLED` | 微信授权总开关。默认 `false`;认证就绪后置 `true` 重启启用 |
+
+---
+
+## 十、代码地图
+
+| 关注点 | 位置 |
+|---|---|
+| admin 端点 | `admin/routers/cps.py` |
+| admin 业务逻辑(群/活动/订单/统计/群内用户) | `admin/repositories/cps.py`(`group_stats` / `group_click_timeseries` / `group_order_daily` / `group_wx_users` / `reconcile_orders`) |
+| admin 出入参 | `admin/schemas/cps.py` |
+| 落地页 + 微信授权 | `api/v1/cps_redirect.py` |
+| 短码生成 / 记点击 / 点击聚合 | `repositories/cps_link.py`(`create_link` / `record_click` / `click_stats_by_group`) |
+| 微信用户 upsert | `repositories/cps_wx_user.py` |
+| 美团集成(转链/拉单) | `integrations/meituan.py`(`get_referral_link` / `query_order`) |
+| 微信网页授权集成 | `integrations/wx_oauth.py`(`build_authorize_url` / `exchange_code` / `get_userinfo`) |
+| 数据模型 | `app/models/cps_*.py` |
+| admin 前端 | shaguabijia-admin-web `src/app/(main)/cps/`(列表页 `page.tsx` + 群详情 `groups/[id]/page.tsx`) |
+
+---
+
+## 十一、部署 / 运维
+
+- 部署:`ssh ecs1` → `deploy server`(app-server + admin 后端)/ `deploy web`(admin-web 前端)。机制 = git reset 到 origin/main + 装依赖 + alembic 迁移 + 重启 + health 自检。
+- 改 `.env`(如开微信开关、换服务号凭证)后:`systemctl restart shaguabijia-app-server shaguabijia-admin` 或重跑 `deploy server`。
+- 数据库迁移:CPS 相关迁移在 `alembic/versions/cps_*`(建表 `cps_tables`/`cps_link_tables`、多平台 `cps_v2_platforms`、活动图 `cps_activity_image`、微信用户 `cps_wx_user`)。
+
+---
+
+## 十二、排障速查
+
+| 现象 | 原因 / 处理 |
+|---|---|
+| 微信授权报 **10005**(没有这些 scope 权限) | 账号不是已认证服务号,或「网页授权获取用户基本信息」接口未获得。确认账号类型 + 接口权限 |
+| 微信授权报 **10003**(redirect_uri 域名不一致) | 服务号后台「网页授权域名」没填 `coupon.shaguabijia.com` 或没保存成功 |
+| 换了服务号后授权失败 | 换 `.env` 的 `WX_MP_APPID/SECRET` + 在**新服务号**后台重配网页授权域名(校验文件可能是新的,要换 `MP_verify_*` 端点返回值) |
+| 落地页在微信里不跳授权 | 开关 `WX_MP_OAUTH_ENABLED` 没开,或不是微信 UA(外部浏览器不跳,正常) |
+| 群详情「群内微信用户」看不到人 | 已按"本群点击"算(`group_wx_users`)。确认该群 cps_click 有带 openid 的记录;美团/京东群昵称头像本就为空 |
+| 统计页订单/佣金是「-」 | 淘宝/京东群无对账(正常);美团群需先点「刷新对账」拉单 |
+| 成交额/佣金不更新 | 对账非实时,要 finance 手动「刷新对账」 |
+| 账号验证 appid/账号类型 | 可用基础 token 调 `cgi-bin/account/getaccountbasicinfo` 查(account_type / 认证 / 主体) |
+
+---
+
+## 十三、已知边界 / 技术债(接手前必读)
+
+1. **下单归因只到「群」级,做不到「人」级**:美团 `query_order` 返回的订单只带 `sid`(群级渠道位),不带 openid。要统计"某个微信用户下没下单/下单额",需把 sid 体系改成**用户级**(每个用户进落地页时用其 openid 实时转链),是较大改动,**未做**。当前微信侧只能统计到"用户级**领券/点击**",下单仍是群级。
+2. **美团/京东群拿不到昵称头像**:无 userinfo 触发点(直接 302 跳转)。只有淘宝群(H5 落地页点复制)能补昵称头像。
+3. **淘宝/京东无对账 API**:只能统计点击/复制,拿不到下单/佣金。
+4. **落地页公网无鉴权**:群发场景,链接谁拿到都能点(刻意如此)。点击 UV 按 (ip, ua) 近似去重,非精确。
+5. **对账非实时**:`cps_order` 是上次「刷新对账」的快照,不点不更新。
+6. **激励奖励金额未统计**:美团订单里 `incentiveOrder` 标记了是否参与激励活动,但 `query_order` 不给激励奖励金额,需另接「奖励活动报表」API(未做)。
+
+---
+
+*本系统由本轮从零搭建:CPS 分发(群/活动/链接/对账/统计)+ 群详情可视化 + 微信网页授权用户级统计。后端在 shaguabijia-app-server,前端在 shaguabijia-admin-web。*

scripts/pull_meituan_coupons.py

@@ -52,6 +52,7 @@ try:
 except Exception:  # noqa: BLE001
     pass
 
+import httpx  # noqa: E402
 from sqlalchemy import delete, func, select  # noqa: E402
 from sqlalchemy.dialects.postgresql import insert as pg_insert  # noqa: E402
 
@@ -85,6 +86,13 @@ DEFAULT_CONCURRENCY = 12   # 并发城市数(实测 15 并发 402 占 3% 可退
 STARTUP_STAGGER = 0.3      # 首批城市启动错峰间隔秒(削平瞬时峰值,实测能压低 402)
 PRUNE_FAIL_RATIO_MAX = 0.05  # 失败城占比超此值则本轮跳过 prune(避免大面积抓取失败误删库)
 
+# 头图大小/类型富集:每城抓完后,按 head_url 去重并发 HEAD 取 Content-Length / Content-Type,
+# 写进 image_size / image_type。图片走美团图片 CDN(p*.meituan.net / img.meituan.net),直连可达、
+# 与 CPS 网关 402 限流无关;任何失败 → None,不阻断入库。城市级已 12 并发,这里每城再开 8,
+# 叠加峰值 ~96 个 HEAD,CDN 扛得住(实测单进程 50 并发稳)。
+IMAGE_META_WORKERS = 8     # 每城 HEAD 并发数
+IMAGE_META_TIMEOUT = 8.0   # 单图 HEAD 超时秒
+
 SOURCES = [
     {"code": "search_waimai", "label": "外卖·搜外卖", "kind": "search", "platform": 1, "keyword": "外卖"},
     {"code": "search_meishi", "label": "外卖·搜美食", "kind": "search", "platform": 1, "keyword": "美食"},
@@ -256,6 +264,9 @@ def _parse_item(item: dict, source: dict, city_id: str) -> dict | None:
         "sell_price_cents": price_cents,
         "original_price_cents": _to_cents(cpd.get("originalPrice")),
         "head_url": ((cpd.get("headUrl") or "").split("@")[0][:512] or None),
+        # 占位;由 _fill_image_meta 抓完本城后并发 HEAD 回填(键必须在,upsert 才会带上这两列)
+        "image_size": None,
+        "image_type": None,
         "sale_volume": cpd.get("saleVolume"),
         "sale_volume_num": _sale_volume_num(cpd.get("saleVolume")),
         "commission_percent": comm_pct,
@@ -268,6 +279,38 @@ def _parse_item(item: dict, source: dict, city_id: str) -> dict | None:
     })
 
 
+def _head_image_meta(client: httpx.Client, url: str) -> tuple[int | None, str | None]:
+    """HEAD 单张头图,取 (字节大小, MIME 类型);任何失败返回 (None, None),不阻断入库。"""
+    try:
+        r = client.head(url, timeout=IMAGE_META_TIMEOUT, follow_redirects=True)
+        if r.status_code >= 400:
+            return None, None
+        cl = r.headers.get("content-length")
+        size = int(cl) if cl and cl.isdigit() else None
+        ctype = (r.headers.get("content-type") or "").split(";")[0].strip()[:32] or None
+        return size, ctype
+    except Exception:  # noqa: BLE001
+        return None, None
+
+
+def _fill_image_meta(rows: list[dict]) -> None:
+    """就地为本城 rows 填 image_size / image_type:按 head_url 去重后并发 HEAD,再回填每行。
+    trust_env=False 直连图片 CDN(绕本机代理,CDN 无需走 CPS 代理)。"""
+    urls = {r["head_url"] for r in rows if r.get("head_url")}
+    if not urls:
+        return
+    meta: dict[str, tuple[int | None, str | None]] = {}
+    with httpx.Client(trust_env=False, timeout=IMAGE_META_TIMEOUT) as client:
+        with ThreadPoolExecutor(max_workers=IMAGE_META_WORKERS) as pool:
+            futs = {pool.submit(_head_image_meta, client, u): u for u in urls}
+            for f in as_completed(futs):
+                meta[futs[f]] = f.result()
+    for r in rows:
+        m = meta.get(r.get("head_url"))
+        if m:
+            r["image_size"], r["image_type"] = m
+
+
 def _pull_one_city(city: dict, index: int, concurrency: int, stagger: float) -> list[dict]:
     """抓单个城市的 3 路券并解析。worker 线程内执行(只抓取+解析,不碰 DB)。
 
@@ -286,6 +329,7 @@ def _pull_one_city(city: dict, index: int, concurrency: int, stagger: float) ->
             p = _parse_item(it, src, cid)
             if p:
                 parsed.append(p)
+    _fill_image_meta(parsed)  # 抓完本城 → 并发 HEAD 回填头图 image_size/image_type
     return parsed