from __future__ import annotations import ipaddress import json import os import socket import urllib.error import urllib.parse import urllib.request from dataclasses import asdict, dataclass from typing import Any DEFAULT_CKAN_ENDPOINT = "https://opendata.muenchen.de/" CKAN_TIMEOUT_SECONDS = 5 ALLOW_LOCAL_CKAN_ENV = "SMOLNALYSIS_ALLOW_LOCAL_CKAN" @dataclass class CkanConnectionStatus: ok: bool base_url: str api_base: str message: str dataset_count: int | None = None def to_dict(self) -> dict[str, Any]: return asdict(self) class CkanEndpointError(ValueError): pass def normalize_ckan_base_url(raw_url: str) -> str: value = raw_url.strip() if not value: raise CkanEndpointError("Enter a CKAN endpoint URL.") parsed = urllib.parse.urlsplit(value) if not parsed.scheme: parsed = urllib.parse.urlsplit(f"https://{value}") scheme = parsed.scheme.lower() if scheme not in {"http", "https"}: raise CkanEndpointError("CKAN endpoint must use http or https.") if not parsed.netloc: raise CkanEndpointError("CKAN endpoint must include a host.") if parsed.username or parsed.password: raise CkanEndpointError("CKAN endpoint must not include credentials.") if parsed.query or parsed.fragment: raise CkanEndpointError("CKAN endpoint must not include query parameters or fragments.") host = (parsed.hostname or "").lower() port = f":{parsed.port}" if parsed.port else "" path = parsed.path.rstrip("/") if path.endswith("/api/3/action"): path = path[: -len("/api/3/action")] elif path.endswith("/api/action"): path = path[: -len("/api/action")] normalized = urllib.parse.urlunsplit((scheme, f"{host}{port}", path or "", "", "")) return f"{normalized}/" def ckan_api_base(base_url: str) -> str: return urllib.parse.urljoin(base_url, "api/3/action") def _local_addresses_allowed() -> bool: return os.environ.get(ALLOW_LOCAL_CKAN_ENV, "").casefold() in {"1", "true", "yes", "on"} def _address_is_private(address: str) -> bool: ip = ipaddress.ip_address(address) return ( ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_reserved or ip.is_unspecified ) def _assert_public_host(base_url: str) -> None: if _local_addresses_allowed(): return parsed = urllib.parse.urlsplit(base_url) host = parsed.hostname if not host: raise CkanEndpointError("CKAN endpoint must include a host.") try: ip = ipaddress.ip_address(host) except ValueError: try: resolved = socket.getaddrinfo(host, parsed.port, type=socket.SOCK_STREAM) except socket.gaierror as exc: raise CkanEndpointError("Could not resolve the CKAN endpoint host.") from exc addresses = {item[4][0] for item in resolved} else: addresses = {str(ip)} if any(_address_is_private(address) for address in addresses): raise CkanEndpointError("CKAN endpoint must resolve to a public address.") def _read_ckan_action(api_base: str, action: str, params: dict[str, Any] | None = None) -> dict[str, Any]: query = urllib.parse.urlencode(params or {}) url = f"{api_base}/{action}" if query: url = f"{url}?{query}" request = urllib.request.Request(url, headers={"Accept": "application/json"}) with urllib.request.urlopen(request, timeout=CKAN_TIMEOUT_SECONDS) as response: body = response.read().decode("utf-8") payload = json.loads(body) if not isinstance(payload, dict): raise CkanEndpointError("CKAN returned an unexpected response.") return payload def read_ckan_action(base_url: str, action: str, params: dict[str, Any] | None = None) -> dict[str, Any]: normalized = normalize_ckan_base_url(base_url) _assert_public_host(normalized) payload = _read_ckan_action(ckan_api_base(normalized), action, params) if payload.get("success") is not True: raise CkanEndpointError(f"CKAN action failed: {action}") result = payload.get("result") return result if isinstance(result, dict) else {"value": result} def package_search(base_url: str, query: str, rows: int = 5, start: int = 0, fq: str = "") -> dict[str, Any]: clean_rows = max(1, min(int(rows), 10)) clean_start = max(0, int(start)) params: dict[str, Any] = {"q": query.strip() or "open data", "rows": clean_rows, "start": clean_start} if fq.strip(): params["fq"] = fq.strip() return read_ckan_action(base_url, "package_search", params) def package_show(base_url: str, package_id: str) -> dict[str, Any]: return read_ckan_action(base_url, "package_show", {"id": package_id}) def tag_search(base_url: str, query: str, rows: int = 10) -> dict[str, Any]: return read_ckan_action(base_url, "tag_search", {"query": query.strip(), "limit": max(1, min(int(rows), 25))}) def group_list(base_url: str, rows: int = 10) -> dict[str, Any]: return read_ckan_action(base_url, "group_list", {"all_fields": True, "limit": max(1, min(int(rows), 25))}) def organization_list(base_url: str, rows: int = 10) -> dict[str, Any]: return read_ckan_action(base_url, "organization_list", {"all_fields": True, "limit": max(1, min(int(rows), 25))}) def validate_ckan_endpoint(raw_url: str) -> CkanConnectionStatus: try: base_url = normalize_ckan_base_url(raw_url) _assert_public_host(base_url) api_base = ckan_api_base(base_url) site_read = _read_ckan_action(api_base, "site_read") if site_read.get("success") is not True: return CkanConnectionStatus(False, base_url, api_base, "CKAN site_read check failed.") package_search = _read_ckan_action(api_base, "package_search", {"rows": 0}) if package_search.get("success") is not True: return CkanConnectionStatus(False, base_url, api_base, "CKAN package_search check failed.") result = package_search.get("result") or {} dataset_count = result.get("count") if isinstance(result, dict) else None return CkanConnectionStatus( True, base_url, api_base, f"Connected to CKAN endpoint. {dataset_count:,} datasets found." if isinstance(dataset_count, int) else "Connected to CKAN endpoint.", dataset_count if isinstance(dataset_count, int) else None, ) except CkanEndpointError as exc: base_url = "" api_base = "" try: base_url = normalize_ckan_base_url(raw_url) api_base = ckan_api_base(base_url) except CkanEndpointError: pass return CkanConnectionStatus(False, base_url, api_base, str(exc)) except (TimeoutError, urllib.error.URLError, OSError): base_url = normalize_ckan_base_url(raw_url) return CkanConnectionStatus(False, base_url, ckan_api_base(base_url), "Could not reach the CKAN endpoint.") except json.JSONDecodeError: base_url = normalize_ckan_base_url(raw_url) return CkanConnectionStatus(False, base_url, ckan_api_base(base_url), "CKAN endpoint did not return JSON.") def default_ckan_status() -> CkanConnectionStatus: base_url = normalize_ckan_base_url(DEFAULT_CKAN_ENDPOINT) return CkanConnectionStatus(False, base_url, ckan_api_base(base_url), "Not connected.")