"""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 (e.g. 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()))
