import logging import urllib.parse from datetime import timedelta, datetime import pytz import xmltodict from pyhamtools import callinfo from requests import 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 DATA_STORE, CACHE_DIR from core.url_data_cache import URLDataCache from data.callsign import Callsign from providers.callsigndata.api_query_callsign_data_provider import APIQueryCallsignDataProvider 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 = ("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(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( 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. logging.debug("HamQTH login details incorrect, failed to look up with HamQTH.") return None except Exception: logging.error("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: logging.debug("Could not look up home call for callsign %s", callsign) # Try looking up each call using the API for lookup_call in calls_to_try: try: response = self._URL_DATA_CACHE.get( 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: self.status = "OK" self.last_update_time = datetime.now(pytz.UTC) self.lookup_count += 1 # Found data, convert it to our object and return it data = xmltodict.parse(response.content)["HamQTH"]["search"] return self.hamqth_response_to_callsign(callsign, data) elif not response.from_cache: logging.warning("HTTP %d looking up callsign %s using HamQTH", response.status_code, lookup_call) except (KeyError, ValueError): continue except ConnectionError: logging.warning(f"Connection error when looking up callsign %s using HamQTH", lookup_call) continue except (ConnectTimeout, ReadTimeout): logging.warning(f"Timeout when looking up callsign %s using HamQTH", lookup_call) continue except Exception: logging.error("Exception when looking up callsign %s using HamQTH", lookup_call, exc_info=True) 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 as e: self.status = "Error" logging.error("Exception when looking up data from HamQTH", e, exc_info=True) # 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["nick"] if "nick" in data else None, qth=data["qth"] if "qth" in data else None, country=data["country"] if "country" in data else None, 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="HOME QTH")