Referencia de API

Documentación autogenerada desde los docstrings del paquete chile_hub.

ChileHub

Punto de entrada principal de la librería.

chile_hub.core.ChileHub

Source code in src/chile_hub/core.py

class ChileHub:
    def __init__(
        self,
        catalog_path: str | Path | None = None,
        *,
        data_dir: str | Path | None = None,
        data_version: str = "latest",
        auto_update: bool = True,
    ) -> None:
        if catalog_path is not None and data_dir is not None:
            raise ValueError("Use catalog_path or data_dir, not both.")

        if catalog_path is not None:
            self.catalog_path = Path(catalog_path)
        elif data_dir is not None:
            self.catalog_path = Path(data_dir) / "dataset_catalog.json"
        elif DATASET_CATALOG_PATH.exists():
            self.catalog_path = DATASET_CATALOG_PATH
        else:
            manager = ChileHubDataManager(data_version=data_version)
            self.catalog_path = (
                manager.ensure_data_dir(auto_update=auto_update) / "dataset_catalog.json"
            )

        self.normalized_dir = self.catalog_path.resolve().parent
        self.root_dir = self.normalized_dir.parents[1]
        self.catalog = self._load_catalog()
        self._df_cache: dict[str, pl.DataFrame] = {}

    def _load_catalog(self) -> dict[str, Any]:
        try:
            with self.catalog_path.open("r", encoding="utf-8") as f:
                return json.load(f)  # type: ignore[no-any-return]  # json.load → dict en runtime
        except FileNotFoundError:
            raise ChileHubDataError(
                f"Catálogo de datasets no encontrado en {self.catalog_path}. "
                f"Usa ChileHub() sin argumentos para descargar el bundle automáticamente, "
                f"o asegúrate de que el directorio '{self.catalog_path.parent}' contiene "
                f"los datos normalizados (ejecuta 'make build' si estás en desarrollo)."
            )

    def _load_json_artifact(self, filename: str) -> dict[str, Any]:
        with (self.normalized_dir / filename).open("r", encoding="utf-8") as f:
            return json.load(f)  # type: ignore[no-any-return]

    @functools.lru_cache(maxsize=1)
    def _load_artifact_manifest(self) -> dict[str, Any]:
        return self._load_json_artifact("artifact_manifest.json")

    @functools.lru_cache(maxsize=1)
    def _load_hub_health(self) -> dict[str, Any]:
        return self._load_json_artifact("hub_health.json")

    @functools.lru_cache(maxsize=1)
    def _load_hub_status(self) -> dict[str, Any]:
        return self._load_json_artifact("hub_status.json")

    @functools.lru_cache(maxsize=1)
    def _load_dataset_status(self) -> dict[str, Any]:
        return self._load_json_artifact("dataset_status.json")

    @functools.lru_cache(maxsize=1)
    def _load_dataset_changelog(self) -> dict[str, Any]:
        return self._load_json_artifact("dataset_changelog.json")

    @functools.lru_cache(maxsize=1)
    def _load_hub_bundle(self) -> dict[str, Any]:
        return self._load_json_artifact("hub_bundle.json")

    @functools.lru_cache(maxsize=1)
    def _load_redistribution_report(self) -> dict[str, Any]:
        return self._load_json_artifact("redistribution_report.json")

    @functools.lru_cache(maxsize=1)
    def _load_provenance_report(self) -> dict[str, Any]:
        return self._load_json_artifact("provenance_report.json")

    @functools.lru_cache(maxsize=1)
    def _load_drift_report(self) -> dict[str, Any]:
        return self._load_json_artifact("drift_report.json")

    @functools.lru_cache(maxsize=1)
    def _load_source_readiness(self) -> dict[str, Any]:
        return self._load_json_artifact("source_readiness.json")

    @functools.lru_cache(maxsize=1)
    def _load_dataset_quality(self) -> dict[str, Any]:
        return self._load_json_artifact("dataset_quality.json")

    @staticmethod
    def _status_rank(status: str) -> int:
        return {"ok": 0, "warn": 1, "error": 2}.get(status, 1)

    @classmethod
    def _max_status(cls, *statuses: str) -> str:
        filtered = [status for status in statuses if status]
        if not filtered:
            return "unknown"
        return max(filtered, key=cls._status_rank)

    def top_issue(self) -> dict[str, Any] | None:
        provenance_by_dataset = {
            entry.get("dataset"): entry for entry in self.provenance().get("datasets", [])
        }
        drift_by_dataset = {
            entry.get("dataset"): entry for entry in self.drift().get("datasets", [])
        }
        freshness_by_dataset = {
            entry.get("dataset"): entry for entry in self.freshness_audit().get("datasets", [])
        }
        entries = []
        for entry in self.summary():
            dataset_name = entry.get("dataset")
            freshness_entry = freshness_by_dataset.get(dataset_name, {})
            provenance = provenance_by_dataset.get(dataset_name, {})
            drift = drift_by_dataset.get(dataset_name, {})
            current_freshness_status = freshness_entry.get("current_freshness_status", "unknown")
            entries.append(
                {
                    "dataset": dataset_name,
                    "warning_count": entry.get("warning_count", 0),
                    "freshness_status": current_freshness_status,
                    "build_freshness_status": entry.get("freshness_status"),
                    "current_freshness_status": current_freshness_status,
                    "drift_status": entry.get("drift_status"),
                    "degradation_status": entry.get("degradation_status"),
                    "source_detail": provenance.get("source_detail", "unknown"),
                    "diagnostic_summary": drift.get(
                        "diagnostic_summary",
                        provenance.get("diagnostic_summary", "Sin observaciones operativas."),
                    ),
                    "recommended_action": drift.get("recommended_action", "Ninguna."),
                }
            )

        return compute_top_issue(entries)  # type: ignore[no-any-return]  # dict en runtime

    def top_issue_table(self) -> str:
        top_issue = self.top_issue()
        if not top_issue:
            return "chile-hub top issue\n\nSin top issue activo.\n"

        rows = [
            ["dataset", top_issue.get("dataset", "unknown")],
            ["attention_priority", str(top_issue.get("attention_priority", "unknown"))],
            ["build_freshness_status", top_issue.get("build_freshness_status", "unknown")],
            ["current_freshness_status", top_issue.get("current_freshness_status", "unknown")],
            ["drift_status", top_issue.get("drift_status", "unknown")],
            ["degradation_status", top_issue.get("degradation_status", "unknown")],
            ["warning_count", str(top_issue.get("warning_count", 0))],
            ["source_detail", top_issue.get("source_detail", "unknown")],
            ["diagnostic_summary", top_issue.get("diagnostic_summary", "unknown")],
            ["recommended_action", top_issue.get("recommended_action", "unknown")],
        ]
        return render_table("chile-hub top issue", ["key", "value"], rows)

    def list_datasets(self) -> list[str]:
        return [entry["dataset"] for entry in self.catalog.get("datasets", [])]

    def get_dataset(self, dataset_name: str | Dataset) -> dict[str, Any]:
        dataset_name = _resolve_dataset_name(dataset_name)
        for entry in self.catalog.get("datasets", []):
            if entry["dataset"] == dataset_name:
                return entry  # type: ignore[no-any-return]  # dict[str, Any] en runtime
        raise ChileHubDatasetError(
            f"Dataset '{dataset_name}' no existe. {_format_available(self.list_datasets(), dataset_name)}"
        )

    def get_output_path(self, dataset_name: str | Dataset, output_type: str = "parquet") -> Path:
        dataset_name = _resolve_dataset_name(dataset_name)
        dataset = self.get_dataset(dataset_name)
        # Resolver alias: si el dataset apunta a otro, usar el canónico
        alias_for = dataset.get("alias_for")
        if alias_for:
            return self.get_output_path(alias_for, output_type)
        outputs = dataset.get("outputs", {})
        if output_type not in outputs:
            available_outputs = sorted(outputs.keys())
            raise ChileHubOutputError(
                f"Output '{output_type}' no existe para '{dataset_name}'. "
                f"{_format_available(available_outputs, output_type)}"
            )
        return self.root_dir / outputs[output_type]  # type: ignore[no-any-return]  # Path en runtime

    def load_polars(self, dataset_name: str | Dataset, validate: bool = False) -> pl.DataFrame:
        """Carga un dataset como DataFrame de Polars.

        Args:
            dataset_name: Nombre del dataset (ej. "comunas", "indicadores").
            validate: Si ``True``, ejecuta ``validate_dataset()`` antes de retornar.
                Lanza ``ChileHubDataError`` si la validación falla.

        Returns:
            DataFrame de Polars con los datos del dataset.

        Raises:
            ChileHubDatasetError: Si el dataset no existe o el archivo es ilegible.
            ChileHubDataError: Si ``validate=True`` y la validación encuentra errores.
        """
        dataset_name = _resolve_dataset_name(dataset_name)
        if dataset_name not in self._df_cache:
            path = self.get_output_path(dataset_name, "parquet")
            try:
                df = pl.read_parquet(path)
            except FileNotFoundError:
                raise ChileHubDatasetError(
                    f"Archivo Parquet no encontrado para '{dataset_name}': {path}"
                )
            except Exception as exc:
                raise ChileHubDatasetError(f"Error al leer Parquet para '{dataset_name}': {exc}")
            self._df_cache[dataset_name] = df

        if validate:
            result = self.validate_dataset(dataset_name)
            if result["status"] == "error":
                raise ChileHubDataError(
                    f"Validación fallida para '{dataset_name}': {'; '.join(result['errors'])}"
                )

        return self._df_cache[dataset_name]

    def cross_view(
        self,
        datasets: list[str | Dataset],
        on: str = "codigo_comuna",
        how: Literal["inner", "left", "right", "full", "semi", "anti", "cross", "outer"] = "left",
    ) -> pl.DataFrame:
        """Retorna un cruce predefinido de datasets vinculados por clave territorial.

        Args:
            datasets: Lista de nombres de datasets a cruzar (ej. ["comunas", "censo_comunal"]).
            on: Clave de join (default: "codigo_comuna").
            how: Tipo de join (default: "left").

        Returns:
            DataFrame de Polars con el cruce de todos los datasets.

        Raises:
            ChileHubDatasetError: Si se pasan menos de 2 datasets.
        """
        if len(datasets) < 2:
            raise ChileHubDatasetError(
                f"cross_view requiere al menos 2 datasets. Recibido: {len(datasets)}"
            )

        dfs = []
        for name in datasets:
            df = self.load_polars(name)
            # Prefijar columnas no-clave con el nombre del dataset para evitar colisiones
            cols_to_prefix = [c for c in df.columns if c != on]
            df = df.select([pl.col(on)] + [pl.col(c).alias(f"{name}_{c}") for c in cols_to_prefix])
            dfs.append(df)

        result = dfs[0]
        for df in dfs[1:]:
            result = result.join(df, on=on, how=how)
        return result

    def resolve_comunas(self, names: list[str]) -> pl.DataFrame:
        """Resuelve nombres de comuna (tipeados por humanos) a códigos CUT.

        Match **determinista**: normaliza cada nombre a su forma ``nombre_comuna_clean``
        (minúsculas, sin acentos, sin ``ñ``) y hace coincidencia exacta contra el
        dataset ``comunas``. No corrige typos ni hace coincidencia difusa (ver
        ``docs/adr/ADR-009-resolutor-nombres-comunales.md``).

        Args:
            names: Lista de nombres de comuna. Para resolver una columna de un
                DataFrame, pásala como ``df["mi_columna"].to_list()``.

        Returns:
            DataFrame Polars con una fila por input (mismo orden, duplicados
            preservados) y columnas: ``input``, ``codigo_comuna``, ``nombre_comuna``,
            ``codigo_region``, ``matched`` (bool). Los no encontrados tienen
            ``matched=False`` y códigos nulos — sin lanzar excepción.

        Examples:
            >>> hub = ChileHub()
            >>> hub.resolve_comunas(["Ñuñoa", "concon", "No Existe"])
        """
        comunas = self.load_polars("comunas")
        lookup = {
            row["nombre_comuna_clean"]: (
                row["codigo_comuna"],
                row["nombre_comuna"],
                row["codigo_region"],
            )
            for row in comunas.iter_rows(named=True)
        }
        rows = []
        for original in names:
            key = normalize_comuna_name(str(original))
            hit = lookup.get(key)
            rows.append(
                {
                    "input": original,
                    "codigo_comuna": hit[0] if hit else None,
                    "nombre_comuna": hit[1] if hit else None,
                    "codigo_region": hit[2] if hit else None,
                    "matched": hit is not None,
                }
            )
        return pl.DataFrame(
            rows,
            schema={
                "input": pl.String,
                "codigo_comuna": pl.String,
                "nombre_comuna": pl.String,
                "codigo_region": pl.String,
                "matched": pl.Boolean,
            },
        )

    def sql(self, query: str) -> pl.DataFrame:
        """Ejecuta una consulta SQL sobre los datasets publicados como vistas DuckDB.

        Cada dataset del catálogo se registra como una vista DuckDB con el mismo
        nombre que el dataset. Las vistas se respaldan con los archivos Parquet
        publicados y se registran de forma lazy en la primera llamada a ``sql()``.

        Args:
            query: Consulta SQL que referencia las vistas con los nombres de los
                datasets (ej. ``"SELECT * FROM comunas LIMIT 5"``).

        Returns:
            DataFrame de Polars con el resultado de la consulta.

        Raises:
            ImportError: Si ``duckdb`` no está instalado.
                Instálalo con ``pip install chile-hub[query]``.

        Examples:
            >>> hub = ChileHub()
            >>> hub.sql("SELECT codigo_comuna, nombre_comuna_clean FROM comunas LIMIT 5")
            >>> hub.sql("SELECT c.codigo_comuna, c.nombre_comuna_clean, cen.poblacion_censada FROM comunas c JOIN censo_comunal cen USING(codigo_comuna) LIMIT 10")
        """
        try:
            import duckdb
        except ImportError:
            raise ImportError(
                "Para usar SQL necesitas instalar duckdb: pip install chile-hub[query]"
            )

        if not hasattr(self, "_sql_views_registered"):
            con = duckdb.connect()
            for dataset_name in Dataset:
                try:
                    name = dataset_name.value
                    path = self.get_output_path(name, "parquet")
                    con.execute(f"CREATE VIEW \"{name}\" AS SELECT * FROM read_parquet('{path}')")  # nosec
                except Exception:  # nosec
                    # Skip datasets whose Parquet isn't available
                    pass
            self._sql_views_registered = True
            self._sql_connection = con
        else:
            con = self._sql_connection

        return cast(pl.DataFrame, con.execute(query).pl())

    @classmethod
    def from_datapackage(cls, path_or_url: str | Path) -> "ChileHub":
        """Abre un ChileHub a partir de un descriptor ``datapackage.json`` externo.

        Resuelve el directorio de datos a partir del descriptor: si el descriptor
        esta en un directorio con los archivos Parquet referenciados en
        ``resources[].path``, esos archivos se usaran directamente. Si no, se
        asume que los paths son relativos al directorio del descriptor.

        Args:
            path_or_url: Ruta local o URL (``http://``/``https://``) a un archivo
                ``datapackage.json``.

        Returns:
            Instancia de ``ChileHub`` configurada para usar los datos del
            descriptor.

        Raises:
            FileNotFoundError: Si el descriptor no existe en la ruta local.
            ChileHubDataError: Si ``path_or_url`` es una URL. El descriptor remoto
                se valida (existe, es un Frictionless Data Package conforme), pero
                ``ChileHub`` todavia no sabe leer Parquet remoto directamente desde
                una URL arbitraria (``__init__`` asume un directorio local) — ver
                ADR-010, Preguntas abiertas. Usa ``ChileHub()`` sin argumentos para
                el bundle publicado (se descarga y cachea automaticamente), o
                ``hub.sql(...)``/``docs/http-access.md`` para consumir el hosting
                HTTP directamente sin pasar por esta API.
            ImportError: Si ``frictionless`` no esta instalado.
                Instalalo con ``pip install chile-hub[validation]``.

        Example:
            >>> hub = ChileHub.from_datapackage("data/normalized/datapackage.json")
            >>> hub.summary()
        """
        try:
            import frictionless
        except ImportError:
            raise ImportError(
                "Para usar from_datapackage necesitas instalar frictionless: "
                "pip install chile-hub[validation]"
            )

        path_or_url_str = str(path_or_url)
        if path_or_url_str.startswith(("http://", "https://")):
            # Valida que el descriptor remoto exista y sea un Frictionless Data
            # Package conforme (levanta si frictionless no puede resolverlo).
            _ = frictionless.Package(path_or_url_str)
            raise ChileHubDataError(
                f"El descriptor remoto '{path_or_url_str}' es valido, pero "
                "from_datapackage(url) todavia no soporta leer Parquet remoto "
                "directamente (ChileHub asume un directorio de datos local). "
                "Usa ChileHub() sin argumentos para el bundle publicado, o "
                "consulta docs/http-access.md para consumir la URL directamente "
                "(ej. con Polars/DuckDB/arrow). Ver ADR-010."
            )

        path = Path(path_or_url)
        if not path.exists():
            raise FileNotFoundError(f"Descriptor no encontrado: {path}")

        _ = frictionless.Package(str(path))
        data_dir = path.parent
        return cls(data_dir=data_dir)

    def frictionless_validate(self, dataset_name: str | None = None) -> dict:
        """Valida el bundle local contra el descriptor ``datapackage.json``.

        Usa ``frictionless.Package.validate_descriptor()`` para verificar la
        estructura del descriptor. Es una validacion de metadatos (no carga
        los datos completos).

        Args:
            dataset_name: Si se especifica, se verifica que el dataset exista
                como recurso en el descriptor. Si es ``None``, se valida el
                descriptor completo.

        Returns:
            Diccionario con:
            - ``valid`` (bool): ``True`` si el descriptor es valido.
            - ``errors`` (list[str]): Errores encontrados (vacio si es valido).
            - ``stats`` (dict): Metadatos del resultado (``tasks``, ``warnings``).

        Raises:
            FileNotFoundError: Si ``datapackage.json`` no existe en el
                directorio de datos.
            ImportError: Si ``frictionless`` no esta instalado.
                Instalalo con ``pip install chile-hub[validation]``.

        Example:
            >>> hub = ChileHub()
            >>> result = hub.frictionless_validate()
            >>> print(result["valid"])
            True
        """
        try:
            import frictionless
        except ImportError:
            raise ImportError(
                "Para validar datapackage.json necesitas instalar frictionless: "
                "pip install chile-hub[validation]"
            )

        descriptor_path = self.normalized_dir / "datapackage.json"
        if not descriptor_path.exists():
            raise FileNotFoundError(
                f"datapackage.json no encontrado en {self.normalized_dir}. "
                "Ejecuta el build primero o descarga un bundle."
            )

        if dataset_name is not None:
            # Validate that the dataset name exists as a resource
            package = frictionless.Package(str(descriptor_path))
            names = [r.name for r in package.resources]
            if dataset_name not in names:
                return {
                    "valid": False,
                    "errors": [
                        f"Dataset '{dataset_name}' no encontrado en datapackage.json. "
                        f"Disponibles: {', '.join(sorted(names))}"
                    ],
                    "stats": {"resources_total": len(names)},
                }
            return {
                "valid": True,
                "errors": [],
                "stats": {"resources_total": len(names), "checked": dataset_name},
            }

        report = frictionless.Package.validate_descriptor(str(descriptor_path))
        errors = [str(e) for e in report.errors] if report.errors else []
        return {
            "valid": report.valid,
            "errors": errors,
            "stats": {
                "tasks": getattr(report, "tasks", []),
                "warnings": getattr(report, "warnings", []),
            },
        }

    def validate_dataset(self, dataset_name: str | Dataset) -> dict:
        """Valida los datos publicados del hub contra su contrato JSON Schema.

        Carga el dataset desde Parquet y lo coteja contra el contrato en
        ``contracts/datasets/{dataset_name}.schema.json``. Retorna un dict
        con ``status``, ``errors`` y ``warnings``.

        Args:
            dataset_name: Nombre del dataset (ej. ``"comunas"``).

        Returns:
            Dict con:
            - ``dataset``: nombre del dataset validado.
            - ``status``: ``"ok"`` o ``"error"``.
            - ``errors``: lista de errores de validación.
            - ``warnings``: lista de advertencias.

        Raises:
            ChileHubDatasetError: Si no existe contrato o dataset.
        """
        dataset_name = _resolve_dataset_name(dataset_name)
        from .contracts import verify_dataset_contract

        contract_path = self.root_dir / "contracts" / "datasets" / f"{dataset_name}.schema.json"
        if not contract_path.exists():
            raise ChileHubDatasetError(
                f"No existe contrato de schema para '{dataset_name}'. "
                f"Datasets disponibles: {self.list_datasets()}"
            )

        with contract_path.open("r", encoding="utf-8") as f:
            contract = json.load(f)

        df = self.load_polars(dataset_name)
        catalog_entry = self.get_dataset(dataset_name)
        outputs = catalog_entry.get("outputs", {}) if catalog_entry else {}

        return verify_dataset_contract(
            dataset_name,
            contract,
            df,
            outputs=outputs,
            root_dir=self.root_dir,
        )

    def validate_user_data(self, df: pl.DataFrame, dataset_name: str | Dataset) -> dict:
        """Valida un DataFrame de usuario contra el contrato de schema del dataset.

        Usa los archivos de contrato en contracts/datasets/*.schema.json, que definen
        required_columns, column_types, primary_key y expected_record_count.

        Args:
            df: DataFrame de Polars a validar.
            dataset_name: Nombre del dataset de referencia (ej. "comunas").

        Returns:
            Dict con:
            - status: "ok" si pasa todas las validaciones, "error" si falla alguna.
            - errors: lista de strings con mensajes de error (vacía si status=="ok").
            - warnings: lista de strings con advertencias no bloqueantes.
            - schema_used: ruta absoluta del schema usado.

        Raises:
            ChileHubDatasetError: Si no existe contrato para el dataset solicitado.
        """
        dataset_name = _resolve_dataset_name(dataset_name)
        schema_path = ROOT_DIR / "contracts" / "datasets" / f"{dataset_name}.schema.json"
        if not schema_path.exists():
            raise ChileHubDatasetError(
                f"No existe contrato de schema para '{dataset_name}'. "
                f"Datasets disponibles: {self.list_datasets()}"
            )

        import json as json_module

        with open(schema_path, "r", encoding="utf-8") as f:
            schema = json_module.load(f)

        errors = []
        warnings = []

        # 1. Validar columnas requeridas (required_columns)
        required_cols = schema.get("required_columns", [])
        df_cols = df.columns
        missing = [c for c in required_cols if c not in df_cols]
        if missing:
            errors.append(f"Columnas requeridas faltantes: {missing}")

        # 2. Validar tipos (column_types)
        column_types = schema.get("column_types", {})
        type_map = {
            "string": ["String", "Utf8", "str"],
            "integer": ["Int64", "Int32", "Int16", "Int8", "UInt32", "UInt16"],
            "number": ["Float64", "Float32"],
            "boolean": ["Boolean"],
            "date": ["Date"],
        }
        for col, expected_type in column_types.items():
            if col not in df_cols:
                continue
            actual_dtype = str(df[col].dtype)
            expected_names = type_map.get(expected_type, [expected_type])
            if actual_dtype not in expected_names:
                errors.append(
                    f"Columna '{col}': se esperaba {expected_type}, se encontró {actual_dtype}"
                )

        # 3. Validar clave primaria (primary_key)
        primary_key = schema.get("primary_key", [])
        if primary_key:
            pk_cols = [c for c in primary_key if c in df_cols]
            if pk_cols:
                if df.select(pk_cols).null_count().sum_horizontal().sum() > 0:
                    errors.append(f"Clave primaria {primary_key} contiene valores nulos")
                pk_df = df.select(pk_cols)
                if pk_df.height != pk_df.unique().height:
                    errors.append(f"Clave primaria {primary_key} tiene valores duplicados")

        # 4. Verificar expected_record_count (solo advertencia)
        expected = schema.get("expected_record_count")
        if expected is not None and df.height != expected:
            warnings.append(
                f"Cantidad de registros ({df.height}) difiere de la esperada ({expected})"
            )

        status = "ok" if not errors else "error"
        return {
            "status": status,
            "errors": errors,
            "warnings": warnings,
            "schema_used": str(schema_path),
        }

    def search_datasets(
        self, query: str = "", source_name: str = "", maturity: str = ""
    ) -> list[dict]:
        """Busca datasets por keyword, fuente, o nivel de madurez.

        Args:
            query: Texto libre para buscar en nombre y descripción.
            source_name: Filtrar por fuente (ej. "INE", "MINSAL"). Coincidencia parcial
                sin distinción de mayúsculas.
            maturity: Filtrar por maturity_status (ej. "stable", "candidate").

        Returns:
            Lista de dicts con información de cada dataset que coincide:
            {"name", "description", "source_name", "record_count", "maturity_status", "fields"}.
        """
        results = []
        query_lower = query.lower().strip() if query else ""
        source_lower = source_name.lower().strip() if source_name else ""
        maturity_lower = maturity.lower().strip() if maturity else ""

        # Cargar source_readiness para maturity_status
        source_readiness = self._load_source_readiness()
        maturity_by_dataset = {
            entry["dataset"]: entry.get("maturity_status", "")
            for entry in source_readiness.get("datasets", [])
        }

        for entry in self.catalog.get("datasets", []):
            name = entry.get("dataset", "")
            desc = entry.get("description", "").lower()

            # Filtro por query
            if query_lower:
                if query_lower not in name.lower() and query_lower not in desc:
                    continue

            # Filtro por fuente
            entry_source = entry.get("source_name", "").lower()
            if source_lower and source_lower not in entry_source:
                continue

            # Filtro por maturity_status
            if maturity_lower:
                entry_maturity = maturity_by_dataset.get(name, "").lower()
                if maturity_lower != entry_maturity:
                    continue

            results.append(
                {
                    "name": name,
                    "description": entry.get("description", ""),
                    "source_name": entry.get("source_name", ""),
                    "record_count": entry.get("record_count", 0),
                    "maturity_status": maturity_by_dataset.get(name, ""),
                    "fields": entry.get("fields", []),
                }
            )

        return results

    def example_usage(self, dataset_name: str | Dataset, kind: str = "python") -> str:
        dataset_name = _resolve_dataset_name(dataset_name)
        dataset = self.get_dataset(dataset_name)
        examples = dataset.get("usage_examples", {})
        if kind not in examples:
            available_examples = sorted(examples.keys())
            raise ChileHubExampleError(
                f"Example '{kind}' no existe para '{dataset_name}'. "
                f"{_format_available(available_examples, kind)}"
            )
        return examples[kind]  # type: ignore[no-any-return]  # str en runtime

    def summary(self) -> list[dict[str, Any]]:
        return [
            {
                "dataset": entry["dataset"],
                "source_mode": entry["source_mode"],
                "record_count": entry["record_count"],
                "join_keys": entry.get("join_keys", []),
                "confidence_tier": entry.get("confidence_tier"),
                "reuse_status": entry.get("reuse_policy", {}).get("status"),
                "reuse_license": entry.get("reuse_policy", {}).get("license"),
                "attribution_required": entry.get("reuse_policy", {}).get("attribution_required"),
                "freshness_status": entry.get("freshness", {}).get("status"),
                "freshness_age_hours": entry.get("freshness", {}).get("age_hours"),
                "coverage_status": entry.get("coverage", {}).get("status"),
                "coverage_ratio": entry.get("coverage", {}).get("coverage_ratio"),
                "validation_status": entry.get("validation_status"),
                "warning_count": len(entry.get("warnings", [])),
                "drift_status": entry.get("drift", {}).get("status"),
                "drift_summary": entry.get("drift", {}).get("summary"),
                "degradation_status": entry.get("degradation", {}).get("status"),
                "degradation_impact": entry.get("degradation", {}).get("impact"),
            }
            for entry in self.catalog.get("datasets", [])
        ]

    def summary_table(self) -> str:
        rows = self.summary()
        table_rows = [
            [
                entry.get("dataset", "unknown"),
                entry.get("source_mode", "unknown"),
                str(entry.get("record_count", "N/D")),
                entry.get("freshness_status", "unknown"),
                entry.get("coverage_status", "unknown"),
                entry.get("validation_status", "unknown"),
                entry.get("drift_status", "unknown"),
                str(entry.get("warning_count", 0)),
            ]
            for entry in rows
        ]
        return render_table(
            "chile-hub summary",
            [
                "dataset",
                "mode",
                "records",
                "freshness",
                "coverage",
                "validation",
                "drift",
                "warnings",
            ],
            table_rows,
        )

    def snapshot_text(self):
        overview = self.overview()
        freshness_audit = self.freshness_audit()
        runtime_status = self.runtime_status_audit()
        freshness_by_dataset = {
            entry.get("dataset"): entry for entry in freshness_audit.get("datasets", [])
        }
        package = overview.get("primary_package") or {}
        top_issue = overview.get("top_issue")
        lines = [
            "chile-hub snapshot",
            f"generated_at_utc: {overview.get('generated_at_utc', 'unknown')}",
            (
                f"status_build: {overview.get('build_overall_status', overview.get('overall_status', 'unknown'))} | "
                f"status_current: {overview.get('current_overall_status', runtime_status.get('current_overall_status', 'unknown'))} | "
                f"datasets={overview.get('dataset_count', 0)} | "
                f"live={overview.get('live_count', 0)} | "
                f"stale={overview.get('stale_count', 0)} | "
                f"drifted={overview.get('drifted_count', 0)} | "
                f"warnings={overview.get('warning_count', 0)}"
            ),
            (
                f"current_freshness: fresh={freshness_audit.get('fresh_count', 0)} | "
                f"stale={freshness_audit.get('stale_count', 0)} | "
                f"unknown={freshness_audit.get('unknown_count', 0)} | "
                f"checked_at={freshness_audit.get('checked_at_utc', 'unknown')}"
            ),
        ]
        if top_issue:
            lines.append(
                f"top_issue: {top_issue.get('dataset')} | "
                f"build={top_issue.get('build_freshness_status', 'unknown')} | "
                f"current={top_issue.get('current_freshness_status', 'unknown')} | "
                f"drift={top_issue.get('drift_status', 'unknown')} | "
                f"warnings={top_issue.get('warning_count', 0)}"
            )
            lines.append(f"top_issue_reason: {top_issue.get('diagnostic_summary', 'unknown')}")
            lines.append(f"top_issue_action: {top_issue.get('recommended_action', 'unknown')}")

        if package:
            lines.append(
                f"package: {package.get('path', 'unknown')} | "
                f"{package.get('package_type', 'unknown')} | "
                f"checksum={package.get('checksum_algorithm', 'unknown')}"
            )
            lines.append(f"verify: {package.get('verification_command', 'unknown')}")

        lines.append("")
        for entry in overview.get("datasets", []):
            runtime_freshness = freshness_by_dataset.get(entry.get("dataset"), {})
            lines.append(
                f"- {entry.get('dataset', 'unknown')}: "
                f"mode={entry.get('source_mode', 'unknown')}, "
                f"validation={entry.get('validation_status', 'unknown')}, "
                f"freshness_build={entry.get('freshness_status', 'unknown')}, "
                f"freshness_now={runtime_freshness.get('current_freshness_status', 'unknown')}, "
                f"coverage={entry.get('coverage_status', 'unknown')}, "
                f"drift={entry.get('drift_status', 'unknown')}"
            )

        return "\n".join(lines) + "\n"

    def snapshot_table(self):
        overview = self.overview()
        freshness_audit = self.freshness_audit()
        freshness_by_dataset = {
            entry.get("dataset"): entry for entry in freshness_audit.get("datasets", [])
        }
        rows = [
            ("generated_at_utc", overview.get("generated_at_utc", "unknown")),
            (
                "build_overall_status",
                overview.get("build_overall_status", overview.get("overall_status", "unknown")),
            ),
            (
                "current_overall_status",
                overview.get("current_overall_status", "unknown"),
            ),
            ("datasets", str(overview.get("dataset_count", 0))),
            ("live", str(overview.get("live_count", 0))),
            ("stale", str(overview.get("stale_count", 0))),
            ("drifted", str(overview.get("drifted_count", 0))),
            ("warnings", str(overview.get("warning_count", 0))),
            ("current_fresh", str(freshness_audit.get("fresh_count", 0))),
            ("current_stale", str(freshness_audit.get("stale_count", 0))),
            ("current_unknown", str(freshness_audit.get("unknown_count", 0))),
            ("audit_checked", freshness_audit.get("checked_at_utc", "unknown")),
        ]
        top_issue = overview.get("top_issue")
        if top_issue:
            rows.extend(
                [
                    ("top_issue", top_issue.get("dataset", "unknown")),
                    (
                        "top_issue_build",
                        top_issue.get("build_freshness_status", "unknown"),
                    ),
                    (
                        "top_issue_current",
                        top_issue.get("current_freshness_status", "unknown"),
                    ),
                    ("top_issue_drift", top_issue.get("drift_status", "unknown")),
                    (
                        "top_issue_reason",
                        top_issue.get("diagnostic_summary", "unknown"),
                    ),
                    (
                        "top_issue_action",
                        top_issue.get("recommended_action", "unknown"),
                    ),
                    (
                        "top_issue_summary",
                        overview.get("top_issue_summary", format_top_issue_summary(top_issue)),
                    ),
                ]
            )

        package = overview.get("primary_package") or {}
        if package:
            rows.extend(
                [
                    ("package_path", package.get("path", "unknown")),
                    ("package_type", package.get("package_type", "unknown")),
                    ("checksum", package.get("checksum_algorithm", "unknown")),
                    ("verify", package.get("verification_command", "unknown")),
                ]
            )

        label_width = max(len(label) for label, _ in rows)
        lines = ["chile-hub snapshot table", ""]
        lines.extend(f"{label.ljust(label_width)} : {value}" for label, value in rows)

        dataset_rows = []
        for entry in overview.get("datasets", []):
            runtime_freshness = freshness_by_dataset.get(entry.get("dataset"), {})
            dataset_rows.append(
                [
                    entry.get("dataset", "unknown"),
                    entry.get("source_mode", "unknown"),
                    entry.get("validation_status", "unknown"),
                    entry.get("freshness_status", "unknown"),
                    runtime_freshness.get("current_freshness_status", "unknown"),
                    entry.get("coverage_status", "unknown"),
                    entry.get("drift_status", "unknown"),
                ]
            )

        lines.append("")
        lines.append(
            render_table(
                "",
                ["dataset", "mode", "validation", "build", "current", "coverage", "drift"],
                dataset_rows,
            )
        )
        return "\n".join(lines) + "\n"

    def artifacts(self, dataset_name=None):
        manifest = self._load_artifact_manifest()
        artifacts = manifest.get("artifacts", [])
        if dataset_name is None:
            return artifacts

        self.get_dataset(dataset_name)
        return [entry for entry in artifacts if entry.get("dataset") == dataset_name]

    def shared_artifacts(self, shared_type=None, format=None):
        artifacts = [entry for entry in self.artifacts() if entry.get("shared_type")]
        if shared_type is not None:
            artifacts = [entry for entry in artifacts if entry.get("shared_type") == shared_type]
        if format is not None:
            artifacts = [entry for entry in artifacts if entry.get("format") == format]
        return artifacts

    def shared_artifacts_table(self, shared_type=None, format=None):
        artifacts = self.shared_artifacts(shared_type, format)
        table_rows = []
        for entry in artifacts:
            size_bytes = entry.get("size_bytes")
            if isinstance(size_bytes, int):
                if size_bytes < 1024:
                    size_label = f"{size_bytes} B"
                else:
                    size_label = f"{size_bytes / 1024:.1f} KB"
            else:
                size_label = "N/D"
            table_rows.append(
                [
                    entry.get("shared_type", "unknown"),
                    entry.get("format", "unknown"),
                    size_label,
                    entry.get("path", "unknown"),
                ]
            )
        return render_table(
            "chile-hub shared artifacts",
            ["shared_type", "format", "size", "path"],
            table_rows,
        )

    def reports(self):
        return self.bundle().get("reports", {})

    def report_index(self):
        rows = []
        for report_key, entry in sorted(self.reports().items()):
            rows.append(
                {
                    "report_key": report_key,
                    "shared_type": entry.get("shared_type"),
                    "format": entry.get("format"),
                    "path": entry.get("path"),
                    "size_bytes": entry.get("size_bytes"),
                    "sha256": entry.get("sha256"),
                }
            )
        return rows

    def report_index_table(self):
        rows = self.report_index()
        table_rows = []
        for entry in rows:
            size_bytes = entry.get("size_bytes")
            if isinstance(size_bytes, int):
                if size_bytes < 1024:
                    size_label = f"{size_bytes} B"
                else:
                    size_label = f"{size_bytes / 1024:.1f} KB"
            else:
                size_label = "N/D"
            table_rows.append(
                [
                    entry.get("report_key", "unknown"),
                    entry.get("shared_type", "unknown"),
                    entry.get("format", "unknown"),
                    size_label,
                    entry.get("path", "unknown"),
                ]
            )
        return render_table(
            "chile-hub report index",
            ["report_key", "shared_type", "format", "size", "path"],
            table_rows,
        )

    def get_report(self, shared_type, format):
        for entry in self.reports().values():
            if entry.get("shared_type") == shared_type and entry.get("format") == format:
                return entry
        raise KeyError(f"Reporte '{shared_type}' con formato '{format}' no existe en el bundle.")

    def overview(self):
        health = self.health()
        bundle = self.bundle()
        packages = self.packages()
        runtime_status = self.runtime_status_audit()
        primary_package = None
        try:
            primary_package = self.primary_package()
        except KeyError:
            primary_package = None
        top_issue = self.top_issue()
        shared_artifacts = self.shared_artifacts()
        return {
            "generated_at_utc": health.get("generated_at_utc"),
            "overall_status": health.get("overall_status"),
            "build_overall_status": health.get("overall_status"),
            "current_overall_status": runtime_status.get("current_overall_status"),
            "dataset_count": health.get("dataset_count"),
            # Los contadores describen el conjunto activo; sin retired_count el
            # resumen no cuadraria con dataset_count (ADR-015).
            "retired_count": health.get("retired_count"),
            "live_count": health.get("live_count"),
            "fallback_count": health.get("fallback_count"),
            "stale_count": health.get("stale_count"),
            "drifted_count": health.get("drifted_count"),
            "degraded_count": health.get("degraded_count"),
            "partial_coverage_count": health.get("partial_coverage_count"),
            "warning_count": health.get("warning_count"),
            "current_fresh_count": runtime_status.get("fresh_count"),
            "current_stale_count": runtime_status.get("stale_count"),
            "current_unknown_count": runtime_status.get("unknown_count"),
            "current_checked_at_utc": runtime_status.get("checked_at_utc"),
            "top_issue": top_issue,
            "top_issue_summary": format_top_issue_summary(top_issue),
            "shared_artifact_count": len(shared_artifacts),
            "package_count": len(packages),
            "primary_package": (
                {
                    "path": primary_package.get("path"),
                    "package_type": primary_package.get("package_type"),
                    "size_bytes": primary_package.get("size_bytes"),
                    "checksum_algorithm": primary_package.get("checksum_algorithm"),
                    "checksum_path": primary_package.get("checksum_path"),
                    "verification_command": primary_package.get("verification_command"),
                }
                if primary_package
                else None
            ),
            "report_keys": sorted(bundle.get("reports", {}).keys()),
            "datasets": [
                {
                    "dataset": entry.get("dataset"),
                    "source_mode": entry.get("source_mode"),
                    "validation_status": entry.get("validation_status"),
                    "freshness_status": entry.get("freshness_status"),
                    "coverage_status": entry.get("coverage_status"),
                    "drift_status": entry.get("drift_status"),
                }
                for entry in health.get("datasets", [])
            ],
        }

    def overview_table(self):
        overview = self.overview()
        rows = [
            ("generated_at_utc", overview.get("generated_at_utc", "unknown")),
            ("build_overall_status", overview.get("build_overall_status", "unknown")),
            (
                "current_overall_status",
                overview.get("current_overall_status", "unknown"),
            ),
            ("datasets", str(overview.get("dataset_count", 0))),
            ("live", str(overview.get("live_count", 0))),
            ("fallback", str(overview.get("fallback_count", 0))),
            ("build_stale", str(overview.get("stale_count", 0))),
            ("current_fresh", str(overview.get("current_fresh_count", 0))),
            ("current_stale", str(overview.get("current_stale_count", 0))),
            ("current_unknown", str(overview.get("current_unknown_count", 0))),
            ("drifted", str(overview.get("drifted_count", 0))),
            ("degraded", str(overview.get("degraded_count", 0))),
            ("partial_coverage", str(overview.get("partial_coverage_count", 0))),
            ("warnings", str(overview.get("warning_count", 0))),
            ("shared_artifacts", str(overview.get("shared_artifact_count", 0))),
            ("packages", str(overview.get("package_count", 0))),
            ("current_checked_at", overview.get("current_checked_at_utc", "unknown")),
        ]
        top_issue = overview.get("top_issue")
        if top_issue:
            rows.extend(
                [
                    ("top_issue", top_issue.get("dataset", "unknown")),
                    (
                        "top_issue_build",
                        top_issue.get("build_freshness_status", "unknown"),
                    ),
                    (
                        "top_issue_current",
                        top_issue.get("current_freshness_status", "unknown"),
                    ),
                    ("top_issue_drift", top_issue.get("drift_status", "unknown")),
                    (
                        "top_issue_reason",
                        top_issue.get("diagnostic_summary", "unknown"),
                    ),
                    (
                        "top_issue_action",
                        top_issue.get("recommended_action", "unknown"),
                    ),
                    (
                        "top_issue_summary",
                        overview.get("top_issue_summary", format_top_issue_summary(top_issue)),
                    ),
                ]
            )

        package = overview.get("primary_package") or {}
        if package:
            rows.extend(
                [
                    ("package_path", package.get("path", "unknown")),
                    ("package_type", package.get("package_type", "unknown")),
                    ("checksum", package.get("checksum_algorithm", "unknown")),
                ]
            )

        label_width = max(len(label) for label, _ in rows)
        lines = ["chile-hub overview", ""]
        lines.extend(f"{label.ljust(label_width)} : {value}" for label, value in rows)

        dataset_rows = [
            [
                entry.get("dataset", "unknown"),
                entry.get("source_mode", "unknown"),
                entry.get("validation_status", "unknown"),
                entry.get("freshness_status", "unknown"),
                entry.get("coverage_status", "unknown"),
                entry.get("drift_status", "unknown"),
            ]
            for entry in overview.get("datasets", [])
        ]

        lines.append("")
        lines.append(
            render_table(
                "",
                ["dataset", "mode", "validation", "build", "coverage", "drift"],
                dataset_rows,
            )
        )
        return "\n".join(lines) + "\n"

    def inventory(self):
        inventory = []
        manifest_artifacts = self.artifacts()
        by_dataset: dict[str, list[dict[str, Any]]] = {}
        for artifact in manifest_artifacts:
            dataset = artifact.get("dataset")
            if not dataset:
                continue
            by_dataset.setdefault(dataset, []).append(artifact)

        for entry in self.catalog.get("datasets", []):
            dataset_name = entry["dataset"]
            artifacts = sorted(
                by_dataset.get(dataset_name, []),
                key=lambda item: (
                    item.get("output_type") or "",
                    item.get("path") or "",
                ),
            )
            published_outputs = [
                artifact["output_type"] for artifact in artifacts if artifact.get("output_type")
            ]
            inventory.append(
                {
                    "dataset": dataset_name,
                    "source_mode": entry.get("source_mode"),
                    "record_count": entry.get("record_count"),
                    "validation_status": entry.get("validation_status"),
                    "confidence_tier": entry.get("confidence_tier"),
                    "reuse_status": entry.get("reuse_policy", {}).get("status"),
                    "reuse_license": entry.get("reuse_policy", {}).get("license"),
                    "attribution_required": entry.get("reuse_policy", {}).get(
                        "attribution_required"
                    ),
                    "freshness_status": entry.get("freshness", {}).get("status"),
                    "freshness_age_hours": entry.get("freshness", {}).get("age_hours"),
                    "coverage_status": entry.get("coverage", {}).get("status"),
                    "coverage_ratio": entry.get("coverage", {}).get("coverage_ratio"),
                    "warning_count": len(entry.get("warnings", [])),
                    "drift_status": entry.get("drift", {}).get("status"),
                    "drift_summary": entry.get("drift", {}).get("summary"),
                    "degradation_status": entry.get("degradation", {}).get("status"),
                    "degradation_impact": entry.get("degradation", {}).get("impact"),
                    "published_outputs": published_outputs,
                    "artifact_count": len(artifacts),
                    "total_size_bytes": sum(
                        artifact.get("size_bytes", 0) for artifact in artifacts
                    ),
                    "artifacts": [
                        {
                            "path": artifact.get("path"),
                            "output_type": artifact.get("output_type"),
                            "size_bytes": artifact.get("size_bytes"),
                        }
                        for artifact in artifacts
                    ],
                }
            )
        return inventory

    def inventory_table(self):
        rows = self.inventory()
        table_rows = []
        for entry in rows:
            outputs = ",".join(entry.get("published_outputs", [])) or "N/D"
            size_bytes = entry.get("total_size_bytes")
            if isinstance(size_bytes, int):
                if size_bytes < 1024:
                    size_label = f"{size_bytes} B"
                else:
                    size_label = f"{size_bytes / 1024:.1f} KB"
            else:
                size_label = "N/D"
            table_rows.append(
                [
                    entry.get("dataset", "unknown"),
                    entry.get("source_mode", "unknown"),
                    str(entry.get("record_count", "N/D")),
                    outputs,
                    size_label,
                    entry.get("freshness_status", "unknown"),
                    entry.get("coverage_status", "unknown"),
                    entry.get("drift_status", "unknown"),
                ]
            )
        return render_table(
            "chile-hub inventory",
            ["dataset", "mode", "records", "outputs", "size", "freshness", "coverage", "drift"],
            table_rows,
        )

    def health(self):
        health = self._load_hub_health()
        if "top_issue_summary" not in health:
            health["top_issue_summary"] = format_top_issue_summary(health.get("top_issue"))
        return health

    def status(self):
        status = self._load_hub_status()
        if "top_issue_summary" not in status:
            status["top_issue_summary"] = format_top_issue_summary(status.get("top_issue"))
        return status

    def dataset_status(self):
        return self._load_dataset_status()

    def dataset_changelog(self):
        return self._load_dataset_changelog()

    def source_readiness(self):
        """Devuelve el reporte de madurez de fuente por dataset."""
        return self._load_source_readiness()

    def check_sources(self, timeout: int = 5) -> list[dict[str, Any]]:
        """Verifica la conectividad de red con las fuentes de datos oficiales."""
        results = []
        for entry in self.catalog.get("datasets", []):
            dataset = entry.get("dataset")
            url = entry.get("source_url")
            source_name = entry.get("source_name")
            if not url:
                results.append(
                    {
                        "dataset": dataset,
                        "source_name": source_name,
                        "url": "N/A",
                        "status": "offline",
                        "status_code": None,
                        "latency_ms": None,
                        "error": "No source URL defined",
                    }
                )
                continue

            try:
                # Intenta HEAD primero
                response = requests.head(url, timeout=timeout, allow_redirects=True)
                if response.status_code >= 400:
                    response.close()
                    response = requests.get(url, timeout=timeout, stream=True)

                status = "online" if response.status_code < 400 else "offline"
                status_code = response.status_code
                latency_ms = round(response.elapsed.total_seconds() * 1000, 2)
                error = None
                response.close()
            except Exception as e:
                status = "offline"
                status_code = None
                latency_ms = None
                error = type(e).__name__

            results.append(
                {
                    "dataset": dataset,
                    "source_name": source_name,
                    "url": url,
                    "status": status,
                    "status_code": status_code,
                    "latency_ms": latency_ms,
                    "error": error,
                }
            )
        return results

    def check_sources_table(self, results: list[dict[str, Any]]) -> str:
        """Formatea el resultado de check_sources como una tabla amigable para terminal."""
        table_rows = []
        for entry in results:
            status = entry.get("status", "unknown")
            code = str(entry.get("status_code")) if entry.get("status_code") is not None else "N/A"
            latency = (
                f"{entry.get('latency_ms')}ms" if entry.get("latency_ms") is not None else "N/A"
            )
            url = entry.get("url", "N/A")
            if len(url) > 48:
                url = url[:45] + "..."
            table_rows.append(
                [
                    entry.get("dataset", "unknown"),
                    status,
                    code,
                    latency,
                    entry.get("source_name", "unknown"),
                    url,
                ]
            )
        return render_table(
            "chile-hub check-sources",
            ["dataset", "status", "code", "latency", "source name", "url"],
            table_rows,
        )

    def dataset_quality(self):
        """Devuelve la tarjeta de puntuación de calidad multidimensional por dataset."""
        return self._load_dataset_quality()

    def status_table(self):
        status = self.status()
        table_rows = [
            ["generated_at_utc", status.get("generated_at_utc", "unknown")],
            ["overall_status", status.get("overall_status", "unknown")],
            ["dataset_count", str(status.get("dataset_count", 0))],
            ["live_count", str(status.get("live_count", 0))],
            ["fallback_count", str(status.get("fallback_count", 0))],
            ["stale_count", str(status.get("stale_count", 0))],
            ["drifted_count", str(status.get("drifted_count", 0))],
            ["degraded_count", str(status.get("degraded_count", 0))],
            ["warning_count", str(status.get("warning_count", 0))],
        ]
        top_issue = status.get("top_issue")
        if top_issue:
            table_rows.extend(
                [
                    ["top_issue", top_issue.get("dataset", "unknown")],
                    ["top_issue_reason", top_issue.get("diagnostic_summary", "unknown")],
                    ["top_issue_action", top_issue.get("recommended_action", "unknown")],
                    [
                        "top_issue_summary",
                        status.get("top_issue_summary", format_top_issue_summary(top_issue)),
                    ],
                ]
            )
        return render_table("chile-hub status", ["key", "value"], table_rows)

    def health_table(self):
        health = self.health()
        lines = ["chile-hub health", ""]
        lines.append(
            "overall="
            f"{health.get('overall_status', 'unknown')} | "
            f"datasets={health.get('dataset_count', 0)} | "
            f"ok={health.get('ok_count', 0)} | "
            f"warn={health.get('warn_count', 0)} | "
            f"error={health.get('error_count', 0)} | "
            f"live={health.get('live_count', 0)} | "
            f"fallback={health.get('fallback_count', 0)} | "
            f"stale={health.get('stale_count', 0)} | "
            f"drifted={health.get('drifted_count', 0)}"
        )

        dataset_rows = [
            [
                entry.get("dataset", "unknown"),
                entry.get("severity", "unknown"),
                entry.get("source_mode", "unknown"),
                entry.get("freshness_status", "unknown"),
                entry.get("validation_status", "unknown"),
                entry.get("publishability_status", "unknown"),
                entry.get("coverage_status", "unknown"),
                entry.get("drift_status", "unknown"),
                str(entry.get("warning_count", 0)),
            ]
            for entry in health.get("datasets", [])
        ]
        lines.append("")
        lines.append(
            render_table(
                "",
                [
                    "dataset",
                    "severity",
                    "mode",
                    "freshness",
                    "validation",
                    "reuse",
                    "coverage",
                    "drift",
                    "warnings",
                ],
                dataset_rows,
            )
        )
        return "\n".join(lines) + "\n"

    def freshness_audit(self):
        checked_at = datetime.now(UTC)
        datasets = []
        fresh_count = 0
        stale_count = 0
        unknown_count = 0

        for entry in self.catalog.get("datasets", []):
            max_age_hours = entry.get("freshness_policy", {}).get("max_age_hours")
            freshness = compute_freshness(entry.get("refreshed_at_utc"), max_age_hours, checked_at)
            current_status = freshness["status"]

            if current_status == "fresh":
                fresh_count += 1
            elif current_status == "stale":
                stale_count += 1
            else:
                unknown_count += 1

            datasets.append(
                {
                    "dataset": entry.get("dataset"),
                    "source_mode": entry.get("source_mode"),
                    "refreshed_at_utc": entry.get("refreshed_at_utc"),
                    "build_freshness_status": entry.get("freshness", {}).get("status"),
                    "current_freshness_status": current_status,
                    "current_age_hours": freshness["age_hours"],
                    "max_age_hours": freshness["max_age_hours"],
                    "freshness_label": entry.get("freshness_policy", {}).get("label"),
                }
            )

        return {
            "checked_at_utc": checked_at.isoformat(),
            "dataset_count": len(datasets),
            "fresh_count": fresh_count,
            "stale_count": stale_count,
            "unknown_count": unknown_count,
            "datasets": datasets,
        }

    def runtime_status_audit(self):
        health = self.health()
        freshness_audit = self.freshness_audit()
        build_overall_status = health.get("overall_status", "unknown")
        runtime_freshness_status = "ok"
        if freshness_audit.get("unknown_count", 0) > 0 or freshness_audit.get("stale_count", 0) > 0:
            runtime_freshness_status = "warn"
        current_overall_status = self._max_status(build_overall_status, runtime_freshness_status)
        return {
            "build_overall_status": build_overall_status,
            "current_overall_status": current_overall_status,
            "fresh_count": freshness_audit.get("fresh_count", 0),
            "stale_count": freshness_audit.get("stale_count", 0),
            "unknown_count": freshness_audit.get("unknown_count", 0),
            "checked_at_utc": freshness_audit.get("checked_at_utc"),
        }

    def runtime_status(self):
        health = self.health()
        runtime_audit = self.runtime_status_audit()
        freshness_by_dataset = {
            entry.get("dataset"): entry for entry in self.freshness_audit().get("datasets", [])
        }
        datasets = []
        for entry in health.get("datasets", []):
            freshness_entry = freshness_by_dataset.get(entry.get("dataset"), {})
            datasets.append(
                {
                    "dataset": entry.get("dataset"),
                    "source_mode": entry.get("source_mode"),
                    "severity": entry.get("severity"),
                    "validation_status": entry.get("validation_status"),
                    "build_freshness_status": entry.get("freshness_status"),
                    "current_freshness_status": freshness_entry.get(
                        "current_freshness_status", "unknown"
                    ),
                    "current_age_hours": freshness_entry.get("current_age_hours"),
                    "max_age_hours": freshness_entry.get("max_age_hours"),
                    "coverage_status": entry.get("coverage_status"),
                    "drift_status": entry.get("drift_status"),
                    "warning_count": entry.get("warning_count", 0),
                }
            )
        top_issue = self.top_issue()
        return {
            "generated_at_utc": health.get("generated_at_utc"),
            "build_overall_status": runtime_audit.get("build_overall_status"),
            "current_overall_status": runtime_audit.get("current_overall_status"),
            "dataset_count": health.get("dataset_count"),
            "live_count": health.get("live_count"),
            "fallback_count": health.get("fallback_count"),
            "fresh_count": runtime_audit.get("fresh_count"),
            "stale_count": runtime_audit.get("stale_count"),
            "unknown_count": runtime_audit.get("unknown_count"),
            "drifted_count": health.get("drifted_count"),
            "warning_count": health.get("warning_count"),
            "checked_at_utc": runtime_audit.get("checked_at_utc"),
            "top_issue": top_issue,
            "top_issue_summary": format_top_issue_summary(top_issue),
            "datasets": datasets,
        }

    def runtime_status_table(self):
        runtime = self.runtime_status()
        lines = ["chile-hub runtime status", ""]
        lines.append(
            f"build={runtime.get('build_overall_status', 'unknown')} | "
            f"current={runtime.get('current_overall_status', 'unknown')} | "
            f"datasets={runtime.get('dataset_count', 0)} | "
            f"live={runtime.get('live_count', 0)} | "
            f"fresh={runtime.get('fresh_count', 0)} | "
            f"stale={runtime.get('stale_count', 0)} | "
            f"unknown={runtime.get('unknown_count', 0)} | "
            f"drifted={runtime.get('drifted_count', 0)} | "
            f"warnings={runtime.get('warning_count', 0)} | "
            f"checked_at={runtime.get('checked_at_utc', 'unknown')}"
        )
        if runtime.get("top_issue"):
            top_issue = runtime["top_issue"]
            lines.append(
                f"top_issue={top_issue.get('dataset', 'unknown')} | "
                f"build={top_issue.get('build_freshness_status', 'unknown')} | "
                f"current={top_issue.get('current_freshness_status', 'unknown')} | "
                f"drift={top_issue.get('drift_status', 'unknown')} | "
                f"warnings={top_issue.get('warning_count', 0)}"
            )
            lines.append(f"top_issue_reason={top_issue.get('diagnostic_summary', 'unknown')}")
            lines.append(f"top_issue_action={top_issue.get('recommended_action', 'unknown')}")
            lines.append(
                f"top_issue_summary={runtime.get('top_issue_summary', format_top_issue_summary(top_issue))}"
            )

        dataset_rows = []
        for entry in runtime.get("datasets", []):
            age = entry.get("current_age_hours")
            age_label = f"{age:.2f}" if isinstance(age, (int, float)) else "N/D"
            max_age = entry.get("max_age_hours")
            max_age_label = str(max_age) if isinstance(max_age, (int, float)) else "N/D"
            dataset_rows.append(
                [
                    entry.get("dataset", "unknown"),
                    entry.get("source_mode", "unknown"),
                    entry.get("severity", "unknown"),
                    entry.get("build_freshness_status", "unknown"),
                    entry.get("current_freshness_status", "unknown"),
                    age_label,
                    max_age_label,
                    entry.get("coverage_status", "unknown"),
                    entry.get("drift_status", "unknown"),
                    str(entry.get("warning_count", 0)),
                ]
            )
        lines.append("")
        lines.append(
            render_table(
                "",
                [
                    "dataset",
                    "mode",
                    "severity",
                    "build",
                    "current",
                    "age_h",
                    "max_h",
                    "coverage",
                    "drift",
                    "warnings",
                ],
                dataset_rows,
            )
        )
        return "\n".join(lines) + "\n"

    def freshness_audit_table(self):
        audit = self.freshness_audit()
        lines = ["chile-hub freshness audit", ""]
        lines.append(
            f"checked_at_utc={audit.get('checked_at_utc')} | "
            f"datasets={audit.get('dataset_count', 0)} | "
            f"fresh={audit.get('fresh_count', 0)} | "
            f"stale={audit.get('stale_count', 0)} | "
            f"unknown={audit.get('unknown_count', 0)}"
        )

        dataset_rows = []
        for entry in audit.get("datasets", []):
            age = entry.get("current_age_hours")
            age_label = f"{age:.2f}" if isinstance(age, (int, float)) else "N/D"
            max_age = entry.get("max_age_hours")
            max_age_label = str(max_age) if isinstance(max_age, (int, float)) else "N/D"
            dataset_rows.append(
                [
                    entry.get("dataset", "unknown"),
                    entry.get("source_mode", "unknown"),
                    entry.get("build_freshness_status", "unknown"),
                    entry.get("current_freshness_status", "unknown"),
                    age_label,
                    max_age_label,
                    entry.get("freshness_label", "N/D"),
                ]
            )
        lines.append("")
        lines.append(
            render_table(
                "",
                ["dataset", "mode", "build", "current", "age_h", "max_h", "label"],
                dataset_rows,
            )
        )
        return "\n".join(lines) + "\n"

    def bundle(self):
        return self._load_hub_bundle()

    def packages(self):
        bundle_packages = self.bundle().get("packages", [])
        if bundle_packages:
            return bundle_packages
        manifest = self._load_artifact_manifest()
        return manifest.get("packages", [])

    def packages_table(self):
        packages = self.packages()
        table_rows = []
        for package in packages:
            size_bytes = package.get("size_bytes")
            if isinstance(size_bytes, int):
                if size_bytes < 1024:
                    size_label = f"{size_bytes} B"
                else:
                    size_label = f"{size_bytes / 1024:.1f} KB"
            else:
                size_label = "N/D"
            table_rows.append(
                [
                    package.get("package_type", "unknown"),
                    size_label,
                    package.get("checksum_algorithm", "unknown"),
                    package.get("path", "unknown"),
                ]
            )
        return render_table(
            "chile-hub packages",
            ["package_type", "size", "checksum", "path"],
            table_rows,
        )

    def primary_package(self, package_type="zip"):
        for package in self.packages():
            if package.get("package_type") == package_type:
                return package
        raise KeyError(f"No existe package_type '{package_type}' en el hub.")

    def package_verification(self, package_type="zip"):
        package = self.primary_package(package_type)
        return {
            "path": package.get("path"),
            "package_type": package.get("package_type"),
            "checksum_algorithm": package.get("checksum_algorithm"),
            "checksum_path": package.get("checksum_path"),
            "verification_command": package.get("verification_command"),
            "sha256": package.get("sha256"),
            "size_bytes": package.get("size_bytes"),
        }

    def redistribution(self):
        return self._load_redistribution_report()

    def redistribution_table(self):
        report = self.redistribution()
        lines = ["chile-hub redistribution", ""]
        lines.append(
            f"ready={report.get('ready_count', 0)} | "
            f"review_terms={report.get('review_terms_count', 0)} | "
            f"unknown={report.get('unknown_count', 0)} | "
            f"datasets={report.get('dataset_count', 0)}"
        )

        dataset_rows = []
        for entry in report.get("datasets", []):
            attribution = "yes" if entry.get("attribution_required") else "no"
            dataset_rows.append(
                [
                    entry.get("dataset", "unknown"),
                    entry.get("publishability_status", "unknown"),
                    entry.get("reuse_status", "unknown"),
                    attribution,
                    entry.get("license", "unknown"),
                ]
            )
        lines.append("")
        lines.append(
            render_table(
                "",
                ["dataset", "status", "reuse_status", "attribution", "license"],
                dataset_rows,
            )
        )
        return "\n".join(lines) + "\n"

    def provenance(self):
        return self._load_provenance_report()

    def provenance_table(self):
        report = self.provenance()
        lines = ["chile-hub provenance", ""]
        lines.append(
            f"datasets={report.get('dataset_count', 0)} | "
            f"live={report.get('live_count', 0)} | "
            f"fallback={report.get('fallback_count', 0)}"
        )

        dataset_rows = [
            [
                entry.get("dataset", "unknown"),
                entry.get("source_mode", "unknown"),
                entry.get("source_detail", "unknown"),
                entry.get("freshness_status", "unknown"),
                str(entry.get("warning_count", 0)),
                entry.get("refreshed_at_utc", "unknown"),
            ]
            for entry in report.get("datasets", [])
        ]
        lines.append("")
        lines.append(
            render_table(
                "",
                ["dataset", "mode", "source", "freshness", "warnings", "refreshed_at_utc"],
                dataset_rows,
            )
        )
        return "\n".join(lines) + "\n"

    def drift(self):
        return self._load_drift_report()

    def drift_table(self):
        report = self.drift()
        lines = ["chile-hub drift", ""]
        lines.append(
            f"datasets={report.get('dataset_count', 0)} | "
            f"drifted={report.get('drifted_count', 0)} | "
            f"healthy={report.get('healthy_count', 0)} | "
            f"fallback={report.get('fallback_count', 0)} | "
            f"partial_coverage={report.get('partial_coverage_count', 0)} | "
            f"degraded={report.get('degraded_count', 0)}"
        )

        dataset_rows = [
            [
                entry.get("dataset", "unknown"),
                entry.get("drift_status", "unknown"),
                entry.get("source_mode", "unknown"),
                entry.get("coverage_status", "unknown"),
                entry.get("degradation_status", "unknown"),
                str(entry.get("warning_count", 0)),
            ]
            for entry in report.get("datasets", [])
        ]
        lines.append("")
        lines.append(
            render_table(
                "",
                ["dataset", "drift", "mode", "coverage", "degradation", "warnings"],
                dataset_rows,
            )
        )
        return "\n".join(lines) + "\n"

