import logging import urllib.parse from datetime import timedelta import xmltodict from pyhamtools import callinfo from requests import ConnectTimeout, ReadTimeout from requests_cache import CachedSession from core.constants import HTTP_HEADERS 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 QRZ(APIQueryCallsignDataProvider): """Callsign data provider for QRZ.com.""" def __init__(self, provider_config): super().__init__("QRZ.com", provider_config, DATA_STORE.callsign_data_qrz) self._QRZ_BASE_URL = "https://xmldata.qrz.com/xml/current/" self._URL_DATA_CACHE = URLDataCache("qrz") # 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/qrz-creds", expire_after=timedelta(minutes=55)) self.status = "Ready" def _perform_new_lookup(self, callsign, lookup_credentials): # If we don't have QRZ credentials, skip this lookup if not ((lookup_credentials.qrz_username and lookup_credentials.qrz_password) or lookup_credentials.qrz_session_key): return None try: # Obtain session key from credentials, by looking it up from username & password if necessary. session_key = None if lookup_credentials.qrz_session_key: session_key = lookup_credentials.qrz_session_key elif lookup_credentials.qrz_username and lookup_credentials.qrz_password: try: login_response = self._CREDENTIALS_CACHE.get( self._QRZ_BASE_URL + "?username=" + urllib.parse.quote_plus(lookup_credentials.qrz_username) + "&password=" + urllib.parse.quote_plus(lookup_credentials.qrz_password) + "&agent=spothole", headers=HTTP_HEADERS).content login_data = xmltodict.parse(login_response) session = login_data.get("QRZDatabase", {}).get("Session", {}) if "Key" in session: session_key = str(session["Key"]) else: # Log this failure at debug level only, not our problem if user entered the wrong password. logging.debug("QRZ.com login details incorrect, failed to look up with QRZ.") return None except Exception: logging.error("Exception when getting QRZ.com session key") return None if not session_key: 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._QRZ_BASE_URL + "?s=" + session_key + "&callsign=" + urllib.parse.quote_plus(lookup_call), headers=HTTP_HEADERS, timeout=10) if response.ok: qrz_response = xmltodict.parse(response.content).get("QRZDatabase", {}) if qrz_response: if "Callsign" in qrz_response: # Found data, convert it to our object and return it return self.qrz_response_to_callsign(callsign, qrz_response.get("Callsign")) elif "Session" in qrz_response and "Error" in qrz_response.get("Session"): # Errors here are normally just "callsign not in database", no need to log that ourselves # above debug level. logging.debug("QRZ returned an error looking up callsign %s: %s", lookup_call, qrz_response.get("Session").get("Error")) elif not response.from_cache: logging.warning("QRZ returned a malformed response looking up callsign %s", lookup_call) elif not response.from_cache: logging.warning("HTTP %d looking up callsign %s using QRZ", lookup_call) except (KeyError, ValueError): continue except ConnectionError: logging.warning(f"Connection error when looking up callsign %s using QRZ", lookup_call) continue except (ConnectTimeout, ReadTimeout): logging.warning(f"Timeout when looking up callsign %s using QRZ.", lookup_call) continue except Exception: logging.error("Exception when looking up callsign %s using QRZ", lookup_call, exc_info=True) continue # Not found in QRZ; 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 QRZ.com", 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 qrz_response_to_callsign(callsign, data): """Convert the "Callsign" block in QRZ's API response to our own Callsign object.""" # Get a name name = None if "name_fmt" in data: name = data["name_fmt"] if "fname" in data: name = data["fname"] if "nick" in data: name = name + " \"" + data["nick"] + "\"" if "name" in data: name = name + " " + data["name"] # 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=name, qth=data["addr2"] if "addr2" 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["cqzone"]) if "cqzone" in data else None, itu_zone=int(data["ituzone"]) if "ituzone" in data else None)