"""
End-to-end pipeline orchestrator for ICEA/DECEA surface meteorology.

Workflow:
    1. Check existing coverage in SQLite (``db.get_coverage``).
    2. Forward pass  — download new data after ``ex_max`` (update).
    3. Backward pass — download history before ``ex_min`` (auto-stop on empty years).
    4. Compute analytics from the quarterly CSV files.
    5. Upsert into SQLite and write the analytics CSV backup to PREPROC_DIR.
    6. Spot-check validation against the quarterly source files.
    7. Remove quarterly intermediate CSVs (if validation passed and cleanup enabled).

Example:
    python pipeline.py --aerodrome SBGR --all-years
    python pipeline.py --aerodrome SBGR --start-year 2020 --end-year 2025 --no-cleanup
"""

import argparse
import io
import sys
import unicodedata
from datetime import date
from pathlib import Path
from typing import Callable, Optional

import pandas as pd
from selenium.webdriver.support.ui import WebDriverWait

import db as _db
from scraper_meteorologia import (
    SITE_MIN_DATE,
    STOP_EMPTY_YEARS,
    make_driver,
    scrape_year,
)
from concat_meteorologia import (
    build_aerodrome_table,
    cleanup_source_files,
    validate_sample,
)

# ── Paths (resolved relative to this file, independent of CWD) ───────────────
_APPS_DIR   = Path(__file__).resolve().parent          # .../meteorologia_aeroportos/_apps/
_BASE_DIR   = _APPS_DIR.parent                         # .../meteorologia_aeroportos/
_REPO_ROOT  = _BASE_DIR.parents[2]                     # dataset/
DADOS_DIR   = _BASE_DIR / "db" / "dados"               # temporary CSV files
DB_PATH     = _BASE_DIR / "db" / "met.db"              # SQLite database
PREPROC_DIR = (
    _REPO_ROOT / "tabelas" / "preproc" / "meteorologia_aeroportos"
)  # permanent analytics CSV backups

# ── Progress milestones ───────────────────────────────────────────────────────
_PROG_START        = 0.05
_PROG_FORWARD_DONE = 0.35
_PROG_ANALYTICS    = 0.72
_PROG_UPSERT       = 0.80
_PROG_VALIDATE     = 0.88
_PROG_CLEANUP      = 0.95
_PROG_DONE         = 1.00


# ---------------------------------------------------------------------------
# Analytics helpers (standalone, no dependency on dashboard.py)
# ---------------------------------------------------------------------------

def _norm(s: str) -> str:
    """Returns *s* lowercased, ASCII-only, with diacritics stripped."""
    return unicodedata.normalize("NFKD", s.lower()).encode("ascii", "ignore").decode("ascii")


def _to_num(s: pd.Series) -> pd.Series:
    """Coerces *s* to float, treating ``'-'`` and blank strings as NaN.

    Args:
        s: A pandas Series of raw string or numeric values.

    Returns:
        Float Series with unparseable values replaced by ``NaN``.
    """
    if pd.api.types.is_numeric_dtype(s):
        return s.astype(float)
    v = (
        s.astype(str)
        .str.strip()
        .replace({"-": None, "nan": None, "NaN": None, "None": None, "": None})
    )
    v = v.str.replace(",", ".", regex=False)
    return pd.to_numeric(v, errors="coerce")


def _agg(
    df: pd.DataFrame,
    prefix: str,
    kw: str,
    fn: str = "mean",
) -> pd.Series:
    """Aggregates columns matching ``<prefix>_*<kw>*``, preferring generic ones.

    Columns whose name contains ``_-_`` are treated as generic (station-wide)
    and preferred over runway/heading-specific columns.  When multiple specific
    columns are present, they are combined using *fn*.

    Args:
        df: Merged wide DataFrame with prefixed column names.
        prefix: Column prefix to filter on (e.g. ``"temp"``).
        kw: Keyword fragment to match in normalised column names.
        fn: Aggregation function: ``"mean"``, ``"min"``, or ``"max"``.

    Returns:
        Float Series aligned with *df*'s index.
    """
    kw_n = _norm(kw)
    cols = [c for c in df.columns if c.startswith(f"{prefix}_") and kw_n in _norm(c)]
    if not cols:
        return pd.Series(dtype=float, index=df.index)
    gen = [c for c in cols if "_-_" in c]
    spc = [c for c in cols if "_-_" not in c]
    if gen:
        s = _to_num(df[gen[0]])
        if s.notna().mean() > 0.05:
            return s
    if spc:
        mat = pd.DataFrame({c: _to_num(df[c]) for c in spc})
        return {"mean": mat.mean, "min": mat.min, "max": mat.max}[fn](axis=1)
    return _to_num(df[cols[0]])


