Agent/examples/tencent_realtime_control/fetch_daily_hourly_cpm.py

#!/usr/bin/env python
"""Fetch one day's overall hourly CPM for real-time ad control."""

from __future__ import annotations

import argparse
import re
from datetime import datetime
from pathlib import Path
from zoneinfo import ZoneInfo

from dotenv import load_dotenv

from odps_source import build_odps_client, fetch_hourly_cpm

SHANGHAI = ZoneInfo("Asia/Shanghai")
ROOT = Path(__file__).resolve().parent


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--date",
        default=datetime.now(SHANGHAI).strftime("%Y%m%d"),
        help="Data date in YYYYMMDD format; defaults to today in Asia/Shanghai.",
    )
    parser.add_argument(
        "--output",
        type=Path,
        help="CSV output path; defaults to outputs/hourly_cpm_<date>.csv.",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    if not re.fullmatch(r"\d{8}", args.date):
        raise ValueError("--date must use YYYYMMDD format")

    load_dotenv(ROOT.parent / "auto_put_ad_mini" / ".env", override=False)
    load_dotenv(Path.cwd() / ".env", override=False)
    frame = fetch_hourly_cpm(build_odps_client(), args.date)
    if frame.empty:
        print(f"No hourly CPM data found for {args.date}.")
        return

    output_path = args.output or ROOT / "outputs" / f"hourly_cpm_{args.date}.csv"
    output_path.parent.mkdir(parents=True, exist_ok=True)
    frame.to_csv(output_path, index=False)

    display = frame.copy()
    display["hour"] = display["hour_of_day"].map(lambda value: f"{value:02d}:00")
    display["cpm"] = display["cpm"].map(lambda value: f"{value:.2f}")
    print(display[["hour", "impressions", "cpm"]].to_string(index=False))
    print(f"hours={len(frame)} latest_partition={frame.iloc[-1]['dt']}")
    print(f"output={output_path}")


if __name__ == "__main__":
    main()