"""Backfill ``cmdb_cost_facts`` (and ``cmdb_cloud_billing_daily``) across
every enabled cloud integration, over a multi-month historical window.

Why this script exists
======================

The platform's cost-ingestion crons run **daily, for yesterday only**:

* ``cmdb.billing_daily_sync`` writes per-(integration, scope, date)
  aggregates into ``cmdb_cloud_billing_daily``.
* ``cmdb.cost_facts_sync`` writes line-item (date × resource × usage
  type) rows into ``cmdb_cost_facts`` — the ground-truth fact table.

Both crons are scoped to "yesterday" (or "today − 1") by design, so a
fresh deploy or a long-stale integration only sees today's data going
forward. The result is that historical months (3 / 4 / 5 months back)
show $0 on every billing surface — the AK spend panel, the
subscription billing-detail drawer, the admin cloud overview cost
ranking — even when the vendor side has the rows.

This script does a one-shot multi-month backfill against ALL enabled
cloud integrations, walking the same ingestion code paths the crons
use so the data semantics are identical to a healthy steady-state
deployment.

What it does
============

For every enabled ``cloud``-kind integration:

1. ``billing_daily``: walks the requested window in 30-day chunks
   (AWS Cost Explorer caps at ~30-day windows per API call for
   resource-level queries; Azure Cost Management has no hard cap but
   responses get unwieldy past 30 days). Each chunk routes through
   :func:`sync_billing_daily_for_integration`, the same function the
   cron uses; idempotent ON DUPLICATE KEY UPDATE so re-running is a
   no-op.
2. ``cost_facts``: walks the requested window in 30-day chunks via
   :func:`sync_cost_facts_for_integration`. Same idempotent upsert
   path the cron writes; the ``fact_key`` SHA256 of
   (date|integration|scope|resource|usage_type) carries the UNIQUE
   constraint so a re-run replaces the row's amount without
   duplicating.
3. After all integrations are processed, runs a single
   :func:`recompute_allocation` over the full window so the
   ``cost_owner_*`` and ``cost_center`` columns reflect every active
   ``CmdbCostAllocationRule`` AND the up-to-date
   ``cmdb_cloud_scopes.owner_person_id`` fallback. Without this step
   a backfilled fact carries owner=None from the ingest path and the
   "未分摊" KPI inflates artificially.

Window contract
===============

* Default: trailing 6 calendar months (today − 6 months, first-of-
  month at the start; today as the end). Six months balances "enough
  trend to spot quarterly drift" against the vendors' retention
  windows — AWS Cost Explorer 14 months, Azure 13 months, Aliyun BSS
  6-12 months. Run with ``--months 12`` for a year if the vendor
  retention allows it.

* Hard ceiling: 14 months (~425 days) — same bound the
  ``/cmdb/cost/backfill`` HTTP endpoint enforces. Older windows would
  spam every cloud's API with mostly-empty queries.

* Chunking: 30-day windows passed to the per-integration call. Inside
  the call the driver may further sub-chunk (AWS resource-level →
  14 days; Aliyun → 1 day; etc.); we don't second-guess the driver.

Idempotent
==========

Re-running with the same window is a no-op for any (date, integration,
scope, resource, usage_type) tuple already present. The
``sync_*_for_integration`` paths use MySQL ``INSERT ... ON DUPLICATE
KEY UPDATE`` so the second run refreshes amounts (in case the vendor
restated a day) without inserting duplicates.

Per-integration lock
====================

Each ``sync_*_for_integration`` call enters the same
``cloud_sync:billing_daily:{id}`` / ``cloud_sync:cost_facts:{id}``
distributed lock the cron and HTTP backfill paths use. If a cron tick
or another admin's backfill is running for the SAME integration, the
script's call returns ``status='skipped'`` with a clear reason and we
log it; the script does NOT fail the whole run for a single skipped
integration.

Production safety
=================

Read-mostly: the script writes ONLY to ``cmdb_cost_facts`` /
``cmdb_cloud_billing_daily`` / (during recompute) ``cmdb_cost_facts``
column updates, plus ``audit_logs`` invocation traces (below). No
mutations to ``identity_persons``, ``cmdb_cloud_scopes``, integration
credentials, or anything else.

Audit: since 2026-07-28 the SERVICE layer
(``sync_cost_facts_for_integration``) writes one
``cost.facts_sync.window_invoked`` audit row (``actor_type=system``,
detail carries window + caller) per explicit-window chunk — a prod
cost-facts reingest ran through this script's path with zero
``audit_logs`` / ``scheduled_job_runs`` footprint and cost a forensic
round to attribute. Expect ~(window / 30d) rows per integration per
run; that is the intended trace, not noise.

Usage
=====

::

  # local dev: default 6-month window
  uv run python scripts/backfill_cost_facts.py

  # prod: explicit URL guard against pointing at the wrong DB
  HELIX_PROD_DB_URL_ASYNC='mysql+aiomysql://...@tidbcloud.com:4000/helix?ssl=true' \
    uv run python scripts/backfill_cost_facts.py --prod

  # custom window — 4 months back
  uv run python scripts/backfill_cost_facts.py --months 4

  # explicit dates (must be within retention)
  uv run python scripts/backfill_cost_facts.py \
    --start 2025-12-01 --end 2026-04-30

  # filter to one integration (debugging)
  uv run python scripts/backfill_cost_facts.py \
    --integration-id <uuid>

  # skip allocation recompute (rarely useful — recompute is fast and
  # without it the owner columns stay stale on backfilled rows)
  uv run python scripts/backfill_cost_facts.py --skip-recompute

  # dry-run prints the plan but issues no vendor API calls
  uv run python scripts/backfill_cost_facts.py --dry-run

Exit codes
==========

0  All integrations processed (some may have been skipped — that's
   reported, not a failure).
1  At least one integration errored. Summary lists which.
2  Environment / configuration error caught before any sync call.
"""
from __future__ import annotations

