taurusx_tracking/fetch_taurusx.py

#!/usr/bin/env python3
"""Fetch recent TaurusX performance data without persisting API credentials."""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import tempfile
import time
import urllib.parse
import urllib.request
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from typing import Any

API_URL = "https://scaler.taurusx.com/openapi/performance_data"
TIMEZONE = "+8"
PACKAGE_SUFFIX_TAGS = {"audience", "test", "retargeting", "rtg"}


def build_token(secret_key: str, timestamp: str) -> str:
    timestamp_hash = hashlib.md5(timestamp.encode()).hexdigest()
    return hashlib.md5((secret_key + timestamp_hash).encode()).hexdigest()


def request_performance(
    access_key: str,
    secret_key: str,
    start_date: date,
    end_date: date,
) -> list[dict[str, Any]]:
    timestamp = str(int(time.time()))
    query = urllib.parse.urlencode(
        {
            "timezone": TIMEZONE,
            "report_type": "event",
            "start_time": start_date.isoformat(),
            "end_time": end_date.isoformat(),
            "dimension_option": "Campaign,Country",
        }
    )
    request = urllib.request.Request(
        f"{API_URL}?{query}",
        headers={
            "access-key": access_key,
            "token": build_token(secret_key, timestamp),
            "timestamp": timestamp,
            "Accept": "application/json",
            "User-Agent": "AgentStage-TaurusX-Report/1.0",
        },
    )
    with urllib.request.urlopen(request, timeout=60) as response:
        payload = json.loads(response.read())
    if payload.get("code") != 200:
        raise RuntimeError(f"TaurusX API returned code {payload.get('code')}: {payload.get('msg', 'unknown error')}")
    rows = (payload.get("data") or {}).get("data_list")
    if not isinstance(rows, list):
        raise RuntimeError("TaurusX API response is missing data.data_list")
    return rows


def extract_pkg(campaign_name: str, country: str) -> str:
    """Extract an Android package or iOS app ID from the TaurusX campaign name."""
    candidates = re.findall(
        r"(?:[A-Za-z][A-Za-z0-9]*\.)+[A-Za-z0-9_]+|id\d{6,}",
        campaign_name,
        flags=re.IGNORECASE,
    )
    if not candidates:
        return ""
    package = max(candidates, key=len)
    parts = package.split("_")
    while len(parts) > 1 and (
        parts[-1].upper() == country.upper()
        or parts[-1].casefold() in PACKAGE_SUFFIX_TAGS
        or parts[-1].isdigit()
    ):
        parts.pop()
    return "_".join(parts)


def build_dataset(
    api_rows: list[dict[str, Any]],
    start_date: date,
    end_date: date,
) -> dict[str, Any]:
    records: list[dict[str, Any]] = []
    for api_row in api_rows:
        campaign_name = str(api_row.get("campaign_name") or "").strip()
        country = str(api_row.get("country") or "").strip().upper()
        records.append(
            {
                "date": str(api_row.get("date") or ""),
                "campaign_id": str(api_row.get("campaign_id") or ""),
                "campaign_name": campaign_name,
                "pkg": extract_pkg(campaign_name, country),
                "country": country,
                "impressions": int(float(api_row.get("impressions") or 0)),
                "clicks": int(float(api_row.get("clicks") or 0)),
                "conversions": int(float(api_row.get("conversions") or 0)),
                "cost": round(float(api_row.get("cost") or 0), 4),
            }
        )

    records.sort(key=lambda row: (row["date"], row["cost"], row["clicks"]), reverse=True)
    fetched_at = datetime.now(timezone(timedelta(hours=8))).isoformat(timespec="seconds")
    return {
        "meta": {
            "source": API_URL,
            "timezone": TIMEZONE,
            "start_date": start_date.isoformat(),
            "end_date": end_date.isoformat(),
            "fetched_at": fetched_at,
            "rows": len(records),
            "campaigns": len({row["campaign_id"] for row in records}),
            "pkg_country_count": len({(row["pkg"].casefold(), row["country"]) for row in records if row["pkg"]}),
            "countries": len({row["country"] for row in records if row["country"]}),
            "unparsed_pkg_rows": sum(not row["pkg"] for row in records),
            "impressions": sum(row["impressions"] for row in records),
            "clicks": sum(row["clicks"] for row in records),
            "conversions": sum(row["conversions"] for row in records),
            "cost": round(sum(row["cost"] for row in records), 4),
        },
        "records": records,
    }


def atomic_write_json(path: Path, payload: dict[str, Any]) -> None:
    with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, delete=False) as handle:
        json.dump(payload, handle, ensure_ascii=False, indent=2)
        handle.write("\n")
        temporary_path = Path(handle.name)
    temporary_path.replace(path)


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--output",
        type=Path,
        default=Path(__file__).with_name("taurusx_data.json"),
        help="Standalone TaurusX dataset output (default: taurusx_data.json next to this script)",
    )
    parser.add_argument("--days", type=int, default=7, choices=range(1, 8), metavar="1-7")
    parser.add_argument("--end-date", type=date.fromisoformat, help="Last report date; defaults to yesterday in UTC+8")
    args = parser.parse_args()

    access_key = os.environ.get("TAURUSX_ACCESS_KEY", "").strip()
    secret_key = os.environ.get("TAURUSX_SECRET_KEY", "").strip()
    if not access_key or not secret_key:
        raise SystemExit("TAURUSX_ACCESS_KEY and TAURUSX_SECRET_KEY environment variables are required")
    if not args.output.parent.is_dir():
        raise SystemExit(f"Output directory not found: {args.output.parent}")

    utc8_today = datetime.now(timezone(timedelta(hours=8))).date()
    end_date = args.end_date or (utc8_today - timedelta(days=1))
    start_date = end_date - timedelta(days=args.days - 1)
    api_rows = request_performance(access_key, secret_key, start_date, end_date)
    dataset = build_dataset(api_rows, start_date, end_date)
    atomic_write_json(args.output, dataset)
    meta = dataset["meta"]
    print(
        f"Updated {args.output}: {meta['rows']} TaurusX rows, "
        f"{meta['campaigns']} campaigns, {meta['start_date']}..{meta['end_date']}"
    )


if __name__ == "__main__":
    main()