load_polars(dataset_name, validate=False)

Carga un dataset como DataFrame de Polars.

Parameters:

Name	Type	Description	Default
dataset_name	str | Dataset	Nombre del dataset (ej. "comunas", "indicadores").	required
validate	bool	Si True, ejecuta validate_dataset() antes de retornar. Lanza ChileHubDataError si la validación falla.	False

Returns:

Type	Description
DataFrame	DataFrame de Polars con los datos del dataset.

Raises:

Type	Description
ChileHubDatasetError	Si el dataset no existe o el archivo es ilegible.
ChileHubDataError	Si validate=True y la validación encuentra errores.

Source code in src/chile_hub/core.py

def load_polars(self, dataset_name: str | Dataset, validate: bool = False) -> pl.DataFrame:
    """Carga un dataset como DataFrame de Polars.

    Args:
        dataset_name: Nombre del dataset (ej. "comunas", "indicadores").
        validate: Si ``True``, ejecuta ``validate_dataset()`` antes de retornar.
            Lanza ``ChileHubDataError`` si la validación falla.

    Returns:
        DataFrame de Polars con los datos del dataset.

    Raises:
        ChileHubDatasetError: Si el dataset no existe o el archivo es ilegible.
        ChileHubDataError: Si ``validate=True`` y la validación encuentra errores.
    """
    dataset_name = _resolve_dataset_name(dataset_name)
    if dataset_name not in self._df_cache:
        path = self.get_output_path(dataset_name, "parquet")
        try:
            df = pl.read_parquet(path)
        except FileNotFoundError:
            raise ChileHubDatasetError(
                f"Archivo Parquet no encontrado para '{dataset_name}': {path}"
            )
        except Exception as exc:
            raise ChileHubDatasetError(f"Error al leer Parquet para '{dataset_name}': {exc}")
        self._df_cache[dataset_name] = df

    if validate:
        result = self.validate_dataset(dataset_name)
        if result["status"] == "error":
            raise ChileHubDataError(
                f"Validación fallida para '{dataset_name}': {'; '.join(result['errors'])}"
            )

    return self._df_cache[dataset_name]

