#!/usr/bin/env python3
"""Sync an FXMacroData dataset into a Smartsheet sheet.

The script reads one FXMacroData dataset (an indicator's release history, an
FX pair, a commodity, a currency's release calendar or CFTC COT positioning),
creates the target sheet with typed columns if it does not exist, and upserts
rows matched on the primary "Key" column: new releases are added, revised
values are updated, unchanged rows are left alone. Run it on a schedule (cron,
Task Scheduler, a CI job) to keep the sheet current.

    export FXMD_API_KEY=...                 # optional for USD indicators, calendar and COT
    export SMARTSHEET_ACCESS_TOKEN=...      # Smartsheet API access token
    python fxmd_smartsheet_sync.py indicator --currency usd --indicator inflation \\
        --workspace-id 1234567890123456

Tokens are read from environment variables only. FXMacroData requests send the
key in the ``X-API-Key`` header and Smartsheet requests send the token as a
Bearer header; neither is ever written to a URL or printed.

Only the Python standard library is used. Docs:
https://fxmacrodata.com/documentation/integrations?utm_source=github&utm_medium=referral&utm_campaign=smartsheet-sync&utm_content=docstring
"""

from __future__ import annotations

import argparse
import json
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple

API_ORIGIN = "https://api.fxmacrodata.com"
USER_AGENT = "fxmacrodata-smartsheet-sync/1.0"
PAGE_LIMIT = 100

# Smartsheet hosts per instance (developers.smartsheet.com, "Base URL").
# Tokens are region-specific: use a token from the same instance.
SMARTSHEET_HOSTS = {
    "us": "https://api.smartsheet.com/2.0",
    "eu": "https://api.smartsheet.eu/2.0",
    "au": "https://api.smartsheet.au/2.0",
    "gov": "https://api.smartsheetgov.com/2.0",
}

# Smartsheet limits: bulk row operations should send at most 500 rows per
# request; a sheet holds at most 500,000 cells; 300 requests per minute per
# token, after which the API answers 429 with errorCode 4003 and asks clients
# to wait at least 60 seconds.
MAX_ROWS_PER_REQUEST = 500
MAX_CELLS_PER_SHEET = 500_000
MAX_SYNC_ROWS = 20_000
RATE_LIMIT_WAIT_SECONDS = 60.0
MAX_RETRIES = 5


class SyncError(Exception):
    """A failure the user can act on; printed without a traceback."""


# --- value helpers ---------------------------------------------------------


