reboucassr/dataset
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

V.1.0

Browse Source
This commit is contained in:
kolmeasi
2026-06-02 08:29:38 -03:00
parent 6cc98a4531
commit 57fc64f120
15 changed files with 1292264 additions and 1908331 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
softwares/test/meteorologia_aeroportos/_apps/pipeline.py
View File

@@ -26,6 +26,8 @@ from typing import Callable, Optional
import pandas as pd
from selenium.webdriver.support.ui import WebDriverWait
import sqlite3
import db as _db
from scraper_meteorologia import (
SITE_MIN_DATE,
@@ -281,6 +283,52 @@ def _run_backward(
return total
# ---------------------------------------------------------------------------
# CSV backup helper
# ---------------------------------------------------------------------------
def _write_csv_backup(
conn: sqlite3.Connection,
aerodrome: str,
preproc_dir: Path,
log: Callable[[str], None],
) -> None:
"""Writes the complete analytics history for *aerodrome* to its CSV backup.
Always queries the **full** history from the database so the backup reflects
the entire dataset, regardless of how much data was collected in this run.
Deletes any previous backup file before writing the updated one.
Args:
conn: Open SQLite connection.
aerodrome: ICAO code (e.g. ``"SBGR"``).
preproc_dir: Root directory for permanent analytics CSV backups.
log: Logging callback.
"""
cov = _db.get_coverage(conn, aerodrome)
if not cov:
return
anl_dir = preproc_dir / aerodrome
anl_dir.mkdir(parents=True, exist_ok=True)
s_str = str(cov[0]).replace("-", "_")
e_str = str(cov[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_full = _db.query_analytics(conn, aerodrome)
if anl_full.empty:
return
anl_out = anl_full.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} ({len(anl_full):,} linhas)")
# ---------------------------------------------------------------------------
# Main pipeline
# ---------------------------------------------------------------------------
@@ -297,6 +345,7 @@ def run_pipeline(
n_samples: int = 20,
do_validate: bool = True,
do_cleanup: bool = True,
update_only: bool = False,
log: Callable[[str], None] = print,
progress: Callable[[float, str], None] = lambda pct, msg: None,
) -> dict:
@@ -315,6 +364,9 @@ def run_pipeline(
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.
update_only: When ``True``, skip the backward (historical) pass and only
download data newer than the latest existing record. Aerodromes
without any existing data are skipped entirely.
log: Logging callback receiving a single string.
progress: Progress callback receiving ``(fraction: float, message: str)``.
@@ -347,9 +399,12 @@ def run_pipeline(
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)
if not update_only:
progress(_PROG_FORWARD_DONE, f"Histórico {aerodrome}")
_run_backward(driver, wait, aerodrome, ex_min, all_years,
str(base), log, progress)
elif update_only:
log(f"[pipeline] {aerodrome}: sem dados existentes e update_only=True — ignorado.")
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
@@ -386,11 +441,18 @@ def run_pipeline(
if cov:
s, e = str(cov[0]), str(cov[1])
stats = _db.aerodrome_stats(conn, aerodrome)
# Regenerate CSV backup if it is missing (e.g. after a partial run)
anl_dir = Path(preproc_dir) / aerodrome
csv_exists = anl_dir.is_dir() and bool(list(anl_dir.glob(f"{aerodrome}_*.csv")))
if not csv_exists:
log("[pipeline] CSV de backup ausente — regenerando do banco…")
_write_csv_backup(conn, aerodrome, Path(preproc_dir), log)
progress(_PROG_DONE, "Concluído")
return dict(rows=stats.get("n_obs", 0), period_start=s, period_end=e,
return dict(rows=stats.get("n_obs", 0), n_upserted=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="",
return dict(rows=0, n_upserted=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)
@@ -398,7 +460,7 @@ def run_pipeline(
if merged.empty:
log("[pipeline] Merged vazio.")
progress(_PROG_DONE, "Concluído")
return dict(rows=0, period_start="", period_end="",
return dict(rows=0, n_upserted=0, period_start="", period_end="",
n_ok=0, n_fail=0, errors=[], db_path=str(db_path))
anl = build_analytics(merged)
@@ -410,20 +472,8 @@ def run_pipeline(
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}")
# Write analytics CSV backup — full history from DB, not just the new records
_write_csv_backup(conn, aerodrome, Path(preproc_dir), log)
progress(_PROG_VALIDATE, "Validando…")
@@ -460,6 +510,7 @@ def run_pipeline(
return dict(
rows = stats.get("n_obs", len(anl)),
n_upserted = n_upserted,
period_start = str(cov_now[0]) if cov_now else "",
period_end = str(cov_now[1]) if cov_now else "",
n_ok = n_ok,
@@ -507,6 +558,9 @@ exemplos:
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")
parser.add_argument("--update-only", action="store_true",
help="Only download data newer than the last existing record "
"(skip backward/historical pass)")
args = parser.parse_args()
if not args.all_years and (args.start_year is None or args.end_year is None):
@@ -524,6 +578,7 @@ exemplos:
n_samples = args.n_samples,
do_validate = not args.no_validate,
do_cleanup = not args.no_cleanup,
update_only = args.update_only,
)
print(f"\n{'=' * 55}")