feat(ad-revenue): 接入穿山甲 GroMore 数据 API 拉取后台收益(预估+收益API)
- 新增穿山甲 GroMore「聚合数据报告 API」对接,按天 T+1 拉取后台 revenue(预估)与 api_revenue(收益API,更接近结算),在广告收益报表大盘 + 按天趋势级展示,与客户端自报 eCPM 折算的预估并列对照;逐条广告事件行不动(仍是客户端预估)。 - 链路:integrations/pangle_report.py(MD5 签名,与官方文档两个测试向量逐字节一致)→ scripts/sync_pangle_revenue.py(拉昨天/--days 回补)→ 新表 ad_pangle_daily_revenue (repositories/ad_pangle_revenue.py:upsert + 按日聚合)→ admin/repositories/ad_revenue.py 汇总,新增 total_pangle_revenue_yuan / total_pangle_api_revenue_yuan / daily[].pangle_*。 - 穿山甲无用户/类型/场景维度:仅全量视图(未按 user/ad_type/feed_scene 过滤)给值,否则置 None;join key 用 ad_unit_id(=客户端配的 104xxx),非 code_id。 - 新增配置 PANGLE_REPORT_USER_ID/ROLE_ID/SECURITY_KEY(≠发奖 m-key)+ site_id→应用映射; 含单测(签名向量+分页解析);已真连穿山甲验证。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
zzhyyyyy
@@ -0,0 +1,151 @@
|
||||
"""每日拉取穿山甲 GroMore 天级收益报表入库(供 admin 广告收益报表的「穿山甲后台收益」)。
|
||||
|
||||
GroMore 数据 API 为 T+1:次日穿山甲约 10:00 出数。建议线上每天 ~10:30 由 systemd timer 跑一次
|
||||
(默认拉【昨天】);穿山甲对历史数据可能订正,故支持回补近 N 天(幂等 upsert,重跑无害)。
|
||||
|
||||
用法:
|
||||
python -m scripts.sync_pangle_revenue # 拉昨天(北京时间)
|
||||
python -m scripts.sync_pangle_revenue --days 3 # 回补昨天起往前 3 天
|
||||
python -m scripts.sync_pangle_revenue --date 2026-06-27 # 指定单天
|
||||
python -m scripts.sync_pangle_revenue --start 2026-06-01 --end 2026-06-27 # 指定区间
|
||||
|
||||
约束(来自官方文档):天级跨度 ≤ 1 个月、不早于 12 个月;「今天」与「今天以前」必须分开查,
|
||||
本脚本默认只拉昨天及更早,不混查今天。未配置 PANGLE_REPORT_* 时直接 no-op 退出。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
from datetime import date, timedelta
|
||||
|
||||
# Windows 控制台按 UTF-8 输出中文/¥
|
||||
try:
|
||||
sys.stdout.reconfigure(encoding="utf-8") # type: ignore[attr-defined]
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
|
||||
from app.core.config import settings
|
||||
from app.core.rewards import cn_today
|
||||
from app.db.session import SessionLocal
|
||||
from app.integrations import pangle_report
|
||||
from app.integrations.pangle_report import PangleReportError
|
||||
from app.repositories import ad_pangle_revenue as repo
|
||||
|
||||
# 拉取维度固定:date×site_id×ad_unit_id。⚠️ 用 ad_unit_id(GroMore 广告位ID = 我们客户端配的 104xxx
|
||||
# = ad_ecpm_record.our_code_id)而**非** code_id —— 实测 code_id 返回的是底层各 ADN 的代码位
|
||||
# (如 983674557 / youlianghui_xinxiliu),对不上我们的口径。指标取预估 + Api 两个收益。
|
||||
_DIMENSIONS = "date,site_id,ad_unit_id"
|
||||
_METRICS = "revenue,api_revenue,ecpm,imp_cnt"
|
||||
|
||||
|
||||
def _to_float(v: object) -> float | None:
|
||||
"""接口数值字段都是字符串;空/非法 → None(api_revenue 缺失时即为 None,前端显示「-」)。"""
|
||||
if v is None or str(v).strip() == "":
|
||||
return None
|
||||
try:
|
||||
return float(str(v).strip())
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def _row_date(row: dict) -> str | None:
|
||||
"""报表行的日期:优先 date 维度,回退 start_date;取前 10 位 'YYYY-MM-DD'。"""
|
||||
raw = row.get("date") or row.get("start_date") or ""
|
||||
raw = str(raw).strip()
|
||||
return raw[:10] if len(raw) >= 10 else None
|
||||
|
||||
|
||||
def _normalize(rows: list[dict]) -> tuple[list[dict], int, set[str]]:
|
||||
"""接口原始行 → 入库行;返回 (入库行, 跳过数, 未映射 site_id 集合)。
|
||||
|
||||
site_id 经 settings 映射成 app_env(prod/test);未映射的仍入库(app_env=site:<id>)但提示,
|
||||
避免静默丢收益。缺 code_id / 日期的行跳过。
|
||||
"""
|
||||
site_map = settings.pangle_report_site_id_to_env
|
||||
out: list[dict] = []
|
||||
skipped = 0
|
||||
unmapped: set[str] = set()
|
||||
for row in rows:
|
||||
rdate = _row_date(row)
|
||||
# 用 ad_unit_id(GroMore 广告位ID = 我们的 104xxx),不是 code_id(底层 ADN 代码位)
|
||||
code_id = str(row.get("ad_unit_id") or "").strip()
|
||||
if not rdate or not code_id:
|
||||
skipped += 1
|
||||
continue
|
||||
site_id = str(row.get("site_id") or "").strip()
|
||||
app_env = site_map.get(site_id)
|
||||
if app_env is None:
|
||||
app_env = f"site:{site_id}" if site_id else "unknown"
|
||||
if site_id:
|
||||
unmapped.add(site_id)
|
||||
out.append({
|
||||
"report_date": rdate,
|
||||
"app_env": app_env,
|
||||
"site_id": site_id or None,
|
||||
"our_code_id": code_id,
|
||||
"adn": "",
|
||||
"revenue_yuan": _to_float(row.get("revenue")) or 0.0,
|
||||
"api_revenue_yuan": _to_float(row.get("api_revenue")),
|
||||
"ecpm": (str(row.get("ecpm")).strip() or None) if row.get("ecpm") is not None else None,
|
||||
"impressions": int(_to_float(row.get("imp_cnt")) or 0),
|
||||
"currency": "cny",
|
||||
})
|
||||
return out, skipped, unmapped
|
||||
|
||||
|
||||
def sync_range(start_date: str, end_date: str) -> None:
|
||||
if not settings.pangle_report_configured:
|
||||
print("PANGLE_REPORT_* 未配置,跳过(no-op)。需在后台「接入中心 → GroMore-API」领取 "
|
||||
"user_id/role_id/Security Key 并填入 .env。")
|
||||
return
|
||||
|
||||
print(f"拉取 GroMore 天级收益 {start_date} ~ {end_date} (dims={_DIMENSIONS}) ...")
|
||||
try:
|
||||
raw = pangle_report.fetch_daily_report(
|
||||
start_date=start_date, end_date=end_date,
|
||||
dimensions=_DIMENSIONS, metrics=_METRICS, time_zone=8,
|
||||
)
|
||||
except PangleReportError as e:
|
||||
print(f"❌ 拉取失败:{e}")
|
||||
sys.exit(1)
|
||||
|
||||
rows, skipped, unmapped = _normalize(raw)
|
||||
if unmapped:
|
||||
print(f"⚠️ 有未映射 site_id(收益仍入库,app_env=site:<id>):{sorted(unmapped)};"
|
||||
f"如需归入 prod/test,补 PANGLE_REPORT_SITE_ID_* 配置。")
|
||||
with SessionLocal() as db:
|
||||
stats = repo.upsert_daily_rows(db, rows)
|
||||
total_rev = round(sum(r["revenue_yuan"] for r in rows), 4)
|
||||
print(f"✅ 完成:接口 {len(raw)} 行 → 入库 {len(rows)} 行(跳过 {skipped}),"
|
||||
f"新增 {stats['inserted']} / 更新 {stats['updated']};预估收益合计 ¥{total_rev}")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
ap = argparse.ArgumentParser(description="拉取穿山甲 GroMore 天级收益报表入库")
|
||||
ap.add_argument("--date", help="指定单天 YYYY-MM-DD")
|
||||
ap.add_argument("--start", help="区间起始 YYYY-MM-DD(与 --end 同用)")
|
||||
ap.add_argument("--end", help="区间结束 YYYY-MM-DD(与 --start 同用)")
|
||||
ap.add_argument("--days", type=int, help="回补:昨天起往前 N 天(含昨天)")
|
||||
args = ap.parse_args()
|
||||
|
||||
yesterday = cn_today() - timedelta(days=1)
|
||||
if args.date:
|
||||
start = end = args.date
|
||||
elif args.start and args.end:
|
||||
start, end = args.start, args.end
|
||||
elif args.days:
|
||||
start = (yesterday - timedelta(days=args.days - 1)).isoformat()
|
||||
end = yesterday.isoformat()
|
||||
else:
|
||||
start = end = yesterday.isoformat()
|
||||
|
||||
# 防呆:跨度 > 31 天直接拒(接口上限 1 个月,超了会返回 114)
|
||||
if (date.fromisoformat(end) - date.fromisoformat(start)).days > 31:
|
||||
print("❌ 区间跨度超过 31 天(接口上限 1 个月),请分批拉。")
|
||||
sys.exit(1)
|
||||
|
||||
sync_range(start, end)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user