#!/usr/bin/env python
"""Run one or more connectors end to end (fetch → store_raw → normalize → validate → staging parquet) and print a
per-spec summary: rows, countries, period range, forecast rows, status/message.

    CA_DATA_DIR=~/countryatlas-data .venv/bin/python scripts/run_connector.py who fred bis ilo [--indicator SLUG]
                                                                             [--mode fetch|normalize] [--concurrency N]

Uses the pipeline's `run_fetch` (same code path as `ca fetch`), so the staging files land in
`staging/<connector>/<dataset>__<code>__<indicator>.parquet` with the .run.json / .meta.json / .issues.json sidecars.
FRED calls are serialised by the connector itself (≥ 1.1 s spacing), whatever the concurrency.
"""
from __future__ import annotations

import argparse
import logging
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))

import polars as pl

from countryatlas.config import settings
from countryatlas.pipeline.fetch import run_fetch
from countryatlas.pipeline.staging import spec_paths
from countryatlas.registry import source_specs


def summarize(connector: str, indicator: str | None) -> list[str]:
    lines = []
    for spec in source_specs(connector=connector, indicator=indicator):
        p = spec_paths(spec)["parquet"]
        label = f"{spec.dataset}:{spec.code} → {spec.indicator_id}"
        if not p.exists():
            lines.append(f"  ✗ {label}: no staging file")
            continue
        df = pl.read_parquet(p)
        if df.is_empty():
            lines.append(f"  ✗ {label}: empty")
            continue
        n_c = df["country_id"].n_unique()
        pmin, pmax = df["period"].min(), df["period"].max()
        n_fc = int(df["is_forecast"].sum())
        n_q = int((df["status"] == "quarantined").sum())
        n_w = int((df["status"] == "warning").sum())
        freq = ",".join(sorted(df["frequency"].unique().to_list()))
        lines.append(
            f"  ✓ {label}: {df.height} rows, {n_c} countries, {pmin}…{pmax} [{freq}]"
            + (f", {n_fc} forecast" if n_fc else "")
            + (f", {n_q} quarantined" if n_q else "")
            + (f", {n_w} warnings" if n_w else "")
        )
    return lines


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("connectors", nargs="+", help="connector ids (who fred bis ilo …)")
    ap.add_argument("--indicator", default=None)
    ap.add_argument("--mode", choices=["fetch", "normalize"], default="fetch")
    ap.add_argument("--concurrency", type=int, default=None)
    ap.add_argument("-v", "--verbose", action="store_true")
    args = ap.parse_args()
    logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO,
                        format="%(asctime)s %(levelname)s %(name)s: %(message)s")
    logging.getLogger("httpx").setLevel(logging.WARNING)
    print(f"data dir: {settings.data_dir}")
    summary = run_fetch(connectors=args.connectors, indicator=args.indicator, mode=args.mode, concurrency=args.concurrency)
    print(f"\nrun {summary.run_id}: {len(summary.ok)} ok / {len(summary.runs)} specs in {summary.duration_s:.0f}s")
    for cid in args.connectors:
        print(f"\n[{cid}]")
        for line in summarize(cid, args.indicator):
            print(line)
        for r in summary.runs:
            if r.connector == cid and r.status not in ("ok",):
                print(f"  ! {r.status.upper()} {r.dataset}: {r.message}")
    return 0 if not summary.failed else 1


if __name__ == "__main__":
    raise SystemExit(main())