import argparse
import asyncio
import json
import logging
import os
import sys
from dataclasses import dataclass, field
from datetime import UTC, date, datetime, timedelta
from typing import Any

# Repo-relative imports — run from the server/ directory or via
# ``uv run python scripts/backfill_cost_facts.py``.
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

from sqlalchemy import select  # noqa: E402
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine  # noqa: E402
from sqlalchemy.orm import sessionmaker  # noqa: E402

# Eagerly import every model module so SQLAlchemy's mapper resolver
# sees every class before any query triggers a configure pass. Same
# pattern as the other backfill scripts under server/scripts/.
from app.approval import models as _approval_models  # noqa: E402, F401
from app.audit import models as _audit_models  # noqa: E402, F401
from app.auth import models as _auth_models  # noqa: E402, F401
from app.compliance import models as _compliance_models  # noqa: E402, F401
from app.core.safe_logging import get_safe_logger
from app.domains.apikey import models as _apikey_models  # noqa: E402, F401
from app.domains.asset import models as _asset_models  # noqa: E402, F401
from app.domains.bailian import models as _bailian_models  # noqa: E402, F401
from app.domains.cmdb.cost_service import (  # noqa: E402
    recompute_allocation,
    sync_cost_facts_for_integration,
)
from app.domains.cmdb.lifecycle_service import (  # noqa: E402
    sync_billing_daily_for_integration,
)
from app.domains.email_group import models as _email_group_models  # noqa: E402, F401
from app.domains.hr import models as _hr_models  # noqa: E402, F401
from app.identity.person_models import IdentityPerson  # noqa: E402, F401

# Importing this package fires every ``@register_driver`` decorator so
# ``get_driver_class("cloud", <provider>)`` resolves at runtime. Without
# it, the per-integration sync paths see a None driver class and raise
# ``CLOUD_DRIVER_UNKNOWN`` for every chunk (caught on the first prod
# real-run — the registry stays empty without this import because the
# script doesn't go through ``app.main``'s startup chain).
from app.integrations import drivers as _drivers  # noqa: E402, F401
from app.integrations.models import Integration  # noqa: E402
from app.notifications import models as _notification_models  # noqa: E402, F401
from app.scheduled_jobs import models as _scheduled_jobs_models  # noqa: E402, F401

logger = get_safe_logger("backfill_cost_facts")
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)

LOCAL_DB_URL = "mysql+aiomysql://helix:helix@localhost:3306/helix"

# Chunk size for per-call windows. 30 days keeps responses inside every
# vendor's preferred window — AWS Cost Explorer resource-level caps
# at 14 days but the driver sub-chunks below 30 anyway; Azure /
# Aliyun handle 30 cleanly. Smaller chunks just multiply API calls
# without benefit.
_CHUNK_DAYS = 30

