feat(admin): 返回领券点位明细

app/admin/repositories/coupon_data.py

@@ -22,6 +22,10 @@ from app.repositories import ad_ecpm as crud_ecpm
 from app.repositories.coupon_state import DEFAULT_PLATFORMS, coupon_id_to_platform
 
 
+_SLOT_OK = ("success", "already_claimed")
+_SLOT_TRIED = ("success", "already_claimed", "failed")
+
+
 def _cn_hour(dt: datetime) -> int:
     """started_at(UTC 口径)→ 北京时间小时(0–23)。naive 当 UTC(sqlite),tz-aware 直接换算(pg)。"""
     if dt.tzinfo is None:
@@ -86,7 +90,13 @@ def _success_rates(rows: list) -> dict:
     }
 
 
-def _session_to_row(r, phone: str | None = None, nickname: str | None = None, ad_revenue_yuan: float = 0.0) -> dict:
+def _session_to_row(
+    r,
+    phone: str | None = None,
+    nickname: str | None = None,
+    ad_revenue_yuan: float = 0.0,
+    point_stats: dict | None = None,
+) -> dict:
     """CouponSession ORM → 明细行 dict(主表「领券数据」与「用户全部领券」抽屉共用)。"""
     return {
         "id": r.id,
@@ -104,11 +114,47 @@ def _session_to_row(r, phone: str | None = None, nickname: str | None = None, ad
         "app_env": r.app_env,
         "started_at": r.started_at,
         "claimed_count": r.claimed_count,
+        "point_success_count": point_stats["succeeded"] if point_stats and point_stats["tried"] else None,
+        "point_total_count": point_stats["tried"] if point_stats and point_stats["tried"] else None,
+        "point_details": point_stats["details"] if point_stats else [],
         "trace_url": r.trace_url,
         "ad_revenue_yuan": ad_revenue_yuan,
     }
 
 
+def _point_stats_by_trace(db: Session, trace_ids: list[str]) -> dict[str, dict]:
+    """一次查询批量返回逐场点位分数和逐券明细。"""
+    if not trace_ids:
+        return {}
+    rows = db.execute(
+        select(
+            CouponClaimRecord.trace_id,
+            CouponClaimRecord.coupon_id,
+            CouponClaimRecord.coupon_name,
+            CouponClaimRecord.status,
+            CouponClaimRecord.reason,
+        )
+        .where(CouponClaimRecord.trace_id.in_(trace_ids))
+        .order_by(CouponClaimRecord.trace_id, CouponClaimRecord.id)
+    ).all()
+    result: dict[str, dict] = {}
+    for trace_id, coupon_id, coupon_name, status, reason in rows:
+        if trace_id is None:
+            continue
+        stats = result.setdefault(trace_id, {"succeeded": 0, "tried": 0, "details": []})
+        if status in _SLOT_TRIED:
+            stats["tried"] += 1
+            if status in _SLOT_OK:
+                stats["succeeded"] += 1
+        stats["details"].append({
+            "coupon_id": coupon_id,
+            "coupon_name": coupon_name,
+            "status": status,
+            "reason": reason,
+        })
+    return result
+
+
 def _empty_result() -> dict:
     return {
         "summary": {
@@ -249,10 +295,17 @@ def coupon_data_report(
             ).all()
         }
     rev_map = crud_ecpm.revenue_yuan_by_trace(db, [r.trace_id for r in page])
+    point_stats_map = _point_stats_by_trace(db, [r.trace_id for r in page])
     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, ad_revenue_yuan=rev_map.get(r.trace_id, 0.0)))
+        items.append(_session_to_row(
+            r,
+            phone,
+            nickname,
+            ad_revenue_yuan=rev_map.get(r.trace_id, 0.0),
+            point_stats=point_stats_map.get(r.trace_id),
+        ))
 
     return {
         "summary": summary,
@@ -275,16 +328,20 @@ def coupon_user_records(db: Session, *, user_id: int, limit: int = 100) -> dict:
         select(func.count()).select_from(CouponSession).where(CouponSession.user_id == user_id)
     ).scalar_one()
     rev_map = crud_ecpm.revenue_yuan_by_trace(db, [r.trace_id for r in rows])
