zhuzihao
b4c27f4d88
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> --------- Co-authored-by: zzhyyyyy <2685922758@qq.com> Reviewed-on: #85 Co-authored-by: zhuzihao <zhuzihao@wonderable.ai> Co-committed-by: zhuzihao <zhuzihao@wonderable.ai>
108 lines
4.1 KiB
Python
108 lines
4.1 KiB
Python
"""广告 eCPM 上报 CRUD(内部收益统计/对账)。
|
|
|
|
客户端在广告展示后(onAdShow)读到 eCPM,经鉴权接口上报,这里落库。鉴权接口已确保
|
|
user 存在(JWT),故不做 UnknownUser 校验。best-effort 上报:丢一两条不影响业务,
|
|
穿山甲后台报表是结算权威兜底。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.rewards import cn_today
|
|
from app.models.ad_ecpm import AdEcpmRecord
|
|
|
|
|
|
def create_ecpm_record(
|
|
db: Session,
|
|
user_id: int,
|
|
*,
|
|
ad_type: str,
|
|
ecpm_raw: str,
|
|
ad_session_id: str | None = None,
|
|
adn: str | None = None,
|
|
slot_id: str | None = None,
|
|
feed_scene: str | None = None,
|
|
app_env: str | None = None,
|
|
our_code_id: str | None = None,
|
|
) -> AdEcpmRecord:
|
|
"""落一条 eCPM 上报记录。report_date 用北京时间当天,供按天聚合。
|
|
|
|
app_env(prod/test)与 our_code_id(我们后台配置的 104xxx 代码位)供广告收益报表按
|
|
应用/代码位聚合;与 adn(实际填充子渠道)/slot_id(底层 rit)是两组不同口径。
|
|
"""
|
|
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
|
|
rec = AdEcpmRecord(
|
|
user_id=user_id,
|
|
ad_type=ad_type,
|
|
ad_session_id=ad_session_id,
|
|
adn=adn,
|
|
slot_id=slot_id,
|
|
feed_scene=feed_scene,
|
|
app_env=app_env,
|
|
our_code_id=our_code_id,
|
|
ecpm_raw=ecpm_raw,
|
|
report_date=cn_today().isoformat(),
|
|
)
|
|
db.add(rec)
|
|
try:
|
|
db.commit()
|
|
except IntegrityError:
|
|
db.rollback()
|
|
# 撞唯一约束 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
|
|
|
|
|
|
def find_by_session(
|
|
db: Session, *, user_id: int, ad_session_id: str | None
|
|
) -> AdEcpmRecord | None:
|
|
"""按广告会话找 eCPM。旧客户端无 ad_session_id 时返回 None。"""
|
|
if not ad_session_id:
|
|
return None
|
|
return db.execute(
|
|
select(AdEcpmRecord).where(
|
|
AdEcpmRecord.user_id == user_id,
|
|
AdEcpmRecord.ad_session_id == ad_session_id,
|
|
)
|
|
).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(
|
|
select(func.count())
|
|
.select_from(AdEcpmRecord)
|
|
.where(
|
|
AdEcpmRecord.user_id == user_id,
|
|
AdEcpmRecord.report_date == cn_today().isoformat(),
|
|
)
|
|
).scalar_one()
|