# Hard cap on retroactive window — matches the HTTP backfill endpoint
# (cost_router.py:backfill_endpoint). Refuse anything older.
_MAX_RETRO_DAYS = 425  # ~14 months


@dataclass
class IntegrationResult:
    integration_id: str
    name: str
    provider: str
    billing_daily: list[dict[str, Any]] = field(default_factory=list)
    cost_facts: list[dict[str, Any]] = field(default_factory=list)
    errors: list[str] = field(default_factory=list)

    @property
    def status(self) -> str:
        if self.errors:
            return "error"
        # Any chunk returned status != ok / skipped / not_supported
        # is treated as a partial failure for the integration.
        statuses = (
            [r.get("status", "unknown") for r in self.billing_daily]
            + [r.get("status", "unknown") for r in self.cost_facts]
        )
        if any(
            s not in ("ok", "skipped", "not_supported", "partial_success")
            for s in statuses
        ):
            return "partial_failure"
        return "ok"


def _resolve_db_url(*, use_prod: bool) -> str:
    """Pick the right DB URL with explicit prod gating.

    Prod requires ``HELIX_PROD_DB_URL_ASYNC`` to be exported so an
    operator can't run ``--prod`` with a stale ``DATABASE_URL``
    pointing at local and clobber zero data with cron-style writes.
    """
    if use_prod:
        url = (
            os.environ.get("HELIX_PROD_DB_URL_ASYNC")
            or os.environ.get("HELIX_PROD_DB_URL")
        )
        if not url:
            print(
                "[fatal] --prod requires HELIX_PROD_DB_URL_ASYNC "
                "(or HELIX_PROD_DB_URL); refusing to fall back to "
                "DATABASE_URL.",
                file=sys.stderr,
            )
            sys.exit(2)
        # Coerce pymysql / sync drivers to aiomysql so this script
        # (which uses AsyncSession end-to-end) doesn't crash mid-run.
        if url.startswith("mysql+pymysql://"):
            url = url.replace("mysql+pymysql://", "mysql+aiomysql://", 1)
        elif url.startswith("mysql://"):
            url = url.replace("mysql://", "mysql+aiomysql://", 1)
        elif url.startswith("mysql+asyncmy://"):
            url = url.replace("mysql+asyncmy://", "mysql+aiomysql://", 1)
        return url
    # Local default — same value docker-compose.yml + tests/conftest.py use.
    return os.environ.get("DATABASE_URL") or LOCAL_DB_URL


def _resolve_window(
    *, months: int, start: str | None, end: str | None,
) -> tuple[date, date]:
    """Pick (start_date, end_date) from CLI args.

    Explicit ``--start`` / ``--end`` win when both are passed (both
    required if either is). Otherwise default to the trailing
    ``--months`` calendar months ending today.
    """
    if start and end:
        try:
            s = date.fromisoformat(start)
            e = date.fromisoformat(end)
        except ValueError as exc:
            print(
                f"[fatal] --start / --end must be ISO YYYY-MM-DD: {exc}",
                file=sys.stderr,
            )
            sys.exit(2)
        return s, e
    if start or end:
        print(
            "[fatal] pass both --start and --end, or neither.",
            file=sys.stderr,
        )
        sys.exit(2)

    today = datetime.now(UTC).date()
    # Walk back ``months`` calendar months from today; anchor at the
    # first of that month so the window covers complete months.
    y, m = today.year, today.month
    for _ in range(months):
        m -= 1
        if m == 0:
            m = 12
            y -= 1
    start_dt = date(y, m, 1)
    return start_dt, today


def _validate_window(start_dt: date, end_dt: date) -> None:
    if start_dt > end_dt:
        print(
            f"[fatal] start ({start_dt}) > end ({end_dt}).",
            file=sys.stderr,
        )
        sys.exit(2)
    today = datetime.now(UTC).date()
    earliest_allowed = today - timedelta(days=_MAX_RETRO_DAYS)
    if start_dt < earliest_allowed:
        print(
            f"[fatal] start_date {start_dt} is older than the supported "
            f"backfill horizon ({earliest_allowed}, ~14 months). Cloud "
            "billing APIs (AWS CE / Azure CM / Aliyun BSS) do not "
            "retain detail older than this.",
            file=sys.stderr,
        )
        sys.exit(2)


