fix(ad): detach late impressions from abandoned coupons

app/api/v1/ad.py

@@ -280,13 +280,21 @@ def ecpm_report(payload: EcpmReportIn, user: CurrentUser, db: DbSession) -> Ecpm
     Bearer 鉴权,user_id 取自 JWT(不信 body)。best-effort:落库即 ok,客户端 fire-and-forget,
     丢一两条不影响业务(穿山甲后台报表是结算权威)。eCPM 与发奖(S2S)是两条独立流,不逐条关联。
     """
+    attributed_trace_id = crud_ecpm.attributable_trace_id(
+        db, feed_scene=payload.feed_scene, trace_id=payload.trace_id
+    )
+    if payload.trace_id and attributed_trace_id is None:
+        logger.info(
+            "detach late coupon ad impression from failed trace user_id=%d trace=%s session=%s",
+            user.id, payload.trace_id, payload.ad_session_id,
+        )
     crud_ecpm.create_ecpm_record(
         db, user.id,
         ad_type=payload.ad_type, ecpm_raw=payload.ecpm,
         ad_session_id=payload.ad_session_id,
         adn=payload.adn, slot_id=payload.slot_id,
         feed_scene=payload.feed_scene,
-        trace_id=payload.trace_id,
+        trace_id=attributed_trace_id,
         app_env=payload.app_env, our_code_id=payload.our_code_id,
     )
     logger.info(

app/repositories/ad_ecpm.py

@@ -13,6 +13,25 @@ from sqlalchemy.orm import Session
 from app.core import rewards
 from app.core.rewards import cn_today
 from app.models.ad_ecpm import AdEcpmRecord
+from app.models.coupon_state import CouponSession
+
+
+def attributable_trace_id(
+    db: Session, *, feed_scene: str | None, trace_id: str | None
+) -> str | None:
+    """返回广告展示允许归属的业务 trace。
+
+    领券任务可能在 Draw 广告异步加载完成前已经失败或被用户放弃。非完成终态先落库、
+    广告回调后到时，收益仍需保留在总广告报表中，但不能再挂到该死亡领券明细，
+    因此清空关联 trace。其它场景、找不到 session、进行中或已完成状态保持原值，
+    由客户端生命周期修复负责主防线。
+    """
+    if feed_scene != "coupon" or not trace_id:
+        return trace_id
+    session_status = db.execute(
+        select(CouponSession.status).where(CouponSession.trace_id == trace_id)
+    ).scalar_one_or_none()
+    return None if session_status in {"failed", "abandoned"} else trace_id
 
 
 def create_ecpm_record(

tests/test_ad_ecpm_trace_revenue.py

@@ -1,12 +1,13 @@
 """ad_ecpm_record.trace_id 落库 + 按 trace 聚合广告收益(元)。"""
 from __future__ import annotations
 
-from datetime import UTC, datetime
+from datetime import UTC, date, datetime
 
 from sqlalchemy import delete
 
 from app.db.session import SessionLocal
 from app.models.ad_ecpm import AdEcpmRecord
+from app.models.coupon_state import CouponSession
 from app.repositories import ad_ecpm as crud_ecpm
 
 
@@ -57,6 +58,39 @@ def test_revenue_yuan_by_trace_empty() -> None:
         db.close()
 
 
+def test_terminal_coupon_trace_is_not_attributable_to_late_impression() -> None:
+    """领券失败或被放弃后才到达的广告展示保留收益记录，但不再关联死亡 trace。"""
+    db = SessionLocal()
+    try:
+        db.add_all([
+            CouponSession(
+                trace_id="failed-before-ad", device_id="d-late-ad", status="failed", app_env="prod",
+                started_at=datetime(2020, 1, 2, tzinfo=UTC), started_date=date(2020, 1, 2),
+            ),
+            CouponSession(
+                trace_id="abandoned-before-ad", device_id="d-late-ad", status="abandoned", app_env="prod",
+                started_at=datetime(2020, 1, 2, tzinfo=UTC), started_date=date(2020, 1, 2),
+            ),
+        ])
+        db.flush()
+
+        assert crud_ecpm.attributable_trace_id(
+            db, feed_scene="coupon", trace_id="failed-before-ad"
+        ) is None
+        assert crud_ecpm.attributable_trace_id(
+            db, feed_scene="coupon", trace_id="abandoned-before-ad"
+        ) is None
+        assert crud_ecpm.attributable_trace_id(
+            db, feed_scene="comparison", trace_id="failed-before-ad"
+        ) == "failed-before-ad"
+        assert crud_ecpm.attributable_trace_id(
+            db, feed_scene="coupon", trace_id="unknown-trace"
+        ) == "unknown-trace"
+    finally:
+        db.rollback()
+        db.close()
+
+
 def test_create_ecpm_record_persists_trace_id() -> None:
     """create_ecpm_record 落 trace_id。"""
     db = SessionLocal()