#!/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 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" 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 match_pkg(campaign_name: str, packages: list[str]) -> str: lowered = campaign_name.casefold() for package in packages: if package.casefold() in lowered: return package return "" def build_dataset( api_rows: list[dict[str, Any]], report_payload: dict[str, Any], start_date: date, end_date: date, ) -> dict[str, Any]: coverage_records = report_payload.get("records") or [] packages = sorted({str(row["pkg"]) for row in coverage_records}, key=len, reverse=True) coverage_by_key = { (str(row["pkg"]).casefold(), str(row["country"]).upper()): row for row in coverage_records } records: list[dict[str, Any]] = [] for api_row in api_rows: campaign_name = str(api_row.get("campaign_name") or "").strip() pkg = match_pkg(campaign_name, packages) country = str(api_row.get("country") or "").strip().upper() coverage = coverage_by_key.get((pkg.casefold(), country)) if pkg and country else None records.append( { "date": str(api_row.get("date") or ""), "campaign_id": str(api_row.get("campaign_id") or ""), "campaign_name": campaign_name, "pkg": pkg, "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), "order_match": coverage is not None, "adpingids": list(coverage.get("adpingids") or []) if coverage else [], "order_statuses": list(coverage.get("order_statuses") or []) if coverage else [], "listed_in_taurusx_orders": bool(coverage and coverage.get("target_present")), } ) 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"]}), "matched_order_rows": sum(row["order_match"] for row in records), "unmatched_order_rows": sum(not row["order_match"] 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("--data-file", type=Path, default=Path(__file__).with_name("data.json")) 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.data_file.is_file(): raise SystemExit(f"Report data file not found: {args.data_file}") 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) report_payload = json.loads(args.data_file.read_text(encoding="utf-8")) api_rows = request_performance(access_key, secret_key, start_date, end_date) report_payload["taurusx_perf"] = build_dataset(api_rows, report_payload, start_date, end_date) atomic_write_json(args.data_file, report_payload) meta = report_payload["taurusx_perf"]["meta"] print( f"Updated {args.data_file}: {meta['rows']} TaurusX rows, " f"{meta['matched_order_rows']} matched orders, {meta['start_date']}..{meta['end_date']}" ) if __name__ == "__main__": main()