cross_view(datasets, on='codigo_comuna', how='left')

Retorna un cruce predefinido de datasets vinculados por clave territorial.

Parameters:

Name	Type	Description	Default
datasets	list[str | Dataset]	Lista de nombres de datasets a cruzar (ej. ["comunas", "censo_comunal"]).	required
on	str	Clave de join (default: "codigo_comuna").	'codigo_comuna'
how	Literal['inner', 'left', 'right', 'full', 'semi', 'anti', 'cross', 'outer']	Tipo de join (default: "left").	'left'

Returns:

Type	Description
DataFrame	DataFrame de Polars con el cruce de todos los datasets.

Raises:

Type	Description
ChileHubDatasetError	Si se pasan menos de 2 datasets.

Source code in src/chile_hub/core.py

def cross_view(
    self,
    datasets: list[str | Dataset],
    on: str = "codigo_comuna",
    how: Literal["inner", "left", "right", "full", "semi", "anti", "cross", "outer"] = "left",
) -> pl.DataFrame:
    """Retorna un cruce predefinido de datasets vinculados por clave territorial.

    Args:
        datasets: Lista de nombres de datasets a cruzar (ej. ["comunas", "censo_comunal"]).
        on: Clave de join (default: "codigo_comuna").
        how: Tipo de join (default: "left").

    Returns:
        DataFrame de Polars con el cruce de todos los datasets.

    Raises:
        ChileHubDatasetError: Si se pasan menos de 2 datasets.
    """
    if len(datasets) < 2:
        raise ChileHubDatasetError(
            f"cross_view requiere al menos 2 datasets. Recibido: {len(datasets)}"
        )

    dfs = []
    for name in datasets:
        df = self.load_polars(name)
        # Prefijar columnas no-clave con el nombre del dataset para evitar colisiones
        cols_to_prefix = [c for c in df.columns if c != on]
        df = df.select([pl.col(on)] + [pl.col(c).alias(f"{name}_{c}") for c in cols_to_prefix])
        dfs.append(df)

    result = dfs[0]
    for df in dfs[1:]:
        result = result.join(df, on=on, how=how)
    return result

