Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 4630fd017b | |||
| 70c2349950 |
@@ -0,0 +1,26 @@
|
||||
"""merge jd_cps_order_fields and coupon_session_origin_package heads
|
||||
|
||||
Revision ID: 761ef181ce7c
|
||||
Revises: coupon_session_origin_package, jd_cps_order_fields
|
||||
Create Date: 2026-07-01 13:52:16.068808
|
||||
|
||||
"""
|
||||
from typing import Sequence, Union
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = '761ef181ce7c'
|
||||
down_revision: Union[str, Sequence[str], None] = ('coupon_session_origin_package', 'jd_cps_order_fields')
|
||||
branch_labels: Union[str, Sequence[str], None] = None
|
||||
depends_on: Union[str, Sequence[str], None] = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
pass
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
pass
|
||||
@@ -136,12 +136,16 @@ def _feed_scene_matches(rec: AdFeedRewardRecord, scene: str | None) -> bool:
|
||||
"""该信息流记录是否落入请求的展示筛选 scene。
|
||||
- scene=="feed":ad_type in ("feed", NULL)(旧数据 NULL 视为 feed,向后兼容)
|
||||
- scene=="draw":ad_type=="draw"
|
||||
- scene=="feed_all":所有信息流(feed/draw/NULL 都要)——业务已全切 Draw 信息流,收益报表把「Draw 信息流」
|
||||
当作整个信息流口径(含历史误标 feed/NULL),用它避免筛选漏历史。
|
||||
- scene 为 None:不筛(两类都要)。
|
||||
"""
|
||||
if scene == "feed":
|
||||
return rec.ad_type in (None, "feed")
|
||||
if scene == "draw":
|
||||
return rec.ad_type == "draw"
|
||||
if scene == "feed_all":
|
||||
return True
|
||||
return True
|
||||
|
||||
|
||||
@@ -184,6 +188,7 @@ def _feed_rows(
|
||||
"record_id": rec.id,
|
||||
"user_id": rec.user_id,
|
||||
"ad_session_id": rec.ad_session_id,
|
||||
"trace_id": rec.trace_id,
|
||||
"app_env": rec.app_env,
|
||||
"our_code_id": rec.our_code_id,
|
||||
"created_at": rec.created_at,
|
||||
@@ -209,6 +214,7 @@ def _feed_rows(
|
||||
"record_id": rec.id,
|
||||
"user_id": rec.user_id,
|
||||
"ad_session_id": rec.ad_session_id,
|
||||
"trace_id": rec.trace_id,
|
||||
"app_env": rec.app_env,
|
||||
"our_code_id": rec.our_code_id,
|
||||
"created_at": rec.created_at,
|
||||
@@ -241,7 +247,7 @@ def audit_rows(
|
||||
rows: list[dict] = []
|
||||
if scene in (None, "reward_video"):
|
||||
rows.extend(_reward_video_rows(db, date=date, user_id=user_id))
|
||||
if scene in (None, "feed", "draw"):
|
||||
if scene in (None, "feed", "draw", "feed_all"):
|
||||
rows.extend(_feed_rows(db, date=date, user_id=user_id, scene=scene))
|
||||
return rows
|
||||
|
||||
|
||||
@@ -3,9 +3,11 @@
|
||||
只读。每行 = 一次广告事件(不再按用户聚合):
|
||||
- **激励视频**:一次观看 = 1 条展示(ad_ecpm)+ 1 条发奖(ad_reward),按 ad_session_id 合并成一行,
|
||||
直接给出 eCPM / 收益 + 状态 / 应发 / 实发 / 一致;点开看该条金币复算因子。
|
||||
- **信息流**:轮播每条展示各一行(impressionId 各自独立);整场发奖(ad_feed_reward,client_event_id)
|
||||
与逐条展示无法对应,单独成「纯发奖」行。
|
||||
- 兜底:有展示无发奖(中途关 / 未达发奖)、有发奖无展示(未上报 eCPM)都各自成行。
|
||||
- **信息流(比价/领券)**:一次比价 / 一次领券 = 一条整场发奖(ad_feed_reward)一行,给出 eCPM /
|
||||
发奖金币 + 应发 / 实发 / 一致;点开看金币复算因子。⚠️ draw 的逐条展示(ad_ecpm,impressionId 各自
|
||||
独立、与整场发奖无公共键、无法归到「哪一次」)**不再单独占行**(2026-07 按「一次比价/领券放一块」调整)——
|
||||
其展示数 / eCPM / 预估收益仍进全量统计(合计 / 趋势 / 分类大盘 / 穿山甲对照),只是主表不逐条铺开。
|
||||
- 兜底:激励视频有展示无发奖(中途关 / 未达发奖)、有发奖无展示(未上报 eCPM)仍各自成行。
|
||||
|
||||
展示与收益来自 ad_ecpm_record(收益 = eCPM元 ÷ 1000);应发 / 实发金币复用金币审计逐条复算
|
||||
(ad_audit.audit_rows,与正式发奖同一公式口径,不另写公式)。合计与对账在全量上统计,
|
||||
@@ -58,14 +60,6 @@ def _date_range(date_from: str, date_to: str) -> list[str]:
|
||||
_AUDIT_SCENES = {"reward_video", "feed", "draw"}
|
||||
|
||||
|
||||
def _event_ad_type(row: dict) -> str:
|
||||
"""纯发奖事件行的 ad_type:信息流行用 audit 带回的真实 ad_type(feed/draw),回退 feed;
|
||||
激励视频行恒 reward_video。不再用 scene 硬映射,避免把 draw 丢成 feed。"""
|
||||
if row["scene"] == "reward_video":
|
||||
return "reward_video"
|
||||
return row.get("ad_type") or "feed"
|
||||
|
||||
|
||||
# 发奖复算明细字段(展开下钻看「金币怎么算出来的」)——从 audit 行原样取这些 key。
|
||||
_REWARD_DETAIL_KEYS = (
|
||||
"record_id", "created_at", "status", "ecpm", "ecpm_factor", "units",
|
||||
@@ -108,9 +102,14 @@ def ad_revenue_report(
|
||||
# 同时保留全量列表,未被展示合并的成「纯发奖」事件。
|
||||
reward_by_session: dict[tuple[int, str], list[dict]] = {}
|
||||
all_reward_rows: list[dict] = []
|
||||
# 报表 ad_type 直接当 audit scene 用(取值一致);未知/无效 ad_type 不取发奖行。draw 在此被
|
||||
# 正确传成 scene="draw",audit 会按 ad_type 筛出 Draw 发奖,不再丢成 feed。
|
||||
audit_scene = ad_type if ad_type in _AUDIT_SCENES else None
|
||||
# 报表 ad_type → audit scene:reward_video/feed 直传;**draw(前端「Draw 信息流」)映射成 feed_all**
|
||||
# ——业务已全切 Draw,把「Draw 信息流」当作整个信息流口径(含历史误标 feed/NULL),否则筛选会漏历史。
|
||||
if ad_type == "draw":
|
||||
audit_scene = "feed_all"
|
||||
elif ad_type in _AUDIT_SCENES:
|
||||
audit_scene = ad_type
|
||||
else:
|
||||
audit_scene = None
|
||||
if ad_type is None or audit_scene is not None:
|
||||
for d in _date_range(date_from, date_to):
|
||||
for row in ad_audit.audit_rows(db, date=d, user_id=user_id, scene=audit_scene):
|
||||
@@ -140,7 +139,10 @@ def ad_revenue_report(
|
||||
)
|
||||
if user_id is not None:
|
||||
stmt = stmt.where(AdEcpmRecord.user_id == user_id)
|
||||
if ad_type is not None:
|
||||
if ad_type == "draw":
|
||||
# draw = 所有信息流展示(业务已全 Draw,含历史误标 feed);展示行只进统计,不占主表行
|
||||
stmt = stmt.where(AdEcpmRecord.ad_type.in_(["draw", "feed"]))
|
||||
elif ad_type is not None:
|
||||
stmt = stmt.where(AdEcpmRecord.ad_type == ad_type)
|
||||
for rec in db.execute(stmt).scalars():
|
||||
rwd = _pop_reward(rec.user_id, rec.ad_session_id)
|
||||
@@ -165,6 +167,8 @@ def ad_revenue_report(
|
||||
),
|
||||
"adn": rec.adn,
|
||||
"slot_id": rec.slot_id,
|
||||
"sub_rewards": [],
|
||||
"sub_count": 1,
|
||||
}
|
||||
if rwd is not None:
|
||||
ev.update({
|
||||
@@ -184,33 +188,88 @@ def ad_revenue_report(
|
||||
})
|
||||
events.append(ev)
|
||||
|
||||
# 3) 未被展示合并的发奖行 → 「纯发奖」事件(信息流整场发奖 / 有发奖无展示)。
|
||||
# 收益恒 0(收益只算展示侧,避免与展示行重复计)。
|
||||
# 3) 未被展示合并的发奖行 → 事件:
|
||||
# - 激励视频(reward_video):逐条成「纯发奖」事件(每次一个 ad_session_id;有发奖无展示等)。
|
||||
# - 信息流(feed/draw):同一次比价/领券的多条广告共享**整场 ad_session_id**(客户端整场复用),
|
||||
# 按 (user_id, ad_session_id) 聚成**一次比价 / 一次领券**父事件;sub_rewards 为组内逐条明细,
|
||||
# 应发/实发取组内合计;业务已全 Draw → 类型统一 "draw"。session 缺失(极少旧数据)各自单独成组。
|
||||
feed_groups: dict[tuple[int, str], list[dict]] = {}
|
||||
for row in all_reward_rows:
|
||||
if row["record_id"] in used_reward_ids:
|
||||
continue
|
||||
if row["scene"] == "reward_video":
|
||||
events.append({
|
||||
"event_key": f"rwd-{row['record_id']}",
|
||||
"report_date": row["_report_date"],
|
||||
"user_id": row["user_id"],
|
||||
"ad_type": "reward_video",
|
||||
"feed_scene": row.get("feed_scene"),
|
||||
"app_env": row.get("app_env"),
|
||||
"our_code_id": row.get("our_code_id"),
|
||||
"created_at": row["created_at"],
|
||||
"hour": _cn_hour(row["created_at"]) if by_hour else None,
|
||||
"has_impression": False,
|
||||
"impressions": 0,
|
||||
"ecpm": row["ecpm"],
|
||||
"revenue_yuan": 0.0,
|
||||
"adn": None,
|
||||
"slot_id": None,
|
||||
"has_reward": True,
|
||||
"status": row["status"],
|
||||
"expected_coin": int(row["expected_coin"]),
|
||||
"actual_coin": int(row["actual_coin"]),
|
||||
"matched": bool(row["matched"]),
|
||||
"reward_detail": _reward_detail(row),
|
||||
"sub_rewards": [],
|
||||
"sub_count": 1,
|
||||
})
|
||||
else:
|
||||
# 聚合单位 = 一次完整比价/领券流程:优先用 trace_id(比价带 comparisonTraceId、领券带 sessionTraceId,
|
||||
# 整个流程不变;即使中途点广告致浮层关闭重弹、ad_session_id 变了,trace_id 仍不变 → 全流程聚成一行)。
|
||||
# 无 trace_id(历史领券未上报 / 旧数据)回退整场 ad_session_id;再无则 record_id 各自成组、不误并。
|
||||
grp_key = row.get("trace_id") or row.get("ad_session_id") or f"_rid-{row['record_id']}"
|
||||
feed_groups.setdefault((row["user_id"], grp_key), []).append(row)
|
||||
|
||||
# 信息流分组 → 「一次比价 / 一次领券」父事件(收益恒 0:收益只算展示侧,避免与展示行重复计)。
|
||||
for (uid, grp_key), group in feed_groups.items():
|
||||
group.sort(key=lambda r: (r["created_at"], r["record_id"]))
|
||||
rep = group[-1] # 代表条(最新一条):时间/场景/应用/代码位取它
|
||||
expected_sum = sum(int(g["expected_coin"]) for g in group)
|
||||
actual_sum = sum(int(g["actual_coin"]) for g in group)
|
||||
# 父行 eCPM:组内各条 eCPM(分)均值(展示用,各条不同);无有效值则取代表条
|
||||
ecpm_fens = [rewards.parse_ecpm_fen(g["ecpm"]) for g in group if g.get("ecpm")]
|
||||
avg_ecpm = str(round(sum(ecpm_fens) / len(ecpm_fens))) if ecpm_fens else rep.get("ecpm")
|
||||
# 主表逐行显示用:这次发奖广告的预估收益之和(发奖侧 eCPM 折算,钳顶同展示侧)。只放进
|
||||
# row_revenue_yuan 给主表逐行展示,不进 revenue_yuan/合计/趋势——避免与展示侧 total 重复计。
|
||||
row_revenue = round(sum(
|
||||
min(rewards.parse_ecpm_yuan(g["ecpm"]), rewards.AD_ECPM_MAX_FEN / 100.0) / 1000.0
|
||||
for g in group if g.get("ecpm")
|
||||
), 6)
|
||||
events.append({
|
||||
"event_key": f"rwd-{row['record_id']}",
|
||||
"report_date": row["_report_date"],
|
||||
"user_id": row["user_id"],
|
||||
"ad_type": _event_ad_type(row),
|
||||
"feed_scene": row.get("feed_scene"),
|
||||
"app_env": row.get("app_env"),
|
||||
"our_code_id": row.get("our_code_id"),
|
||||
"created_at": row["created_at"],
|
||||
"hour": _cn_hour(row["created_at"]) if by_hour else None,
|
||||
"event_key": f"feedgrp-{uid}-{grp_key}",
|
||||
"report_date": rep["_report_date"],
|
||||
"user_id": uid,
|
||||
"ad_type": "draw", # 业务已全切 Draw 信息流,聚合行统一 draw
|
||||
"feed_scene": rep.get("feed_scene"),
|
||||
"app_env": rep.get("app_env"),
|
||||
"our_code_id": rep.get("our_code_id"),
|
||||
"created_at": rep["created_at"],
|
||||
"hour": _cn_hour(rep["created_at"]) if by_hour else None,
|
||||
"has_impression": False,
|
||||
"impressions": 0,
|
||||
"ecpm": row["ecpm"],
|
||||
"ecpm": avg_ecpm,
|
||||
"revenue_yuan": 0.0,
|
||||
"row_revenue_yuan": row_revenue,
|
||||
"adn": None,
|
||||
"slot_id": None,
|
||||
"has_reward": True,
|
||||
"status": row["status"],
|
||||
"expected_coin": int(row["expected_coin"]),
|
||||
"actual_coin": int(row["actual_coin"]),
|
||||
"matched": bool(row["matched"]),
|
||||
"reward_detail": _reward_detail(row),
|
||||
"status": rep["status"], # 代表状态(逐条见展开)
|
||||
"expected_coin": expected_sum,
|
||||
"actual_coin": actual_sum,
|
||||
"matched": all(bool(g["matched"]) for g in group),
|
||||
"reward_detail": None,
|
||||
"sub_rewards": [_reward_detail(g) for g in group],
|
||||
"sub_count": len(group),
|
||||
})
|
||||
|
||||
# 「场景」作为全局筛选(与 user_id/ad_type 一致):同时作用于明细、合计与 daily/hourly 趋势。
|
||||
@@ -331,9 +390,20 @@ def ad_revenue_report(
|
||||
is_today = date_from == date_to == rewards.cn_today().isoformat()
|
||||
dau = admin_stats.today_dau(db) if is_today else None
|
||||
|
||||
# 主表「逐行」= 单次广告行为(2026-07 按「一次比价/领券放一块」聚合):激励视频 = 一次观看一行(展示+发奖
|
||||
# 按 ad_session_id 合并);一次比价 / 一次领券 = 该次整场多条广告按 ad_session_id 聚成一行(展开看逐条)。
|
||||
# 信息流(draw/feed)的逐条展示(ad_ecpm,impressionId 各自独立、与整场发奖无公共键)不再单独占行
|
||||
# ——其展示数 / eCPM / 预估收益已计入上面的全量统计(total_*、daily / hourly、type_stats、穿山甲对照),
|
||||
# 只是主表不逐条铺开;逐条明细在父行展开里看(sub_rewards)。合计 / 趋势 / 分类大盘均基于全量 events,
|
||||
# 不受此过滤影响;total / 分页只作用于主表行。
|
||||
main_rows = [
|
||||
e for e in events
|
||||
if not (e["ad_type"] in ("draw", "feed") and e["has_impression"] and not e["has_reward"])
|
||||
]
|
||||
|
||||
return {
|
||||
"total": len(events),
|
||||
"truncated": len(events) > offset + limit,
|
||||
"total": len(main_rows),
|
||||
"truncated": len(main_rows) > offset + limit,
|
||||
"total_impressions": total_impressions,
|
||||
"total_revenue_yuan": total_revenue_yuan,
|
||||
# 穿山甲后台收益合计(元):预估 revenue + 收益Api;非全量视图(带 user/类型/场景过滤)或无数据为 None。
|
||||
@@ -347,5 +417,5 @@ def ad_revenue_report(
|
||||
"hourly": hourly,
|
||||
"type_stats": type_stats,
|
||||
"dau": dau,
|
||||
"items": events[offset:offset + limit],
|
||||
"items": main_rows[offset:offset + limit],
|
||||
}
|
||||
|
||||
@@ -806,6 +806,12 @@ def get_user_overview(db: Session, user_id: int) -> dict | None:
|
||||
}
|
||||
|
||||
|
||||
def _as_utc_naive(value: datetime) -> datetime:
|
||||
"""窗口入参 → UTC naive(= _as_utc 去时区),与库里按 naive UTC 存取的 created_at 同口径比较。
|
||||
历史遗留:_window_conds 一直引用本函数却未定义(自定义区间会 NameError),此处补上。"""
|
||||
return _as_utc(value).replace(tzinfo=None)
|
||||
|
||||
|
||||
def _window_conds(col, date_from: datetime | None, date_to: datetime | None) -> list:
|
||||
"""把 [date_from, date_to] 转成对 col(created_at)的过滤条件;都为 None = 全量(注册至今)。"""
|
||||
conds = []
|
||||
@@ -895,6 +901,13 @@ def user_reward_stats(
|
||||
}
|
||||
|
||||
|
||||
def _cn_wall_to_utc(dt: datetime) -> datetime:
|
||||
"""coin_transaction 存的是北京 wall-clock(naive,见 wallet.grant_coins「存北京 wall-clock」),转成 UTC naive,
|
||||
与广告表(func.now() UTC)统一 —— 让本函数按同一绝对时刻排序、且前端 apiTime(把无时区时间当 UTC 再 +8 展示)
|
||||
口径一致;否则签到会比实际多显示 8 小时(北京时间又被 +8)。"""
|
||||
return dt.replace(tzinfo=rewards.CN_TZ).astimezone(timezone.utc).replace(tzinfo=None)
|
||||
|
||||
|
||||
def user_coin_records(
|
||||
db: Session,
|
||||
user_id: int,
|
||||
@@ -915,6 +928,9 @@ def user_coin_records(
|
||||
offset = max(cursor or 0, 0)
|
||||
fetch = offset + limit + 1
|
||||
rows: list[dict] = []
|
||||
# coin_transaction 存北京 wall-clock(其余表存 UTC);签到窗口边界 +8h 对齐北京,过滤/计数才不偏移 8 小时
|
||||
signin_from = date_from + timedelta(hours=8) if date_from is not None else None
|
||||
signin_to = date_to + timedelta(hours=8) if date_to is not None else None
|
||||
|
||||
for rec in db.execute(
|
||||
select(AdRewardRecord)
|
||||
@@ -958,7 +974,7 @@ def user_coin_records(
|
||||
.where(
|
||||
CoinTransaction.user_id == user_id,
|
||||
CoinTransaction.biz_type == "signin",
|
||||
*_window_conds(CoinTransaction.created_at, date_from, date_to),
|
||||
*_window_conds(CoinTransaction.created_at, signin_from, signin_to),
|
||||
)
|
||||
.order_by(CoinTransaction.created_at.desc())
|
||||
.limit(fetch)
|
||||
@@ -966,7 +982,8 @@ def user_coin_records(
|
||||
rows.append({
|
||||
"source": "signin",
|
||||
"source_label": "签到",
|
||||
"created_at": rec.created_at,
|
||||
# 北京 wall-clock → UTC,与广告记录统一(前端 apiTime 会 +8 回北京展示,不然签到会多 8 小时)
|
||||
"created_at": _cn_wall_to_utc(rec.created_at),
|
||||
"ecpm": None,
|
||||
"coin": rec.amount,
|
||||
})
|
||||
@@ -992,7 +1009,7 @@ def user_coin_records(
|
||||
+ _count(
|
||||
CoinTransaction, CoinTransaction.user_id == user_id,
|
||||
CoinTransaction.biz_type == "signin",
|
||||
*_window_conds(CoinTransaction.created_at, date_from, date_to),
|
||||
*_window_conds(CoinTransaction.created_at, signin_from, signin_to),
|
||||
)
|
||||
)
|
||||
return rows[offset:offset + limit], (offset + limit if has_more else None), total
|
||||
|
||||
@@ -94,6 +94,11 @@ class AdRevenueRow(BaseModel):
|
||||
impressions: int = Field(..., description="本行展示条数:有展示=1 / 纯发奖=0(供日汇总、趋势图复用)")
|
||||
ecpm: str | None = Field(None, description="eCPM 原始值(分/千次);展示行取展示值,纯发奖行取发奖采用值")
|
||||
revenue_yuan: float = Field(..., description="本次展示预估收益(元)= eCPM元 ÷ 1000;纯发奖行=0")
|
||||
row_revenue_yuan: float | None = Field(
|
||||
None,
|
||||
description="主表逐行展示用的预估收益(元):一次比价/领券聚合行=该次发奖广告 eCPM 折算之和;"
|
||||
"其它行为空(前端回退取 revenue_yuan)。不进合计/趋势,避免与展示侧重复计",
|
||||
)
|
||||
adn: str | None = Field(None, description="实际填充 ADN 子渠道(pangle/gdt…);纯发奖行为空")
|
||||
slot_id: str | None = Field(None, description="底层 mediation rit(非我们配置的广告位 ID);纯发奖行为空")
|
||||
# ── 发奖侧 ──
|
||||
@@ -106,6 +111,15 @@ class AdRevenueRow(BaseModel):
|
||||
None,
|
||||
description="发奖复算明细(eCPM/因子1/份数/LT/因子2/应发/实发/一致);点行展开下钻用,纯展示为空",
|
||||
)
|
||||
sub_rewards: list[AdRevenueRecord] = Field(
|
||||
default_factory=list,
|
||||
description="一次比价/领券聚合行的组内逐条发奖明细(同一整场 ad_session_id 的多条广告);"
|
||||
"点行展开渲染多行。激励视频/纯展示行为空(单条看 reward_detail)",
|
||||
)
|
||||
sub_count: int = Field(
|
||||
1,
|
||||
description="本行聚合的发奖条数:一次比价/领券=该次广告条数(≥1);激励视频/纯展示=1",
|
||||
)
|
||||
|
||||
|
||||
class AdRevenueReportOut(BaseModel):
|
||||
|
||||
@@ -17,6 +17,7 @@ class AdminUserListItem(BaseModel):
|
||||
status: str
|
||||
debug_trace_enabled: bool = False
|
||||
wechat_openid: str | None = None
|
||||
wechat_nickname: str | None = None
|
||||
created_at: datetime
|
||||
last_login_at: datetime
|
||||
|
||||
|
||||
Reference in New Issue
Block a user