fix(ad): eCPM 上报撞会话唯一约束不再抛 500,按全局口径兜底去重

create_ecpm_record 撞 uq_ad_ecpm_record_session(只含 ad_session_id、全局唯一)
时,改用与约束同口径的全局查找 _find_by_session_global(仅按 ad_session_id)兜底,
并去掉末尾的 raise。原逻辑兜底用带 user_id 的 find_by_session:不同 user 上报了
同一 ad_session_id 时二次查找仍为 None → raise → 接口 500,违背该接口
best-effort / fire-and-forget 约定(丢一两条不影响业务,穿山甲后台才是结算权威)。
find_by_session 签名与行为不变,另两个调用方(ad_reward、ad.py)不受影响。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

alembic/versions/7db22acee504_merge_invite_cash_analytics_device_heads.py

@@ -0,0 +1,26 @@
+"""merge invite_cash + analytics + device heads
+
+Revision ID: 7db22acee504
+Revises: c2874d2bf705, device_first_protected_at, invite_cash_compare_reward
+Create Date: 2026-06-27 03:06:52.594401
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '7db22acee504'
+down_revision: Union[str, Sequence[str], None] = ('c2874d2bf705', 'device_first_protected_at', 'invite_cash_compare_reward')
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass

app/repositories/ad_ecpm.py

@@ -53,11 +53,16 @@ def create_ecpm_record(
         db.commit()
     except IntegrityError:
         db.rollback()
-        if ad_session_id:
-            existing = find_by_session(db, user_id=user_id, ad_session_id=ad_session_id)
-            if existing is not None:
-                return existing
-        raise
+        # 撞唯一约束 uq_ad_ecpm_record_session(全局按 ad_session_id、不含 user_id):并发同会话重复上报,
+        # 或同一 ad_session_id 已被先到的上报占用。本接口 fire-and-forget、best-effort —— 丢一条不影响业务
+        # (穿山甲后台才是结算权威),绝不向客户端抛 500。兜底查找须与唯一约束**同口径**(只按 ad_session_id、
+        # 不带 user_id):否则不同 user 上报了同一 ad_session_id 时,带 user_id 的查找会漏掉那条别人的记录 →
+        # 旧逻辑在此 raise 成 500(本应静默吞掉)。
+        existing = _find_by_session_global(db, ad_session_id)
+        if existing is not None:
+            return existing
+        # 极少:rollback 后既存记录又查不到(并发删除 / 竞态)。吞掉、返回未入库的内存对象(调用方不读返回值)。
+        return rec
     db.refresh(rec)
     return rec
 
@@ -76,6 +81,20 @@ def find_by_session(
     ).scalar_one_or_none()
 
 
+def _find_by_session_global(db: Session, ad_session_id: str | None) -> AdEcpmRecord | None:
+    """按 ad_session_id **全局**查找(与唯一约束 uq_ad_ecpm_record_session 同口径,不含 user_id)。
+
+    仅 create_ecpm_record 撞约束后兜底用:此时撞的是全局会话约束,既存记录可能属于**另一个 user**,
+    带 user_id 的 find_by_session 会漏掉它、导致误判「查无 → raise 500」。其它业务查「某 user 的某次
+    展示 eCPM」仍用 find_by_session(带 user_id,语义更准),不走这里。
+    """
+    if not ad_session_id:
+        return None
+    return db.execute(
+        select(AdEcpmRecord).where(AdEcpmRecord.ad_session_id == ad_session_id)
+    ).scalar_one_or_none()
+
+
 def count_today(db: Session, user_id: int) -> int:
     """该用户今日(北京时间)上报的 eCPM 条数,排查/对账辅助用。"""
     return db.execute(

app/schemas/compare_record.py

@@ -11,7 +11,7 @@ from __future__ import annotations
 
 from datetime import datetime
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, field_validator
 
 
 # ===== 上报请求 =====
@@ -21,8 +21,36 @@ class ComparisonItemIn(BaseModel):
 
     name: str
     qty: int = 1
+    # specs 仅供比价记录展示(admin-web 详情 / app 记录页都按字符串数组渲染并 join)。
+    # pricebot(2026-06-27 嵌套规格统一)起 calibration 的 specs 可能是规格对象
+    # [{name, qty, sub_specs}] 而非字符串 → 下面的 before-validator 统一拍平成可读字符串,
+    # 兼容新旧两种输入、保持 list[str] 契约不变(下游零改动)。
+    # ⚠️ 两个坑都踩过, 必须"拍平"而非别的: ① 直接声明 list[str] 不拍平 → 对象 specs 整条
+    # 422 被拒、不入库(同下方 platform_results list→dict 同类事故); ② 仅放宽成裸 list 又会让
+    # 下游 join 出 "[object Object]"/对象 toString 的乱码。
     specs: list[str] | None = None
 
+    @field_validator("specs", mode="before")
+    @classmethod
+    def _flatten_specs(cls, v: object) -> object:
+        """pricebot 规格对象 [{name, qty, sub_specs}] → 可读字符串数组; 字符串元素原样保留;
+        非 list 原样交还(让 pydantic 照常报类型错)。嵌套规格拼成 '主项(子1,子2)'。"""
+        if not isinstance(v, list):
+            return v
+        out: list[str] = []
+        for it in v:
+            if isinstance(it, str):
+                s = it
+            elif isinstance(it, dict):
+                name = str(it.get("name") or "").strip()
+                subs = [str(x).strip() for x in (it.get("sub_specs") or []) if str(x).strip()]
+                s = f"{name}({','.join(subs)})" if name and subs else (name or ",".join(subs))
+            else:
+                continue
+            if s:
+                out.append(s)
+        return out
+
 
 class AppliedCouponIn(BaseModel):
     """单笔已用优惠(来自 comparison_results[].applied_coupons)。amount 单位:元、正数。"""