"""Machine-readable Geofabrik catalog resolution for English-language markets.""" from __future__ import annotations import json import re import urllib.error import urllib.request from collections import defaultdict from collections.abc import Callable, Mapping from dataclasses import dataclass from types import MappingProxyType from typing import Any, Final from urllib.parse import urlsplit from ..errors import PbfDownloadRejected from ..markets import EnglishMarket, GeoBounds _INDEX_URL: Final = "https://download.geofabrik.de/index-v1-nogeom.json" _GEOFABRIK_HOST: Final = "download.geofabrik.de" _MAX_INDEX_BYTES: Final = 4 * 1024 * 1024 _MAX_FEATURES: Final = 10_000 _MAX_SHARDS: Final = 4_096 _FEATURE_ID = re.compile(r"^[a-z0-9][a-z0-9-]{0,127}$") _FEATURE_PATH = re.compile(r"^[a-z0-9][a-z0-9-]{0,127}(?:/[a-z0-9][a-z0-9-]{0,127}){0,15}$") _ISO_ALPHA2 = re.compile(r"^[A-Z]{2}$") _ISO_SUBDIVISION = re.compile(r"^[A-Z]{2}-[A-Z0-9]{1,8}$") class _RejectRedirects(urllib.request.HTTPRedirectHandler): def redirect_request( self, req: urllib.request.Request, fp: Any, code: int, msg: str, headers: Any, newurl: str, ) -> urllib.request.Request | None: del req, fp, code, msg, headers, newurl return None def _default_opener() -> urllib.request.OpenerDirector: return urllib.request.build_opener( urllib.request.ProxyHandler({}), _RejectRedirects(), ) def _nonblank(value: object, name: str, maximum: int = 256) -> str: if not isinstance(value, str) or not value.strip(): raise PbfDownloadRejected(f"{name} must be a nonblank string") result = value.strip() if len(result) > maximum: raise PbfDownloadRejected(f"{name} exceeds the bounded length") return result def _normalized_feature_id(value: object) -> tuple[str, str]: raw = _nonblank(value, "feature id", 2_048) if not _FEATURE_PATH.fullmatch(raw): raise PbfDownloadRejected("Geofabrik feature id has an invalid shape") normalized = raw.replace("/", "--") if not _FEATURE_ID.fullmatch(normalized): raise PbfDownloadRejected("Geofabrik normalized feature id has an invalid shape") return raw, normalized def _nearest_downloadable_ancestor( path: str, raw_to_normalized: Mapping[str, str], ) -> str | None: parts = path.split("/") for end in range(len(parts) - 1, 0, -1): candidate = "/".join(parts[:end]) normalized = raw_to_normalized.get(candidate) if normalized is not None: return normalized return None def _normalized_parent_id( raw_feature_id: str, value: str | None, *, raw_to_normalized: Mapping[str, str], ) -> str | None: feature_id = raw_to_normalized[raw_feature_id] if "/" in raw_feature_id: path_parent = _nearest_downloadable_ancestor(raw_feature_id, raw_to_normalized) if path_parent is not None: return path_parent if value is None: return None parent = _nonblank(value, "Geofabrik parent id", 2_048) if not _FEATURE_PATH.fullmatch(parent): raise PbfDownloadRejected("Geofabrik parent id has an invalid shape") normalized = raw_to_normalized.get(parent) if normalized is None and "/" in parent: normalized = _nearest_downloadable_ancestor(parent, raw_to_normalized) if normalized is None: raise PbfDownloadRejected("Geofabrik hierarchical parent has no downloadable ancestor") if normalized is None: normalized = parent if normalized == feature_id: raise PbfDownloadRejected("Geofabrik catalog contains a self-parent") return normalized def _string_tuple( value: object, *, name: str, pattern: re.Pattern[str], ) -> tuple[str, ...]: if value is None: return () if isinstance(value, str): raw = (value,) elif isinstance(value, (list, tuple)): raw = tuple(value) else: raise PbfDownloadRejected(f"{name} must be a string or sequence of strings") normalized: list[str] = [] for item in raw: if not isinstance(item, str) or not pattern.fullmatch(item): raise PbfDownloadRejected(f"{name} contains an invalid value") normalized.append(item) return tuple(sorted(set(normalized))) def _approved_pbf_url(value: object) -> str: url = _nonblank(value, "Geofabrik PBF URL", 1024) parsed = urlsplit(url) if ( parsed.scheme != "https" or parsed.hostname != _GEOFABRIK_HOST or parsed.username is not None or parsed.password is not None or parsed.query or parsed.fragment or not parsed.path.endswith("-latest.osm.pbf") ): raise PbfDownloadRejected("Geofabrik catalog contains an unapproved PBF URL") return url @dataclass(frozen=True, slots=True) class CatalogFeature: """One validated PBF-bearing feature from the Geofabrik index.""" feature_id: str parent_id: str | None display_name: str pbf_url: str iso_alpha2: tuple[str, ...] = () iso_subdivision: tuple[str, ...] = () def __post_init__(self) -> None: if not isinstance(self.feature_id, str) or not _FEATURE_ID.fullmatch(self.feature_id): raise PbfDownloadRejected("Geofabrik feature id has an invalid shape") if self.parent_id is not None and ( not isinstance(self.parent_id, str) or not _FEATURE_ID.fullmatch(self.parent_id) ): raise PbfDownloadRejected("Geofabrik parent id has an invalid shape") object.__setattr__(self, "display_name", _nonblank(self.display_name, "feature name")) object.__setattr__(self, "pbf_url", _approved_pbf_url(self.pbf_url)) object.__setattr__( self, "iso_alpha2", _string_tuple(self.iso_alpha2, name="iso_alpha2", pattern=_ISO_ALPHA2), ) object.__setattr__( self, "iso_subdivision", _string_tuple( self.iso_subdivision, name="iso_subdivision", pattern=_ISO_SUBDIVISION, ), ) @dataclass(frozen=True, slots=True) class MarketSourceShard: """One bounded Geofabrik source selected for one English-language market.""" market: EnglishMarket feature: CatalogFeature country_bounds: GeoBounds | None = None def __post_init__(self) -> None: if not isinstance(self.market, EnglishMarket): raise PbfDownloadRejected("market must be EnglishMarket") if not isinstance(self.feature, CatalogFeature): raise PbfDownloadRejected("feature must be CatalogFeature") if self.country_bounds is not None and not isinstance(self.country_bounds, GeoBounds): raise PbfDownloadRejected("country_bounds must be GeoBounds or null") if self.market.requires_country_filter != (self.country_bounds is not None): raise PbfDownloadRejected("country-filter metadata is inconsistent") @property def shard_id(self) -> str: return f"{self.market.country_code.casefold()}-{self.feature.feature_id}" class GeofabrikCatalog: """Immutable validated index with deterministic country and shard resolution.""" def __init__(self, features: tuple[CatalogFeature, ...]) -> None: if not features or len(features) > _MAX_FEATURES: raise PbfDownloadRejected("Geofabrik catalog feature count is outside bounds") by_id = {feature.feature_id: feature for feature in features} if len(by_id) != len(features): raise PbfDownloadRejected("Geofabrik catalog feature ids are not unique") children: dict[str, list[CatalogFeature]] = defaultdict(list) for feature in features: if feature.parent_id is not None: children[feature.parent_id].append(feature) self._features = tuple(sorted(features, key=lambda item: item.feature_id)) self._by_id = MappingProxyType(by_id) self._children = MappingProxyType( { key: tuple(sorted(values, key=lambda item: item.feature_id)) for key, values in children.items() } ) @classmethod def from_mapping(cls, value: object) -> "GeofabrikCatalog": if not isinstance(value, Mapping): raise PbfDownloadRejected("Geofabrik index must be an object") raw_features = value.get("features") if not isinstance(raw_features, list) or not 1 <= len(raw_features) <= _MAX_FEATURES: raise PbfDownloadRejected("Geofabrik index has an invalid feature list") pending: list[tuple[str, str, Mapping[str, object], Mapping[str, object]]] = [] raw_to_normalized: dict[str, str] = {} normalized_ids: set[str] = set() for raw_feature in raw_features: if not isinstance(raw_feature, Mapping): raise PbfDownloadRejected("Geofabrik feature must be an object") properties = raw_feature.get("properties") if not isinstance(properties, Mapping): raise PbfDownloadRejected("Geofabrik feature properties must be an object") urls = properties.get("urls") if not isinstance(urls, Mapping) or "pbf" not in urls: continue raw_feature_id, feature_id = _normalized_feature_id(properties.get("id")) if raw_feature_id in raw_to_normalized: raise PbfDownloadRejected("Geofabrik catalog feature ids are not unique") if feature_id in normalized_ids: raise PbfDownloadRejected("Geofabrik normalized feature ids collide") raw_to_normalized[raw_feature_id] = feature_id normalized_ids.add(feature_id) pending.append((raw_feature_id, feature_id, properties, urls)) parsed: list[CatalogFeature] = [] for raw_feature_id, feature_id, properties, urls in pending: raw_parent = properties.get("parent") if raw_parent is not None and not isinstance(raw_parent, str): raise PbfDownloadRejected("Geofabrik parent must be a string or null") parsed.append( CatalogFeature( feature_id=feature_id, parent_id=_normalized_parent_id( raw_feature_id, raw_parent, raw_to_normalized=raw_to_normalized, ), display_name=_nonblank(properties.get("name"), "feature name"), pbf_url=_approved_pbf_url(urls.get("pbf")), iso_alpha2=_string_tuple( properties.get("iso3166-1:alpha2"), name="iso3166-1:alpha2", pattern=_ISO_ALPHA2, ), iso_subdivision=_string_tuple( properties.get("iso3166-2"), name="iso3166-2", pattern=_ISO_SUBDIVISION, ), ) ) return cls(tuple(parsed)) @classmethod def fetch( cls, *, timeout_seconds: float = 30.0, opener: Any | None = None, ) -> "GeofabrikCatalog": if ( isinstance(timeout_seconds, bool) or not isinstance(timeout_seconds, (int, float)) or not 5 <= float(timeout_seconds) <= 120 ): raise PbfDownloadRejected("catalog timeout is outside the approved range") request = urllib.request.Request( _INDEX_URL, headers={ "Accept": "application/json", "Accept-Encoding": "identity", "User-Agent": "osm-lead-source-service/0.4", }, method="GET", ) active_opener = opener or _default_opener() try: with active_opener.open(request, timeout=float(timeout_seconds)) as response: if response.status != 200 or response.geturl() != _INDEX_URL: raise PbfDownloadRejected("Geofabrik index changed URL or status") content = response.read(_MAX_INDEX_BYTES + 1) except PbfDownloadRejected: raise except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError, OSError) as exc: raise PbfDownloadRejected("Geofabrik index request failed") from exc if len(content) > _MAX_INDEX_BYTES: raise PbfDownloadRejected("Geofabrik index exceeds the bounded size") try: value = json.loads(content) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise PbfDownloadRejected("Geofabrik index is invalid JSON") from exc return cls.from_mapping(value) def get(self, feature_id: str) -> CatalogFeature: try: return self._by_id[feature_id] except KeyError: raise PbfDownloadRejected(f"Geofabrik feature is unavailable: {feature_id}") from None def children(self, feature_id: str) -> tuple[CatalogFeature, ...]: return self._children.get(feature_id, ()) def root_for_market(self, market: EnglishMarket) -> CatalogFeature: if not isinstance(market, EnglishMarket): raise PbfDownloadRejected("market must be EnglishMarket") if market.source_feature_id is not None: return self.get(market.source_feature_id) matches = tuple( feature for feature in self._features if market.country_code in feature.iso_alpha2 ) if len(matches) != 1: raise PbfDownloadRejected( f"market {market.country_code} does not resolve to exactly one Geofabrik root" ) return matches[0] @staticmethod def _child_belongs_to_market(child: CatalogFeature, market: EnglishMarket) -> bool: if child.iso_alpha2: return market.country_code in child.iso_alpha2 if child.iso_subdivision: prefix = f"{market.country_code}-" return any(value.startswith(prefix) for value in child.iso_subdivision) return True def country_children( self, feature: CatalogFeature, market: EnglishMarket, ) -> tuple[CatalogFeature, ...]: if market.requires_country_filter: return () return tuple( child for child in self.children(feature.feature_id) if self._child_belongs_to_market(child, market) ) def plan_market_leaf_shards( self, market: EnglishMarket, ) -> tuple[MarketSourceShard, ...]: """Resolve the smallest available non-overseas PBF shards for a market.""" root = self.root_for_market(market) if market.requires_country_filter: return (MarketSourceShard(market, root, market.source_bounds),) planned: list[MarketSourceShard] = [] visiting: set[str] = set() def add(feature: CatalogFeature) -> None: if len(planned) >= _MAX_SHARDS: raise PbfDownloadRejected("market shard plan exceeds the bounded count") if feature.feature_id in visiting: raise PbfDownloadRejected("Geofabrik catalog contains a cycle") visiting.add(feature.feature_id) children = self.country_children(feature, market) if children: for child in children: add(child) else: planned.append(MarketSourceShard(market, feature)) visiting.remove(feature.feature_id) add(root) return tuple(planned) def plan_market_shards( self, market: EnglishMarket, *, max_shard_bytes: int, size_lookup: Callable[[CatalogFeature], int], ) -> tuple[MarketSourceShard, ...]: """Split a market recursively until each selected PBF is within the size bound.""" if ( isinstance(max_shard_bytes, bool) or not isinstance(max_shard_bytes, int) or not 32 * 1024 * 1024 <= max_shard_bytes <= 4 * 1024 * 1024 * 1024 ): raise PbfDownloadRejected("max_shard_bytes is outside the approved range") root = self.root_for_market(market) if market.requires_country_filter: return (MarketSourceShard(market, root, market.source_bounds),) planned: list[MarketSourceShard] = [] visiting: set[str] = set() def add(feature: CatalogFeature) -> None: if len(planned) >= _MAX_SHARDS: raise PbfDownloadRejected("market shard plan exceeds the bounded count") if feature.feature_id in visiting: raise PbfDownloadRejected("Geofabrik catalog contains a cycle") visiting.add(feature.feature_id) size = size_lookup(feature) if isinstance(size, bool) or not isinstance(size, int) or size <= 0: raise PbfDownloadRejected("Geofabrik size lookup returned an invalid value") children = self.country_children(feature, market) if size <= max_shard_bytes: planned.append(MarketSourceShard(market, feature)) elif children: for child in children: add(child) else: raise PbfDownloadRejected( f"market {market.country_code} has an oversized unsplittable source: " f"{feature.feature_id}" ) visiting.remove(feature.feature_id) add(root) return tuple(planned) __all__ = [ "CatalogFeature", "GeofabrikCatalog", "MarketSourceShard", ]