def build_analytics(merged: pd.DataFrame) -> pd.DataFrame:
    """Derives the 10 clean analytic variables from the 84-column merged CSV.

    Args:
        merged: Raw merged DataFrame produced by
            :func:`concat_meteorologia.build_aerodrome_table`.

    Returns:
        DataFrame with columns ``_dt`` (``datetime64``) plus
        ``T``, ``Td``, ``UR``, ``QNH``, ``WS``, ``WG``, ``WD``,
        ``VIS``, ``TETO``, ``PREC`` (all float).
    """
    dt_col = next((c for c in merged.columns if "Data" in c and "Hora" in c), None)
    dt = (
        pd.to_datetime(merged[dt_col], format="%d/%m/%Y - %H:%M:%S", errors="coerce")
        if dt_col
        else pd.Series(dtype="datetime64[ns]")
    )

    qnh  = (
        _to_num(merged["pres_QNH"])
        if "pres_QNH" in merged.columns
        else _agg(merged, "pres", "QNH")
    )
    vis  = next(
        (
            _to_num(merged[c])
            for c in merged.columns
            if "visib" in c.lower() and "predominante" in _norm(c)
        ),
        pd.Series(dtype=float, index=merged.index),
    )
    prec = next(
        (
            _to_num(merged[c])
            for c in merged.columns
            if "prec_precipita" in c.lower() and "dura" not in c.lower()
        ),
        pd.Series(dtype=float, index=merged.index),
    )

    return pd.DataFrame({
        "_dt":  dt.values,
        "T":    _agg(merged, "temp", "Bulbo_Seco").values,
        "Td":   _agg(merged, "temp", "Orvalho").values,
        "UR":   _agg(merged, "temp", "Umidade").values,
        "QNH":  qnh.values,
        "WS":   _agg(merged, "vent", "Velocidade").values,
        "WG":   _agg(merged, "vent", "Rajada", "max").values,
        "WD":   _agg(merged, "vent", "Dire").values,
        "VIS":  vis.values,
        "TETO": _agg(merged, "teto", "Teto", "min").values,
        "PREC": prec.values,
    })


# ---------------------------------------------------------------------------
# Two-pass scraping
# ---------------------------------------------------------------------------

def _run_forward(
    driver: object,
    wait: WebDriverWait,
    aerodrome: str,
    ex_max: date,
    dados_dir: str,
    log: Callable[[str], None],
) -> int:
    """Downloads new observations after *ex_max* (update pass).

    Args:
        driver: Selenium WebDriver instance.
        wait: Configured WebDriverWait bound to *driver*.
        aerodrome: ICAO code to scrape.
        ex_max: Latest date already stored; scraping starts just after this.
        dados_dir: Directory where quarterly CSVs are written.
        log: Logging callback.

    Returns:
        Number of non-empty quarters downloaded.
    """
    today = date.today()
    if ex_max >= today:
        log("[pipeline] Dados já atualizados até hoje.")
        return 0

    log(f"\n[pipeline] === Passe 1: atualização {ex_max.year} → {today.year} ===")
    total = 0
    for year in range(today.year, ex_max.year - 1, -1):
        log(f"\n{'=' * 55}\nAno {year}  —  {aerodrome} (forward)\n{'=' * 55}")
        total += scrape_year(driver, wait, aerodrome, year, dados_dir,
                             stop_before=ex_max, log=log)
    return total


