shaguabijia-app-server/app/repositories/ad_pangle_revenue.py

"""穿山甲 GroMore 天级收益 读写(`ad_pangle_daily_revenue` 表)。

`scripts/sync_pangle_revenue` 拉数后调 `upsert_daily_rows` 落库(同一(日期×应用×代码位×广告源)
幂等覆盖,T+1 订正可重跑);admin 广告收益报表调 `aggregate_by_date` 取「穿山甲后台收益」做
汇总/趋势级展示。穿山甲无用户维度,故这里不涉及 user_id。
"""
from __future__ import annotations

from typing import Any, TypedDict

from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app.models.ad_pangle_revenue import AdPangleDailyRevenue

# upsert 时可被覆盖更新的列(唯一键之外的业务列)。
_UPDATABLE = ("site_id", "revenue_yuan", "api_revenue_yuan", "ecpm", "impressions", "currency")


class PangleDateAgg(TypedDict):
    date: str
    revenue_yuan: float
    api_revenue_yuan: float | None
    impressions: int


def upsert_daily_rows(db: Session, rows: list[dict[str, Any]]) -> dict[str, int]:
    """按唯一键 (report_date, app_env, our_code_id, adn) 逐行 upsert。

    每个 row 须含:report_date, app_env, our_code_id;可选 adn(默认"")、site_id、
    revenue_yuan、api_revenue_yuan、ecpm、impressions、currency。返回 {inserted, updated}。
    规模很小(代码位数×天数),逐行 select-then-write 足够,且 SQLite/PG 通用。
    """
    inserted = updated = 0
    for row in rows:
        adn = row.get("adn") or ""
        existing = db.execute(
            select(AdPangleDailyRevenue).where(
                AdPangleDailyRevenue.report_date == row["report_date"],
                AdPangleDailyRevenue.app_env == row["app_env"],
                AdPangleDailyRevenue.our_code_id == row["our_code_id"],
                AdPangleDailyRevenue.adn == adn,
            )
        ).scalar_one_or_none()

        if existing is None:
            db.add(AdPangleDailyRevenue(
                report_date=row["report_date"],
                app_env=row["app_env"],
                site_id=row.get("site_id"),
                our_code_id=row["our_code_id"],
                adn=adn,
                revenue_yuan=float(row.get("revenue_yuan") or 0.0),
                api_revenue_yuan=row.get("api_revenue_yuan"),
                ecpm=row.get("ecpm"),
                impressions=int(row.get("impressions") or 0),
                currency=row.get("currency") or "cny",
            ))
            inserted += 1
        else:
            for col in _UPDATABLE:
                if col in row:
                    setattr(existing, col, row[col])
            updated += 1
    db.commit()
    return {"inserted": inserted, "updated": updated}


def aggregate_by_date(
    db: Session,
    *,
    date_from: str,
    date_to: str,
    app_env: str | None = None,
    our_code_id: str | None = None,
) -> list[PangleDateAgg]:
    """按日期汇总穿山甲收益(闭区间,北京时间),供报表趋势 + 合计。

    revenue_yuan = Σrevenue;api_revenue_yuan = Σapi_revenue(SQL SUM 忽略 NULL,
    全为空则返回 None,前端显示「-」)。可选按应用 / 代码位过滤。返回按日期升序。
    """
    stmt = (
        select(
            AdPangleDailyRevenue.report_date,
            func.sum(AdPangleDailyRevenue.revenue_yuan),
            func.sum(AdPangleDailyRevenue.api_revenue_yuan),
            func.sum(AdPangleDailyRevenue.impressions),
        )
        .where(
            AdPangleDailyRevenue.report_date >= date_from,
            AdPangleDailyRevenue.report_date <= date_to,
        )
        .group_by(AdPangleDailyRevenue.report_date)
        .order_by(AdPangleDailyRevenue.report_date)
    )
    if app_env is not None:
        stmt = stmt.where(AdPangleDailyRevenue.app_env == app_env)
    if our_code_id is not None:
        stmt = stmt.where(AdPangleDailyRevenue.our_code_id == our_code_id)

    out: list[PangleDateAgg] = []
    for report_date, rev, api_rev, imp in db.execute(stmt).all():
        out.append(PangleDateAgg(
            date=report_date,
            revenue_yuan=round(float(rev or 0.0), 6),
            api_revenue_yuan=(round(float(api_rev), 6) if api_rev is not None else None),
            impressions=int(imp or 0),
        ))
    return out