from __future__ import annotations import logging import urllib.parse from datetime import datetime, timedelta from typing import Any 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 data.lookup_credentials import LookupCredentials 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: dict[str, Any]) -> None: 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: str, lookup_credentials: LookupCredentials | None) -> Callsign | None: # 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: str, data: dict[str, Any]) -> Callsign: """Convert the "Callsign" block in HamQTH's API response to our own Callsign object.""" # Check for sensible latitudes lat = None lon = None if ( data.get("latitude") is not None and data.get("longitude") is not None 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 data.get("grid") 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 data.get("adif") is not None else None, cq_zone=int(data["cq"]) if data.get("cq") is not None else None, itu_zone=int(data["itu"]) if data.get("itu") is not None else None, location_source=LocationSourceForCallsign.HOME_QTH, )