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

app/admin/repositories/coupon_data.py

@@ -95,7 +95,7 @@ def _session_to_row(
     phone: str | None = None,
     nickname: str | None = None,
     ad_revenue_yuan: float = 0.0,
-    point_score: tuple[int, int] | None = None,
+    point_stats: dict | None = None,
 ) -> dict:
     """CouponSession ORM → 明细行 dict(主表「领券数据」与「用户全部领券」抽屉共用)。"""
     return {
@@ -114,29 +114,45 @@ def _session_to_row(
         "app_env": r.app_env,
         "started_at": r.started_at,
         "claimed_count": r.claimed_count,
-        "point_success_count": point_score[0] if point_score is not None else None,
-        "point_total_count": point_score[1] if point_score is not None else None,
+        "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_scores_by_trace(db: Session, trace_ids: list[str]) -> dict[str, tuple[int, int]]:
-    """批量统计逐场券点位分数:成功(含已领) / 尝试(不含 skipped)。"""
+def _point_stats_by_trace(db: Session, trace_ids: list[str]) -> dict[str, dict]:
+    """一次查询批量返回逐场点位分数和逐券明细。"""
     if not trace_ids:
         return {}
-    succeeded = func.sum(case((CouponClaimRecord.status.in_(_SLOT_OK), 1), else_=0))
-    tried = func.sum(case((CouponClaimRecord.status.in_(_SLOT_TRIED), 1), else_=0))
     rows = db.execute(
-        select(CouponClaimRecord.trace_id, succeeded, tried)
+        select(
+            CouponClaimRecord.trace_id,
+            CouponClaimRecord.coupon_id,
+            CouponClaimRecord.coupon_name,
+            CouponClaimRecord.status,
+            CouponClaimRecord.reason,
+        )
         .where(CouponClaimRecord.trace_id.in_(trace_ids))
-        .group_by(CouponClaimRecord.trace_id)
+        .order_by(CouponClaimRecord.trace_id, CouponClaimRecord.id)
     ).all()
-    return {
-        trace_id: (int(success_count or 0), int(total_count or 0))
-        for trace_id, success_count, total_count in rows
-        if trace_id is not None and int(total_count or 0) > 0
-    }
+    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:
@@ -279,7 +295,7 @@ def coupon_data_report(
             ).all()
         }
     rev_map = crud_ecpm.revenue_yuan_by_trace(db, [r.trace_id for r in page])
-    point_score_map = _point_scores_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)
@@ -288,7 +304,7 @@ def coupon_data_report(
             phone,
             nickname,
             ad_revenue_yuan=rev_map.get(r.trace_id, 0.0),
-            point_score=point_score_map.get(r.trace_id),
+            point_stats=point_stats_map.get(r.trace_id),
         ))
 
     return {
@@ -312,13 +328,13 @@ 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_score_map = _point_scores_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),
-                point_score=point_score_map.get(r.trace_id),
+                point_stats=point_stats_map.get(r.trace_id),
             )
             for r in rows
         ],

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):
     """一条领券明细(一次领券任务)。"""
 
@@ -75,6 +84,9 @@ class CouponDataRow(BaseModel):
     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

@@ -1,7 +1,7 @@
 """admin 领券明细逐场点位分数。"""
 from datetime import date
 
-from app.admin.repositories.coupon_data import _point_scores_by_trace
+from app.admin.repositories.coupon_data import _point_stats_by_trace
 from app.db.session import SessionLocal
 from app.models.coupon_state import CouponClaimRecord
 
@@ -17,20 +17,28 @@ def test_point_scores_by_trace() -> None:
                 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()
 
-        assert _point_scores_by_trace(db, [trace]) == {trace: (2, 3)}
+        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_scores_omit_trace_without_attempts() -> None:
-    """仅有 skipped 或完全无记录时不伪造 0/0。"""
+def test_point_stats_keep_skipped_detail_without_faking_score() -> None:
+    """仅有 skipped 时保留明细，但分数仍为 0/0，由 schema 层展示为空。"""
     db = SessionLocal()
     trace = "point-score-skipped"
     try:
@@ -43,7 +51,11 @@ def test_point_scores_omit_trace_without_attempts() -> None:
         ))
         db.flush()
 
-        assert _point_scores_by_trace(db, [trace, "missing-trace"]) == {}
+        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()