def _chunks(
    start_dt: date, end_dt: date, *, chunk_days: int = _CHUNK_DAYS,
) -> list[tuple[date, date]]:
    """Split [start, end] into ``chunk_days``-day windows.

    ``chunk_days`` is operator-tunable (``--chunk-days``) because the
    per-scope guard inside ``sync_cost_facts_for_integration``
    (``_COST_FACTS_PER_SCOPE_TIMEOUT_SEC`` = 300s) is sized for the
    DAILY cron's 1-3 day window, not a 30-day backfill chunk. The two
    heaviest prod AWS accounts (632330501419, 385595570414) blew that
    timeout on every 30-day chunk in the 2026-07-28 backfill — the
    chunk came back ``partial_failure`` and those accounts' historical
    rows silently kept ``amount_usd_list = NULL``. Shrink the chunk
    (``--chunk-days 7``) for heavy tenants instead of raising the
    platform timeout, so the daily cron's guard stays untouched.
    """
    out: list[tuple[date, date]] = []
    cur = start_dt
    while cur <= end_dt:
        chunk_end = min(cur + timedelta(days=chunk_days - 1), end_dt)
        out.append((cur, chunk_end))
        cur = chunk_end + timedelta(days=1)
    return out


def _collect_chunk_failure(
    *, surface: str, start: date, end: date, payload: dict[str, Any],
) -> str | None:
    """Render a one-line error for a chunk whose RETURNED status is not
    a success shape, naming the offending scopes.

    Why this exists: ``sync_*_for_integration`` never raises on a
    per-scope timeout / driver error — it returns
    ``status='partial_failure'`` with the detail in ``errors`` /
    ``per_scope``. The pre-fix script only appended to
    ``result.errors`` from the ``except`` branch, so a returned
    ``partial_failure`` printed one INFO line, left ``errors`` empty,
    and the script **exited 0** while a whole account's list measure
    stayed NULL (2026-07-28 prod backfill: 8 AWS chunks, 2 accounts,
    found only by a manual facts-vs-daily parity query). Surfacing the
    scope names here + the exit-code change below makes that loud.
    """
    status = str(payload.get("status") or "unknown")
    if status in ("ok", "skipped", "not_supported", "partial_success", "dry_run"):
        return None
    failed_scopes = [
        f"{row.get('scope') or '<tenant>'}({row.get('status')})"
        for row in (payload.get("per_scope") or [])
        if str(row.get("status")) not in ("ok", "skipped", "not_supported")
    ]
    detail = payload.get("reason") or "; ".join(
        str(x) for x in (payload.get("errors") or [])
    )
    parts = [f"{surface} {start}..{end}: status={status}"]
    if failed_scopes:
        parts.append(f"scopes={', '.join(failed_scopes[:8])}")
    if detail:
        parts.append(f"detail={str(detail)[:400]}")
    return " | ".join(parts)


async def _list_cloud_integrations(
    db: AsyncSession, *, only_id: str | None,
) -> list[Integration]:
    """Enabled cloud-kind integrations.

    ``Integration`` has no ``deleted_at`` column — soft-disable is
    expressed via ``enabled=False``. Filter on enabled rows so a row
    explicitly disabled by an admin doesn't get hammered with vendor
    API calls on every backfill run.
    """
    q = select(Integration).where(
        Integration.kind == "cloud",
        Integration.enabled.is_(True),
    )
    if only_id:
        import uuid
        try:
            q = q.where(Integration.id == str(uuid.UUID(only_id)))
        except (ValueError, TypeError):
            print(
                f"[fatal] --integration-id {only_id!r} is not a UUID.",
                file=sys.stderr,
            )
            sys.exit(2)
    rows = list((await db.execute(q)).scalars().all())
    return rows


