fix(admin): 按任务白名单聚合常规任务金币 (#157)

## 背景

大盘“常规任务金币”原先采用“全部正向金币减排除清单”的反向口径。线上新增 `feed_ad_reward_coupon` / `feed_ad_reward_comparison` 后未同步加入排除清单，导致领券和比价奖励误计入常规任务金币。

## 修改

- 改为明确白名单：`signin`、历史 `signin_boost`、全部 `task_` 任务、`price_report_reward`、`feedback_reward`
- 未知新 `biz_type` 默认不进入常规任务桶
- 增加覆盖领券、比价、广告、邀请、管理员及未知类型的回归测试
- 顺带修复改动文件已有的 Ruff `UP017`

## 验证

- 线上只读 PostgreSQL：新口径全量为 108,022，领券/比价误计差额为 450,675
- Ruff：通过
- `pytest tests/test_admin_read.py tests/test_cps_admin.py -q`：23 passed

---------

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

app/admin/repositories/coupon_data.py

@@ -21,6 +21,9 @@ from app.models.user import User
 from app.repositories import ad_ecpm as crud_ecpm
 from app.repositories.coupon_state import DEFAULT_PLATFORMS, coupon_id_to_platform
 
+_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)。"""
@@ -86,7 +89,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 +113,60 @@ 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 else None,
+        "point_total_count": point_stats["tried"] if point_stats else None,
         "trace_url": r.trace_url,
         "ad_revenue_yuan": ad_revenue_yuan,
     }
 
 
+def _point_scores_by_trace(db: Session, trace_ids: list[str]) -> dict[str, dict[str, int]]:
+    """聚合查询批量返回逐场点位分数，不加载逐券明细。"""
+    if not trace_ids:
+        return {}
+    succeeded = func.sum(case((CouponClaimRecord.status.in_(_SLOT_OK), 1), else_=0))
+    rows = db.execute(
+        select(
+            CouponClaimRecord.trace_id,
+            succeeded.label("succeeded"),
+            func.count().label("tried"),
+        )
+        .where(
+            CouponClaimRecord.trace_id.in_(trace_ids),
+            CouponClaimRecord.status.in_(_SLOT_TRIED),
+        )
+        .group_by(CouponClaimRecord.trace_id)
+    ).all()
+    return {
+        trace_id: {"succeeded": int(success_count or 0), "tried": int(tried or 0)}
+        for trace_id, success_count, tried in rows
+        if trace_id is not None
+    }
+
+
+def coupon_point_details(db: Session, *, trace_id: str) -> list[dict]:
+    """按单个 trace 查询逐券结果；仅在后台用户点击分数时调用。"""
+    rows = db.execute(
+        select(
+            CouponClaimRecord.coupon_id,
+            CouponClaimRecord.coupon_name,
+            CouponClaimRecord.status,
+            CouponClaimRecord.reason,
+        )
+        .where(CouponClaimRecord.trace_id == trace_id)
+        .order_by(CouponClaimRecord.id)
+    ).all()
+    return [
+        {
+            "coupon_id": coupon_id,
+            "coupon_name": coupon_name,
+            "status": status,
+            "reason": reason,
+        }
+        for coupon_id, coupon_name, status, reason in rows
+    ]
+
+
 def _empty_result() -> dict:
     return {
         "summary": {
@@ -249,10 +307,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_scores_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,
@@ -276,15 +341,17 @@ def coupon_user_records(db: Session, *, user_id: int, limit: int = 100) -> dict:
     ).scalar_one()
     rev_map = crud_ecpm.revenue_yuan_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),
+            )
+            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/repositories/stats.py

@@ -6,10 +6,10 @@ user.last_login_at / comparison_record.status / withdraw_order.status)要加索
 from __future__ import annotations
 
 from collections import Counter
-from datetime import date, datetime, time, timedelta, timezone
+from datetime import UTC, date, datetime, time, timedelta, timezone
 from decimal import ROUND_HALF_UP, Decimal, InvalidOperation
 
-from sqlalchemy import case, func, select
+from sqlalchemy import case, func, or_, select
 from sqlalchemy.orm import Session
 
 from app.admin.repositories.coupon_data import _percentile
@@ -37,14 +37,13 @@ REWARD_VIDEO_BIZ_TYPES = ("reward_video", "ad_reward")
 # ad_reward 是激励视频,单独成桶、不再混进领券奖励(历史误并会把激励视频金币双计进领券)。
 COUPON_REWARD_BIZ_TYPES = ("coupon", "coupon_reward")
 COMPARISON_REWARD_BIZ_TYPES = ("comparison", "compare_reward", "comparison_reward")
-EXCLUDED_REWARD_BIZ_TYPES = ("invite_inviter", "invite_invitee", "admin_grant")
+# 常规任务必须按明确来源相加；不能从全部正向流水反减排除项，否则新增广告/运营
-UNCLASSIFIED_FEED_BIZ_TYPES = ("feed_ad_reward",)
+# biz_type 时会在排除清单更新前自动混入该桶。task_ 前缀在查询处单独覆盖现有及未来任务。
-REGULAR_TASK_EXCLUDED_BIZ_TYPES = (
+REGULAR_TASK_EXACT_BIZ_TYPES = (
-    *REWARD_VIDEO_BIZ_TYPES,
+    "signin",
-    *COUPON_REWARD_BIZ_TYPES,
+    "signin_boost",
-    *COMPARISON_REWARD_BIZ_TYPES,
+    "price_report_reward",
-    *EXCLUDED_REWARD_BIZ_TYPES,
+    "feedback_reward",
-    *UNCLASSIFIED_FEED_BIZ_TYPES,
 )
 MEITUAN_CPS_INVALID_STATUSES = ("4", "5")
 MEITUAN_CPS_SETTLED_STATUS = "6"
@@ -61,7 +60,7 @@ def _beijing_today_start_utc() -> datetime:
     """北京时间今天 0 点对应的 UTC 时刻(DAU / 今日新增按北京时区切天)。"""
     now_bj = datetime.now(_BEIJING)
     start_bj = now_bj.replace(hour=0, minute=0, second=0, microsecond=0)
-    return start_bj.astimezone(timezone.utc)
+    return start_bj.astimezone(UTC)
 
 
 def today_dau(db: Session) -> int:
@@ -94,8 +93,8 @@ def _period_bounds(date_from: date, date_to: date) -> tuple[datetime, datetime,
     """
     start_bj = datetime.combine(date_from, time.min, tzinfo=_BEIJING)
     end_bj = datetime.combine(date_to + timedelta(days=1), time.min, tzinfo=_BEIJING)