resolve_comunas(names)

Resuelve nombres de comuna (tipeados por humanos) a códigos CUT.

Match determinista: normaliza cada nombre a su forma nombre_comuna_clean (minúsculas, sin acentos, sin ñ) y hace coincidencia exacta contra el dataset comunas. No corrige typos ni hace coincidencia difusa (ver docs/adr/ADR-009-resolutor-nombres-comunales.md).

Parameters:

Name	Type	Description	Default
names	list[str]	Lista de nombres de comuna. Para resolver una columna de un DataFrame, pásala como df["mi_columna"].to_list().	required

Returns:

Type	Description
DataFrame	DataFrame Polars con una fila por input (mismo orden, duplicados
DataFrame	preservados) y columnas: input, codigo_comuna, nombre_comuna,
DataFrame	codigo_region, matched (bool). Los no encontrados tienen
DataFrame	matched=False y códigos nulos — sin lanzar excepción.

Examples:

>>> hub = ChileHub()
>>> hub.resolve_comunas(["Ñuñoa", "concon", "No Existe"])

Source code in src/chile_hub/core.py

def resolve_comunas(self, names: list[str]) -> pl.DataFrame:
    """Resuelve nombres de comuna (tipeados por humanos) a códigos CUT.

    Match **determinista**: normaliza cada nombre a su forma ``nombre_comuna_clean``
    (minúsculas, sin acentos, sin ``ñ``) y hace coincidencia exacta contra el
    dataset ``comunas``. No corrige typos ni hace coincidencia difusa (ver
    ``docs/adr/ADR-009-resolutor-nombres-comunales.md``).

    Args:
        names: Lista de nombres de comuna. Para resolver una columna de un
            DataFrame, pásala como ``df["mi_columna"].to_list()``.

    Returns:
        DataFrame Polars con una fila por input (mismo orden, duplicados
        preservados) y columnas: ``input``, ``codigo_comuna``, ``nombre_comuna``,
        ``codigo_region``, ``matched`` (bool). Los no encontrados tienen
        ``matched=False`` y códigos nulos — sin lanzar excepción.

    Examples:
        >>> hub = ChileHub()
        >>> hub.resolve_comunas(["Ñuñoa", "concon", "No Existe"])
    """
    comunas = self.load_polars("comunas")
    lookup = {
        row["nombre_comuna_clean"]: (
            row["codigo_comuna"],
            row["nombre_comuna"],
            row["codigo_region"],
        )
        for row in comunas.iter_rows(named=True)
    }
    rows = []
    for original in names:
        key = normalize_comuna_name(str(original))
        hit = lookup.get(key)
        rows.append(
            {
                "input": original,
                "codigo_comuna": hit[0] if hit else None,
                "nombre_comuna": hit[1] if hit else None,
                "codigo_region": hit[2] if hit else None,
                "matched": hit is not None,
            }
        )
    return pl.DataFrame(
        rows,
        schema={
            "input": pl.String,
            "codigo_comuna": pl.String,
            "nombre_comuna": pl.String,
            "codigo_region": pl.String,
            "matched": pl.Boolean,
        },
    )

