mirror of
https://git.ianrenton.com/ian/spothole.git
synced 2026-08-05 18:11:41 +00:00
56 lines
2.2 KiB
Python
56 lines
2.2 KiB
Python
import gzip
|
|
import logging
|
|
|
|
from pyhamtools import LookupLib, Callinfo
|
|
|
|
from core.data_store import DATA_STORE
|
|
from core.utils import get_callsign_object_from_pyhamtools_callinfo
|
|
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, DATA_STORE.callsign_data_clublogxml)
|
|
|
|
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 _perform_new_lookup(self, callsign, lookup_credentials):
|
|
if self._callinfo:
|
|
return get_callsign_object_from_pyhamtools_callinfo(callsign, self._callinfo)
|
|
else:
|
|
return None
|