"""CKAN configuration catalog helper utilities."""
from __future__ import annotations
import json
from copy import deepcopy
from importlib import resources
from typing import Any, cast
def _load_catalog_from_disk() -> dict[str, Any]:
data_file = resources.files("ckan_mcp.data").joinpath("ckan_config_selection.json")
with data_file.open("r", encoding="utf-8") as fp:
data = json.load(fp)
return cast(dict[str, Any], data)
CKAN_CONFIG_SELECTION: dict[str, Any] = _load_catalog_from_disk()
class CkanConfigCatalog:
"""Helper for working with curated CKAN endpoint selections."""
def __init__(self, catalog: dict[str, Any]):
self._catalog = deepcopy(catalog)
def list_countries(self) -> list[str]:
"""Return the list of configured countries."""
countries = self._catalog.get("countries", {})
if not isinstance(countries, dict):
return []
return sorted(countries.keys())
def resolve_country(self, country: str) -> str | None:
"""Return the canonical country name for a loose query."""
if not country:
return None
normalized = country.strip().lower()
countries = self._catalog.get("countries", {})
if not isinstance(countries, dict):
return None
for candidate in countries.keys():
if isinstance(candidate, str) and candidate.lower() == normalized:
return candidate
return None
def list_locations(self, country: str) -> list[str]:
"""Return locations for a given country."""
canonical_country = self.resolve_country(country)
if canonical_country is None:
return []
countries = self._catalog.get("countries", {})
if not isinstance(countries, dict):
return []
country_entry = countries.get(canonical_country)
if not isinstance(country_entry, dict):
return []
locations = country_entry.get("locations", {})
if not isinstance(locations, dict):
return []
return sorted(locations.keys())
def resolve_location(self, country: str, location: str) -> tuple[str, str] | None:
"""Resolve country and location names to their canonical forms."""
canonical_country = self.resolve_country(country)
if canonical_country is None or not location:
return None
normalized_location = location.strip().lower()
countries = self._catalog.get("countries", {})
if not isinstance(countries, dict):
return None
country_entry = countries.get(canonical_country)
if not isinstance(country_entry, dict):
return None
locations = country_entry.get("locations", {})
if not isinstance(locations, dict):
return None
for candidate in locations.keys():
if isinstance(candidate, str) and candidate.lower() == normalized_location:
return canonical_country, candidate
return None
def get_location_entry(self, country: str, location: str) -> dict[str, Any] | None:
"""Return metadata for a specific location within a country."""
resolved = self.resolve_location(country, location)
if resolved is None:
return None
canonical_country, canonical_location = resolved
countries = self._catalog.get("countries", {})
if not isinstance(countries, dict):
return None
country_entry = countries.get(canonical_country)
if not isinstance(country_entry, dict):
return None
locations = country_entry.get("locations", {})
if not isinstance(locations, dict):
return None
entry = locations.get(canonical_location)
if not isinstance(entry, dict):
return None
return {
"country": canonical_country,
"location": canonical_location,
**entry,
}
def describe(self, country: str | None = None) -> dict[str, Any]:
"""
Return a human-friendly description of the catalog.
When no country is provided, the response lists all available CKAN portals with details.
When a country is provided, the response lists the locations within it.
"""
if not country:
# Return comprehensive overview of all available CKAN portals
all_portals = []
countries = self._catalog.get("countries", {})
if not isinstance(countries, dict):
return {
"message": "CKAN configuration catalog is empty.",
"portals": [],
"usage_hint": "Call ckan_api_initialise(country='<country>', location='<location>') to select a portal",
}
for country_name, country_data in countries.items():
if not isinstance(country_data, dict):
continue
locations = country_data.get("locations", {})
if not isinstance(locations, dict):
continue
for location_name, location_data in locations.items():
if not isinstance(location_data, dict):
continue
portal_info = {
"country": country_name,
"location": location_name,
"base_url": location_data["base_url"],
"description": f"{location_name}, {country_name}",
}
# Add special characteristics if available
if "overrides" in location_data:
overrides = location_data["overrides"]
characteristics = []
if overrides.get("action_transport") == "get":
characteristics.append("GET requests only")
if overrides.get("datastore_id_alias"):
characteristics.append("datastore ID alias supported")
if overrides.get("requires_api_key"):
characteristics.append("API key required")
if overrides.get("helper_prompt"):
characteristics.append("special requirements")
if characteristics:
portal_info["characteristics"] = characteristics
all_portals.append(portal_info)
return {
"message": "Available CKAN data portals - select one by providing country and location:",
"portals": all_portals,
"usage_hint": "Call ckan_api_initialise(country='<country>', location='<location>') to select a portal",
}
canonical_country = self.resolve_country(country)
if canonical_country is None:
# Return comprehensive overview since country wasn't found
return self.describe() # Recursively call with no country to show all options
return {
"message": f"Select a location for {canonical_country}.",
"country": canonical_country,
"locations": self.list_locations(canonical_country),
}
