import logging import urllib.parse from datetime import datetime, timedelta import pytz import xmltodict from pyhamtools import callinfo from requests.exceptions import ConnectionError, ConnectTimeout, ReadTimeout from requests_cache import CachedSession from core.config import SERVER_OWNER_CALLSIGN from core.constants import HTTP_HEADERS, SOFTWARE_VERSION from core.data_store import CACHE_DIR, DATA_STORE from core.enums import Continent from core.url_data_cache import URLDataCache from data.callsign import Callsign, LocationSourceForCallsign from providers.callsigndata.api_query_callsign_data_provider import ( APIQueryCallsignDataProvider, ) logger = logging.getLogger(__name__) class HamQTH(APIQueryCallsignDataProvider): """Callsign data provider for HamQTH.""" def __init__(self, provider_config): super().__init__("HamQTH", provider_config, DATA_STORE.callsign_data_hamqth) self._HAMQTH_BASE_URL = "https://www.hamqth.com/xml.php" self._PRG = f"Spothole v{SOFTWARE_VERSION} operated by {SERVER_OWNER_CALLSIGN}".replace(" ", "_") self._URL_DATA_CACHE = URLDataCache("hamqth") # Separate URL cache for session key lookups. Once a session key is returned from logging in with a username # and password, this is valid for an hour, so our cache stores this specifically for 55 minutes. self._CREDENTIALS_CACHE = CachedSession(f"{CACHE_DIR}/urls/hamqth-creds", expire_after=timedelta(minutes=55)) def _perform_new_lookup(self, callsign, lookup_credentials): # If we don't have HamQTH credentials, skip this lookup Return None so we don't *cache* the lack of data, because # # someone might provide credentials next time around. if not lookup_credentials or not ( (lookup_credentials.hamqth_username and lookup_credentials.hamqth_password) or lookup_credentials.hamqth_session_id ): return None try: # Obtain session key from credentials, by looking it up from username & password if necessary. session_id = None if lookup_credentials.hamqth_session_id: session_id = lookup_credentials.hamqth_session_id elif lookup_credentials.hamqth_username and lookup_credentials.hamqth_password: try: session_data = self._CREDENTIALS_CACHE.get( f"{self._HAMQTH_BASE_URL}?u={urllib.parse.quote_plus(lookup_credentials.hamqth_username)}&p={urllib.parse.quote_plus(lookup_credentials.hamqth_password)}", headers=HTTP_HEADERS, ).content dict_data = xmltodict.parse(session_data) if "session_id" in dict_data["HamQTH"]["session"]: session_id = str(dict_data["HamQTH"]["session"]["session_id"]) else: # Log this failure at debug level only, not our problem if user entered the wrong password. logger.debug("HamQTH login details incorrect, failed to look up with HamQTH.") return None except Exception: logger.exception("Exception when getting HamQTH session key") return None if not session_id: return None # Try the call as given, then fall back to the base call (strips /P, /M etc.) calls_to_try = [callsign] try: home_call = callinfo.Callinfo.get_homecall(callsign) if home_call != callsign: calls_to_try.append(home_call) except ValueError: logger.debug(f"Could not look up home call for callsign {callsign}") # Try looking up each call using the API for lookup_call in calls_to_try: try: response = self._URL_DATA_CACHE.get( f"{self._HAMQTH_BASE_URL}?id={session_id}&callsign={urllib.parse.quote_plus(lookup_call)}&prg={self._PRG}", headers=HTTP_HEADERS, timeout=10, ) if response.ok: # Found data, convert it to our object and return it data = xmltodict.parse(response.content)["HamQTH"]["search"] self.status = "OK" self.last_update_time = datetime.now(pytz.UTC) self.lookup_count += 1 return self.hamqth_response_to_callsign(callsign, data) elif not response.from_cache: logger.warning(f"HTTP {response.status_code} looking up callsign {lookup_call} using HamQTH") except (KeyError, ValueError): continue except ConnectionError: logger.warning(f"Connection error when looking up callsign {lookup_call} using HamQTH") continue except (ConnectTimeout, ReadTimeout): logger.warning(f"Timeout when looking up callsign {lookup_call} using HamQTH") continue except Exception: logger.exception(f"Exception when looking up callsign {lookup_call} using HamQTH") continue # Not found in HamQTH; return a Callsign object with no data so we cache that and don't keep retrying return Callsign(call=callsign) except Exception: self.status = "Error" logger.exception("Exception when looking up data from HamQTH") # Return None, this won't be cached so we will be asked to query data again for this call next time. return None @staticmethod def hamqth_response_to_callsign(callsign, data): """Convert the "Callsign" block in HamQTH's API response to our own Callsign object.""" # Check for sensible latitudes lat = None lon = None if ( "latitude" in data and "longitude" in data and (float(data["latitude"]) != 0 or float(data["longitude"]) != 0) and -89.9 < float(data["latitude"]) < 89.9 ): lat = float(data["latitude"]) lon = float(data["longitude"]) # Check for sensible grids grid = None if "grid" in data and not data["grid"].startswith("AA00"): grid = data["grid"] return Callsign( call=callsign, home_call=callinfo.Callinfo.get_homecall(callsign), name=data.get("nick", None), qth=data.get("qth", None), country=data.get("country", None), continent=Continent(data["continent"]) if "continent" in data else None, latitude=lat, longitude=lon, grid=grid, dxcc_id=int(data["adif"]) if "adif" in data else None, cq_zone=int(data["cq"]) if "cq" in data else None, itu_zone=int(data["itu"]) if "itu" in data else None, location_source=LocationSourceForCallsign.HOME_QTH, )