async def _run_billing_daily_chunks(
    integ: Integration,
    chunks: list[tuple[date, date]],
    *,
    dry_run: bool,
    result: IntegrationResult,
) -> None:
    for (s, e) in chunks:
        if dry_run:
            logger.info(
                "  [dry-run] billing_daily %s %s..%s",
                integ.name, s, e,
            )
            result.billing_daily.append({
                "status": "dry_run", "start": s.isoformat(), "end": e.isoformat(),
            })
            continue
        try:
            # ``dispatch_alerts=False`` — historical-window replay.
            # Without this, evaluate_scope_alert_rules + _alert_spike
            # treat each chunk's end_date as "today is end_date" and
            # fire a Feishu card per (scope, crossed-threshold,
            # historical-month) tuple. Prod incident 2026-05-16 issued
            # ~7 cards per scope across 60 scopes for this reason.
            # Slot claims still land — see
            # :func:`sync_billing_daily_for_integration` docstring.
            r = await sync_billing_daily_for_integration(
                str(integ.id), start_date=s, end_date=e,
                dispatch_alerts=False,
            )
            result.billing_daily.append({
                "start": s.isoformat(), "end": e.isoformat(), **r,
            })
            logger.info(
                "  billing_daily %s %s..%s → %s",
                integ.name, s, e, r.get("status"),
            )
            failure = _collect_chunk_failure(
                surface="billing_daily", start=s, end=e, payload=r,
            )
            if failure:
                logger.error("  [chunk-failure] %s", failure)
                result.errors.append(failure)
        except Exception as exc:  # noqa: BLE001 — log + continue per chunk
            msg = f"billing_daily {s}..{e}: {type(exc).__name__}: {exc!s}"
            logger.exception("  [error] %s", msg)
            result.errors.append(msg)


async def _run_cost_facts_chunks(
    integ: Integration,
    chunks: list[tuple[date, date]],
    *,
    dry_run: bool,
    result: IntegrationResult,
) -> None:
    for (s, e) in chunks:
        if dry_run:
            logger.info(
                "  [dry-run] cost_facts %s %s..%s",
                integ.name, s, e,
            )
            result.cost_facts.append({
                "status": "dry_run", "start": s.isoformat(), "end": e.isoformat(),
            })
            continue
        try:
            r = await sync_cost_facts_for_integration(
                str(integ.id), start_date=s, end_date=e,
            )
            result.cost_facts.append({
                "start": s.isoformat(), "end": e.isoformat(), **r,
            })
            logger.info(
                "  cost_facts   %s %s..%s → %s (rows=%s)",
                integ.name, s, e, r.get("status"), r.get("rows_upserted"),
            )
            failure = _collect_chunk_failure(
                surface="cost_facts", start=s, end=e, payload=r,
            )
            if failure:
                logger.error("  [chunk-failure] %s", failure)
                result.errors.append(failure)
        except Exception as exc:  # noqa: BLE001 — log + continue per chunk
            msg = f"cost_facts {s}..{e}: {type(exc).__name__}: {exc!s}"
            logger.exception("  [error] %s", msg)
            result.errors.append(msg)


