"""Descarga única y congelado de los cinco archivos de datos del Estudio 01. Es el único módulo del estudio con acceso a la red. Hace una sola petición GET por URL, con un User-Agent que identifica el estudio, y guarda cada cuerpo byte a byte, sin decodificar ni normalizar nada. Cualquier fallo (código distinto de 200, redirección, cuerpo vacío, codificación de contenido, error de red) detiene el script sin escribir ningún archivo de datos y sin reintentar; el intento queda registrado en data/download-log.json. Si los archivos ya existen, el script se niega a correr: los datos se descargan una sola vez. Uso, desde la raíz del repositorio: .venv/bin/python studies/estudio-01/download.py """ from __future__ import annotations import hashlib import http.client import json import sys import urllib.error import urllib.request from dataclasses import dataclass from datetime import UTC, datetime from pathlib import Path from typing import Any USER_AGENT = "travidence-estudio-01/1.0 (+https://travidence.com)" TIMEOUT_SECONDS = 120 DATA_DIR = Path(__file__).resolve().parent / "data" LOG_PATH = DATA_DIR / "download-log.json" SUMS_PATH = DATA_DIR / "SHA256SUMS" EXIT_FAILED = 1 EXIT_ALREADY_FROZEN = 2 @dataclass(frozen=True) class Source: number: int filename: str url: str SOURCES = ( Source( 1, "01-h10-daily-rates.csv", "https://www.federalreserve.gov/datadownload/Output.aspx?rel=H10" "&series=60f32914ab61dfab590e0e470153e3ae&lastobs=&from=01/04/1999&to=09/30/2026" "&filetype=csv&label=include&layout=seriescolumn&type=package", ), Source( 2, "02-h10-hist-euro.htm", "https://www.federalreserve.gov/releases/h10/hist/dat00_eu.htm", ), Source( 3, "03-fomc-openmarket.htm", "https://www.federalreserve.gov/monetarypolicy/openmarket.htm", ), Source( 4, "04-fomc-openmarket-archive.htm", "https://www.federalreserve.gov/monetarypolicy/openmarket_archive.htm", ), Source( 5, "05-ecb-dfr.csv", "https://data-api.ecb.europa.eu/service/data/FM/B.U2.EUR.4F.KR.DFR.LEV?format=csvdata", ), ) # Encabezados de la respuesta que se registran en el log cuando vienen. LOGGED_HEADERS = ( ("content_type", "Content-Type"), ("content_length", "Content-Length"), ("content_encoding", "Content-Encoding"), ("content_disposition", "Content-Disposition"), ("last_modified", "Last-Modified"), ("etag", "ETag"), ) class _NoRedirect(urllib.request.HTTPRedirectHandler): """Una sola petición por URL: una redirección cuenta como fallo, no se sigue.""" def redirect_request(self, req, fp, code, msg, headers, newurl): return None _OPENER = urllib.request.build_opener(_NoRedirect) def _utc_now() -> str: return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ") def _header_fields(status: int, headers: Any) -> dict[str, Any]: fields: dict[str, Any] = {"http_status": status} for key, header in LOGGED_HEADERS: value = headers.get(header) if value is not None: fields[key] = value return fields def _failure_reason(record: dict[str, Any], body: bytes) -> str | None: if record["http_status"] != 200: return f"HTTP {record['http_status']}" encoding = record.get("content_encoding", "identity") if encoding.lower() != "identity": return f"Content-Encoding {encoding!r}: el cuerpo no es el archivo tal cual" if not body: return "cuerpo vacío (0 bytes)" declared = record.get("content_length") if declared is not None and declared.isdigit() and int(declared) != len(body): return f"Content-Length {declared} distinto de los {len(body)} bytes recibidos" return None def fetch(source: Source) -> tuple[dict[str, Any], bytes | None]: """Una petición GET. Devuelve el registro del log y el cuerpo (None si falló).""" request = urllib.request.Request( source.url, headers={"User-Agent": USER_AGENT, "Accept": "*/*", "Accept-Encoding": "identity"}, method="GET", ) record: dict[str, Any] = { "number": source.number, "filename": source.filename, "url": source.url, "started_utc": _utc_now(), } body: bytes | None = None try: with _OPENER.open(request, timeout=TIMEOUT_SECONDS) as response: record.update(_header_fields(response.status, response.headers)) body = response.read() except urllib.error.HTTPError as error: record.update(_header_fields(error.code, error.headers)) record["failure"] = f"HTTP {error.code}" error.close() except (urllib.error.URLError, http.client.HTTPException, OSError) as error: record["http_status"] = None record["failure"] = f"{type(error).__name__}: {error}" record["finished_utc"] = _utc_now() if body is not None: record["bytes"] = len(body) record["sha256"] = hashlib.sha256(body).hexdigest() reason = _failure_reason(record, body) if reason is not None: record["failure"] = reason body = None return record, body def _append_log(run: dict[str, Any]) -> None: runs: list[dict[str, Any]] = [] if LOG_PATH.exists(): runs = json.loads(LOG_PATH.read_text(encoding="utf-8"))["runs"] run["run"] = len(runs) + 1 runs.append(run) text = json.dumps({"runs": runs}, indent=2, ensure_ascii=False) + "\n" LOG_PATH.write_bytes(text.encode("utf-8")) def main() -> int: DATA_DIR.mkdir(parents=True, exist_ok=True) present = [s.filename for s in SOURCES if (DATA_DIR / s.filename).exists()] if SUMS_PATH.exists(): present.append(SUMS_PATH.name) if present: print(f"Los datos ya están congelados, no se descarga de nuevo: {present}", file=sys.stderr) return EXIT_ALREADY_FROZEN run: dict[str, Any] = { "started_utc": _utc_now(), "user_agent": USER_AGENT, "python": sys.version.split()[0], "requests": [], } bodies: dict[str, bytes] = {} failed_at: Source | None = None for source in SOURCES: record, body = fetch(source) run["requests"].append(record) if body is None: failed_at = source print(f"FALLO en el archivo {source.number}: {record['failure']}", file=sys.stderr) break bodies[source.filename] = body run["finished_utc"] = _utc_now() if failed_at is not None: run["status"] = "failed" run["not_attempted"] = [s.filename for s in SOURCES if s.number > failed_at.number] _append_log(run) print(f"Intento registrado en {LOG_PATH}; no se escribió ningún archivo de datos.") return EXIT_FAILED sums = [] for source in SOURCES: with open(DATA_DIR / source.filename, "xb") as handle: handle.write(bodies[source.filename]) sums.append(f"{hashlib.sha256(bodies[source.filename]).hexdigest()} {source.filename}\n") SUMS_PATH.write_bytes("".join(sums).encode("ascii")) run["status"] = "ok" _append_log(run) for record in run["requests"]: print(f"{record['number']} {record['filename']} {record['bytes']} B {record['sha256']}") return 0 if __name__ == "__main__": sys.exit(main())