Daily CSV export

"""Export a workflow's runs for a given day (Europe/Madrid) as one flat CSV."""
import argparse
import asyncio
import csv
import logging
import os
import sys
from collections import deque
from datetime import date, datetime, time, timedelta, timezone
from zoneinfo import ZoneInfo

import httpx
from anyformat.sdk import AsyncClient, RateLimited

BASE_URL = "https://api.anyformat.ai"
MADRID = ZoneInfo("Europe/Madrid")
RATE_PER_MIN = 55       # API cap is 60/min; leave headroom for the 429 retry
CONCURRENCY = 8

logger = logging.getLogger("daily_csv_export")


class TokenBucket:
    """Sliding-window '<= N requests per 60s' gate. Await before every HTTP call."""

    def __init__(self, per_minute: int) -> None:
        self._n = per_minute
        self._times: deque[float] = deque()
        self._lock = asyncio.Lock()

    async def acquire(self) -> None:
        async with self._lock:
            loop = asyncio.get_running_loop()
            now = loop.time()
            while self._times and now - self._times[0] >= 60.0:
                self._times.popleft()
            if len(self._times) >= self._n:
                await asyncio.sleep(60.0 - (now - self._times[0]))
            self._times.append(loop.time())


def parse_day(text: str) -> date:
    """`today`, `yesterday`, or an ISO `YYYY-MM-DD`. Interpreted in Madrid."""
    today = datetime.now(MADRID).date()
    if text == "today":
        return today
    if text == "yesterday":
        return today - timedelta(days=1)
    return date.fromisoformat(text)


async def gated(bucket: TokenBucket, coro_factory):
    """One `bucket.acquire()` per outbound HTTP call, with one 429 retry."""
    while True:
        await bucket.acquire()
        try:
            return await coro_factory()
        except RateLimited as e:
            logger.warning("429 rate limited; sleeping %.1fs", e.retry_after or 5.0)
            await asyncio.sleep(e.retry_after or 5.0)


async def list_run_ids(
    http: httpx.AsyncClient,
    api_key: str,
    workflow_id: str,
    after_iso: str,
    before_iso: str,
    bucket: TokenBucket,
) -> list[str]:
    """Paginate GET /v3/workflows/{id}/runs/ with server-side date filtering.

    The SDK doesn't (yet) forward `document_packet_created_after` / `_before`,
    so we call the endpoint directly; the SDK covers the per-run fetch below."""
    ids: list[str] = []
    cursor: str | None = None
    headers = {"Authorization": f"Bearer {api_key}"}
    while True:
        params = {
            "limit": 100,
            "document_packet_created_after": after_iso,
            "document_packet_created_before": before_iso,
        }
        if cursor:
            params["cursor"] = cursor
        logger.info("GET /v3/workflows/%s/runs/ params=%s", workflow_id, params)

        async def _fetch() -> dict:
            r = await http.get(
                f"{BASE_URL}/v3/workflows/{workflow_id}/runs/",
                params=params,
                headers=headers,
            )
            if r.status_code == 429:
                raise RateLimited(429, retry_after=float(r.headers.get("Retry-After") or 5.0))
            r.raise_for_status()
            return r.json()

        body = await gated(bucket, _fetch)
        items = body.get("items") or []
        cursor = body.get("next_cursor")
        logger.info("  -> %d runs on this page; next_cursor=%s", len(items), cursor)
        ids.extend(item["id"] for item in items)
        if not cursor:
            return ids


async def fetch_run(client: AsyncClient, run_id: str, bucket: TokenBucket, sem: asyncio.Semaphore):
    async with sem:
        logger.info("GET /v3/runs/%s/", run_id)
        run = await gated(bucket, lambda: client.get_run(run_id))
        logger.info("  -> status=%s", run.status)
        return run


def scalar_cells(run) -> dict[str, object]:
    """Top-level scalar values from the (single) extraction. Skips nested
    list-of-object fields such as line items; v2's CSV did the same.

    Reads `result.raw["extractions"][0]["fields"]` rather than the typed
    `result.fields`: the typed view is silently empty on any wire drift,
    which shows up as a CSV with no columns."""
    extractions = run.result.raw.get("extractions") or []
    if not extractions:
        return {}
    return {
        name: val.get("value")
        for name, val in (extractions[0].get("fields") or {}).items()
        if isinstance(val, dict) and "value" in val
    }