sql(query)

Ejecuta una consulta SQL sobre los datasets publicados como vistas DuckDB.

Cada dataset del catálogo se registra como una vista DuckDB con el mismo nombre que el dataset. Las vistas se respaldan con los archivos Parquet publicados y se registran de forma lazy en la primera llamada a sql().

Parameters:

Name	Type	Description	Default
query	str	Consulta SQL que referencia las vistas con los nombres de los datasets (ej. "SELECT * FROM comunas LIMIT 5").	required

Returns:

Type	Description
DataFrame	DataFrame de Polars con el resultado de la consulta.

Raises:

Type	Description
ImportError	Si duckdb no está instalado. Instálalo con pip install chile-hub[query].

Examples:

>>> hub = ChileHub()
>>> hub.sql("SELECT codigo_comuna, nombre_comuna_clean FROM comunas LIMIT 5")
>>> hub.sql("SELECT c.codigo_comuna, c.nombre_comuna_clean, cen.poblacion_censada FROM comunas c JOIN censo_comunal cen USING(codigo_comuna) LIMIT 10")

Source code in src/chile_hub/core.py

def sql(self, query: str) -> pl.DataFrame:
    """Ejecuta una consulta SQL sobre los datasets publicados como vistas DuckDB.

    Cada dataset del catálogo se registra como una vista DuckDB con el mismo
    nombre que el dataset. Las vistas se respaldan con los archivos Parquet
    publicados y se registran de forma lazy en la primera llamada a ``sql()``.

    Args:
        query: Consulta SQL que referencia las vistas con los nombres de los
            datasets (ej. ``"SELECT * FROM comunas LIMIT 5"``).

    Returns:
        DataFrame de Polars con el resultado de la consulta.

    Raises:
        ImportError: Si ``duckdb`` no está instalado.
            Instálalo con ``pip install chile-hub[query]``.

    Examples:
        >>> hub = ChileHub()
        >>> hub.sql("SELECT codigo_comuna, nombre_comuna_clean FROM comunas LIMIT 5")
        >>> hub.sql("SELECT c.codigo_comuna, c.nombre_comuna_clean, cen.poblacion_censada FROM comunas c JOIN censo_comunal cen USING(codigo_comuna) LIMIT 10")
    """
    try:
        import duckdb
    except ImportError:
        raise ImportError(
            "Para usar SQL necesitas instalar duckdb: pip install chile-hub[query]"
        )

    if not hasattr(self, "_sql_views_registered"):
        con = duckdb.connect()
        for dataset_name in Dataset:
            try:
                name = dataset_name.value
                path = self.get_output_path(name, "parquet")
                con.execute(f"CREATE VIEW \"{name}\" AS SELECT * FROM read_parquet('{path}')")  # nosec
            except Exception:  # nosec
                # Skip datasets whose Parquet isn't available
                pass
        self._sql_views_registered = True
        self._sql_connection = con
    else:
        con = self._sql_connection

    return cast(pl.DataFrame, con.execute(query).pl())

from_datapackage(path_or_url) classmethod

Abre un ChileHub a partir de un descriptor datapackage.json externo.

Resuelve el directorio de datos a partir del descriptor: si el descriptor esta en un directorio con los archivos Parquet referenciados en resources[].path, esos archivos se usaran directamente. Si no, se asume que los paths son relativos al directorio del descriptor.

Parameters:

