spothole/providers/callsigndata/qrz.py

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.enums import Continent, LocationSourceForCallsign
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."""

        # I have encountered a user passing multiple callsigns to the QRZ lookup function in a way that QRZ actually
        # does handle, and returns a list of callsign data. If we encounter that, just take the first one, as our own
        # functions can't deal with multiple calls this way.
        if isinstance(data, list):
            data = data[0]
            callsign = data["call"]

        # 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=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,
            location_source=LocationSourceForCallsign.HOME_QTH,
        )