#!/usr/bin/env python3 """Python 3.10+: resume one CSV export with a persisted idempotency key. Run one process per state file. Keep state and downloaded customer data private. """ import argparse import csv import hashlib import json import os from pathlib import Path import sys import time import urllib.error import urllib.parse import urllib.request import uuid BASE = "https://api.returning.ai/v1/loyalty-exports" COLUMNS = ["client_ref", "event_id", "recorded_at_utc", "operation", "points_delta"] def save_state(path, state): temporary = path.with_name(path.name + ".tmp") with open(temporary, "w", encoding="utf-8") as handle: os.chmod(temporary, 0o600) json.dump(state, handle, indent=2) os.replace(temporary, path) def export_csv(key, body, state_path, output_path, max_wait=600): # The fingerprint prevents accidental resume with another community or key. fingerprint = hashlib.sha256(key.encode()).hexdigest() if state_path.exists(): state = json.loads(state_path.read_text()) if state["body"] != body or state["key_fingerprint"] != fingerprint: raise RuntimeError("State belongs to different inputs or an API key. Resume with the original values; use a new state file only for an intentional new export.") else: state = {"body": body, "key_fingerprint": fingerprint, "idempotency_key": str(uuid.uuid4())} save_state(state_path, state) # Persist before sending the first request. deadline = time.monotonic() + max_wait def wait(seconds): if time.monotonic() + seconds >= deadline: raise RuntimeError("Wait limit reached. Rerun with the same arguments and state file to resume.") time.sleep(seconds) def api(path="", method="GET", payload=None): headers = {"Authorization": "Bearer " + key, "Content-Type": "application/json"} if method == "POST": headers["Idempotency-Key"] = state["idempotency_key"] for attempt in range(4): request = urllib.request.Request(BASE + path, headers=headers, method=method, data=json.dumps(payload).encode() if payload is not None else None) try: with urllib.request.urlopen(request, timeout=30) as response: return json.load(response)["data"] except urllib.error.HTTPError as error: try: details = json.load(error) except (ValueError, UnicodeError): details = {} error.close() meta = details.get("meta", {}) code = meta.get("code", "UNSPECIFIED") if attempt == 3 or (error.code != 429 and not 500 <= error.code < 600): raise RuntimeError(f"API HTTP {error.code}, code={code}, requestId={meta.get('requestId', 'unavailable')}. Keep the state file for recovery.") from None retry_after = error.headers.get("Retry-After", "") wait(max(30 if error.code == 429 else 0, int(retry_after) if retry_after.isdigit() else 0, 5 * 2 ** attempt)) except (urllib.error.URLError, TimeoutError): if attempt == 3: raise RuntimeError("Network request failed. Rerun with the same state file; do not create a new idempotency key.") from None wait(5 * 2 ** attempt) if "export_id" not in state: job = api(method="POST", payload=body) state["export_id"] = job["export_id"] save_state(state_path, state) export_id = state["export_id"] print("Export ID:", export_id) interval = 5 while True: job = api("/" + export_id) if job["status"] == "completed": break if job["status"] not in ("queued", "processing"): raise RuntimeError(f"Export ended as {job['status']}. Check status and the recovery guide before requesting another export.") wait(interval) interval = min(interval * 2, 30) temporary = output_path.with_name(output_path.name + ".part") try: for attempt in range(2): link = api("/" + export_id + "/download") if urllib.parse.urlsplit(link["url"]).scheme != "https": raise RuntimeError("Download URL must use HTTPS.") try: # Deliberately no Authorization header on the download request. with urllib.request.urlopen(link["url"], timeout=60) as response, open(temporary, "wb") as handle: os.chmod(temporary, 0o600) while chunk := response.read(1024 * 1024): handle.write(chunk) break except (urllib.error.URLError, TimeoutError) as error: if isinstance(error, urllib.error.HTTPError): error.close() if attempt == 1: raise RuntimeError("Download failed. Rerun with the same state file to request a fresh link.") from None with open(temporary, newline="", encoding="utf-8-sig") as handle: rows = csv.reader(handle) if next(rows, None) != COLUMNS: raise RuntimeError("Unexpected CSV columns; keep the export ID and contact support.") count = sum(1 for _ in rows) if count != job["total_rows"] or temporary.stat().st_size != job["file_size_bytes"]: raise RuntimeError("Downloaded file does not match status row count or byte size. Rerun with the same state file.") os.replace(temporary, output_path) print(f"Saved {count} rows to {output_path}") finally: temporary.unlink(missing_ok=True) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--start", required=True, help="Inclusive UTC timestamp ending in Z") parser.add_argument("--end", required=True, help="Exclusive UTC timestamp ending in Z") parser.add_argument("--client-ref", action="append", help="Repeat to select customers; omit for all") parser.add_argument("--state", type=Path, required=True, help="Private state file for this intended export") parser.add_argument("--out", type=Path, required=True, help="Output CSV path; replaced after verification") parser.add_argument("--max-wait", type=int, default=600, help="Client polling budget in seconds, not an API guarantee") args = parser.parse_args() key = os.environ.get("RETURNING_API_KEY") if not key: parser.error("Set RETURNING_API_KEY to your Community API key.") if args.max_wait <= 0 or args.state.resolve() == args.out.resolve(): parser.error("Use a positive wait budget and distinct state/output paths.") body = {"dataset": "points_transactions", "format": "csv", "start_at": args.start, "end_at": args.end} if args.client_ref: body["client_refs"] = sorted(args.client_ref) try: export_csv(key, body, args.state, args.out, args.max_wait) except (RuntimeError, OSError, ValueError, KeyError, csv.Error) as error: # Network exceptions are converted above so signed URLs are never printed. print(str(error), file=sys.stderr) return 1 return 0 if __name__ == "__main__": sys.exit(main())