spothole/providers/callsigndata/hamqth.py

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:
                        # 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:
                        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")