-    start_utc = start_bj.astimezone(timezone.utc)
+    start_utc = start_bj.astimezone(UTC)
-    end_utc = end_bj.astimezone(timezone.utc)
+    end_utc = end_bj.astimezone(UTC)
     return (
         start_utc,
         end_utc,
@@ -505,7 +504,10 @@ def dashboard_overview(
     period_regular_task_coin_total = _sum(
         CoinTransaction.amount,
         *period_coin_conds,
-        CoinTransaction.biz_type.notin_(REGULAR_TASK_EXCLUDED_BIZ_TYPES),
+        or_(
+            CoinTransaction.biz_type.in_(REGULAR_TASK_EXACT_BIZ_TYPES),
+            CoinTransaction.biz_type.like(r"task\_%", escape="\\"),
+        ),
     )
     period_cps_orders = list(
         db.execute(

app/admin/routers/coupon_data.py

@@ -18,6 +18,8 @@ from app.admin.schemas.coupon_data import (
     CouponDataOut,
     CouponDataRow,
     CouponDataSummary,
+    CouponPointDetail,
+    CouponPointDetailsOut,
     CouponSlotRow,
     CouponSlotsOut,
     CouponUserRecordsOut,
@@ -122,6 +124,22 @@ def get_coupon_slots(
     )
 
 
+@router.get(
+    "/point-details",
+    response_model=CouponPointDetailsOut,
+    summary="按 trace 查询单次领券任务的逐券点位明细",
+)
+def get_coupon_point_details(
+    db: AdminDb,
+    trace_id: Annotated[str, Query(min_length=1, max_length=64, description="领券 trace_id")],
+) -> CouponPointDetailsOut:
+    items = coupon_data.coupon_point_details(db, trace_id=trace_id)
+    return CouponPointDetailsOut(
+        trace_id=trace_id,
+        items=[CouponPointDetail(**item) for item in items],
+    )
+
+
 @router.get(
     "/user-records",
     response_model=CouponUserRecordsOut,

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,12 @@ 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);无逐券埋点为空"
+    )
     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"
@@ -89,6 +104,13 @@ class CouponDataOut(BaseModel):
     items: list[CouponDataRow] = Field(..., description="逐条领券明细(当前页)")
 
 
+class CouponPointDetailsOut(BaseModel):
+    """单次领券任务的逐券点位结果，供点击分数时按需加载。"""
+
+    trace_id: str
+    items: list[CouponPointDetail] = Field(default_factory=list)
+
+
 class CouponUserRecordsOut(BaseModel):
     """某用户全部领券记录(点手机号抽屉用):total=该用户领券总次数,items=记录列表(UserRecordsDrawer 渲染)。"""
 

tests/test_admin_read.py

@@ -455,3 +455,62 @@ def test_period_coupon_reward_excludes_reward_video(
     coins = r.json()["period"]["coins"]
     assert coins["coupon_reward_coin_total"] == 0        # 激励视频不计入领券奖励
     assert coins["reward_video_coin_total"] == 50        # 仍计入激励视频卡
+
+
+def test_period_regular_task_coin_uses_explicit_allowlist(
+    admin_client: TestClient, admin_token: str
+) -> None:
+    """常规任务金币只加明确任务来源，领券/比价及未知新类型不得自动混入。"""
+    from datetime import datetime
+
+    from app.models.wallet import CoinTransaction
+
+    d = "2021-06-18"
+    included = {
+        "signin": 100,
+        "signin_boost": 200,
+        "task_enable_notification": 300,
+        "task_other": 400,
+        "price_report_reward": 500,
+        "feedback_reward": 600,
+    }
+    excluded = {
+        "feed_ad_reward_coupon": 700,
+        "feed_ad_reward_comparison": 800,
+        "feed_ad_reward": 900,
+        "reward_video": 1000,
+        "admin_grant": 1100,
+        "invite_inviter": 1200,
+        "future_unknown_reward": 1300,
+    }
+    db = SessionLocal()
+    try:
+        uid = user_repo.upsert_user_for_login(
+            db, phone="13800008804", register_channel="sms"
+        ).id
+        balance = 0
+        rows = []
+        for index, (biz_type, amount) in enumerate((included | excluded).items(), start=1):
+            balance += amount
+            rows.append(CoinTransaction(
+                user_id=uid,
+                amount=amount,
+                balance_after=balance,
+                biz_type=biz_type,
+                ref_id=f"regular-task-{index}",
+                created_at=datetime(2021, 6, 18, 12, 0, index),
+            ))
+        db.add_all(rows)
+        db.commit()
+    finally:
+        db.close()
+
+    response = admin_client.get(
+        "/admin/api/stats/overview",
+        params={"date_from": d, "date_to": d},
+        headers=_auth(admin_token),
+    )
+    assert response.status_code == 200, response.text
+    coins = response.json()["period"]["coins"]
+    assert coins["regular_task_coin_total"] == sum(included.values())
+    assert coins["task_coin_total"] == 700

tests/test_coupon_point_score.py

@@ -0,0 +1,161 @@
+"""admin 领券明细逐场点位分数与按需明细。"""
+from datetime import UTC, date, datetime
+
+from fastapi.testclient import TestClient
+from sqlalchemy import delete
+
+from app.admin.main import admin_app
+from app.admin.repositories import admin_user as admin_repo
+from app.admin.repositories.coupon_data import (
+    _point_scores_by_trace,
+    coupon_data_report,
+    coupon_point_details,
+)
+from app.admin.security import create_admin_token
+from app.db.session import SessionLocal
+from app.models.coupon_state import CouponClaimRecord, CouponSession
+
+
+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_scores_by_trace(db, [trace])[trace]
+        assert stats["succeeded"] == 2
+        assert stats["tried"] == 3
+        details = coupon_point_details(db, trace_id=trace)
+        assert [item["status"] for item in details] == [
+            "success", "already_claimed", "failed", "skipped"
+        ]
+        assert details[2]["reason"] == "测试失败"
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_skipped_detail_does_not_create_a_score() -> None:
+    """仅有 skipped 时按需明细仍可查到，但列表没有虚假的 0/0 分数。"""
+    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()
+
+        scores = _point_scores_by_trace(db, [trace, "missing-trace"])
+        assert trace not in scores
+        assert "missing-trace" not in scores
+        assert coupon_point_details(db, trace_id=trace)[0]["status"] == "skipped"
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_coupon_data_report_returns_scores_without_embedding_details() -> None:
+    """主列表只返回聚合分数，逐券记录必须走按 trace 的明细查询。"""
+    db = SessionLocal()
+    trace = "point-score-report"
+    report_date = date(2020, 1, 4)
+    try:
+        db.add(CouponSession(
+            trace_id=trace,
+            device_id="score-report-device",
+            status="completed",
+            app_env="prod",
+            platforms=["meituan-waimai"],
+            started_at=datetime(2020, 1, 4, tzinfo=UTC),
+            started_date=report_date,
+        ))
+        db.add_all([
+            CouponClaimRecord(
+                device_id="score-report-device",
+                coupon_id=f"mt-report-{status}",
+                claim_date=report_date,
+                status=status,
+                trace_id=trace,
+            )
+            for status in ("success", "failed")
+        ])
+        db.flush()
+
+        report = coupon_data_report(
+            db,
+            date_from=report_date.isoformat(),
+            date_to=report_date.isoformat(),
+            app_env="prod",
+        )
+        row = next(item for item in report["items"] if item["trace_id"] == trace)
+        assert row["point_success_count"] == 1
+        assert row["point_total_count"] == 2
+        assert "point_details" not in row
+        assert len(coupon_point_details(db, trace_id=trace)) == 2
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_coupon_point_details_endpoint() -> None:
+    """前端点击使用的接口按约定返回 trace_id 和逐券 items。"""
+    db = SessionLocal()
+    trace = "point-details-endpoint"
+    try:
+        admin = admin_repo.get_by_username(db, "point_details_admin")
+        if admin is None:
+            admin = admin_repo.create_admin(
+                db,
+                username="point_details_admin",
+                password="pass1234",
+                role="super_admin",
+            )
+        token, _expires_at = create_admin_token(admin_id=admin.id, role=admin.role)
+        db.add(CouponClaimRecord(
+            device_id="point-details-endpoint-device",
+            coupon_id="mt-point-details-endpoint",
+            coupon_name="接口测试券",
+            claim_date=date(2020, 1, 5),
+            status="failed",
+            reason="接口测试失败",
+            trace_id=trace,
+        ))
+        db.commit()
+
+        response = TestClient(admin_app).get(
+            "/admin/api/coupon-data/point-details",
+            params={"trace_id": trace},
+            headers={"Authorization": f"Bearer {token}"},
+        )
+        assert response.status_code == 200, response.text
+        assert response.json() == {
+            "trace_id": trace,
+            "items": [{
+                "coupon_id": "mt-point-details-endpoint",
+                "coupon_name": "接口测试券",
+                "status": "failed",
+                "reason": "接口测试失败",
+            }],
+        }
+    finally:
+        db.rollback()
+        db.execute(delete(CouponClaimRecord).where(CouponClaimRecord.trace_id == trace))
+        db.commit()
+        db.close()