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.constants import HTTP_HEADERS from core.data_store import CACHE_DIR, DATA_STORE from core.url_data_cache import URLDataCache from data.callsign import Callsign from providers.callsigndata.api_query_callsign_data_provider import ( APIQueryCallsignDataProvider, ) logger = logging.getLogger(__name__) 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(f"{CACHE_DIR}/urls/qrz-creds", expire_after=timedelta(minutes=55)) def _perform_new_lookup(self, callsign, lookup_credentials): # If we don't have QRZ 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.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( f"{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. logger.debug("QRZ.com login details incorrect, failed to look up with QRZ.") return None except Exception: logger.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: 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._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: qrz_data = qrz_response.get("Callsign") 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 return self.qrz_response_to_callsign(callsign, qrz_data) 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. logger.debug( f"QRZ returned an error looking up callsign {lookup_call}: {qrz_response.get('Session').get('Error')}" ) elif not response.from_cache: logger.warning(f"QRZ returned a malformed response looking up callsign {lookup_call}") elif not response.from_cache: logger.warning(f"HTTP {response.status_code} looking up callsign {lookup_call} using QRZ") except (KeyError, ValueError): continue except ConnectionError: logger.warning(f"Connection error when looking up callsign {lookup_call} using QRZ") continue except (ConnectTimeout, ReadTimeout): logger.warning(f"Timeout when looking up callsign {lookup_call} using QRZ.") continue except Exception: logger.exception(f"Exception when looking up callsign {lookup_call} using QRZ") 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: self.status = "Error" logger.exception("Exception when looking up data from QRZ.com") # 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 = f'{name} "{data["nick"]}"' if "name" in data: name = f"{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.get("addr2", None), country=data.get("country", None), continent=data.get("continent", 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, location_source="HOME QTH", )