Name	Type	Description	Default
path_or_url	str | Path	Ruta local o URL (http:///https://) a un archivo datapackage.json.	required

Returns:

Type	Description
ChileHub	Instancia de ChileHub configurada para usar los datos del
ChileHub	descriptor.

Raises:

Type	Description
FileNotFoundError	Si el descriptor no existe en la ruta local.
ChileHubDataError	Si path_or_url es una URL. El descriptor remoto se valida (existe, es un Frictionless Data Package conforme), pero ChileHub todavia no sabe leer Parquet remoto directamente desde una URL arbitraria (__init__ asume un directorio local) — ver ADR-010, Preguntas abiertas. Usa ChileHub() sin argumentos para el bundle publicado (se descarga y cachea automaticamente), o hub.sql(...)/docs/http-access.md para consumir el hosting HTTP directamente sin pasar por esta API.
ImportError	Si frictionless no esta instalado. Instalalo con pip install chile-hub[validation].

Example

hub = ChileHub.from_datapackage("data/normalized/datapackage.json") hub.summary()

Source code in src/chile_hub/core.py

@classmethod
def from_datapackage(cls, path_or_url: str | Path) -> "ChileHub":
    """Abre un ChileHub a partir de un descriptor ``datapackage.json`` externo.

    Resuelve el directorio de datos a partir del descriptor: si el descriptor
    esta en un directorio con los archivos Parquet referenciados en
    ``resources[].path``, esos archivos se usaran directamente. Si no, se
    asume que los paths son relativos al directorio del descriptor.

    Args:
        path_or_url: Ruta local o URL (``http://``/``https://``) a un archivo
            ``datapackage.json``.

    Returns:
        Instancia de ``ChileHub`` configurada para usar los datos del
        descriptor.

    Raises:
        FileNotFoundError: Si el descriptor no existe en la ruta local.
        ChileHubDataError: Si ``path_or_url`` es una URL. El descriptor remoto
            se valida (existe, es un Frictionless Data Package conforme), pero
            ``ChileHub`` todavia no sabe leer Parquet remoto directamente desde
            una URL arbitraria (``__init__`` asume un directorio local) — ver
            ADR-010, Preguntas abiertas. Usa ``ChileHub()`` sin argumentos para
            el bundle publicado (se descarga y cachea automaticamente), o
            ``hub.sql(...)``/``docs/http-access.md`` para consumir el hosting
            HTTP directamente sin pasar por esta API.
        ImportError: Si ``frictionless`` no esta instalado.
            Instalalo con ``pip install chile-hub[validation]``.

    Example:
        >>> hub = ChileHub.from_datapackage("data/normalized/datapackage.json")
        >>> hub.summary()
    """
    try:
        import frictionless
    except ImportError:
        raise ImportError(
            "Para usar from_datapackage necesitas instalar frictionless: "
            "pip install chile-hub[validation]"
        )

    path_or_url_str = str(path_or_url)
    if path_or_url_str.startswith(("http://", "https://")):
        # Valida que el descriptor remoto exista y sea un Frictionless Data
        # Package conforme (levanta si frictionless no puede resolverlo).
        _ = frictionless.Package(path_or_url_str)
        raise ChileHubDataError(
            f"El descriptor remoto '{path_or_url_str}' es valido, pero "
            "from_datapackage(url) todavia no soporta leer Parquet remoto "
            "directamente (ChileHub asume un directorio de datos local). "
            "Usa ChileHub() sin argumentos para el bundle publicado, o "
            "consulta docs/http-access.md para consumir la URL directamente "
            "(ej. con Polars/DuckDB/arrow). Ver ADR-010."
        )

    path = Path(path_or_url)
    if not path.exists():
        raise FileNotFoundError(f"Descriptor no encontrado: {path}")

    _ = frictionless.Package(str(path))
    data_dir = path.parent
    return cls(data_dir=data_dir)

frictionless_validate(dataset_name=None)

Valida el bundle local contra el descriptor datapackage.json.

Usa frictionless.Package.validate_descriptor() para verificar la estructura del descriptor. Es una validacion de metadatos (no carga los datos completos).

Parameters:

Name	Type	Description	Default
dataset_name	str | None	Si se especifica, se verifica que el dataset exista como recurso en el descriptor. Si es None, se valida el descriptor completo.	None

Returns:

Type	Description
dict	Diccionario con:
dict	valid (bool): True si el descriptor es valido.
dict	errors (list[str]): Errores encontrados (vacio si es valido).
dict	stats (dict): Metadatos del resultado (tasks, warnings).

Raises:

Type	Description
FileNotFoundError	Si datapackage.json no existe en el directorio de datos.
ImportError	Si frictionless no esta instalado. Instalalo con pip install chile-hub[validation].

Example

hub = ChileHub() result = hub.frictionless_validate() print(result["valid"]) True

Source code in src/chile_hub/core.py

def frictionless_validate(self, dataset_name: str | None = None) -> dict:
    """Valida el bundle local contra el descriptor ``datapackage.json``.

    Usa ``frictionless.Package.validate_descriptor()`` para verificar la
    estructura del descriptor. Es una validacion de metadatos (no carga
    los datos completos).

    Args:
        dataset_name: Si se especifica, se verifica que el dataset exista
            como recurso en el descriptor. Si es ``None``, se valida el
            descriptor completo.

    Returns:
        Diccionario con:
        - ``valid`` (bool): ``True`` si el descriptor es valido.
        - ``errors`` (list[str]): Errores encontrados (vacio si es valido).
        - ``stats`` (dict): Metadatos del resultado (``tasks``, ``warnings``).

    Raises:
        FileNotFoundError: Si ``datapackage.json`` no existe en el
            directorio de datos.
        ImportError: Si ``frictionless`` no esta instalado.
            Instalalo con ``pip install chile-hub[validation]``.

    Example:
        >>> hub = ChileHub()
        >>> result = hub.frictionless_validate()
        >>> print(result["valid"])
        True
    """
    try:
        import frictionless
    except ImportError:
        raise ImportError(
            "Para validar datapackage.json necesitas instalar frictionless: "
            "pip install chile-hub[validation]"
        )

    descriptor_path = self.normalized_dir / "datapackage.json"
    if not descriptor_path.exists():
        raise FileNotFoundError(
            f"datapackage.json no encontrado en {self.normalized_dir}. "
            "Ejecuta el build primero o descarga un bundle."
        )

    if dataset_name is not None:
        # Validate that the dataset name exists as a resource
        package = frictionless.Package(str(descriptor_path))
        names = [r.name for r in package.resources]
        if dataset_name not in names:
            return {
                "valid": False,
                "errors": [
                    f"Dataset '{dataset_name}' no encontrado en datapackage.json. "
                    f"Disponibles: {', '.join(sorted(names))}"
                ],
                "stats": {"resources_total": len(names)},
            }
        return {
            "valid": True,
            "errors": [],
            "stats": {"resources_total": len(names), "checked": dataset_name},
        }

    report = frictionless.Package.validate_descriptor(str(descriptor_path))
    errors = [str(e) for e in report.errors] if report.errors else []
    return {
        "valid": report.valid,
        "errors": errors,
        "stats": {
            "tasks": getattr(report, "tasks", []),
            "warnings": getattr(report, "warnings", []),
        },
    }

validate_dataset(dataset_name)

Valida los datos publicados del hub contra su contrato JSON Schema.

Carga el dataset desde Parquet y lo coteja contra el contrato en contracts/datasets/{dataset_name}.schema.json. Retorna un dict con status, errors y warnings.

Parameters:

Name	Type	Description	Default
dataset_name	str | Dataset	Nombre del dataset (ej. "comunas").	required

Returns:

Type	Description
dict	Dict con:
dict	dataset: nombre del dataset validado.
dict	status: "ok" o "error".
dict	errors: lista de errores de validación.
dict	warnings: lista de advertencias.

Raises:

Type	Description
ChileHubDatasetError	Si no existe contrato o dataset.

Source code in src/chile_hub/core.py

def validate_dataset(self, dataset_name: str | Dataset) -> dict:
    """Valida los datos publicados del hub contra su contrato JSON Schema.

    Carga el dataset desde Parquet y lo coteja contra el contrato en
    ``contracts/datasets/{dataset_name}.schema.json``. Retorna un dict
    con ``status``, ``errors`` y ``warnings``.

    Args:
        dataset_name: Nombre del dataset (ej. ``"comunas"``).

    Returns:
        Dict con:
        - ``dataset``: nombre del dataset validado.
        - ``status``: ``"ok"`` o ``"error"``.
        - ``errors``: lista de errores de validación.
        - ``warnings``: lista de advertencias.

    Raises:
        ChileHubDatasetError: Si no existe contrato o dataset.
    """
    dataset_name = _resolve_dataset_name(dataset_name)
    from .contracts import verify_dataset_contract

    contract_path = self.root_dir / "contracts" / "datasets" / f"{dataset_name}.schema.json"
    if not contract_path.exists():
        raise ChileHubDatasetError(
            f"No existe contrato de schema para '{dataset_name}'. "
            f"Datasets disponibles: {self.list_datasets()}"
        )

    with contract_path.open("r", encoding="utf-8") as f:
        contract = json.load(f)

    df = self.load_polars(dataset_name)
    catalog_entry = self.get_dataset(dataset_name)
    outputs = catalog_entry.get("outputs", {}) if catalog_entry else {}

    return verify_dataset_contract(
        dataset_name,
        contract,
        df,
        outputs=outputs,
        root_dir=self.root_dir,
    )

validate_user_data(df, dataset_name)

Valida un DataFrame de usuario contra el contrato de schema del dataset.

Usa los archivos de contrato en contracts/datasets/*.schema.json, que definen required_columns, column_types, primary_key y expected_record_count.

Parameters:

Name	Type	Description	Default
df	DataFrame	DataFrame de Polars a validar.	required
dataset_name	str | Dataset	Nombre del dataset de referencia (ej. "comunas").	required

Returns:

Type	Description
dict	Dict con:
dict	status: "ok" si pasa todas las validaciones, "error" si falla alguna.
dict	errors: lista de strings con mensajes de error (vacía si status=="ok").
dict	warnings: lista de strings con advertencias no bloqueantes.
dict	schema_used: ruta absoluta del schema usado.

Raises:

Type	Description
ChileHubDatasetError	Si no existe contrato para el dataset solicitado.

Source code in src/chile_hub/core.py

def validate_user_data(self, df: pl.DataFrame, dataset_name: str | Dataset) -> dict:
    """Valida un DataFrame de usuario contra el contrato de schema del dataset.

    Usa los archivos de contrato en contracts/datasets/*.schema.json, que definen
    required_columns, column_types, primary_key y expected_record_count.

    Args:
        df: DataFrame de Polars a validar.
        dataset_name: Nombre del dataset de referencia (ej. "comunas").

    Returns:
        Dict con:
        - status: "ok" si pasa todas las validaciones, "error" si falla alguna.
        - errors: lista de strings con mensajes de error (vacía si status=="ok").
        - warnings: lista de strings con advertencias no bloqueantes.
        - schema_used: ruta absoluta del schema usado.

    Raises:
        ChileHubDatasetError: Si no existe contrato para el dataset solicitado.
    """
    dataset_name = _resolve_dataset_name(dataset_name)
    schema_path = ROOT_DIR / "contracts" / "datasets" / f"{dataset_name}.schema.json"
    if not schema_path.exists():
        raise ChileHubDatasetError(
            f"No existe contrato de schema para '{dataset_name}'. "
            f"Datasets disponibles: {self.list_datasets()}"
        )

    import json as json_module

    with open(schema_path, "r", encoding="utf-8") as f:
        schema = json_module.load(f)

    errors = []
    warnings = []

    # 1. Validar columnas requeridas (required_columns)
    required_cols = schema.get("required_columns", [])
    df_cols = df.columns
    missing = [c for c in required_cols if c not in df_cols]
    if missing:
        errors.append(f"Columnas requeridas faltantes: {missing}")

    # 2. Validar tipos (column_types)
    column_types = schema.get("column_types", {})
    type_map = {
        "string": ["String", "Utf8", "str"],
        "integer": ["Int64", "Int32", "Int16", "Int8", "UInt32", "UInt16"],
        "number": ["Float64", "Float32"],
        "boolean": ["Boolean"],
        "date": ["Date"],
    }
    for col, expected_type in column_types.items():
        if col not in df_cols:
            continue
        actual_dtype = str(df[col].dtype)
        expected_names = type_map.get(expected_type, [expected_type])
        if actual_dtype not in expected_names:
            errors.append(
                f"Columna '{col}': se esperaba {expected_type}, se encontró {actual_dtype}"
            )

    # 3. Validar clave primaria (primary_key)
    primary_key = schema.get("primary_key", [])
    if primary_key:
        pk_cols = [c for c in primary_key if c in df_cols]
        if pk_cols:
            if df.select(pk_cols).null_count().sum_horizontal().sum() > 0:
                errors.append(f"Clave primaria {primary_key} contiene valores nulos")
            pk_df = df.select(pk_cols)
            if pk_df.height != pk_df.unique().height:
                errors.append(f"Clave primaria {primary_key} tiene valores duplicados")

    # 4. Verificar expected_record_count (solo advertencia)
    expected = schema.get("expected_record_count")
    if expected is not None and df.height != expected:
        warnings.append(
            f"Cantidad de registros ({df.height}) difiere de la esperada ({expected})"
        )

    status = "ok" if not errors else "error"
    return {
        "status": status,
        "errors": errors,
        "warnings": warnings,
        "schema_used": str(schema_path),
    }

search_datasets(query='', source_name='', maturity='')

Busca datasets por keyword, fuente, o nivel de madurez.

Parameters:

Name	Type	Description	Default
query	str	Texto libre para buscar en nombre y descripción.	''
source_name	str	Filtrar por fuente (ej. "INE", "MINSAL"). Coincidencia parcial sin distinción de mayúsculas.	''
maturity	str	Filtrar por maturity_status (ej. "stable", "candidate").	''

Returns:

Type	Description
list[dict]	Lista de dicts con información de cada dataset que coincide:
list[dict]	{"name", "description", "source_name", "record_count", "maturity_status", "fields"}.

Source code in src/chile_hub/core.py

def search_datasets(
    self, query: str = "", source_name: str = "", maturity: str = ""
) -> list[dict]:
    """Busca datasets por keyword, fuente, o nivel de madurez.

    Args:
        query: Texto libre para buscar en nombre y descripción.
        source_name: Filtrar por fuente (ej. "INE", "MINSAL"). Coincidencia parcial
            sin distinción de mayúsculas.
        maturity: Filtrar por maturity_status (ej. "stable", "candidate").

    Returns:
        Lista de dicts con información de cada dataset que coincide:
        {"name", "description", "source_name", "record_count", "maturity_status", "fields"}.
    """
    results = []
    query_lower = query.lower().strip() if query else ""
    source_lower = source_name.lower().strip() if source_name else ""
    maturity_lower = maturity.lower().strip() if maturity else ""

    # Cargar source_readiness para maturity_status
    source_readiness = self._load_source_readiness()
    maturity_by_dataset = {
        entry["dataset"]: entry.get("maturity_status", "")
        for entry in source_readiness.get("datasets", [])
    }

    for entry in self.catalog.get("datasets", []):
        name = entry.get("dataset", "")
        desc = entry.get("description", "").lower()

        # Filtro por query
        if query_lower:
            if query_lower not in name.lower() and query_lower not in desc:
                continue

        # Filtro por fuente
        entry_source = entry.get("source_name", "").lower()
        if source_lower and source_lower not in entry_source:
            continue

        # Filtro por maturity_status
        if maturity_lower:
            entry_maturity = maturity_by_dataset.get(name, "").lower()
            if maturity_lower != entry_maturity:
                continue

        results.append(
            {
                "name": name,
                "description": entry.get("description", ""),
                "source_name": entry.get("source_name", ""),
                "record_count": entry.get("record_count", 0),
                "maturity_status": maturity_by_dataset.get(name, ""),
                "fields": entry.get("fields", []),
            }
        )

    return results

source_readiness()

Devuelve el reporte de madurez de fuente por dataset.

Source code in src/chile_hub/core.py

def source_readiness(self):
    """Devuelve el reporte de madurez de fuente por dataset."""
    return self._load_source_readiness()

check_sources(timeout=5)

Verifica la conectividad de red con las fuentes de datos oficiales.

Source code in src/chile_hub/core.py

def check_sources(self, timeout: int = 5) -> list[dict[str, Any]]:
    """Verifica la conectividad de red con las fuentes de datos oficiales."""
    results = []
    for entry in self.catalog.get("datasets", []):
        dataset = entry.get("dataset")
        url = entry.get("source_url")
        source_name = entry.get("source_name")
        if not url:
            results.append(
                {
                    "dataset": dataset,
                    "source_name": source_name,
                    "url": "N/A",
                    "status": "offline",
                    "status_code": None,
                    "latency_ms": None,
                    "error": "No source URL defined",
                }
            )
            continue

        try:
            # Intenta HEAD primero
            response = requests.head(url, timeout=timeout, allow_redirects=True)
            if response.status_code >= 400:
                response.close()
                response = requests.get(url, timeout=timeout, stream=True)

            status = "online" if response.status_code < 400 else "offline"
            status_code = response.status_code
            latency_ms = round(response.elapsed.total_seconds() * 1000, 2)
            error = None
            response.close()
        except Exception as e:
            status = "offline"
            status_code = None
            latency_ms = None
            error = type(e).__name__

        results.append(
            {
                "dataset": dataset,
                "source_name": source_name,
                "url": url,
                "status": status,
                "status_code": status_code,
                "latency_ms": latency_ms,
                "error": error,
            }
        )
    return results

check_sources_table(results)

Formatea el resultado de check_sources como una tabla amigable para terminal.

Source code in src/chile_hub/core.py

def check_sources_table(self, results: list[dict[str, Any]]) -> str:
    """Formatea el resultado de check_sources como una tabla amigable para terminal."""
    table_rows = []
    for entry in results:
        status = entry.get("status", "unknown")
        code = str(entry.get("status_code")) if entry.get("status_code") is not None else "N/A"
        latency = (
            f"{entry.get('latency_ms')}ms" if entry.get("latency_ms") is not None else "N/A"
        )
        url = entry.get("url", "N/A")
        if len(url) > 48:
            url = url[:45] + "..."
        table_rows.append(
            [
                entry.get("dataset", "unknown"),
                status,
                code,
                latency,
                entry.get("source_name", "unknown"),
                url,
            ]
        )
    return render_table(
        "chile-hub check-sources",
        ["dataset", "status", "code", "latency", "source name", "url"],
        table_rows,
    )

dataset_quality()

Devuelve la tarjeta de puntuación de calidad multidimensional por dataset.

Source code in src/chile_hub/core.py

def dataset_quality(self):
    """Devuelve la tarjeta de puntuación de calidad multidimensional por dataset."""
    return self._load_dataset_quality()

Gestión de datos

chile_hub.data_manager.ChileHubDataManager

Source code in src/chile_hub/data_manager.py

class ChileHubDataManager:
    def __init__(
        self,
        *,
        data_version: str = "latest",
        repository: str = DEFAULT_REPOSITORY,
        cache_dir: str | Path | None = None,
        session: requests.Session | None = None,
    ) -> None:
        self.data_version = data_version
        self.repository = repository
        self.cache_root = Path(
            cache_dir or os.environ.get(ENV_CACHE_DIR) or user_cache_dir("chile-hub", "chile-hub")
        )
        self.session = session or requests.Session()

    @property
    def version_cache_dir(self) -> Path:
        return self.cache_root / self.data_version

    @property
    def normalized_dir(self) -> Path:
        return self.version_cache_dir / "data" / "normalized"

    @property
    def marker_path(self) -> Path:
        return self.version_cache_dir / ".verified.json"

    @property
    def update_check_path(self) -> Path:
        """Ruta del estado local del chequeo periódico de actualizaciones."""
        return self.version_cache_dir / ".update_check.json"

    def status(self) -> dict[str, Any]:
        """Estado del caché local de datos: versión, rutas y si está listo para usarse."""
        catalog_path = self.normalized_dir / "dataset_catalog.json"
        marker = self._read_json(self.marker_path)
        return {
            "cache_root": str(self.cache_root),
            "data_version": self.data_version,
            "normalized_dir": str(self.normalized_dir),
            "is_ready": catalog_path.exists() and self.marker_path.exists(),
            "dataset_catalog": str(catalog_path),
            "verified": bool(marker),
            "release": marker.get("release") if marker else None,
        }

    def ensure_data_dir(self, *, auto_update: bool = True) -> Path:
        """Garantiza que el directorio de datos normalizados existe, descargándolo si es necesario.

        Args:
            auto_update: Si es True, descarga automáticamente el bundle cuando no hay caché local.

        Returns:
            Path al directorio normalized/ con los datos listos para consumir.

        Raises:
            ChileHubDataError: Si auto_update es False y no existe caché local verificado.
        """
        if (self.normalized_dir / "dataset_catalog.json").exists() and self.marker_path.exists():
            self._check_for_update_if_due()
            return self.normalized_dir
        if not auto_update:
            raise ChileHubDataError(
                "No verified chile-hub data cache found. Run `chile-hub cache update` "
                "or pass ChileHub(data_dir='/path/to/data/normalized')."
            )
        self.update()
        return self.normalized_dir

    def update(self) -> Path:
        """Descarga y verifica el bundle de datos desde GitHub Releases.

        El proceso es: resolver release → descargar checksum → descargar bundle
        (hasheando en tránsito) → verificar SHA-256 → extraer → escribir marcador
        de verificación. Si el hash no coincide, el bundle se descarta sin tocar
        el directorio normalized/.

        Returns:
            Path al directorio normalized/ con los datos extraídos.

        Raises:
            ChileHubDataError: Si el checksum no coincide o el bundle no contiene el catálogo.
        """
        release = self._resolve_release()
        assets = self._assets_by_name(release)
        bundle = self._require_asset(assets, DEFAULT_BUNDLE_NAME)
        checksum = self._require_asset(assets, DEFAULT_CHECKSUM_NAME)

        self.version_cache_dir.mkdir(parents=True, exist_ok=True)
        bundle_path = self.version_cache_dir / DEFAULT_BUNDLE_NAME
        checksum_path = self.version_cache_dir / DEFAULT_CHECKSUM_NAME

        # Descargar checksum primero
        self._download(checksum.url, checksum_path)
        expected_sha256 = self._read_checksum(checksum_path)

        # Descargar bundle hasheando en tránsito para eliminar ventana TOCTOU
        sha256_hash = hashlib.sha256()
        with self.session.get(bundle.url, stream=True, timeout=120) as response:
            response.raise_for_status()
            with tempfile.NamedTemporaryFile(
                dir=str(self.version_cache_dir), delete=False, suffix=".tmp"
            ) as tmp:
                for chunk in response.iter_content(chunk_size=8192):
                    if chunk:
                        sha256_hash.update(chunk)
                        tmp.write(chunk)
                tmp_path = Path(tmp.name)

        actual_sha256 = sha256_hash.hexdigest()
        if actual_sha256 != expected_sha256:
            tmp_path.unlink(missing_ok=True)
            raise ChileHubDataError(
                f"Checksum mismatch for {DEFAULT_BUNDLE_NAME}: "
                f"expected {expected_sha256}, got {actual_sha256}"
            )

        # Renombrar atómicamente al path final solo si el hash coincide
        tmp_path.replace(bundle_path)

        self._extract_bundle(bundle_path)
        if not (self.normalized_dir / "dataset_catalog.json").exists():
            raise ChileHubDataError(
                f"Downloaded bundle did not contain {self.normalized_dir / 'dataset_catalog.json'}"
            )

        self.marker_path.write_text(
            json.dumps(
                {
                    "release": {
                        "tag_name": release.get("tag_name"),
                        "html_url": release.get("html_url"),
                    },
                    "sha256": actual_sha256,
                    "bundle": DEFAULT_BUNDLE_NAME,
                },
                ensure_ascii=False,
                indent=2,
            )
            + "\n",
            encoding="utf-8",
        )
        return self.normalized_dir

    def clear(self) -> None:
        """Elimina el caché local de datos, forzando una descarga fresca en el próximo uso.

        Por seguridad, solo opera dentro del directorio de caché esperado (platformdirs).
        Si el directorio configurado está fuera de ese árbol, levanta ChileHubDataError.

        Raises:
            ChileHubDataError: Si cache_root no está bajo el directorio de caché esperado.
        """
        # Validar que cache_root es un subdirectorio esperado
        expected_parent = user_cache_dir("chile-hub")
        cache_path = Path(self.cache_root).resolve()
        if not str(cache_path).startswith(str(Path(expected_parent).resolve())):
            raise ChileHubDataError(
                f"Por seguridad, 'cache clear' solo opera dentro del directorio de cache "
                f"esperado ({expected_parent}). El directorio configurado es {cache_path}. "
                f"Verifica la variable de entorno CHILE_HUB_CACHE_DIR."
            )
        if not cache_path.exists():
            return  # nothing to clear
        shutil.rmtree(str(cache_path))

    def _check_for_update_if_due(self) -> None:
        """Avisa semanalmente si el bundle ``latest`` quedó desactualizado.

        Es una comprobación de mejor esfuerzo: no envía telemetría ni interrumpe
        el consumo de datos si GitHub, la red o el estado local no están disponibles.
        """
        if self.data_version != "latest" or self._update_checks_disabled():
            return

        state = self._read_json(self.update_check_path)
        if not self._is_update_check_due(state):
            return

        checked_at = datetime.now(UTC)
        try:
            release = self._resolve_release(timeout=UPDATE_CHECK_TIMEOUT)
            latest_tag = release.get("tag_name")
            if not isinstance(latest_tag, str) or not latest_tag:
                raise ValueError("GitHub release response does not include tag_name")

            marker = self._read_json(self.marker_path)
            current_tag = marker.get("release", {}).get("tag_name")
            self._write_json(
                self.update_check_path,
                {
                    "checked_at_utc": checked_at.isoformat(),
                    "latest_tag": latest_tag,
                    "status": "ok",
                },
            )
        except Exception:
            # Esta consulta es opcional: recordar el intento evita reintentos en
            # cada inicialización si el usuario está sin conectividad.
            try:
                self._write_json(
                    self.update_check_path,
                    {
                        "checked_at_utc": checked_at.isoformat(),
                        "status": "unavailable",
                    },
                )
            except OSError:
                pass
            return

        if isinstance(current_tag, str) and current_tag and current_tag != latest_tag:
            warnings.warn(
                self._format_update_notice(current_tag=current_tag, latest_tag=latest_tag),
                ChileHubUpdateWarning,
                stacklevel=3,
            )

    @staticmethod
    def _update_checks_disabled() -> bool:
        value = os.environ.get(ENV_DISABLE_UPDATE_CHECK, "")
        return value.strip().lower() in {"1", "true", "yes", "on"}

    @staticmethod
    def _is_update_check_due(state: dict[str, Any]) -> bool:
        checked_at = state.get("checked_at_utc")
        if not isinstance(checked_at, str):
            return True
        try:
            parsed = datetime.fromisoformat(checked_at)
        except ValueError:
            return True
        if parsed.tzinfo is None:
            parsed = parsed.replace(tzinfo=UTC)

        now = datetime.now(UTC)
        return parsed > now or now - parsed >= UPDATE_CHECK_INTERVAL

    @staticmethod
    def _preferred_language() -> str:
        """Retorna ``en`` solo para una preferencia inglesa explícita; español es el fallback."""
        language = os.environ.get(ENV_LANGUAGE)
        if not language:
            language = (
                os.environ.get("LC_ALL")
                or os.environ.get("LC_MESSAGES")
                or os.environ.get("LANG")
                or ""
            )
        return "en" if language.lower().startswith("en") else "es"

    @classmethod
    def _format_update_notice(cls, *, current_tag: str, latest_tag: str) -> str:
        if cls._preferred_language() == "en":
            return (
                f"A new chile-hub data release is available: {latest_tag}\n"
                f"Cached version: {current_tag}\n\n"
                "Update it with:\n"
                "    chile-hub cache update\n\n"
                "chile-hub is an independent project developed and maintained by one person, "
                "without institutional sponsors or affiliations. This independence lets it "
                "prioritize verifiable data and deliver an objective, impartial, useful, "
                "high-quality tool.\n\n"
                "If chile-hub is useful to you, consider supporting its development and "
                f"maintenance financially:\n{SUPPORT_URL}\n{BUY_ME_A_COFFEE_URL}"
            )
        return (
            f"Hay una nueva versión de los datos de chile-hub: {latest_tag}\n"
            f"Versión almacenada localmente: {current_tag}\n\n"
            "Actualízala ejecutando:\n"
            "    chile-hub cache update\n\n"
            "chile-hub es un proyecto independiente, desarrollado y mantenido por una sola "
            "persona, sin patrocinadores institucionales ni afiliaciones. Esta independencia "
            "permite priorizar datos verificables y ofrecer una herramienta objetiva, "
            "imparcial, útil y de calidad.\n\n"
            "Si chile-hub te resulta útil, puedes apoyar económicamente su desarrollo y "
            f"mantenimiento:\n{SUPPORT_URL}\n{BUY_ME_A_COFFEE_URL}"
        )

    def _resolve_release(self, *, timeout: float = 30) -> dict[str, Any]:
        suffix = (
            "releases/latest"
            if self.data_version == "latest"
            else f"releases/tags/{self.data_version}"
        )
        url = f"https://api.github.com/repos/{self.repository}/{suffix}"
        response = self.session.get(
            url,
            headers={"Accept": "application/vnd.github+json"},
            timeout=timeout,
        )
        if response.status_code != 200:
            raise ChileHubDataError(
                f"Could not resolve chile-hub release '{self.data_version}' "
                f"from {url}: HTTP {response.status_code}"
            )
        return response.json()  # type: ignore[no-any-return]  # requests.Response.json → dict en runtime

    @staticmethod
    def _assets_by_name(release: dict[str, Any]) -> dict[str, ReleaseAsset]:
        return {
            asset["name"]: ReleaseAsset(
                name=asset["name"],
                url=asset["browser_download_url"],
            )
            for asset in release.get("assets", [])
            if asset.get("name") and asset.get("browser_download_url")
        }

    @staticmethod
    def _require_asset(assets: dict[str, ReleaseAsset], name: str) -> ReleaseAsset:
        if name not in assets:
            available = ", ".join(sorted(assets)) or "none"
            raise ChileHubDataError(f"Release asset '{name}' not found. Available: {available}")
        return assets[name]

    def _download(self, url: str, destination: Path) -> None:
        with self.session.get(url, stream=True, timeout=120) as response:
            if response.status_code != 200:
                raise ChileHubDataError(f"Could not download {url}: HTTP {response.status_code}")
            with destination.open("wb") as f:
                for chunk in response.iter_content(chunk_size=1024 * 1024):
                    if chunk:
                        f.write(chunk)

    def _extract_bundle(self, bundle_path: Path) -> None:
        if self.normalized_dir.exists():
            shutil.rmtree(self.normalized_dir)
        with zipfile.ZipFile(bundle_path) as archive:
            archive.extractall(self.version_cache_dir)

    @staticmethod
    def _read_checksum(path: Path) -> str:
        line = path.read_text(encoding="utf-8").strip().splitlines()[0]
        return line.split()[0].lower()

    @staticmethod
    def _sha256(path: Path) -> str:
        digest = hashlib.sha256()
        with path.open("rb") as f:
            for chunk in iter(lambda: f.read(1024 * 1024), b""):
                digest.update(chunk)
        return digest.hexdigest()

    @staticmethod
    def _read_json(path: Path) -> dict[str, Any]:
        if not path.exists():
            return {}
        return json.loads(path.read_text(encoding="utf-8"))  # type: ignore[no-any-return]

    @staticmethod
    def _write_json(path: Path, payload: dict[str, Any]) -> None:
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(
            json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
            encoding="utf-8",
        )

update_check_path property

Ruta del estado local del chequeo periódico de actualizaciones.

status()

Estado del caché local de datos: versión, rutas y si está listo para usarse.

Source code in src/chile_hub/data_manager.py

def status(self) -> dict[str, Any]:
    """Estado del caché local de datos: versión, rutas y si está listo para usarse."""
    catalog_path = self.normalized_dir / "dataset_catalog.json"
    marker = self._read_json(self.marker_path)
    return {
        "cache_root": str(self.cache_root),
        "data_version": self.data_version,
        "normalized_dir": str(self.normalized_dir),
        "is_ready": catalog_path.exists() and self.marker_path.exists(),
        "dataset_catalog": str(catalog_path),
        "verified": bool(marker),
        "release": marker.get("release") if marker else None,
    }

ensure_data_dir(*, auto_update=True)

Garantiza que el directorio de datos normalizados existe, descargándolo si es necesario.

Parameters:

Name	Type	Description	Default
auto_update	bool	Si es True, descarga automáticamente el bundle cuando no hay caché local.	True

Returns:

Type	Description
Path	Path al directorio normalized/ con los datos listos para consumir.

Raises:

Type	Description
ChileHubDataError	Si auto_update es False y no existe caché local verificado.

Source code in src/chile_hub/data_manager.py

def ensure_data_dir(self, *, auto_update: bool = True) -> Path:
    """Garantiza que el directorio de datos normalizados existe, descargándolo si es necesario.

    Args:
        auto_update: Si es True, descarga automáticamente el bundle cuando no hay caché local.

    Returns:
        Path al directorio normalized/ con los datos listos para consumir.

    Raises:
        ChileHubDataError: Si auto_update es False y no existe caché local verificado.
    """
    if (self.normalized_dir / "dataset_catalog.json").exists() and self.marker_path.exists():
        self._check_for_update_if_due()
        return self.normalized_dir
    if not auto_update:
        raise ChileHubDataError(
            "No verified chile-hub data cache found. Run `chile-hub cache update` "
            "or pass ChileHub(data_dir='/path/to/data/normalized')."
        )
    self.update()
    return self.normalized_dir

update()

Descarga y verifica el bundle de datos desde GitHub Releases.

El proceso es: resolver release → descargar checksum → descargar bundle (hasheando en tránsito) → verificar SHA-256 → extraer → escribir marcador de verificación. Si el hash no coincide, el bundle se descarta sin tocar el directorio normalized/.

Returns:

Type	Description
Path	Path al directorio normalized/ con los datos extraídos.

Raises:

Type	Description
ChileHubDataError	Si el checksum no coincide o el bundle no contiene el catálogo.

Source code in src/chile_hub/data_manager.py

def update(self) -> Path:
    """Descarga y verifica el bundle de datos desde GitHub Releases.

    El proceso es: resolver release → descargar checksum → descargar bundle
    (hasheando en tránsito) → verificar SHA-256 → extraer → escribir marcador
    de verificación. Si el hash no coincide, el bundle se descarta sin tocar
    el directorio normalized/.

    Returns:
        Path al directorio normalized/ con los datos extraídos.

    Raises:
        ChileHubDataError: Si el checksum no coincide o el bundle no contiene el catálogo.
    """
    release = self._resolve_release()
    assets = self._assets_by_name(release)
    bundle = self._require_asset(assets, DEFAULT_BUNDLE_NAME)
    checksum = self._require_asset(assets, DEFAULT_CHECKSUM_NAME)

    self.version_cache_dir.mkdir(parents=True, exist_ok=True)
    bundle_path = self.version_cache_dir / DEFAULT_BUNDLE_NAME
    checksum_path = self.version_cache_dir / DEFAULT_CHECKSUM_NAME

    # Descargar checksum primero
    self._download(checksum.url, checksum_path)
    expected_sha256 = self._read_checksum(checksum_path)

    # Descargar bundle hasheando en tránsito para eliminar ventana TOCTOU
    sha256_hash = hashlib.sha256()
    with self.session.get(bundle.url, stream=True, timeout=120) as response:
        response.raise_for_status()
        with tempfile.NamedTemporaryFile(
            dir=str(self.version_cache_dir), delete=False, suffix=".tmp"
        ) as tmp:
            for chunk in response.iter_content(chunk_size=8192):
                if chunk:
                    sha256_hash.update(chunk)
                    tmp.write(chunk)
            tmp_path = Path(tmp.name)

    actual_sha256 = sha256_hash.hexdigest()
    if actual_sha256 != expected_sha256:
        tmp_path.unlink(missing_ok=True)
        raise ChileHubDataError(
            f"Checksum mismatch for {DEFAULT_BUNDLE_NAME}: "
            f"expected {expected_sha256}, got {actual_sha256}"
        )

    # Renombrar atómicamente al path final solo si el hash coincide
    tmp_path.replace(bundle_path)

    self._extract_bundle(bundle_path)
    if not (self.normalized_dir / "dataset_catalog.json").exists():
        raise ChileHubDataError(
            f"Downloaded bundle did not contain {self.normalized_dir / 'dataset_catalog.json'}"
        )

    self.marker_path.write_text(
        json.dumps(
            {
                "release": {
                    "tag_name": release.get("tag_name"),
                    "html_url": release.get("html_url"),
                },
                "sha256": actual_sha256,
                "bundle": DEFAULT_BUNDLE_NAME,
            },
            ensure_ascii=False,
            indent=2,
        )
        + "\n",
        encoding="utf-8",
    )
    return self.normalized_dir

clear()

Elimina el caché local de datos, forzando una descarga fresca en el próximo uso.

Por seguridad, solo opera dentro del directorio de caché esperado (platformdirs). Si el directorio configurado está fuera de ese árbol, levanta ChileHubDataError.

Raises:

Type	Description
ChileHubDataError	Si cache_root no está bajo el directorio de caché esperado.

Source code in src/chile_hub/data_manager.py

def clear(self) -> None:
    """Elimina el caché local de datos, forzando una descarga fresca en el próximo uso.

    Por seguridad, solo opera dentro del directorio de caché esperado (platformdirs).
    Si el directorio configurado está fuera de ese árbol, levanta ChileHubDataError.

    Raises:
        ChileHubDataError: Si cache_root no está bajo el directorio de caché esperado.
    """
    # Validar que cache_root es un subdirectorio esperado
    expected_parent = user_cache_dir("chile-hub")
    cache_path = Path(self.cache_root).resolve()
    if not str(cache_path).startswith(str(Path(expected_parent).resolve())):
        raise ChileHubDataError(
            f"Por seguridad, 'cache clear' solo opera dentro del directorio de cache "
            f"esperado ({expected_parent}). El directorio configurado es {cache_path}. "
            f"Verifica la variable de entorno CHILE_HUB_CACHE_DIR."
        )
    if not cache_path.exists():
        return  # nothing to clear
    shutil.rmtree(str(cache_path))

Datasets

chile_hub.datasets

Enumeración de todos los datasets curados por ChileHub.

Úsalo en vez de strings mágicos para obtener autocompletado en IDE y detección temprana de errores por typos:

>>> from chile_hub.datasets import Dataset
>>> hub.load_polars(Dataset.COMUNAS)
>>> Dataset.from_string("comunas")
<Dataset.COMUNAS: 'comunas'>

Dataset

Bases: str, Enum

Enumeración de todos los datasets curados por ChileHub.

Hereda de str para compatibilidad total: Dataset.COMUNAS es un string "comunas" en runtime, así que puede pasarse a cualquier función que espere str.

Source code in src/chile_hub/datasets.py

class Dataset(str, enum.Enum):
    """Enumeración de todos los datasets curados por ChileHub.

    Hereda de ``str`` para compatibilidad total: ``Dataset.COMUNAS`` es
    un string ``"comunas"`` en runtime, así que puede pasarse a cualquier
    función que espere ``str``.
    """

    REGIONES = "regiones"
    PROVINCIAS = "provincias"
    COMUNAS = "comunas"
    COMUNAS_ENRIQUECIDAS = "comunas_enriquecidas"
    INDICADORES = "indicadores"
    CENSO_COMUNAL = "censo_comunal"
    CENSO_HOGARES_VIVIENDAS = "censo_hogares_viviendas"
    ESTABLECIMIENTOS_SALUD = "establecimientos_salud"
    ESTABLECIMIENTOS_EDUCACIONALES = "establecimientos_educacionales"
    DISTRITOS_ELECTORALES = "distritos_electorales"
    PARTIDOS_POLITICOS = "partidos_politicos"
    AUTORIDADES_ELECTAS = "autoridades_electas"
    FINANZAS_MUNICIPALES = "finanzas_municipales"
    RESULTADOS_EDUCACIONALES = "resultados_educacionales"
    INDICADORES_URBANOS_SIEDU = "indicadores_urbanos_siedu"
    POBREZA_COMUNAL = "pobreza_comunal"
    CONSUMO_ELECTRICO_COMUNAL = "consumo_electrico_comunal"
    EMPRESAS = "empresas"
    PERFIL_TERRITORIAL_COMUNAL = "perfil_territorial_comunal"

    @classmethod
    def from_string(cls, name: str) -> Dataset:
        """Resuelve un string a Dataset, con sugerencia si no coincide exactamente.

        Args:
            name: Nombre del dataset (ej. ``"comunas"``).

        Returns:
            Miembro de ``Dataset`` correspondiente.

        Raises:
            ValueError: Si el string no coincide con ningún dataset conocido.
        """
        try:
            return cls(name)
        except ValueError:
            matches = get_close_matches(name, [m.value for m in cls], n=1)
            hint = f" Quizás quisiste decir '{matches[0]}'." if matches else ""
            raise ValueError(
                f"Dataset '{name}' no es válido.{hint} Valores posibles: {', '.join(cls.values())}"
            )

    @classmethod
    def values(cls) -> list[str]:
        """Retorna la lista de valores string de todos los datasets.

        Returns:
            Lista de strings con los nombres canónicos de los datasets.
        """
        return [m.value for m in cls]

from_string(name) classmethod

Resuelve un string a Dataset, con sugerencia si no coincide exactamente.

Parameters:

Name	Type	Description	Default
name	str	Nombre del dataset (ej. "comunas").	required

Returns:

Type	Description
Dataset	Miembro de Dataset correspondiente.

Raises:

Type	Description
ValueError	Si el string no coincide con ningún dataset conocido.

Source code in src/chile_hub/datasets.py

@classmethod
def from_string(cls, name: str) -> Dataset:
    """Resuelve un string a Dataset, con sugerencia si no coincide exactamente.

    Args:
        name: Nombre del dataset (ej. ``"comunas"``).

    Returns:
        Miembro de ``Dataset`` correspondiente.

    Raises:
        ValueError: Si el string no coincide con ningún dataset conocido.
    """
    try:
        return cls(name)
    except ValueError:
        matches = get_close_matches(name, [m.value for m in cls], n=1)
        hint = f" Quizás quisiste decir '{matches[0]}'." if matches else ""
        raise ValueError(
            f"Dataset '{name}' no es válido.{hint} Valores posibles: {', '.join(cls.values())}"
        )

values() classmethod

Retorna la lista de valores string de todos los datasets.

Returns:

Type	Description
list[str]	Lista de strings con los nombres canónicos de los datasets.

Source code in src/chile_hub/datasets.py

@classmethod
def values(cls) -> list[str]:
    """Retorna la lista de valores string de todos los datasets.

    Returns:
        Lista de strings con los nombres canónicos de los datasets.
    """
    return [m.value for m in cls]

Validación de contratos

chile_hub.contracts

Validación de datasets contra contratos JSON Schema.

Provee la función verify_dataset_contract(), que valida un DataFrame de Polars contra un contrato {dataset}.schema.json y retorna un dict con el resultado.

Migrado desde scripts/verify_pipeline.py para que la librería tenga esta capacidad disponible en runtime.

contract_type(dtype)

Convierte un tipo de columna Polars a su nombre canónico en JSON Schema.

Parameters:

Name	Type	Description	Default
dtype	Any	Tipo de columna Polars (ej. pl.String, pl.Int32, pl.Float64).	required

Returns:

Type	Description
str	Nombre canónico: "string", "integer", "float", "date", "boolean".

Source code in src/chile_hub/contracts.py

def contract_type(dtype: Any) -> str:
    """Convierte un tipo de columna Polars a su nombre canónico en JSON Schema.

    Args:
        dtype: Tipo de columna Polars (ej. pl.String, pl.Int32, pl.Float64).

    Returns:
        Nombre canónico: "string", "integer", "float", "date", "boolean".
    """
    dtype_name = str(dtype)
    if dtype_name == "String":
        return "string"
    if dtype_name in {
        "Int8",
        "Int16",
        "Int32",
        "Int64",
        "UInt8",
        "UInt16",
        "UInt32",
        "UInt64",
    }:
        return "integer"
    if dtype_name in {"Float32", "Float64"}:
        return "float"
    if dtype_name == "Date":
        return "date"
    if dtype_name == "Boolean":
        return "boolean"
    return dtype_name.lower()

verify_dataset_contract(dataset_name, contract, df, outputs=None, root_dir=None, *, strict=False)

Valida un DataFrame contra un contrato JSON Schema de ChileHub.

Parameters:

Name	Type	Description	Default
dataset_name	str	Nombre del dataset (se coteja contra contract["dataset"]).	required
contract	dict[str, Any]	Dict del contrato JSON Schema cargado.	required
df	DataFrame	DataFrame de Polars a validar.	required
outputs	dict[str, str] | None	Mapa {tipo: ruta_relativa} del catálogo (para verificar publish_outputs). Opcional.	None
root_dir	str | Path | None	Directorio raíz del proyecto. Requerido si se pasa outputs (para resolver las rutas relativas). Por defecto se infiere desde __file__.	None
strict	bool	Si True, las discrepancias en expected_record_count se reportan como error. Si False, son solo advertencia.	False

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Resultado de validación con dataset, status,
dict[str, Any]	errors y warnings.

Source code in src/chile_hub/contracts.py

def verify_dataset_contract(
    dataset_name: str,
    contract: dict[str, Any],
    df: pl.DataFrame,
    outputs: dict[str, str] | None = None,
    root_dir: str | Path | None = None,
    *,
    strict: bool = False,
) -> dict[str, Any]:
    """Valida un DataFrame contra un contrato JSON Schema de ChileHub.

    Args:
        dataset_name: Nombre del dataset (se coteja contra ``contract["dataset"]``).
        contract: Dict del contrato JSON Schema cargado.
        df: DataFrame de Polars a validar.
        outputs: Mapa ``{tipo: ruta_relativa}`` del catálogo (para verificar
            ``publish_outputs``). Opcional.
        root_dir: Directorio raíz del proyecto. Requerido si se pasa ``outputs``
            (para resolver las rutas relativas). Por defecto se infiere desde
            ``__file__``.
        strict: Si ``True``, las discrepancias en ``expected_record_count`` se
            reportan como error. Si ``False``, son solo advertencia.

    Returns:
        dict[str, Any]: Resultado de validación con ``dataset``, ``status``,
        ``errors`` y ``warnings``.
    """
    if root_dir is None:
        root_dir = Path(__file__).resolve().parents[2]
    root_dir = Path(root_dir)

    errors: list[str] = []
    warnings: list[str] = []

    # 1. Coherencia del nombre del dataset
    if contract.get("dataset") != dataset_name:
        errors.append(
            f"El contrato tiene dataset='{contract.get('dataset')}', se esperaba '{dataset_name}'"
        )

    # 2. Columnas requeridas
    required_columns = contract.get("required_columns", [])
    missing_columns = [c for c in required_columns if c not in df.columns]
    if missing_columns:
        errors.append(f"Faltan columnas requeridas: {', '.join(missing_columns)}")

    # 3. Tipos de columna
    for column, expected_type in contract.get("column_types", {}).items():
        if column not in df.schema:
            errors.append(f"Columna '{column}' declarada en contract_type no existe en los datos")
            continue
        actual_type = contract_type(df.schema[column])
        if actual_type != expected_type:
            errors.append(f"Columna '{column}': tipo {actual_type}, se esperaba {expected_type}")

    # 4. Clave primaria
    primary_key = contract.get("primary_key", [])
    if primary_key:
        missing_key_columns = [c for c in primary_key if c not in df.columns]
        if missing_key_columns:
            errors.append(f"Columnas de clave primaria faltantes: {', '.join(missing_key_columns)}")
        else:
            # Valores nulos en PK
            null_mask = df.select(primary_key).null_count()
            if null_mask.sum_horizontal().item() > 0:
                errors.append(f"Clave primaria {primary_key} contiene valores nulos")
            # Unicidad de PK
            if df.select(primary_key).n_unique() != df.height:
                errors.append(
                    f"Clave primaria {primary_key} no es única "
                    f"({df.height - df.select(primary_key).n_unique()} duplicados)"
                )

    # 5. Columnas de ancho fijo (códigos CUT, etc.)
    for column, width in contract.get("fixed_width_columns", {}).items():
        if column not in df.schema:
            errors.append(f"Columna de ancho fijo '{column}' no existe en los datos")
            continue
        if contract_type(df.schema[column]) != "string":
            errors.append(
                f"Columna de ancho fijo '{column}' debe ser string "
                f"(tipo actual: {df.schema[column]})"
            )
        else:
            invalid_count = df.filter(
                pl.col(column).is_null() | (pl.col(column).str.len_chars() != width)
            ).height
            if invalid_count:
                errors.append(
                    f"Columna '{column}': {invalid_count} valores fuera del "
                    f"ancho esperado de {width} caracteres"
                )

    # 6. Conteo de registros esperado
    coverage_policy = contract.get("coverage_policy")
    expected_count = contract.get("expected_record_count")
    if expected_count is not None and df.height != expected_count:
        msg = (
            f"Registros: {df.height}, se esperaban {expected_count} "
            f"(coverage_policy={coverage_policy})"
        )
        if coverage_policy == "full" and strict:
            errors.append(msg)
        else:
            warnings.append(msg)

    # 7. Outputs publicables
    if outputs:
        for output_type in contract.get("publish_outputs", []):
            relative_path = outputs.get(output_type)
            if not relative_path:
                errors.append(f"El catálogo no tiene entrada para output '{output_type}'")
            else:
                output_path = root_dir / relative_path
                if not output_path.exists():
                    errors.append(f"Output '{output_type}' no existe: {relative_path}")

    status = "ok" if not errors else "error"
    return {
        "dataset": dataset_name,
        "status": status,
        "errors": errors,
        "warnings": warnings,
    }

Excepciones

chile_hub.exceptions

Jerarquía pública de excepciones de chile-hub.

ChileHubError

Bases: Exception

Clase base para errores de ejecución esperados de chile-hub.

Source code in src/chile_hub/exceptions.py

class ChileHubError(Exception):
    """Clase base para errores de ejecución esperados de chile-hub."""

ChileHubDataError

Bases: ChileHubError, RuntimeError

Se lanza cuando los datos de una release no pueden resolverse o verificarse.

Source code in src/chile_hub/exceptions.py

class ChileHubDataError(ChileHubError, RuntimeError):
    """Se lanza cuando los datos de una release no pueden resolverse o verificarse."""

ChileHubDatasetError

Bases: _ChileHubKeyError

Se lanza cuando un nombre de dataset no está registrado en el catálogo.

Source code in src/chile_hub/exceptions.py

class ChileHubDatasetError(_ChileHubKeyError):
    """Se lanza cuando un nombre de dataset no está registrado en el catálogo."""

ChileHubOutputError

Bases: _ChileHubKeyError

Se lanza cuando un tipo de salida de dataset no está disponible.

Source code in src/chile_hub/exceptions.py

class ChileHubOutputError(_ChileHubKeyError):
    """Se lanza cuando un tipo de salida de dataset no está disponible."""

ChileHubExampleError

Bases: _ChileHubKeyError

Se lanza cuando un tipo de ejemplo no está disponible para un dataset.

Source code in src/chile_hub/exceptions.py

class ChileHubExampleError(_ChileHubKeyError):
    """Se lanza cuando un tipo de ejemplo no está disponible para un dataset."""