def _run_backward(
    driver: object,
    wait: WebDriverWait,
    aerodrome: str,
    ex_min: date,
    all_years: bool,
    dados_dir: str,
    log: Callable[[str], None],
    progress: Callable[[float, str], None],
) -> int:
    """Downloads historical data before *ex_min* (backward pass).

    Stops automatically after :data:`STOP_EMPTY_YEARS` consecutive empty years
    when *all_years* is ``True``.

    Args:
        driver: Selenium WebDriver instance.
        wait: Configured WebDriverWait bound to *driver*.
        aerodrome: ICAO code to scrape.
        ex_min: Earliest date already stored; scraping goes back from here.
        all_years: When ``True``, enable the consecutive-empty-year stop logic.
        dados_dir: Directory where quarterly CSVs are written.
        log: Logging callback.
        progress: Progress callback ``(fraction, message)``.

    Returns:
        Number of non-empty quarters downloaded.
    """
    start_year = ex_min.year - 1
    if start_year < SITE_MIN_DATE.year:
        log("[pipeline] Histórico já vai até o limite do site.")
        return 0

    log(f"\n[pipeline] === Passe 2: histórico {start_year} → {SITE_MIN_DATE.year} ===")
    total = 0
    consecutive_empty = 0
    years = list(range(start_year, SITE_MIN_DATE.year - 1, -1))

    for i, year in enumerate(years):
        pct = _PROG_FORWARD_DONE + (_PROG_ANALYTICS - _PROG_FORWARD_DONE) * (
            i / max(len(years), 1)
        )
        progress(pct, f"Histórico {aerodrome} {year}")
        log(f"\n{'=' * 55}\nAno {year}  —  {aerodrome} (backward)\n{'=' * 55}")

        n = scrape_year(driver, wait, aerodrome, year, dados_dir,
                        stop_before=None, log=log)
        total += n

        if n == 0:
            consecutive_empty += 1
            log(f"  [sem dados] ({consecutive_empty}/{STOP_EMPTY_YEARS})")
            if all_years and consecutive_empty >= STOP_EMPTY_YEARS:
                log(f"\n{STOP_EMPTY_YEARS} anos consecutivos sem dados — parando.")
                break
        else:
            consecutive_empty = 0

    return total


# ---------------------------------------------------------------------------
# Main pipeline
# ---------------------------------------------------------------------------