async def main() -> int:
    logging.basicConfig(
        level=logging.INFO,
        stream=sys.stderr,
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
    )

    ap = argparse.ArgumentParser()
    ap.add_argument("--workflow", required=True)
    ap.add_argument("--day", default="yesterday", help="today | yesterday | YYYY-MM-DD")
    ap.add_argument("--out", default="runs.csv")
    args = ap.parse_args()

    day = parse_day(args.day)
    start_local = datetime.combine(day, time.min, MADRID)
    end_local = start_local + timedelta(days=1)
    start_utc = start_local.astimezone(timezone.utc)
    end_utc = end_local.astimezone(timezone.utc)
    # API accepts naive ISO 8601 as UTC. Drop the offset for a clean value.
    after_iso = start_utc.strftime("%Y-%m-%dT%H:%M:%S")
    before_iso = end_utc.strftime("%Y-%m-%dT%H:%M:%S")

    logger.info(
        "window day=%s Madrid=[%s, %s) UTC=[%s, %s)",
        day, start_local.isoformat(), end_local.isoformat(), after_iso, before_iso,
    )

    api_key = os.environ["ANYFORMAT_API_KEY"]
    bucket = TokenBucket(RATE_PER_MIN)
    sem = asyncio.Semaphore(CONCURRENCY)

    async with httpx.AsyncClient(timeout=60.0) as http, AsyncClient(api_key=api_key) as client:
        run_ids = await list_run_ids(http, api_key, args.workflow, after_iso, before_iso, bucket)
        logger.info("total runs in window: %d", len(run_ids))
        runs = await asyncio.gather(
            *(fetch_run(client, rid, bucket, sem) for rid in run_ids)
        )

    processed = [r for r in runs if r.status == "processed" and r.result]
    logger.info("processed=%d (of %d fetched)", len(processed), len(runs))

    # CSV: union of extracted-field columns across runs, first-seen order.
    prefix = ["run_id", "packet_id"]
    columns: list[str] = []
    seen: set[str] = set(prefix)
    rows: list[dict] = []
    for r in processed:
        cells = scalar_cells(r)
        for c in cells:
            if c not in seen:
                seen.add(c)
                columns.append(c)
        rows.append({"run_id": r.id, "packet_id": r.document_packet_id, **cells})

    header = [*prefix, *columns]
    with open(args.out, "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=header, extrasaction="ignore")
        w.writeheader()
        w.writerows(rows)

    logger.info("wrote %d rows -> %s", len(rows), args.out)
    return 0


if __name__ == "__main__":
    raise SystemExit(asyncio.run(main()))

export ANYFORMAT_API_KEY=af_...
python daily_csv_export.py --workflow wf_123 --day yesterday --out yesterday.csv
python daily_csv_export.py --workflow wf_123 --day 2026-07-14 --out 2026-07-14.csv

INFO daily_csv_export: window day=2026-07-14 Madrid=[2026-07-14T00:00:00+02:00, 2026-07-15T00:00:00+02:00) UTC=[2026-07-13T22:00:00, 2026-07-14T22:00:00)
INFO daily_csv_export: GET /v3/workflows/wf_123/runs/ params={'limit': 100, 'document_packet_created_after': '2026-07-13T22:00:00', 'document_packet_created_before': '2026-07-14T22:00:00'}
INFO daily_csv_export:   -> 100 runs on this page; next_cursor=eyJ...
INFO daily_csv_export: GET /v3/workflows/wf_123/runs/ params={'limit': 100, ..., 'cursor': 'eyJ...'}
INFO daily_csv_export:   -> 34 runs on this page; next_cursor=None
INFO daily_csv_export: total runs in window: 134
INFO daily_csv_export: GET /v3/runs/aaaa.../
INFO daily_csv_export:   -> status=processed
...
INFO daily_csv_export: wrote 134 rows -> yesterday.csv