393 lines
16 KiB
Python
393 lines
16 KiB
Python
"""
|
|
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()
|