def run_pipeline(
    aerodrome: str,
    dados_dir: Path = DADOS_DIR,
    db_path: Path = DB_PATH,
    preproc_dir: Path = PREPROC_DIR,
    all_years: bool = True,
    start_year: Optional[int] = None,
    end_year: Optional[int] = None,
    headless: bool = True,
    n_samples: int = 20,
    do_validate: bool = True,
    do_cleanup: bool = True,
    log: Callable[[str], None] = print,
    progress: Callable[[float, str], None] = lambda pct, msg: None,
) -> dict:
    """Runs the full scrape → analytics → SQLite → validate → cleanup pipeline.

    Args:
        aerodrome: ICAO code to process (e.g. ``"SBGR"``).
        dados_dir: Directory for temporary quarterly CSV files.
        db_path: Path to the SQLite database file.
        preproc_dir: Root directory for permanent analytics CSV backups.
            Each aerodrome gets a sub-directory: ``<preproc_dir>/<aerodrome>/``.
        all_years: When ``True``, scrape the full available history.
        start_year: First year to scrape (used when *all_years* is ``False``).
        end_year: Last year to scrape (used when *all_years* is ``False``).
        headless: Run Chrome in headless mode (``True``) or visible (``False``).
        n_samples: Number of spot-check samples for validation.
        do_validate: Enable spot-check validation against source CSVs.
        do_cleanup: Remove quarterly CSVs after a successful validation.
        log: Logging callback receiving a single string.
        progress: Progress callback receiving ``(fraction: float, message: str)``.

    Returns:
        Dict with keys ``rows``, ``period_start``, ``period_end``,
        ``n_ok``, ``n_fail``, ``errors``, ``db_path``.
    """
    base = Path(dados_dir)
    base.mkdir(parents=True, exist_ok=True)
    db_path = Path(db_path)

    # 1. Existing coverage ────────────────────────────────────────────────────
    conn = _db.get_connection(db_path)
    _db.ensure_schema(conn)
    coverage = _db.get_coverage(conn, aerodrome)

    if coverage:
        ex_min, ex_max = coverage
        log(f"\n[pipeline] Cobertura existente: {ex_min} → {ex_max}")
    else:
        log(f"\n[pipeline] Sem dados anteriores para {aerodrome}.")
        ex_min = ex_max = None

    progress(_PROG_START, f"Iniciando scraping {aerodrome}")

    # 2. Scraping ─────────────────────────────────────────────────────────────
    driver = make_driver(headless=headless)
    wait   = WebDriverWait(driver, 60)

    try:
        if coverage:
            _run_forward(driver, wait, aerodrome, ex_max, str(base), log)
            progress(_PROG_FORWARD_DONE, f"Histórico {aerodrome}")
            _run_backward(driver, wait, aerodrome, ex_min, all_years,
                          str(base), log, progress)
        else:
            log(f"\n[pipeline] === Coleta inicial: {date.today().year} → {SITE_MIN_DATE.year} ===")
            end_yr   = end_year   if not all_years else date.today().year
            start_yr = start_year if not all_years else SITE_MIN_DATE.year
            consec   = 0
            years    = list(range(end_yr, start_yr - 1, -1))

            for i, year in enumerate(years):
                pct = _PROG_START + (_PROG_FORWARD_DONE + 0.30) * (i / max(len(years), 1))
                progress(pct, f"Scraping {aerodrome} {year}")
                log(f"\n{'=' * 55}\nAno {year}  —  {aerodrome}\n{'=' * 55}")
                n = scrape_year(driver, wait, aerodrome, year, str(base), log=log)
                if n == 0:
                    consec += 1
                    log(f"  [sem dados] ({consec}/{STOP_EMPTY_YEARS})")
                    if all_years and consec >= STOP_EMPTY_YEARS:
                        log(f"\n{STOP_EMPTY_YEARS} anos consecutivos — parando.")
                        break
                else:
                    consec = 0

    except KeyboardInterrupt:
        log("\n[pipeline] Interrompido.")
    finally:
        driver.quit()

    progress(_PROG_ANALYTICS, "Construindo analytics…")

    # 3. Concat → analytics ───────────────────────────────────────────────────
    new_files = [f for f in base.glob("Dados de Superfície*.csv") if f.is_file()]

    if not new_files:
        cov = _db.get_coverage(conn, aerodrome)
        if cov:
            s, e = str(cov[0]), str(cov[1])
            stats = _db.aerodrome_stats(conn, aerodrome)
            progress(_PROG_DONE, "Concluído")
            return dict(rows=stats.get("n_obs", 0), period_start=s, period_end=e,
                        n_ok=0, n_fail=0, errors=[], db_path=str(db_path))
        log("[pipeline] Nenhum dado disponível.")
        return dict(rows=0, period_start="", period_end="",
                    n_ok=0, n_fail=0, errors=[], db_path=str(db_path))

    _, merged = build_aerodrome_table(new_files, log=log)

    if merged.empty:
        log("[pipeline] Merged vazio.")
        progress(_PROG_DONE, "Concluído")
        return dict(rows=0, period_start="", period_end="",
                    n_ok=0, n_fail=0, errors=[], db_path=str(db_path))

    anl = build_analytics(merged)
    anl = anl.dropna(subset=["_dt"]).reset_index(drop=True)

    progress(_PROG_UPSERT, "Inserindo no banco SQLite…")

    # 4. Upsert into SQLite ───────────────────────────────────────────────────
    n_upserted = _db.upsert_analytics(conn, aerodrome, anl)
    log(f"\n[pipeline] {n_upserted} linhas upsertadas no SQLite ({db_path.name})")

    # Write analytics CSV backup to tabelas/preproc/meteorologia_aeroportos/<ICAO>/
    anl_dir = Path(preproc_dir) / aerodrome
    anl_dir.mkdir(parents=True, exist_ok=True)
    cov_final = _db.get_coverage(conn, aerodrome)
    if cov_final:
        s_str = str(cov_final[0]).replace("-", "_")
        e_str = str(cov_final[1]).replace("-", "_")
        csv_backup = anl_dir / f"{aerodrome}_{s_str}_{e_str}.csv"
        for old in anl_dir.glob(f"{aerodrome}_*.csv"):
            old.unlink()
        anl_out = anl.copy()
        anl_out["_dt"] = anl_out["_dt"].dt.strftime("%Y-%m-%d %H:%M:%S")
        anl_out.to_csv(csv_backup, index=False, encoding="utf-8-sig")
        log(f"[pipeline] Backup CSV: {csv_backup}")

    progress(_PROG_VALIDATE, "Validando…")

    # 5. Validation ───────────────────────────────────────────────────────────
    n_ok = n_fail = 0
    val_errors: list[str] = []
    validated = True

    if do_validate and new_files:
        log(f"\n[pipeline] Validando {n_samples} amostras…")
        n_ok, n_fail, val_errors = validate_sample(merged, new_files, n=n_samples, log=log)
        log(f"[pipeline] Validação: {n_ok} OK | {n_fail} divergências")
        for e in val_errors:
            log(e)
        if n_fail > 0:
            validated = False

    progress(_PROG_CLEANUP, "Cleanup…")

    # 6. Cleanup ──────────────────────────────────────────────────────────────
    if do_cleanup and new_files:
        if validated:
            log(f"\n[pipeline] Removendo {len(new_files)} CSVs trimestrais…")
            cleanup_source_files(new_files, log=log)
        else:
            log("[pipeline] Validação com falhas — CSVs trimestrais mantidos.")

    cov_now = _db.get_coverage(conn, aerodrome)
    stats   = _db.aerodrome_stats(conn, aerodrome)
    conn.close()

    progress(_PROG_DONE, "Concluído")
    log("\n[pipeline] Pipeline finalizado.")

    return dict(
        rows         = stats.get("n_obs", len(anl)),
        period_start = str(cov_now[0]) if cov_now else "",
        period_end   = str(cov_now[1]) if cov_now else "",
        n_ok         = n_ok,
        n_fail       = n_fail,
        errors       = val_errors,
        db_path      = str(db_path),
    )


