kerykeion.fetch_geonames

  1# -*- coding: utf-8 -*-
  2"""
  3    This is part of Kerykeion (C) 2025 Giacomo Battaglia
  4"""
  5
  6
  7import logging
  8from datetime import timedelta
  9from requests import Request
 10from requests_cache import CachedSession
 11from typing import Union
 12
 13
 14class FetchGeonames:
 15    """
 16    Class to handle requests to the GeoNames API
 17
 18    Args:
 19    - city_name (str): Name of the city
 20    - country_code (str): Two letters country code
 21    - username (str, optional): GeoNames username, defaults to "century.boy".
 22    - cache_expire_after_days (int, optional): Cache expiration time in days, defaults to 30.
 23    """
 24
 25    def __init__(
 26        self,
 27        city_name: str,
 28        country_code: str,
 29        username: str = "century.boy",
 30        cache_expire_after_days=30,
 31    ):
 32        self.session = CachedSession(
 33            cache_name="cache/kerykeion_geonames_cache",
 34            backend="sqlite",
 35            expire_after=timedelta(days=cache_expire_after_days),
 36        )
 37
 38        self.username = username
 39        self.city_name = city_name
 40        self.country_code = country_code
 41        self.base_url = "http://api.geonames.org/searchJSON"
 42        self.timezone_url = "http://api.geonames.org/timezoneJSON"
 43
 44    def __get_timezone(self, lat: Union[str, float, int], lon: Union[str, float, int]) -> dict[str, str]:
 45        """
 46        Get the timezone for a given latitude and longitude
 47        """
 48        # Dictionary that will be returned:
 49        timezone_data = {}
 50
 51        params = {"lat": lat, "lng": lon, "username": self.username}
 52
 53        prepared_request = Request("GET", self.timezone_url, params=params).prepare()
 54        logging.debug(f"Requesting data from GeoName timezones: {prepared_request.url}")
 55
 56        try:
 57            response = self.session.send(prepared_request)
 58            response_json = response.json()
 59
 60        except Exception as e:
 61            logging.error(f"Error fetching {self.timezone_url}: {e}")
 62            return {}
 63
 64        try:
 65            timezone_data["timezonestr"] = response_json["timezoneId"]
 66
 67        except Exception as e:
 68            logging.error(f"Error serializing data maybe wrong username? Details: {e}")
 69            return {}
 70
 71        if hasattr(response, "from_cache"):
 72            timezone_data["from_tz_cache"] = response.from_cache  # type: ignore
 73
 74        return timezone_data
 75
 76    def __get_contry_data(self, city_name: str, country_code: str) -> dict[str, str]:
 77        """
 78        Get the city data *whitout timezone* for a given city and country name
 79        """
 80        # Dictionary that will be returned:
 81        city_data_whitout_tz = {}
 82
 83        params = {
 84            "q": city_name,
 85            "country": country_code,
 86            "username": self.username,
 87            "maxRows": 1,
 88            "style": "SHORT",
 89            "featureClass": ["A", "P"],
 90        }
 91
 92        prepared_request = Request("GET", self.base_url, params=params).prepare()
 93        logging.debug(f"Requesting data from geonames basic: {prepared_request.url}")
 94
 95        try:
 96            response = self.session.send(prepared_request)
 97            response_json = response.json()
 98            logging.debug(f"Response from GeoNames: {response_json}")
 99
100        except Exception as e:
101            logging.error(f"Error in fetching {self.base_url}: {e}")
102            return {}
103
104        try:
105            city_data_whitout_tz["name"] = response_json["geonames"][0]["name"]
106            city_data_whitout_tz["lat"] = response_json["geonames"][0]["lat"]
107            city_data_whitout_tz["lng"] = response_json["geonames"][0]["lng"]
108            city_data_whitout_tz["countryCode"] = response_json["geonames"][0]["countryCode"]
109
110        except Exception as e:
111            logging.error(f"Error serializing data maybe wrong username? Details: {e}")
112            return {}
113
114        if hasattr(response, "from_cache"):
115            city_data_whitout_tz["from_country_cache"] = response.from_cache  # type: ignore
116
117        return city_data_whitout_tz
118
119    def get_serialized_data(self) -> dict[str, str]:
120        """
121        Returns all the data necessary for the Kerykeion calculation.
122
123        Returns:
124            dict[str, str]: _description_
125        """
126        city_data_response = self.__get_contry_data(self.city_name, self.country_code)
127        try:
128            timezone_response = self.__get_timezone(city_data_response["lat"], city_data_response["lng"])
129
130        except Exception as e:
131            logging.error(f"Error in fetching timezone: {e}")
132            return {}
133
134        return {**timezone_response, **city_data_response}
135
136
137if __name__ == "__main__":
138    from kerykeion.utilities import setup_logging
139    setup_logging(level="debug")
140
141    geonames = FetchGeonames("Montichiari", "IT")
142    print(geonames.get_serialized_data())