""" Orquestrador do pipeline completo de meteorologia ICEA/DECEA. Fluxo: 1. Verifica cobertura existente no banco SQLite (db.get_coverage) 2. Passe 1 — forward: baixa dados novos após ex_max (stop_before=ex_max) 3. Passe 2 — backward: baixa histórico antes de ex_min (auto-stop sem limit) 4. Computa analytics a partir dos CSVs trimestrais 5. Upserta no SQLite + salva CSV de analytics como backup 6. Valida por amostragem (CSVs trimestrais ainda presentes) 7. Remove CSVs trimestrais intermediários (se validação OK e --cleanup) Uso: 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, ) DB_FILENAME = "met.db" # --------------------------------------------------------------------------- # Analytics inline (independente do dashboard.py) # --------------------------------------------------------------------------- def _norm(s: str) -> str: return unicodedata.normalize("NFKD", s.lower()).encode("ascii", "ignore").decode("ascii") def _to_num(s: pd.Series) -> pd.Series: 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: 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: """Computa as 10 variáveis analíticas limpas a partir do merged bruto de 84 colunas.""" 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, }) # --------------------------------------------------------------------------- # Scraping em dois passes # --------------------------------------------------------------------------- def _run_forward(driver, wait, aerodrome, ex_max: date, dados_dir: str, log: Callable) -> int: """Passe 1: baixa dados novos posteriores a ex_max (atualização).""" 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, wait, aerodrome, ex_min: date, all_years: bool, dados_dir: str, log: Callable, progress: Callable) -> int: """Passe 2: baixa histórico anterior a ex_min (extensão regressiva com auto-stop).""" 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 = 0.10 + 0.55 * (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 # --------------------------------------------------------------------------- # Pipeline principal # --------------------------------------------------------------------------- def run_pipeline( aerodrome: str, dados_dir: str = "dados", 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: """ Pipeline completo: scraping → analytics → SQLite → validação → cleanup. Returns dict: rows, period_start, period_end, n_ok, n_fail, errors, db_path """ base = Path(dados_dir) base.mkdir(exist_ok=True) db_path = base / DB_FILENAME # ── 1. Cobertura existente no SQLite ───────────────────────────────────── 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(0.05, f"Iniciando scraping {aerodrome}") # ── 2. Scraping ─────────────────────────────────────────────────────────── driver = make_driver(headless=headless) wait = WebDriverWait(driver, 60) try: if coverage: # Modo incremental: dois passes _run_forward(driver, wait, aerodrome, ex_max, str(base), log) progress(0.35, f"Histórico {aerodrome}") _run_backward(driver, wait, aerodrome, ex_min, all_years, str(base), log, progress) else: # Primeira coleta: único passe regressivo com auto-stop 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 = 0.05 + 0.65 * (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(0.72, "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: # Sem novos dados, apenas reporta cobertura existente cov = _db.get_coverage(conn, aerodrome) if cov: s, e = str(cov[0]), str(cov[1]) stats = _db.aerodrome_stats(conn, aerodrome) progress(1.0, "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(1.0, "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(0.80, "Inserindo no banco SQLite…") # ── 4. Upsert no SQLite ─────────────────────────────────────────────────── n_upserted = _db.upsert_analytics(conn, aerodrome, anl) log(f"\n[pipeline] {n_upserted} linhas upsertadas no SQLite ({db_path.name})") # Salva CSV de analytics como backup legível anl_dir = base / aerodrome anl_dir.mkdir(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" # Remove arquivos de backup anteriores 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.name}") progress(0.88, "Validando…") # ── 5. Validação ───────────────────────────────────────────────────────── 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(0.95, "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(1.0, "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 standalone # --------------------------------------------------------------------------- def main() -> None: 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="dados", metavar="DIR") 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 = args.dados_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()