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-08 13:16:49 -03:00
parent 57fc64f120
commit 1dec3bf4c5
3382 changed files with 4474661 additions and 214 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
softwares/test/meteorologia_aeroportos/_apps/__pycache__/scraper_meteorologia.cpython-313.pyc
View File

Binary file not shown.

478
softwares/test/meteorologia_aeroportos/_apps/collect_all.py
View File

@@ -1,36 +1,42 @@
"""
Bulk data collection for all aerodromes in the ICEA surface-data catalog.
Designed as a one-time migration script to bootstrap the database with the
full available history for every aerodrome on the portal. A JSON status
file (``db/collect_all_status.json``) tracks progress so interrupted runs
can be resumed with ``--resume``.
Each aerodrome is collected in an isolated subprocess with a configurable
timeout. If Chrome hangs or the ICEA site times out, only that subprocess
is killed; the next aerodrome starts with a clean Chrome instance.
A JSON status file (``db/collect_all_status.json``) persists progress so
interrupted runs resume automatically with ``--resume``.
Usage:
# First time: fetch catalog, then collect everything
# Step 1 — populate catalog (one time)
python _apps/collect_all.py --fetch-catalog
# Step 2 — collect all pending aerodromes (resumes automatically)
python _apps/collect_all.py
# Resume after interruption (skip aerodromes already with data)
python _apps/collect_all.py --resume
# Force re-collection of all aerodromes
# Force re-collection of all (even those already done)
python _apps/collect_all.py --force
# Specific aerodromes only
python _apps/collect_all.py --aerodrome SBSP SBGL SBRJ
# Dry run / test (first 3 aerodromes only)
python _apps/collect_all.py --max-aerodromes 3
# Test with first 3 aerodromes, visible Chrome
python _apps/collect_all.py --max-aerodromes 3 --no-headless
# Visible Chrome (useful for debugging portal issues)
python _apps/collect_all.py --no-headless --aerodrome SBSP
# Increase per-aerodrome timeout to 3 hours (default: 2h)
python _apps/collect_all.py --timeout 180
# Check collection status
python _apps/collect_all.py --status
"""
import argparse
import io
import json
import subprocess
import sys
import time
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path
from typing import Callable, Optional
@@ -45,6 +51,11 @@ PREPROC_DIR = _REPO_ROOT / "tabelas" / "preproc" / "meteorologia_aeroportos"
STATUS_FILE = _BASE_DIR / "db" / "collect_all_status.json"
DEFAULT_LOG = _BASE_DIR / "db" / "collect_all.log"
# Default per-aerodrome timeout: 2 hours (most take 30–90 min)
_TIMEOUT_DEFAULT_MIN = 120
# Sleep between aerodromes to avoid ICEA rate-limiting
_SLEEP_BETWEEN_S = 10
# ── Status helpers ────────────────────────────────────────────────────────────
@@ -59,21 +70,17 @@ def _load_status() -> dict:
def _save_status(status: dict) -> None:
"""Persists the status dict to the JSON status file.
Args:
status: Dict mapping ICAO codes to their collection state.
"""
"""Atomically persists the status dict to the JSON status file."""
STATUS_FILE.parent.mkdir(parents=True, exist_ok=True)
STATUS_FILE.write_text(
json.dumps(status, indent=2, ensure_ascii=False), encoding="utf-8"
)
tmp = STATUS_FILE.with_suffix(".tmp")
tmp.write_text(json.dumps(status, indent=2, ensure_ascii=False), encoding="utf-8")
tmp.replace(STATUS_FILE)
# ── Logging ───────────────────────────────────────────────────────────────────
def _make_logger(log_path: Optional[Path]) -> Callable[[str], None]:
"""Returns a logger that writes to stdout and optionally to a file.
"""Returns a logger that writes to stdout and, optionally, to a file.
Args:
log_path: File to append log lines to, or ``None`` for stdout only.
@@ -82,7 +89,7 @@ def _make_logger(log_path: Optional[Path]) -> Callable[[str], None]:
A callable that accepts a single string message.
"""
def _log(msg: str) -> None:
print(msg)
print(msg, flush=True)
if log_path:
try:
with log_path.open("a", encoding="utf-8") as f:
@@ -95,7 +102,7 @@ def _make_logger(log_path: Optional[Path]) -> Callable[[str], None]:
# ── Catalog helpers ───────────────────────────────────────────────────────────
def fetch_and_store_catalog(headless: bool = True, log: Callable = print) -> int:
"""Downloads the aerodrome catalog from the ICEA site and upserts it.
"""Downloads the aerodrome catalog from the ICEA site and stores it.
Args:
headless: Run Chrome in headless mode.
@@ -115,17 +122,11 @@ def fetch_and_store_catalog(headless: bool = True, log: Callable = print) -> int
_db.ensure_schema(conn)
n = _db.upsert_aerodromes(conn, catalog)
conn.close()
log(f" {n} registros gravados no catálogo.")
log(f" {n} registros gravados.")
return n
def _get_catalog() -> list[dict]:
"""Returns all entries from the local aerodrome catalog table.
Returns:
List of ``{"icao", "nome", "uf"}`` dicts.
"""
import db as _db
conn = _db.get_connection(DB_PATH)
_db.ensure_schema(conn)
@@ -135,11 +136,6 @@ def _get_catalog() -> list[dict]:
def _aerodromes_with_data() -> set[str]:
"""Returns the set of ICAO codes that have at least one observation row.
Returns:
Set of ICAO code strings.
"""
import db as _db
conn = _db.get_connection(DB_PATH)
_db.ensure_schema(conn)
@@ -148,6 +144,114 @@ def _aerodromes_with_data() -> set[str]:
return aeros
# ── Subprocess runner ─────────────────────────────────────────────────────────
def _run_subprocess(
icao: str,
headless: bool,
n_samples: int,
do_cleanup: bool,
timeout_s: int,
log: Callable[[str], None],
) -> tuple[str, dict]:
"""Runs the pipeline for *icao* in an isolated subprocess.
Each call spawns a fresh Python + Chrome process. On timeout the child
process is forcibly killed, leaving no hanging Chrome instances in the
parent's address space.
Args:
icao: ICAO aerodrome code.
headless: Pass ``--no-headless`` to pipeline when ``False``.
n_samples: Validation sample count (0 = skip validation).
do_cleanup: Remove intermediate CSVs after collection.
timeout_s: Hard kill timeout in seconds.
log: Logging callback for forwarding subprocess output.
Returns:
Tuple ``(event, result)`` where *event* is one of
``"completed"``, ``"no_data"``, ``"timeout"``, ``"failed"``
and *result* is a dict with collection statistics.
"""
cmd = [
sys.executable,
str(_APPS_DIR / "pipeline.py"),
"--aerodrome", icao,
"--all-years",
"--dados-dir", str(DADOS_DIR),
"--db-path", str(DB_PATH),
"--preproc-dir", str(PREPROC_DIR),
"--n-samples", str(n_samples),
]
if not headless:
cmd.append("--no-headless")
if n_samples == 0:
cmd.append("--no-validate")
if not do_cleanup:
cmd.append("--no-cleanup")
# On Windows, CREATE_NEW_PROCESS_GROUP lets us kill the whole tree on timeout
kwargs: dict = dict(
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
encoding="utf-8",
errors="replace",
cwd=str(_BASE_DIR),
)
if sys.platform == "win32":
kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
proc = subprocess.Popen(cmd, **kwargs)
# Read output line-by-line in real time
output_lines: list[str] = []
timed_out = False
start = time.monotonic()
try:
for line in proc.stdout: # type: ignore[union-attr]
line = line.rstrip()
log(line)
output_lines.append(line)
if time.monotonic() - start > timeout_s:
timed_out = True
break
except Exception:
pass
if timed_out:
log(f" [TIMEOUT] {icao} — encerrando processo após {timeout_s // 60} min…")
if sys.platform == "win32":
subprocess.run(
f"taskkill /F /T /PID {proc.pid}",
shell=True, capture_output=True,
)
else:
proc.kill()
proc.wait()
return "timeout", {}
proc.wait()
# Determine result from DB (subprocess already wrote to DB)
try:
import db as _db
conn = _db.get_connection(DB_PATH)
cov = _db.get_coverage(conn, icao)
stats = _db.aerodrome_stats(conn, icao)
conn.close()
if cov:
return "completed", {
"rows": stats.get("n_obs", 0),
"period_start": str(cov[0]),
"period_end": str(cov[1]),
}
return "no_data", {}
except Exception as exc:
return "failed", {"error": str(exc)[:200]}
# ── Core collection loop ──────────────────────────────────────────────────────
def collect_all(
@@ -156,88 +260,82 @@ def collect_all(
force: bool = False,
max_aerodromes: Optional[int] = None,
headless: bool = True,
n_samples: int = 5,
n_samples: int = 0,
do_cleanup: bool = True,
timeout_min: int = _TIMEOUT_DEFAULT_MIN,
sleep_between: int = _SLEEP_BETWEEN_S,
log: Callable[[str], None] = print,
on_start: Optional[Callable[[str, int, int, str], None]] = None,
on_done: Optional[Callable[[str, str, dict], None]] = None,
) -> dict:
"""Collects full historical data for every aerodrome in the catalog.
For each aerodrome, the full pipeline (``all_years=True``) is run.
Progress is saved to ``db/collect_all_status.json`` after every aerodrome
so the run can be resumed if interrupted.
Each aerodrome runs in an isolated subprocess. If Chrome hangs the
subprocess is killed after *timeout_min* minutes and the loop continues
with the next aerodrome.
Args:
target_aerodromes: Specific ICAO codes to process, or ``None`` for all.
resume: Skip aerodromes that already have data in the database.
force: Re-collect aerodromes even if they already have data.
Overrides *resume*.
max_aerodromes: Cap the number of aerodromes processed (useful for
testing).
target_aerodromes: Specific ICAO codes, or ``None`` for all catalog.
resume: Skip aerodromes that already have data (default ``True``).
force: Re-collect even if data exists (overrides *resume*).
max_aerodromes: Cap total processed (useful for testing).
headless: Run Chrome without a visible window.
n_samples: Spot-check sample size per aerodrome (0 disables validation).
do_cleanup: Remove intermediate quarterly CSVs after processing.
n_samples: Spot-check sample size (0 = skip validation, faster).
do_cleanup: Remove intermediate quarterly CSVs after collection.
timeout_min: Per-aerodrome hard timeout in minutes.
sleep_between: Seconds to pause between aerodromes (anti-throttle).
log: Logging callback.
on_start: Optional callback fired before each aerodrome starts.
Signature: ``on_start(icao, idx, total, nome)``.
on_done: Optional callback fired after each aerodrome finishes.
Signature: ``on_done(icao, event, result)`` where *event* is one
of ``"completed"``, ``"skipped"``, ``"failed"``,
``"interrupted"``.
on_start: Optional callback ``(icao, idx, total, nome)`` before each.
on_done: Optional callback ``(icao, event, result)`` after each.
Returns:
Summary dict with keys ``n_total``, ``n_done``, ``n_skipped``,
``n_failed``, ``elapsed_s``.
Summary dict with ``n_total``, ``n_done``, ``n_skipped``,
``n_timeout``, ``n_failed``, ``elapsed_s``.
"""
from pipeline import run_pipeline
status = _load_status()
catalog = _get_catalog()
if not catalog:
log(
"[collect_all] Catálogo vazio.\n"
" Execute: python _apps/collect_all.py --fetch-catalog"
)
return {"n_total": 0, "n_done": 0, "n_skipped": 0, "n_failed": 0, "elapsed_s": 0}
log("[collect_all] Catálogo vazio — execute com --fetch-catalog primeiro.")
return {"n_total": 0, "n_done": 0, "n_skipped": 0,
"n_timeout": 0, "n_failed": 0, "elapsed_s": 0}
# Filter to requested aerodromes
if target_aerodromes:
icao_set = {a.upper() for a in target_aerodromes}
catalog = [d for d in catalog if d["icao"] in icao_set]
missing = icao_set - {d["icao"] for d in catalog}
catalog = [d for d in catalog if d["icao"] in icao_set]
missing = icao_set - {d["icao"] for d in catalog}
if missing:
log(f"[collect_all] AVISO — não encontrados no catálogo: {sorted(missing)}")
log(f"[collect_all] AVISO — não encontrados: {sorted(missing)}")
if max_aerodromes:
catalog = catalog[:max_aerodromes]
existing_data = _aerodromes_with_data() if (resume and not force) else set()
existing = _aerodromes_with_data() if (resume and not force) else set()
timeout_s = timeout_min * 60
total = len(catalog)
n_done = 0
n_skip = 0
n_fail = 0
n_done = n_skip = n_timeout = n_fail = 0
started = datetime.now(timezone.utc)
log(f"\n{'='*60}")
log(f"collect_all — {total} aeródromos — {started.strftime('%Y-%m-%d %H:%M UTC')}")
log(f"resume={resume} force={force} headless={headless} n_samples={n_samples}")
log(f"{'='*60}\n")
log(f"\n{'='*65}")
log(f"collect_all — {total} aeródromos — "
f"{started.strftime('%Y-%m-%d %H:%M UTC')}")
log(f"resume={resume} headless={headless} "
f"timeout={timeout_min}min sleep={sleep_between}s")
log(f"{'='*65}\n")
for idx, aero_dict in enumerate(catalog, 1):
icao = aero_dict["icao"]
nome = aero_dict.get("nome", "")
icao = aero_dict["icao"]
nome = aero_dict.get("nome", "")
label = f"{icao} ({nome})" if nome else icao
log(f"\n[{idx:>4}/{total}] {label}")
if on_start:
on_start(icao, idx, total, nome)
# Resume logic: skip aerodromes that already have data
if not force and icao in existing_data:
log(f" SKIP dados existentes no banco")
# Skip if already has data (resume mode)
if not force and icao in existing:
log(" SKIP dados existentes no banco")
status[icao] = {"status": "skipped", "reason": "data_exists"}
_save_status(status)
if on_done:
@@ -245,79 +343,68 @@ def collect_all(
n_skip += 1
continue
# Mark as in-progress
status[icao] = {
"status": "in_progress",
"started_at": datetime.now(timezone.utc).isoformat(),
}
_save_status(status)
try:
result = run_pipeline(
aerodrome = icao,
dados_dir = DADOS_DIR,
db_path = DB_PATH,
preproc_dir = PREPROC_DIR,
all_years = True,
headless = headless,
n_samples = n_samples,
do_validate = n_samples > 0,
do_cleanup = do_cleanup,
log = log,
progress = lambda pct, msg: None,
)
log(f" Iniciando subprocesso (timeout {timeout_min} min)…")
event, result = _run_subprocess(
icao, headless, n_samples, do_cleanup, timeout_s, log
)
ts = datetime.now(timezone.utc).isoformat()
if event == "completed":
rows = result.get("rows", 0)
p_min = result.get("period_start", "")
p_max = result.get("period_end", "")
log(f" DONE {rows:,} observações | {p_min} → {p_max}")
p_max = result.get("period_end", "")
log(f" DONE {rows:,} obs | {p_min} → {p_max}")
status[icao] = {
"status": "completed",
"rows": rows,
"period_start": p_min,
"period_end": p_max,
"completed_at": datetime.now(timezone.utc).isoformat(),
"status": "completed", "rows": rows,
"period_start": p_min, "period_end": p_max,
"completed_at": ts,
}
if on_done:
on_done(icao, "completed", result)
n_done += 1
except KeyboardInterrupt:
log("\n[collect_all] Interrompido pelo usuário.")
status[icao]["status"] = "interrupted"
_save_status(status)
if on_done:
on_done(icao, "interrupted", {})
break
elif event == "no_data":
log(" SEM DADOS (aeródromo sem observações no portal)")
status[icao] = {"status": "no_data", "completed_at": ts}
n_skip += 1
except Exception as exc:
err = str(exc)[:300]
elif event == "timeout":
log(f" TIMEOUT após {timeout_min} min")
status[icao] = {"status": "timeout", "failed_at": ts}
n_timeout += 1
else:
err = result.get("error", "desconhecido")
log(f" ERRO {err}")
status[icao] = {
"status": "failed",
"error": err,
"failed_at": datetime.now(timezone.utc).isoformat(),
}
if on_done:
on_done(icao, "failed", {"error": err})
status[icao] = {"status": "failed", "error": err, "failed_at": ts}
n_fail += 1
if on_done:
on_done(icao, event, result)
_save_status(status)
elapsed = (datetime.now(timezone.utc) - started).total_seconds()
# Pause between aerodromes to avoid overwhelming ICEA
if idx < total and sleep_between > 0:
time.sleep(sleep_between)
log(f"\n{'='*60}")
log(
f"Resultado | total={total} done={n_done} "
f"skipped={n_skip} failed={n_fail} "
f"elapsed={elapsed / 3600:.1f}h"
)
log(f"Status gravado em: {STATUS_FILE}")
log(f"{'='*60}\n")
elapsed = (datetime.now(timezone.utc) - started).total_seconds()
log(f"\n{'='*65}")
log(f"Resultado | done={n_done} skipped={n_skip} "
f"timeout={n_timeout} failed={n_fail} "
f"elapsed={elapsed / 3600:.1f}h")
log(f"Status: {STATUS_FILE}")
log(f"{'='*65}\n")
return {
"n_total": total,
"n_done": n_done,
"n_skipped": n_skip,
"n_timeout": n_timeout,
"n_failed": n_fail,
"elapsed_s": elapsed,
}
@@ -326,25 +413,24 @@ def collect_all(
# ── Status report ─────────────────────────────────────────────────────────────
def print_status_report() -> None:
"""Prints a human-readable summary of the collect_all status file."""
"""Prints a human-readable summary of the status file."""
status = _load_status()
if not status:
print("Nenhum status encontrado. Execute collect_all.py primeiro.")
return
from collections import Counter
counts: Counter = Counter(v.get("status", "unknown") for v in status.values())
print(f"\n=== collect_all status: {STATUS_FILE} ===\n")
counts: Counter = Counter(v.get("status", "?") for v in status.values())
print(f"\n=== Status da coleta — {STATUS_FILE.name} ===\n")
for state, n in sorted(counts.items()):
print(f" {state:<15} {n:>5}")
print(f" {'TOTAL':<15} {sum(counts.values()):>5}")
print(f" {state:<20} {n:>5}")
print(f" {'TOTAL':<20} {sum(counts.values()):>5}")
failed = [(k, v) for k, v in status.items() if v.get("status") == "failed"]
failed = [(k, v) for k, v in status.items()
if v.get("status") in ("failed", "timeout")]
if failed:
print(f"\n Falhas ({len(failed)}):")
for icao, info in failed:
print(f" {icao}: {info.get('error','?')[:80]}")
print(f"\n Falhas / timeouts ({len(failed)}):")
for icao, info in failed[:20]:
print(f" {icao}: {info.get('error', info.get('status'))[:70]}")
# ── CLI ───────────────────────────────────────────────────────────────────────
@@ -353,83 +439,53 @@ def main() -> None:
"""Parses arguments and runs the bulk collection or a status report."""
try:
if hasattr(sys.stdout, "buffer"):
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
sys.stdout = io.TextIOWrapper(
sys.stdout.buffer, encoding="utf-8", errors="replace", line_buffering=True
)
if hasattr(sys.stderr, "buffer"):
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")
sys.stderr = io.TextIOWrapper(
sys.stderr.buffer, encoding="utf-8", errors="replace"
)
except Exception:
pass
parser = argparse.ArgumentParser(
description="Bulk ICEA aerodrome data collection (one-time migration)",
description="Coleta em lote de todos os aeródromos ICEA (migração inicial)",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
examples:
# Step 1 — populate catalog
python _apps/collect_all.py --fetch-catalog
# Step 2 — collect all (skip aerodromes that already have data)
python _apps/collect_all.py
# Resume after interruption
python _apps/collect_all.py --resume
# Re-collect everything from scratch
python _apps/collect_all.py --force
# Specific aerodromes only
python _apps/collect_all.py --aerodrome SBSP SBGL SBRJ
# Test run: first 3 aerodromes, visible Chrome
python _apps/collect_all.py --max-aerodromes 3 --no-headless
# Show collection status
python _apps/collect_all.py --status
exemplos:
python _apps/collect_all.py --fetch-catalog # popula catálogo
python _apps/collect_all.py # coleta todos os pendentes
python _apps/collect_all.py --aerodrome SBSP SBGL # só esses dois
python _apps/collect_all.py --max-aerodromes 3 # teste (primeiros 3)
python _apps/collect_all.py --timeout 180 # timeout 3h por aeródromo
python _apps/collect_all.py --status # ver progresso
""",
)
parser.add_argument(
"--fetch-catalog", action="store_true",
help="Fetch/update aerodrome catalog from the ICEA site before collecting",
)
parser.add_argument(
"--aerodrome", nargs="+", metavar="ICAO",
help="Collect only these ICAO codes (default: all catalog entries)",
)
parser.add_argument(
"--resume", action="store_true", default=True,
help="Skip aerodromes that already have data in the database (default: on)",
)
parser.add_argument(
"--no-resume", dest="resume", action="store_false",
help="Re-attempt aerodromes even if they already have data",
)
parser.add_argument(
"--force", action="store_true",
help="Re-collect aerodromes even if they already have data (stronger than --no-resume)",
)
parser.add_argument(
"--max-aerodromes", type=int, metavar="N",
help="Limit processing to the first N catalog entries (testing)",
)
parser.add_argument(
"--no-headless", action="store_true",
help="Open Chrome in visible mode (useful for debugging)",
)
parser.add_argument(
"--n-samples", type=int, default=5,
help="Spot-check validation sample size per aerodrome (0 = skip, default: 5)",
)
parser.add_argument(
"--no-cleanup", action="store_true",
help="Keep intermediate quarterly CSVs after processing",
)
parser.add_argument(
"--log", metavar="PATH", default=str(DEFAULT_LOG),
help=f"Log file path (default: {DEFAULT_LOG})",
)
parser.add_argument(
"--status", action="store_true",
help="Print a summary of the collect_all status file and exit",
)
parser.add_argument("--fetch-catalog", action="store_true",
help="Busca/atualiza catálogo no site ICEA antes de coletar")
parser.add_argument("--aerodrome", nargs="+", metavar="ICAO",
help="Coletar só estes códigos ICAO (padrão: todos os pendentes)")
parser.add_argument("--resume", action="store_true", default=True,
help="Pular aeródromos já com dados no banco (padrão: ativo)")
parser.add_argument("--no-resume", dest="resume", action="store_false",
help="Re-tentar aeródromos mesmo com dados existentes")
parser.add_argument("--force", action="store_true",
help="Re-coletar tudo, inclusive aeródromos já com dados")
parser.add_argument("--max-aerodromes", type=int, metavar="N",
help="Limitar a N aeródromos (útil para teste)")
parser.add_argument("--no-headless", action="store_true",
help="Abrir Chrome visível (debug)")
parser.add_argument("--timeout", type=int, default=_TIMEOUT_DEFAULT_MIN,
metavar="MIN",
help=f"Timeout por aeródromo em minutos (padrão: {_TIMEOUT_DEFAULT_MIN})")
parser.add_argument("--sleep", type=int, default=_SLEEP_BETWEEN_S,
metavar="S",
help=f"Pausa entre aeródromos em segundos (padrão: {_SLEEP_BETWEEN_S})")
parser.add_argument("--log", metavar="PATH", default=str(DEFAULT_LOG),
help=f"Arquivo de log persistente (padrão: {DEFAULT_LOG.name})")
parser.add_argument("--status", action="store_true",
help="Mostrar resumo do status e sair")
args = parser.parse_args()
if args.status:
@@ -447,8 +503,10 @@ examples:
force = args.force,
max_aerodromes = args.max_aerodromes,
headless = not args.no_headless,
n_samples = args.n_samples,
do_cleanup = not args.no_cleanup,
n_samples = 0,
do_cleanup = True,
timeout_min = args.timeout,
sleep_between = args.sleep,
log = log,
)

