mirror of
https://git.ianrenton.com/ian/spothole.git
synced 2026-09-20 06:17:41 +00:00
79 lines
2.7 KiB
Python
79 lines
2.7 KiB
Python
import gzip
|
|
import logging
|
|
|
|
from pyhamtools import Callinfo, LookupLib
|
|
|
|
from core.data_store import DATA_STORE
|
|
from core.utils import get_callsign_object_from_pyhamtools_callinfo
|
|
from data.callsign import Callsign
|
|
from providers.callsigndata.file_download_callsign_data_provider import (
|
|
FileDownloadCallsignDataProvider,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
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
|
|
logger.warning(
|
|
"Clublog XML callsign data provider configured but no api key was provided, this has been disabled."
|
|
)
|
|
|
|
super().__init__(
|
|
"Clublog XML",
|
|
provider_config,
|
|
f"{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:
|
|
logger.exception("Exception when loading Clublog XML.")
|
|
return False
|
|
|
|
def _perform_new_lookup(self, callsign, lookup_credentials):
|
|
callsign_data = Callsign(call=callsign)
|
|
|
|
try:
|
|
if self._callinfo:
|
|
callsign_data = get_callsign_object_from_pyhamtools_callinfo(callsign, self._callinfo)
|
|
self.status = "OK"
|
|
self.lookup_count += 1
|
|
else:
|
|
return None
|
|
|
|
except Exception:
|
|
self.status = "Error"
|
|
logger.exception("Exception when looking up data from Clublog XML data")
|
|
|
|
return callsign_data
|