# ---------------------------------------------------------------------------
# CLI entry point
# ---------------------------------------------------------------------------

def main() -> None:
    """Parses CLI arguments and runs the pipeline."""
    try:
        if hasattr(sys.stdout, "buffer"):
            sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
        if hasattr(sys.stderr, "buffer"):
            sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")
    except Exception:
        pass

    parser = argparse.ArgumentParser(
        description="Pipeline meteorologia ICEA/DECEA → SQLite",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
exemplos:
  python pipeline.py --aerodrome SBGR --all-years
  python pipeline.py --aerodrome SBGR --start-year 2020 --end-year 2025
  python pipeline.py --aerodrome SBSP --all-years --no-headless --no-cleanup
        """,
    )
    parser.add_argument("--aerodrome",    default="SBGR",        metavar="ICAO")
    parser.add_argument("--all-years",    action="store_true")
    parser.add_argument("--start-year",   type=int,              metavar="ANO")
    parser.add_argument("--end-year",     type=int,              metavar="ANO")
    parser.add_argument("--dados-dir",    default=str(DADOS_DIR),metavar="DIR",
                        help="Directory for temporary quarterly CSVs")
    parser.add_argument("--db-path",      default=str(DB_PATH),  metavar="PATH",
                        help="Path to the SQLite database file")
    parser.add_argument("--preproc-dir",  default=str(PREPROC_DIR), metavar="DIR",
                        help="Root directory for permanent analytics CSV backups")
    parser.add_argument("--no-headless",  action="store_true")
    parser.add_argument("--no-validate",  action="store_true")
    parser.add_argument("--n-samples",    type=int, default=20)
    parser.add_argument("--no-cleanup",   action="store_true")
    args = parser.parse_args()

    if not args.all_years and (args.start_year is None or args.end_year is None):
        parser.error("Use --all-years ou informe --start-year e --end-year")

    result = run_pipeline(
        aerodrome   = args.aerodrome,
        dados_dir   = Path(args.dados_dir),
        db_path     = Path(args.db_path),
        preproc_dir = Path(args.preproc_dir),
        all_years   = args.all_years,
        start_year  = args.start_year,
        end_year    = args.end_year,
        headless    = not args.no_headless,
        n_samples   = args.n_samples,
        do_validate = not args.no_validate,
        do_cleanup  = not args.no_cleanup,
    )

    print(f"\n{'=' * 55}")
    print(f"  Linhas      : {result['rows']}")
    print(f"  Período     : {result['period_start']} → {result['period_end']}")
    print(f"  Validação   : {result['n_ok']} OK | {result['n_fail']} divergências")
    print(f"  Banco       : {result['db_path']}")


if __name__ == "__main__":
    main()