Refactor of caching & data storage part 10 #118

2026-08-06 10:31:42 +00:00 · 2026-08-02 10:31:31 +01:00
parent 2157bf114e
commit 11a236e668
15 changed files with 452 additions and 170 deletions
@@ -0,0 +1,68 @@
+import gzip
+import logging
+
+from pyhamtools import LookupLib, Callinfo
+
+from data.callsign import Callsign
+from providers.callsigndata.file_download_callsign_data_provider import FileDownloadCallsignDataProvider
+
+
+class ClublogXML(FileDownloadCallsignDataProvider):
+    """Callsign data provider for ClubLog's Country File, which provides basic callsign to DXCC entity mapping."""
+
+    POLL_INTERVAL_DAYS = 30
+    DATA_URL = "https://cdn.clublog.org/cty.php"
+    CACHE_PATH_ZIPPED = "cache/cty.xml.gz"
+    CACHE_PATH_UNZIPPED = "cache/cty.xml"
+    _callinfo = None
+
+    def __init__(self, provider_config):
+        # API key required for this provider
+        self._api_key = provider_config.get("api-key", "")
+        if self._api_key == "":
+            provider_config["enabled"] = False
+            logging.warning(
+                "Clublog XML callsign data provider configured but no api key was provided, this has been disabled.")
+
+        super().__init__("Clublog XML", provider_config, self.DATA_URL + "?api=" + self._api_key,
+                         self.CACHE_PATH_ZIPPED, self.POLL_INTERVAL_DAYS)
+
+    def _handle_file(self, path):
+        try:
+            # The download from Clublog is gzipped so we need to uncompress that and re-save as a separate file that
+            # the LookupLib can actually use.
+            with gzip.open(path, "rb") as uncompressed:
+                file_content = uncompressed.read()
+            assert isinstance(file_content, bytes)
+            with open(self.CACHE_PATH_UNZIPPED, "wb") as f:
+                f.write(file_content)
+                f.flush()
+
+            # Now load the data
+            lookuplib = LookupLib(lookuptype="clublogxml", filename=self.CACHE_PATH_UNZIPPED)
+            self._callinfo = Callinfo(lookuplib)
+            return True
+
+        except Exception as e:
+            logging.error("Exception when loading Clublog XML.", e, exc_info=True)
+            return False
+
+    def lookup(self, callsign, lookup_credentials):
+        # Lookup credentials are not required for this source.
+        # Lat/lon will only be centre of country or capital city from this source
+        ll = self._callinfo.get_lat_long(callsign)
+        lat = None
+        lon = None
+        if ll and "latitude" in ll and "longitude" in ll:
+            lat = float(ll["latitude"])
+            lon = float(ll["longitude"])
+
+        return Callsign(call=callsign,
+                        home_call=self._callinfo.get_homecall(callsign),
+                        country=self._callinfo.get_country_name(callsign),
+                        dxcc_id=self._callinfo.get_adif_id(callsign),
+                        continent=self._callinfo.get_continent(callsign),
+                        cq_zone=self._callinfo.get_cqz(callsign),
+                        itu_zone=self._callinfo.get_ituz(callsign),
+                        latitude=lat,
+                        longitude=lon)