#!/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() 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.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": impressions, "clicks": clicks, "conversions": conversions, "cost": cost, "cpm": round(cost / impressions * 1000, 6) if impressions else None, "cpc": round(cost / clicks, 6) if clicks else None, "cpi": round(cost / conversions, 6) if conversions else None, } ) records.sort(key=lambda row: (row["date"], row["cost"], row["clicks"]), reverse=True) total_impressions = sum(row["impressions"] for row in records) total_clicks = sum(row["clicks"] for row in records) total_conversions = sum(row["conversions"] for row in records) total_cost = round(sum(row["cost"] for row in records), 4) 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": total_impressions, "clicks": total_clicks, "conversions": total_conversions, "cost": total_cost, "cpm": round(total_cost / total_impressions * 1000, 6) if total_impressions else None, "cpc": round(total_cost / total_clicks, 6) if total_clicks else None, "cpi": round(total_cost / total_conversions, 6) if total_conversions else None, }, "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()