spothole/providers/callsigndata/hamqth.py

import logging
import urllib.parse
from datetime import datetime, timedelta
from typing import Any

import pytz
import xmltodict
from pyhamtools import callinfo
from requests.exceptions import ConnectionError, 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 CACHE_DIR, DATA_STORE
from core.enums import Continent
from core.url_data_cache import URLDataCache
from data.callsign import Callsign, LocationSourceForCallsign
from data.lookup_credentials import LookupCredentials
from providers.callsigndata.api_query_callsign_data_provider import (
    APIQueryCallsignDataProvider,
)

logger = logging.getLogger(__name__)


class HamQTH(APIQueryCallsignDataProvider):
    """Callsign data provider for HamQTH."""

    def __init__(self, provider_config: dict[str, Any]) -> None:
        super().__init__("HamQTH", provider_config, DATA_STORE.callsign_data_hamqth)
        self._HAMQTH_BASE_URL = "https://www.hamqth.com/xml.php"
        self._PRG = f"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(f"{CACHE_DIR}/urls/hamqth-creds", expire_after=timedelta(minutes=55))

    def _perform_new_lookup(self, callsign: str, lookup_credentials: LookupCredentials | None) -> Callsign | None:
        # 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(
                        f"{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.
                        logger.debug("HamQTH login details incorrect, failed to look up with HamQTH.")
                        return None
                except Exception:
                    logger.exception("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:
                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._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:
                        logger.warning(f"HTTP {response.status_code} looking up callsign {lookup_call} using HamQTH")

                except (KeyError, ValueError):
                    continue
                except ConnectionError:
                    logger.warning(f"Connection error when looking up callsign {lookup_call} using HamQTH")
                    continue
                except (ConnectTimeout, ReadTimeout):
                    logger.warning(f"Timeout when looking up callsign {lookup_call} using HamQTH")
                    continue
                except Exception:
                    logger.exception(f"Exception when looking up callsign {lookup_call} using HamQTH")
                    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:
            self.status = "Error"
            logger.exception("Exception when looking up data from HamQTH")
            # 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: str, data: dict[str, Any]) -> Callsign:
        """Convert the "Callsign" block in HamQTH's API response to our own Callsign object."""

        # Check for sensible latitudes
        lat = None
        lon = None
        if (
            data.get("latitude") is not None
            and data.get("longitude") is not None
            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 data.get("grid") and not data["grid"].startswith("AA00"):
            grid = data["grid"]

        return Callsign(
            call=callsign,
            home_call=callinfo.Callinfo.get_homecall(callsign),
            name=data.get("nick", None),
            qth=data.get("qth", 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 data.get("adif") is not None else None,
            cq_zone=int(data["cq"]) if data.get("cq") is not None else None,
            itu_zone=int(data["itu"]) if data.get("itu") is not None else None,
            location_source=LocationSourceForCallsign.HOME_QTH,
        )