def _epoch_iso(value: Any) -> Optional[str]:
    if isinstance(value, bool):
        return None
    if isinstance(value, (int, float)):
        return datetime.fromtimestamp(value, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    if isinstance(value, str) and value.strip():
        return value.strip()
    return None


def _num(value: Any) -> Optional[float]:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return None
    if value != value or value in (float("inf"), float("-inf")):
        return None
    return value


def _text(value: Any) -> Optional[str]:
    if value is None:
        return None
    text = str(value).strip()
    return text or None


def _date(value: Any) -> Optional[str]:
    text = _text(value)
    if text and len(text) >= 10:
        try:
            datetime.strptime(text[:10], "%Y-%m-%d")
            return text[:10]
        except ValueError:
            return None
    return None


def _key(*parts: Any) -> Optional[str]:
    """Row key ``part:part:date`` in lower case, or None when the date is missing.

    Indicator keys (``usd:inflation:2026-09-01``) match the ``record_key`` of
    FXMacroData's release webhooks, so a Bridge webhook workflow and this
    script can maintain the same sheet without duplicating rows.
    """

    texts = [(_text(p) or "").lower() for p in parts]
    return ":".join(texts) if all(texts) else None


# --- datasets --------------------------------------------------------------

# Column specs: (title, Smartsheet column type, extra column fields).
# Smartsheet's API cannot create DATETIME columns (system columns only), so
# UTC timestamps are ISO-8601 text in TEXT_NUMBER columns. The primary column
# must be TEXT_NUMBER; it holds the Key used for upserts.
Column = Tuple[str, str, Dict[str, Any]]
KEY = "Key"
IMPORTANCE_OPTIONS = ["high", "medium", "low"]

DATASETS: Dict[str, Dict[str, Any]] = {
    "indicator": {
        "requires": ("currency", "indicator"),
        "free": True,
        "path": lambda a: f"/v1/announcements/{a.currency.lower()}/{a.indicator.lower()}",
        "query": {"fields": "compact"},
        "paged": True,
        "sheet_name": lambda a: f"FXMacroData {a.currency.upper()} {a.indicator.lower()}",
        "columns": [
            (KEY, "TEXT_NUMBER", {"primary": True}),
            ("Date", "DATE", {}),
            ("Value", "TEXT_NUMBER", {}),
            ("Previous", "TEXT_NUMBER", {}),
            ("Change", "TEXT_NUMBER", {}),
            ("Released at (UTC)", "TEXT_NUMBER", {}),
            ("Currency", "TEXT_NUMBER", {}),
            ("Indicator", "TEXT_NUMBER", {}),
        ],
        "row": lambda r, a: {
            KEY: _key(a.currency, a.indicator, r.get("date")),
            "Date": _date(r.get("date")),
            "Value": _num(r.get("val")),
            "Previous": _num(r.get("previous_value")),
            "Change": _num(r.get("change_from_previous")),
            "Released at (UTC)": _epoch_iso(r.get("announcement_datetime")),
            "Currency": a.currency.upper(),
            "Indicator": a.indicator.lower(),
        },
    },
    "forex": {
        "requires": ("base", "quote"),
        "free": False,
        "path": lambda a: f"/v1/forex/{a.base.lower()}/{a.quote.lower()}",
        "query": {},
        "paged": True,
        "sheet_name": lambda a: f"FXMacroData {a.base.upper()}{a.quote.upper()}",
        "columns": [
            (KEY, "TEXT_NUMBER", {"primary": True}),
            ("Date", "DATE", {}),
            ("Rate", "TEXT_NUMBER", {}),
            ("Open", "TEXT_NUMBER", {}),
            ("High", "TEXT_NUMBER", {}),
            ("Low", "TEXT_NUMBER", {}),
            ("Close", "TEXT_NUMBER", {}),
            ("Pair", "TEXT_NUMBER", {}),
        ],
        "row": lambda r, a: {
            KEY: _key("fx", f"{a.base}{a.quote}", r.get("date")),
            "Date": _date(r.get("date")),
            "Rate": _num(r.get("val")),
            "Open": _num(r.get("open")),
            "High": _num(r.get("high")),
            "Low": _num(r.get("low")),
            "Close": _num(r.get("close")),
            "Pair": f"{a.base.upper()}/{a.quote.upper()}",
        },
    },
    "commodity": {
        "requires": ("commodity",),
        "free": False,
        "path": lambda a: f"/v1/commodities/{a.commodity.lower()}",
        "query": {},
        "paged": True,
        "sheet_name": lambda a: f"FXMacroData {a.commodity.lower()}",
        "columns": [
            (KEY, "TEXT_NUMBER", {"primary": True}),
            ("Date", "DATE", {}),
            ("Value", "TEXT_NUMBER", {}),
            ("Commodity", "TEXT_NUMBER", {}),
        ],
        "row": lambda r, a: {
            KEY: _key("commodity", a.commodity, r.get("date")),
            "Date": _date(r.get("date")),
            "Value": _num(r.get("val")),
            "Commodity": a.commodity.lower(),
        },
    },
    "calendar": {
        "requires": ("currency",),
        "free": True,
        "path": lambda a: f"/v1/calendar/{a.currency.lower()}",
        "query": {},
        "paged": False,
        "sheet_name": lambda a: f"FXMacroData {a.currency.upper()} release calendar",
        "columns": [
            (KEY, "TEXT_NUMBER", {"primary": True}),
            ("Release", "TEXT_NUMBER", {}),
            ("Name", "TEXT_NUMBER", {}),
            ("Scheduled date", "DATE", {}),
            ("Scheduled at (UTC)", "TEXT_NUMBER", {}),
            ("Reference period", "TEXT_NUMBER", {}),
            ("Importance", "PICKLIST", {"options": IMPORTANCE_OPTIONS}),
            ("Currency", "TEXT_NUMBER", {}),
        ],
        "row": lambda r, a: {
            KEY: _text(r.get("calendar_event_id"))
            or f"{a.currency.lower()}_{_text(r.get('release')) or ''}_{_text(r.get('announcement_datetime')) or ''}",
            "Release": _text(r.get("release")),
            "Name": _text(r.get("name")),
            "Scheduled date": _date(_epoch_iso(r.get("announcement_datetime"))),
            "Scheduled at (UTC)": _epoch_iso(r.get("announcement_datetime")),
            "Reference period": _text(r.get("reference_period")),
            "Importance": (_text(r.get("event_importance")) or "").lower() or None,
            "Currency": a.currency.upper(),
        },
    },
    "cot": {
        "requires": ("currency",),
        "free": True,
        "path": lambda a: f"/v1/cot/{a.currency.lower()}",
        "query": {},
        "paged": True,
        "sheet_name": lambda a: f"FXMacroData {a.currency.upper()} COT",
        "columns": [
            (KEY, "TEXT_NUMBER", {"primary": True}),
            ("Date", "DATE", {}),
            ("Open interest", "TEXT_NUMBER", {}),
            ("Non-commercial net", "TEXT_NUMBER", {}),
            ("Commercial net", "TEXT_NUMBER", {}),
            ("Leveraged funds net", "TEXT_NUMBER", {}),
            ("Asset manager net", "TEXT_NUMBER", {}),
            ("Dealer net", "TEXT_NUMBER", {}),
            ("Released at (UTC)", "TEXT_NUMBER", {}),
        ],
        "row": lambda r, a: {
            KEY: _key(a.currency, "cot", r.get("date")),
            "Date": _date(r.get("date")),
            "Open interest": _num(r.get("open_interest")),
            "Non-commercial net": _num(r.get("noncommercial_net")),
            "Commercial net": _num(r.get("commercial_net")),
            "Leveraged funds net": _num(r.get("leveraged_funds_net")),
            "Asset manager net": _num(r.get("asset_manager_net")),
            "Dealer net": _num(r.get("dealer_net")),
            "Released at (UTC)": _epoch_iso(r.get("announcement_datetime")),
        },
    },
}

# --- HTTP ------------------------------------------------------------------

Opener = Callable[[urllib.request.Request, float], Any]
Sleeper = Callable[[float], None]


def _default_opener(request: urllib.request.Request, timeout: float) -> Any:
    # Only fixed https hosts are opened: API_ORIGIN and the SMARTSHEET_HOSTS
    # entry chosen with --region. No user-supplied URL is ever requested.
    return urllib.request.urlopen(request, timeout=timeout)  # nosec B310  # noqa: S310


def _read_json(response: Any) -> Any:
    raw = response.read()
    try:
        return json.loads(raw.decode("utf-8") or "null")
    except (UnicodeDecodeError, json.JSONDecodeError):
        return None


def _error_body(err: urllib.error.HTTPError) -> Dict[str, Any]:
    try:
        body = json.loads(err.read().decode("utf-8") or "null")
    except Exception:  # noqa: BLE001 - best-effort detail only
        body = None
    return body if isinstance(body, dict) else {}


def _error_detail(body: Dict[str, Any], err: urllib.error.HTTPError) -> str:
    for key in ("message", "detail", "error"):
        value = body.get(key)
        if isinstance(value, str):
            return value
        if isinstance(value, dict) and isinstance(value.get("message"), str):
            return value["message"]
    return err.reason if isinstance(err.reason, str) else ""


def _retry_after(err: urllib.error.HTTPError) -> Optional[float]:
    headers = getattr(err, "headers", None)
    value = headers.get("Retry-After") if headers is not None else None
    try:
        return max(float(value), 0.0) if value is not None else None
    except (TypeError, ValueError):
        return None


class Client:
    """Small HTTP wrapper with Smartsheet-aware retries."""

    def __init__(self, opener: Opener = _default_opener, sleep: Sleeper = time.sleep) -> None:
        self.opener = opener
        self.sleep = sleep
        self.requests = 0

    def send(
        self,
        url: str,
        *,
        method: str = "GET",
        headers: Dict[str, str],
        body: Any = None,
        service: str,
        timeout: float = 60.0,
    ) -> Any:
        data = None
        all_headers = {"Accept": "application/json", "User-Agent": USER_AGENT, **headers}
        if body is not None:
            data = json.dumps(body).encode("utf-8")
            all_headers["Content-Type"] = "application/json"
        attempt = 0
        while True:
            request = urllib.request.Request(url, data=data, headers=all_headers, method=method)
            self.requests += 1
            try:
                return _read_json(self.opener(request, timeout))
            except urllib.error.HTTPError as err:
                payload = _error_body(err)
                # A rate-limited or unavailable request was not processed, so
                # any method can be retried. Other 5xx errors are retried only
                # for GET and PUT, so a POST that may have landed never adds
                # duplicate rows or sheets.
                retryable = (
                    err.code in (429, 503)
                    or payload.get("errorCode") == 4003
                    or (err.code >= 500 and method in ("GET", "PUT"))
                )
                if retryable and attempt < MAX_RETRIES:
                    wait = _retry_after(err)
                    if err.code == 429 or payload.get("errorCode") == 4003:
                        floor = RATE_LIMIT_WAIT_SECONDS if service == "Smartsheet" else 1.0
                        wait = max(wait or 0.0, floor * (2 ** attempt))
                    else:
                        wait = wait if wait is not None else float(2 ** attempt)
                    attempt += 1
                    print(
                        f"{service} HTTP {err.code}; retrying in {wait:.0f}s "
                        f"(attempt {attempt} of {MAX_RETRIES}).",
                        file=sys.stderr,
                    )
                    self.sleep(wait)
                    continue
                raise _HTTPFailure(err.code, _error_detail(payload, err), payload) from None
            except urllib.error.URLError as err:
                raise SyncError(f"Could not reach {service}: {err.reason}") from None


class _HTTPFailure(Exception):
    def __init__(self, code: int, detail: str, body: Dict[str, Any]) -> None:
        super().__init__(detail)
        self.code = code
        self.detail = detail
        self.body = body


# --- FXMacroData -----------------------------------------------------------


def fetch_rows(args: argparse.Namespace, api_key: Optional[str], client: Client) -> List[Dict[str, Any]]:
    spec = DATASETS[args.dataset]
    headers: Dict[str, str] = {}
    if api_key:
        # Header, not query string: keys stay out of proxy and server logs.
        headers["X-API-Key"] = api_key
    rows: List[Dict[str, Any]] = []
    offset = 0
    for _ in range(1000):
        query = dict(spec["query"])
        if args.start_date:
            query["start_date"] = args.start_date
        if args.end_date:
            query["end_date"] = args.end_date
        if spec["paged"]:
            query.update({"limit": PAGE_LIMIT, "offset": offset})
        url = API_ORIGIN + spec["path"](args)
        if query:
            url += "?" + urllib.parse.urlencode(query)
        try:
            body = client.send(url, headers=headers, service="FXMacroData", timeout=30.0)
        except _HTTPFailure as err:
            hint = ""
            if err.code in (401, 403) and not api_key:
                hint = " Set FXMD_API_KEY; only USD indicators, calendar and COT are free."
            raise SyncError(f"FXMacroData HTTP {err.code}: {err.detail}{hint}".rstrip()) from None
        data = body.get("data") if isinstance(body, dict) else None
        if not isinstance(data, list):
            raise SyncError("FXMacroData returned an unexpected response.")
        for row in data:
            if len(rows) >= args.max_rows:
                break
            if isinstance(row, dict):
                rows.append(row)
        pagination = body.get("pagination") or {}
        if not spec["paged"] or not data or len(rows) >= args.max_rows or pagination.get("has_more") is not True:
            break
        next_offset = pagination.get("next_offset")
        offset = next_offset if isinstance(next_offset, int) else offset + len(data)
    return rows


def build_records(args: argparse.Namespace, rows: Iterable[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Map API rows to sheet records, dropping rows without a key and duplicate keys."""

    spec = DATASETS[args.dataset]
    records: List[Dict[str, Any]] = []
    seen = set()
    for row in rows:
        record = spec["row"](row, args)
        key = record.get(KEY)
        if not key or key in seen:
            continue
        seen.add(key)
        records.append(record)
    # The API returns newest first; add oldest first so the sheet reads in order.
    records.reverse()
    cells = len(records) * len(spec["columns"])
    if cells > MAX_CELLS_PER_SHEET:
        raise SyncError(
            f"{len(records):,} rows x {len(spec['columns'])} columns exceeds Smartsheet's "
            f"{MAX_CELLS_PER_SHEET:,}-cell sheet limit. Use --max-rows or --start-date."
        )
    return records


# --- Smartsheet ------------------------------------------------------------


class Smartsheet:
    def __init__(self, base: str, token: str, client: Client) -> None:
        self.base = base
        self.headers = {"Authorization": f"Bearer {token}"}
        self.client = client

    def call(self, method: str, path: str, query: Optional[Dict[str, Any]] = None, body: Any = None) -> Any:
        url = self.base + path
        if query:
            url += "?" + urllib.parse.urlencode(query)
        try:
            result = self.client.send(url, method=method, headers=self.headers, body=body, service="Smartsheet")
        except _HTTPFailure as err:
            code = err.body.get("errorCode")
            ref = err.body.get("refId")
            suffix = f" (errorCode {code}" + (f", refId {ref})" if ref else ")") if code else ""
            hint = ""
            if err.code == 401:
                hint = " Check SMARTSHEET_ACCESS_TOKEN and that --region matches the account's instance."
            elif err.code == 403:
                hint = " The token's user needs Editor (or higher) access to the sheet, folder or workspace."
            elif err.code == 404:
                hint = " Check the sheet, folder or workspace ID."
            detail = err.detail.strip().rstrip(".")
            raise SyncError(f"Smartsheet HTTP {err.code}: {detail}{suffix}.{hint}") from None
        if isinstance(result, dict) and result.get("resultCode") not in (None, 0):
            # resultCode 3 is a partial success; the row writes here never ask for it.
            raise SyncError(f"Smartsheet reported {result.get('message') or 'a partial failure'}.")
        return result

    # Sheet discovery uses the token-paginated children endpoints
    # (GET /workspaces/{id}/children, GET /folders/{id}/children).
    def find_sheet(self, container: str, container_id: str, name: str) -> Optional[int]:
        last_key: Optional[str] = None
        for _ in range(1000):
            query: Dict[str, Any] = {"childrenResourceTypes": "sheets", "maxItems": 100}
            if last_key:
                query["lastKey"] = last_key
            page = self.call("GET", f"/{container}/{container_id}/children", query) or {}
            for item in page.get("data") or []:
                if not isinstance(item, dict):
                    continue
                if str(item.get("resourceType") or "sheet").lower() != "sheet":
                    continue
                if item.get("name") == name and isinstance(item.get("id"), int):
                    return item["id"]
            last_key = page.get("lastKey")
            if not last_key:
                return None
        return None

    def create_sheet(self, container: str, container_id: str, name: str, columns: Sequence[Column]) -> int:
        body = {"name": name, "columns": [_column_body(c) for c in columns]}
        result = self.call("POST", f"/{container}/{container_id}/sheets", body=body) or {}
        sheet_id = (result.get("result") or {}).get("id")
        if not isinstance(sheet_id, int):
            raise SyncError("Smartsheet did not return the new sheet's ID.")
        return sheet_id

    def get_sheet(self, sheet_id: int) -> Dict[str, Any]:
        """Return the sheet with every row, following page-based paging if the API applies it."""

        sheet = self.call("GET", f"/sheets/{sheet_id}") or {}
        rows = list(sheet.get("rows") or [])
        total = sheet.get("totalRowCount")
        page_size = len(rows)
        page = 1
        while isinstance(total, int) and page_size and len(rows) < total and page < 1000:
            page += 1
            more = self.call("GET", f"/sheets/{sheet_id}", {"page": page, "pageSize": page_size}) or {}
            batch = more.get("rows") or []
            known = {r.get("id") for r in rows}
            fresh = [r for r in batch if r.get("id") not in known]
            if not fresh:
                break
            rows.extend(fresh)
        sheet["rows"] = rows
        return sheet

    def add_columns(self, sheet_id: int, columns: List[Tuple[int, Column]]) -> List[Dict[str, Any]]:
        body = [dict(_column_body(col), index=index) for index, col in columns]
        result = self.call("POST", f"/sheets/{sheet_id}/columns", body=body) or {}
        added = result.get("result") or []
        return added if isinstance(added, list) else [added]

    def add_rows(self, sheet_id: int, rows: List[Dict[str, Any]]) -> None:
        self.call("POST", f"/sheets/{sheet_id}/rows", body=rows)

    def update_rows(self, sheet_id: int, rows: List[Dict[str, Any]]) -> None:
        self.call("PUT", f"/sheets/{sheet_id}/rows", body=rows)


def _column_body(column: Column) -> Dict[str, Any]:
    title, col_type, extra = column
    body: Dict[str, Any] = {"title": title, "type": col_type}
    body.update(extra)
    return body


def _same(existing: Any, new: Any) -> bool:
    if existing is None or existing == "":
        return new is None or new == ""
    if isinstance(new, (int, float)) and isinstance(existing, (int, float)):
        return abs(float(existing) - float(new)) <= 1e-12 * max(1.0, abs(float(new)))
    return str(existing) == str(new)


def plan_upsert(
    sheet: Dict[str, Any], records: List[Dict[str, Any]], column_ids: Dict[str, int]
) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
    """Return (rows to add, rows to update) for records keyed on the Key column."""

    key_id = column_ids[KEY]
    existing: Dict[str, Tuple[int, Dict[int, Any]]] = {}
    for row in sheet.get("rows") or []:
        cells = {c.get("columnId"): c.get("value") for c in row.get("cells") or [] if isinstance(c, dict)}
        key = cells.get(key_id)
        if key is not None and str(key) not in existing and isinstance(row.get("id"), int):
            existing[str(key)] = (row["id"], cells)

    adds: List[Dict[str, Any]] = []
    updates: List[Dict[str, Any]] = []
    for record in records:
        match = existing.get(str(record[KEY]))
        if match is None:
            cells = [
                {"columnId": column_ids[title], "value": value}
                for title, value in record.items()
                if value is not None and title in column_ids
            ]
            adds.append({"toBottom": True, "cells": cells})
            continue
        row_id, current = match
        changed = []
        for title, value in record.items():
            if title == KEY or title not in column_ids:
                continue
            if not _same(current.get(column_ids[title]), value):
                changed.append({"columnId": column_ids[title], "value": "" if value is None else value})
        if changed:
            updates.append({"id": row_id, "cells": changed})
    return adds, updates


def _batches(items: List[Dict[str, Any]], size: int) -> Iterable[List[Dict[str, Any]]]:
    for start in range(0, len(items), size):
        yield items[start : start + size]


def ensure_columns(api: Smartsheet, sheet_id: int, sheet: Dict[str, Any], columns: Sequence[Column]) -> Dict[str, int]:
    present = {c.get("title"): c for c in sheet.get("columns") or [] if isinstance(c, dict)}
    key_col = present.get(KEY)
    if key_col is None:
        raise SyncError(
            f'The sheet has no "{KEY}" column. Point --sheet-id at a sheet this tool created, '
            f'or rename the primary column to "{KEY}".'
        )
    if key_col.get("type") not in (None, "TEXT_NUMBER"):
        raise SyncError(f'The "{KEY}" column must be a Text/Number column.')
    missing = [col for col in columns if col[0] not in present]
    if missing:
        start = len(present)
        added = api.add_columns(sheet_id, [(start + i, col) for i, col in enumerate(missing)])
        for col in added:
            if isinstance(col, dict):
                present[col.get("title")] = col
        still = [col[0] for col in missing if col[0] not in present]
        if still:
            raise SyncError(f"Smartsheet did not add the columns: {', '.join(still)}.")
    return {title: col["id"] for title, col in present.items() if isinstance(col.get("id"), int)}


def sync(args: argparse.Namespace, records: List[Dict[str, Any]], api: Smartsheet) -> Dict[str, Any]:
    spec = DATASETS[args.dataset]
    name = args.sheet_name or spec["sheet_name"](args)
    created = False
    sheet_id: Optional[int] = args.sheet_id
    if sheet_id is None:
        container, container_id = ("workspaces", args.workspace_id) if args.workspace_id else ("folders", args.folder_id)
        sheet_id = api.find_sheet(container, str(container_id), name)
        if sheet_id is None:
            sheet_id = api.create_sheet(container, str(container_id), name, spec["columns"])
            created = True
    sheet = api.get_sheet(sheet_id)
    column_ids = ensure_columns(api, sheet_id, sheet, spec["columns"])
    adds, updates = plan_upsert(sheet, records, column_ids)
    existing_rows = len(sheet.get("rows") or [])
    columns_after = len(column_ids)
    if (existing_rows + len(adds)) * columns_after > MAX_CELLS_PER_SHEET:
        raise SyncError(
            f"Adding {len(adds):,} rows would take the sheet past Smartsheet's "
            f"{MAX_CELLS_PER_SHEET:,}-cell limit. Use --start-date or a new sheet."
        )
    for batch in _batches(updates, args.batch_size):
        api.update_rows(sheet_id, batch)
    for batch in _batches(adds, args.batch_size):
        api.add_rows(sheet_id, batch)
    return {
        "sheet_id": sheet_id,
        "sheet_name": sheet.get("name") or name,
        "permalink": sheet.get("permalink"),
        "created": created,
        "added": len(adds),
        "updated": len(updates),
        "unchanged": len(records) - len(adds) - len(updates),
    }


# --- CLI -------------------------------------------------------------------


def _valid_date(value: str) -> str:
    try:
        if len(value) != 10:
            raise ValueError(value)
        datetime.strptime(value, "%Y-%m-%d")
    except ValueError:
        raise argparse.ArgumentTypeError("dates must be YYYY-MM-DD") from None
    return value


def _bounded(low: int, high: int) -> Callable[[str], int]:
    def parse(value: str) -> int:
        try:
            number = int(value)
        except ValueError:
            raise argparse.ArgumentTypeError("must be a whole number") from None
        if not low <= number <= high:
            raise argparse.ArgumentTypeError(f"must be between {low} and {high}")
        return number

    return parse


def _object_id(value: str) -> int:
    if not value.isdigit():
        raise argparse.ArgumentTypeError("Smartsheet IDs are numbers")
    return int(value)


def parse_args(argv: Optional[Sequence[str]] = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Sync an FXMacroData dataset into a Smartsheet sheet.")
    parser.add_argument("dataset", choices=sorted(DATASETS))
    parser.add_argument("--currency")
    parser.add_argument("--indicator")
    parser.add_argument("--base")
    parser.add_argument("--quote")
    parser.add_argument("--commodity")
    parser.add_argument("--start-date", type=_valid_date)
    parser.add_argument("--end-date", type=_valid_date)
    parser.add_argument("--max-rows", type=_bounded(1, MAX_SYNC_ROWS), default=5000)
    target = parser.add_mutually_exclusive_group()
    target.add_argument("--sheet-id", type=_object_id, help="Existing sheet to upsert into.")
    target.add_argument("--workspace-id", type=_object_id, help="Find or create the sheet in this workspace.")
    target.add_argument("--folder-id", type=_object_id, help="Find or create the sheet in this folder.")
    parser.add_argument("--sheet-name", help="Sheet name to find or create (default depends on the dataset).")
    parser.add_argument("--region", choices=sorted(SMARTSHEET_HOSTS), default=os.environ.get("SMARTSHEET_REGION", "us"))
    parser.add_argument(
        "--batch-size",
        type=_bounded(1, MAX_ROWS_PER_REQUEST),
        default=MAX_ROWS_PER_REQUEST,
        help="Rows per add/update request (max 500).",
    )
    parser.add_argument("--dry-run", action="store_true", help="Print the rows as JSON instead of writing to Smartsheet.")
    args = parser.parse_args(argv)
    if args.region not in SMARTSHEET_HOSTS:
        parser.error(f"--region must be one of {', '.join(sorted(SMARTSHEET_HOSTS))}")
    missing = [name for name in DATASETS[args.dataset]["requires"] if not getattr(args, name)]
    if missing:
        parser.error(f"{args.dataset} needs --{' --'.join(missing)}")
    if not args.dry_run and not (args.sheet_id or args.workspace_id or args.folder_id):
        parser.error("set --sheet-id, --workspace-id or --folder-id (or use --dry-run)")
    return args


def main(
    argv: Optional[Sequence[str]] = None,
    *,
    opener: Opener = _default_opener,
    sleep: Sleeper = time.sleep,
) -> int:
    args = parse_args(argv)
    api_key = (os.environ.get("FXMD_API_KEY") or "").strip() or None
    token = (os.environ.get("SMARTSHEET_ACCESS_TOKEN") or "").strip()
    if not args.dry_run and not token:
        print(
            "Set SMARTSHEET_ACCESS_TOKEN to a Smartsheet API access token "
            "(Account > Personal settings > API access).",
            file=sys.stderr,
        )
        return 2
    client = Client(opener, sleep)
    try:
        rows = fetch_rows(args, api_key, client)
        records = build_records(args, rows)
        if args.dry_run:
            json.dump(records, sys.stdout, indent=2)
            sys.stdout.write("\n")
            return 0
        api = Smartsheet(SMARTSHEET_HOSTS[args.region], token, client)
        result = sync(args, records, api)
    except SyncError as err:
        print(str(err), file=sys.stderr)
        return 1
    action = "Created" if result["created"] else "Updated"
    print(
        f"{action} sheet {result['sheet_name']!r} ({result['sheet_id']}): "
        f"{result['added']} added, {result['updated']} updated, {result['unchanged']} unchanged."
    )
    if result.get("permalink"):
        print(result["permalink"])
    return 0


if __name__ == "__main__":
    sys.exit(main())