19
softwares/test/meteorologia_aeroportos/_apps/scraper_meteorologia.py
View File

@@ -104,13 +104,26 @@ def make_driver(headless: bool = True) -> webdriver.Chrome:
opts.add_argument("--disable-dev-shm-usage")
opts.add_argument("--window-size=1920,1080")
opts.add_argument("--disable-blink-features=AutomationControlled")
# Stability: prevent memory leaks and render hangs in long scraping sessions
opts.add_argument("--disable-extensions")
opts.add_argument("--disable-plugins")
opts.add_argument("--disable-images")
opts.add_argument("--blink-settings=imagesEnabled=false")
opts.add_argument("--disk-cache-size=1")
opts.add_argument("--media-cache-size=1")
opts.add_argument("--disable-background-networking")
opts.add_argument("--disable-default-apps")
opts.add_argument("--disable-translate")
opts.add_argument("--disable-sync")
opts.add_argument("--mute-audio")
opts.add_argument("--metrics-recording-only")
opts.add_experimental_option("excludeSwitches", ["enable-automation"])
service = Service(ChromeDriverManager().install())
driver = webdriver.Chrome(service=service, options=opts)
driver.set_page_load_timeout(300)
driver.set_script_timeout(120)
driver.set_page_load_timeout(600) # 10 min — ICEA can be slow
driver.set_script_timeout(180) # 3 min for JS execution
try:
driver.command_executor.client_config.timeout = 300
driver.command_executor.client_config.timeout = 600
except Exception:
pass
return driver