+    point_stats_map = _point_stats_by_trace(db, [r.trace_id for r in rows])
     return {
-        "items": [_session_to_row(r, ad_revenue_yuan=rev_map.get(r.trace_id, 0.0)) for r in rows],
+        "items": [
+            _session_to_row(
+                r,
+                ad_revenue_yuan=rev_map.get(r.trace_id, 0.0),
+                point_stats=point_stats_map.get(r.trace_id),
+            )
+            for r in rows
+        ],
         "total": int(total),
     }
 
 
-_SLOT_OK = ("success", "already_claimed")
-_SLOT_TRIED = ("success", "already_claimed", "failed")
-
-
 def coupon_slot_report(
     db: Session, *, date_from: str, date_to: str, app_env: str | None = None
 ) -> dict:

app/admin/schemas/coupon_data.py

@@ -49,6 +49,15 @@ class CouponDataHourly(BaseModel):
     avg_elapsed_ms: int | None = None
 
 
+class CouponPointDetail(BaseModel):
+    """一次领券任务中的单券点位结果。"""
+
+    coupon_id: str
+    coupon_name: str | None = None
+    status: str = Field(..., description="success / already_claimed / failed / skipped")
+    reason: str | None = None
+
+
 class CouponDataRow(BaseModel):
     """一条领券明细(一次领券任务)。"""
 
@@ -69,6 +78,15 @@ class CouponDataRow(BaseModel):
     app_env: str | None = None
     started_at: datetime = Field(..., description="发起时刻(明细「时间」列)")
     claimed_count: int | None = None
+    point_success_count: int | None = Field(
+        None, description="本次成功券点位数(success+already_claimed);无逐券埋点为空"
+    )
+    point_total_count: int | None = Field(
+        None, description="本次尝试券点位数(success+already_claimed+failed,不含 skipped);无逐券埋点为空"
+    )
+    point_details: list[CouponPointDetail] = Field(
+        default_factory=list, description="本次逐券点位结果，用于后台点击分数查看成功/失败明细"
+    )
     trace_url: str | None = Field(None, description="pricebot 公网 trace 链接(仅 completed 有);admin 渲染可点链接,无则显示可复制 trace_id")
     ad_revenue_yuan: float = Field(
         0.0, description="本次领券看的信息流广告预估收益(元);按 trace_id 聚合 ad_ecpm_record"

tests/test_coupon_point_score.py

@@ -0,0 +1,61 @@
+"""admin 领券明细逐场点位分数。"""
+from datetime import date
+
+from app.admin.repositories.coupon_data import _point_stats_by_trace
+from app.db.session import SessionLocal
+from app.models.coupon_state import CouponClaimRecord
+
+
+def test_point_scores_by_trace() -> None:
+    """已领算成功、失败算尝试、跳过不进分母。"""
+    db = SessionLocal()
+    trace = "point-score-trace"
+    try:
+        db.add_all([
+            CouponClaimRecord(
+                device_id="score-device",
+                coupon_id=f"mt-score-{status}",
+                claim_date=date(2020, 1, 2),
+                status=status,
+                coupon_name=f"测试点位-{status}",
+                reason="测试失败" if status == "failed" else None,
+                trace_id=trace,
+            )
+            for status in ("success", "already_claimed", "failed", "skipped")
+        ])
+        db.flush()
+
+        stats = _point_stats_by_trace(db, [trace])[trace]
+        assert stats["succeeded"] == 2
+        assert stats["tried"] == 3
+        assert [item["status"] for item in stats["details"]] == [
+            "success", "already_claimed", "failed", "skipped"
+        ]
+        assert stats["details"][2]["reason"] == "测试失败"
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_point_stats_keep_skipped_detail_without_faking_score() -> None:
+    """仅有 skipped 时保留明细，但分数仍为 0/0，由 schema 层展示为空。"""
+    db = SessionLocal()
+    trace = "point-score-skipped"
+    try:
+        db.add(CouponClaimRecord(
+            device_id="score-device-skipped",
+            coupon_id="mt-score-skipped-only",
+            claim_date=date(2020, 1, 2),
+            status="skipped",
+            trace_id=trace,
+        ))
+        db.flush()
+
+        stats = _point_stats_by_trace(db, [trace, "missing-trace"])
+        assert stats[trace]["succeeded"] == 0
+        assert stats[trace]["tried"] == 0
+        assert stats[trace]["details"][0]["status"] == "skipped"
+        assert "missing-trace" not in stats
+    finally:
+        db.rollback()
+        db.close()