import time
import requests

API = "https://api.sequenzy.com/api/v1"
HEADERS = {"Authorization": "Bearer YOUR_API_KEY"}
BATCH_SIZE = 5000

def sync_contacts(rows, run_id):
    import_ids = []
    for i in range(0, len(rows), BATCH_SIZE):
        batch = rows[i : i + BATCH_SIZE]
        resp = requests.post(
            f"{API}/subscribers/imports",
            headers=HEADERS,
            json={
                "duplicateStrategy": "merge",
                "optInMode": "confirmed",
                # The key is scoped to the request content: retrying the same
                # batch after a timeout replays the original import, while a
                # corrected batch under the same key imports as a new one.
                "idempotencyKey": f"{run_id}-batch-{i // BATCH_SIZE}",
                "subscribers": [
                    {
                        "email": row["email"],
                        "externalId": row["customer_id"],
                        "firstName": row["first_name"],
                        "customAttributes": {
                            "plan": row["plan"],
                            "mrr": row["mrr"],
                        },
                        "createdAt": row["signed_up_at"],
                    }
                    for row in batch
                ],
            },
            timeout=60,
        )
        if resp.status_code in (429, 503):
            time.sleep(int(resp.headers.get("Retry-After", "5")))
            continue  # rate limit or overlapping enqueue; the keyed retry is safe
        resp.raise_for_status()
        import_ids.append(resp.json()["import"]["id"])
    return import_ids

def wait_for_imports(import_ids, timeout_seconds=1800):
    for import_id in import_ids:
        deadline = time.monotonic() + timeout_seconds
        while True:
            data = requests.get(
                f"{API}/subscribers/imports/{import_id}", headers=HEADERS, timeout=30
            ).json()["import"]
            # Imports report status "running", "completed", or "blocked" -
            # there is no "failed" status; row failures appear in failedCount
            # on a completed import.
            if data["status"] in ("completed", "blocked"):
                break
            if time.monotonic() > deadline:
                raise TimeoutError(f"{import_id} still {data['status']}")
            time.sleep(5)
        if data["status"] == "blocked":
            raise RuntimeError(f"{import_id} was blocked pending review")
        if data.get("failedCount"):
            raise RuntimeError(f"{import_id}: {data['failedReasons']}")

EVENT_BATCH_SIZE = 25

def sync_events(rows):
    for i in range(0, len(rows), EVENT_BATCH_SIZE):
        batch = rows[i : i + EVENT_BATCH_SIZE]
        resp = requests.post(
            f"{API}/subscribers/events/imports",
            headers=HEADERS,
            json={
                "events": [
                    {
                        # null identifiers are accepted as absent. An
                        # externalId-only row must already exist from contact sync.
                        "externalId": row["customer_id"],
                        "email": row["email"],
                        "name": "warehouse.purchase",
                        "eventId": row["order_id"],
                        "occurredAt": row["ordered_at"],
                        # Row data must live inside properties - keys outside
                        # the documented fields are ignored.
                        "properties": {"amount": row["amount"], "sku": row["sku"]},
                    }
                    for row in batch
                ],
            },
            timeout=120,
        )
        if resp.status_code == 429:
            time.sleep(int(resp.headers.get("Retry-After", "5")))
            continue
        resp.raise_for_status()
        summary = resp.json()
        if summary["failed"]:
            raise RuntimeError(f"{summary['failed']} events failed: {summary['failures'][:5]}")
        if summary["sideEffectFailed"]:
            raise RuntimeError(
                f"{summary['sideEffectFailed']} event receipts need downstream recovery: "
                f"{summary['sideEffectFailures'][:5]}"
            )