async def main_async(args: argparse.Namespace) -> int:
    db_url = _resolve_db_url(use_prod=args.prod)
    start_dt, end_dt = _resolve_window(
        months=args.months, start=args.start, end=args.end,
    )
    _validate_window(start_dt, end_dt)
    chunks = _chunks(start_dt, end_dt, chunk_days=max(1, args.chunk_days))
    logger.info(
        "Window: %s..%s (%d days, %d chunks)",
        start_dt, end_dt, (end_dt - start_dt).days + 1, len(chunks),
    )
    if args.prod:
        logger.warning("PROD MODE — writes go to the production database.")

    # TiDB Cloud needs SSL + ``?ssl=true`` stripped from the URL (the async
    # driver cannot parse that compatibility query parameter). The app's engine in
    # ``app.core.database`` already does this; we reuse the helper to
    # stay symmetric. ``--prod`` callers pass an explicit URL via env
    # var and override what's in ``settings``; for that case we build
    # an engine fresh with the same SSL handling. Local-dev URL
    # ``mysql://helix:helix@localhost:3306/...`` has neither
    # ``tidbcloud.com`` nor ``ssl=true``, so connect_args is empty.
    from app.core.database_tls import database_engine_options
    sanitised_url, connect_args = database_engine_options(db_url)
    engine = create_async_engine(
        sanitised_url, pool_pre_ping=True, connect_args=connect_args,
    )
    Session = sessionmaker(  # noqa: N806
        engine, expire_on_commit=False, class_=AsyncSession,
    )
    results: list[IntegrationResult] = []
    try:
        async with Session() as db:
            integs = await _list_cloud_integrations(
                db, only_id=args.integration_id,
            )
        if not integs:
            print("[fatal] no enabled cloud integrations to backfill.",
                  file=sys.stderr)
            return 2
        logger.info("Found %d cloud integration(s)", len(integs))
        for integ in integs:
            logger.info(
                "==> %s (%s, id=%s)",
                integ.name, integ.provider, integ.id,
            )
            result = IntegrationResult(
                integration_id=str(integ.id),
                name=integ.name,
                provider=integ.provider,
            )
            await _run_billing_daily_chunks(
                integ, chunks, dry_run=args.dry_run, result=result,
            )
            await _run_cost_facts_chunks(
                integ, chunks, dry_run=args.dry_run, result=result,
            )
            results.append(result)
            logger.info("  → %s", result.status)

        # Allocation recompute over the full window — runs once at the
        # end, not per-integration, since the recompute walks every
        # active rule across all facts in the date range. Skipping
        # leaves owner_columns stale on backfilled rows.
        #
        # ``recompute_allocation`` accepts ``days`` (counted back from
        # today UTC), not a date range, so we translate the window:
        # if the backfill started N days ago, walk back the same N
        # (rounded up) so every backfilled fact is in scope. The
        # function caps at the requested ``days`` window — older facts
        # outside this run are left untouched.
        if not args.dry_run and not args.skip_recompute:
            today = datetime.now(UTC).date()
            window_days = max(1, (today - start_dt).days + 1)
            logger.info(
                "Recomputing allocation over trailing %d days",
                window_days,
            )
            async with Session() as db:
                rc = await recompute_allocation(db, days=window_days)
                await db.commit()
                logger.info("  allocation recompute → %s", rc)
        elif args.skip_recompute:
            logger.info("Skipping allocation recompute (--skip-recompute)")
    finally:
        await engine.dispose()

    print("\n── Summary ──")
    summary_rows = [
        {
            "name": r.name,
            "provider": r.provider,
            "status": r.status,
            "billing_daily_chunks": len(r.billing_daily),
            "cost_facts_chunks": len(r.cost_facts),
            "errors": r.errors,
        }
        for r in results
    ]
    print(json.dumps(summary_rows, indent=2, default=str))
    # Exit non-zero on ANY non-ok integration, not just on raised
    # exceptions. A returned ``partial_failure`` (per-scope timeout /
    # driver error inside the sync path) leaves that scope's measures
    # unwritten — reporting success there is the silent-gap shape the
    # 2026-07-28 prod backfill hit. ``_collect_chunk_failure`` has
    # already put the offending scopes into ``r.errors``, so the
    # operator sees WHICH account to re-run (with a smaller
    # ``--chunk-days``) rather than a bare non-zero exit.
    failed = [r for r in results if r.errors or r.status != "ok"]
    if failed:
        print(
            "\n[FAILED] {} integration(s) did not fully backfill: {}".format(
                len(failed),
                ", ".join(f"{r.name}({r.status})" for r in failed),
            )
        )
        print(
            "Re-run the affected integration with a smaller window, e.g.\n"
            "  --integration-id <uuid> --chunk-days 7 --start ... --end ...\n"
            "then confirm with scripts/verify_cost_dual_measure_backfill.py."
        )
        return 1
    return 0


def main() -> int:
    parser = argparse.ArgumentParser(
        description=(
            "Backfill cmdb_cost_facts + cmdb_cloud_billing_daily over a "
            "multi-month window across every enabled cloud integration."
        ),
    )
    parser.add_argument(
        "--prod", action="store_true",
        help="Run against prod (requires HELIX_PROD_DB_URL_ASYNC).",
    )
    parser.add_argument(
        "--months", type=int, default=6,
        help="Trailing months to backfill (default: 6).",
    )
    parser.add_argument(
        "--start", type=str, default=None,
        help="Explicit window start YYYY-MM-DD (use with --end).",
    )
    parser.add_argument(
        "--end", type=str, default=None,
        help="Explicit window end YYYY-MM-DD (use with --start).",
    )
    parser.add_argument(
        "--integration-id", type=str, default=None,
        help="Restrict to one integration UUID (debugging).",
    )
    parser.add_argument(
        "--chunk-days", type=int, default=_CHUNK_DAYS,
        help=(
            f"Days per vendor-API chunk (default: {_CHUNK_DAYS}). Shrink "
            "to 7 for heavy tenants whose per-scope cost-facts query "
            "exceeds the 300s per-scope guard on 30-day windows."
        ),
    )
    parser.add_argument(
        "--dry-run", action="store_true",
        help="Print the plan; do not call any vendor billing API.",
    )
    parser.add_argument(
        "--skip-recompute", action="store_true",
        help="Skip the post-backfill allocation recompute.",
    )
    args = parser.parse_args()
    return asyncio.run(main_async(args))


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