Export Campaign Reports to CSV

import csv
import json
import os
from datetime import date
from pathlib import Path
from urllib.parse import urlencode
from urllib.request import Request, urlopen

start = date.fromisoformat(os.environ["HARDAL_START_DATE"])
end = date.fromisoformat(os.environ["HARDAL_END_DATE"])
if end < start:
    raise ValueError("End date must not precede start date")

params = {
    "tenantId": os.environ["HARDAL_TENANT_ID"],
    "timeframe": "custom", "startDate": str(start), "endDate": str(end),
    "timezone": "Europe/Istanbul", "groupBy": "campaign",
    "sortBy": "campaign", "sortDirection": "asc", "limit": 500,
}
if os.environ.get("HARDAL_SOURCE_ID"):
    params["sourceId"] = os.environ["HARDAL_SOURCE_ID"]

rows, offset, context = [], 0, None
while True:
    if offset > 10000:
        raise RuntimeError("Pagination limit reached; narrow the report before exporting")
    request = Request(
        "https://api.hardal.ai/v1/reports/acquisition?"
        + urlencode({**params, "offset": offset}),
        headers={"Authorization": "Bearer " + os.environ["HARDAL_API_KEY"]},
    )
    with urlopen(request, timeout=60) as response:
        result = json.load(response)
    if result.get("success") is not True:
        raise RuntimeError("The report request did not succeed")
    data = result["data"]
    current = {"range": data["range"], "controls": data["controls"],
               "currency": data.get("currency")}
    if context is not None and context != current:
        raise RuntimeError("Report settings changed during export; run it again")
    context = current
    table = data["trafficDistribution"]
    batch = table["rows"]
    pagination = table["pagination"]
    if pagination["offset"] != offset:
        raise RuntimeError("Unexpected page offset; export stopped")
    total = pagination["total"]
    if not isinstance(total, int) or total < 0:
        raise RuntimeError("The response has no valid total row count")
    rows.extend(batch)
    offset += len(batch)
    if offset >= total:
        break
    if not batch:
        raise RuntimeError("Incomplete result; no further rows were returned")

def csv_value(value):
    if isinstance(value, str) and value.lstrip().startswith(("=", "+", "-", "@")):
        return "'" + value
    return value

fields = ["campaign", "sessions", "conversions", "conversion_rate", "revenue"]
output = Path(f"hardal-campaigns-{start}-{end}.csv")
with output.open("w", newline="", encoding="utf-8") as handle:
    writer = csv.DictWriter(handle, fieldnames=fields)
    writer.writeheader()
    writer.writerows({field: csv_value(row.get(field)) for field in fields} for row in rows)
metadata = {"request": params, **context}
output.with_suffix(".json").write_text(json.dumps(metadata, indent=2), encoding="utf-8")
print("Saved", len(rows), "rows to", output)