shaguabijia-app-server/app/integrations/pangle_report.py

"""穿山甲 GroMore「聚合数据报告 API」客户端 —— 按天拉取收益报表(只读、T+1)。

⚠️ 这是 **GroMore 数据拉取 API**,与发奖回调验签(integrations/pangle.py 的 m-key)是
两套完全不同的凭证与用途。凭证在后台「接入中心 → GroMore-API → 聚合数据报告 API」领取:
媒体账号 user_id、子账号 role_id、Security Key(secure_key)。

鉴权(文档 v2.x「方法一」):
  1. 去掉请求参数里的 `sign` 字段与值为空的字段;
  2. 其余参数按 key 字典序升序,拼成 `k1=v1&k2=v2&...&kn=vn`;
  3. 末尾直接拼接 security_key(无分隔符),对整串做 MD5,取 32 位小写十六进制 = sign。
  签名有 3 分钟过期(天级用 `timestamp` 秒级时间戳;小时级才用 `current_time` 字符串)。

数据口径要点(来自官方文档):
  - 只返回【GroMore 聚合代码位】在 GroMore 内的数据(=我们 useMediation 的口径),
    查不到穿山甲 SDK 自身的数据;
  - **不提供分用户/设备维度**(官方 FAQ 明确拒绝),最细到 日期×应用×代码位×广告源;
  - `revenue` = 预估收益(元,所有 ADN 都有);`api_revenue` = 收益Api(各 ADN 经 Reporting
    回传、按实时汇率折算账号币种,更接近结算),需后台为该 ADN 配置 Reporting 才有、且不支持当天;
  - 「今天」与「今天以前」必须分开查;天级跨度 ≤ 1 个月、不早于 12 个月。
"""
from __future__ import annotations

import hashlib
import logging
import time
from collections.abc import Mapping
from typing import Any

import httpx

from app.core.config import settings

logger = logging.getLogger("shagua.pangle_report")

HOST = "https://www.csjplatform.com"
# 天级收益报表(路径与文档代码示例一致;另有小时级 get_hour_report_data,本服务只用天级)。
DAILY_PATH = "/union_media/open/api/mediation/get_daily_income_report_data"
VERSION = "2.0"
SIGN_TYPE = "MD5"
PAGE_LIMIT = 5000          # 文档上限;一次尽量多取,减少翻页
DEFAULT_TIMEOUT = 30.0
_MAX_PAGES = 200           # 翻页安全阀(5000×200=100w 行,远超我们规模),防异常 has_next 死循环


class PangleReportError(Exception):
    """GroMore 报表接口调用失败(未配置 / 网络 / 业务码非 100)。"""


def build_sign(params: Mapping[str, Any], security_key: str) -> str:
    """按文档「方法一」生成 sign:去 sign/空值 → key 升序 → k=v& 拼接 → 末尾接 secure_key → MD5(32 位小写)。

    与官方 Python/Java 示例逐字节一致(见 tests/test_pangle_report.py 的两个测试向量)。
    """
    items = [
        (str(k), str(v))
        for k, v in params.items()
        if k != "sign" and v is not None and str(v) != ""
    ]
    items.sort(key=lambda kv: kv[0])
    raw = "&".join(f"{k}={v}" for k, v in items)
    return hashlib.md5((raw + security_key).encode("utf-8")).hexdigest()


def fetch_daily_report(
    *,
    start_date: str,
    end_date: str,
    # ⚠️ 用 ad_unit_id(GroMore 广告位ID = 我们的 104xxx),不要用 code_id(底层各 ADN 代码位,对不上)
    dimensions: str = "date,site_id,ad_unit_id",
    metrics: str = "revenue,api_revenue,ecpm,imp_cnt",
    site_ids: str | None = None,
    code_ids: str | None = None,
    os: str | None = None,
    network: str | None = None,
    time_zone: int = 8,
    limit: int = PAGE_LIMIT,
    timeout: float = DEFAULT_TIMEOUT,
) -> list[dict[str, Any]]:
    """拉一个日期区间(北京时间,time_zone=8)的天级收益报表,自动翻页,返回 report_list 全量行。

    每行是字符串字典(接口原值),形如
    {"start_date": "2026-06-27", "site_id": "5830519", "ad_unit_id": "104142227",
     "revenue": "1.23", "api_revenue": "1.05", "ecpm": "0.80", "imp_cnt": "1537", ...}。
    业务码非 100 直接抛 PangleReportError(由调用方/脚本兜底,不静默吞)。
    """
    if not settings.pangle_report_configured:
        raise PangleReportError(
            "PANGLE_REPORT_USER_ID / ROLE_ID / SECURITY_KEY 未配置,无法拉取 GroMore 报表"
        )

    base_params: dict[str, Any] = {
        "user_id": settings.PANGLE_REPORT_USER_ID,
        "role_id": settings.PANGLE_REPORT_ROLE_ID,
        "version": VERSION,
        "sign_type": SIGN_TYPE,
        "start_date": start_date,
        "end_date": end_date,
        "dimensions": dimensions,
        "metrics": metrics,
        "time_zone": time_zone,
        "limit": limit,
    }
    # 可选过滤(空则不传,避免进签名串)
    for key, val in (
        ("site_ids", site_ids), ("code_ids", code_ids),
        ("os", os), ("network", network),
    ):
        if val:
            base_params[key] = val

    rows: list[dict[str, Any]] = []
    offset = 0
    try:
        with httpx.Client(timeout=timeout) as client:
            for _ in range(_MAX_PAGES):
                params = dict(base_params)
                params["offset"] = offset
                # timestamp 每页临请求时取最新(3 分钟过期),并参与签名
                params["timestamp"] = int(time.time())
                params["sign"] = build_sign(params, settings.PANGLE_REPORT_SECURITY_KEY)

                resp = client.get(HOST + DAILY_PATH, params=params)
                resp.raise_for_status()
                body = resp.json()

                code = str(body.get("code"))
                if code != "100":
                    raise PangleReportError(
                        f"GroMore 报表业务失败 code={code} message={body.get('message')!r} "
                        f"(101=验签失败/102=userid无效/107=无权限/118=无收益查看权限,详见文档状态码)"
                    )

                data = body.get("data") or {}
                page = data.get("report_list") or []
                rows.extend(page)

                # has_next=1 还有下一页;无该字段则按本页是否取满判断
                has_next = str(data.get("has_next", "")) == "1"
                if not page or not has_next:
                    break
                offset += len(page)
    except httpx.HTTPError as e:
        raise PangleReportError(f"GroMore 报表请求异常: {e}") from e

    logger.info(
        "GroMore 天级报表拉取完成 %s~%s dims=%s 行数=%d", start_date, end_date, dimensions, len(rows)
    )
    return rows