w1cdn/spothole
1
0
Fork 0
mirror of https://git.ianrenton.com/ian/spothole.git synced 2026-09-20 22:37:44 +00:00
Code Issues 13 Packages Projects Releases Wiki Activity

Use ruff linter to fix issues and provide consistent formatting

Browse Source
This commit is contained in:
Ian Renton
2026-08-15 08:25:54 +01:00
parent 7391c28cd0
commit af3f82c14d
121 changed files with 1989 additions and 996 deletions
Download Patch File Download Diff File Expand all files Collapse all files
providers/callsigndata/qrz.py
+45 -34
View File
@@ -1,6 +1,6 @@
import logging
import urllib.parse
from datetime import timedelta, datetime
from datetime import datetime, timedelta
import pytz
import xmltodict
@@ -9,10 +9,12 @@ 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.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
from providers.callsigndata.api_query_callsign_data_provider import (
APIQueryCallsignDataProvider,
)
class QRZ(APIQueryCallsignDataProvider):
@@ -24,14 +26,14 @@ class QRZ(APIQueryCallsignDataProvider):
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))
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):
if not lookup_credentials or not (
(lookup_credentials.qrz_username and lookup_credentials.qrz_password) or lookup_credentials.qrz_session_key
):
return None
try:
@@ -43,7 +45,8 @@ class QRZ(APIQueryCallsignDataProvider):
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
headers=HTTP_HEADERS,
).content
login_data = xmltodict.parse(login_response)
session = login_data.get("QRZDatabase", {}).get("Session", {})
if "Key" in session:
@@ -66,14 +69,16 @@ class QRZ(APIQueryCallsignDataProvider):
if home_call != callsign:
calls_to_try.append(home_call)
except ValueError:
logging.debug("Could not look up home call for callsign %s", callsign)
logging.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)
headers=HTTP_HEADERS,
timeout=10,
)
if response.ok:
qrz_response = xmltodict.parse(response.content).get("QRZDatabase", {})
if qrz_response:
@@ -88,24 +93,25 @@ class QRZ(APIQueryCallsignDataProvider):
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"))
logging.debug(
f"QRZ returned an error looking up callsign {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)
logging.warning(f"QRZ returned a malformed response looking up callsign {lookup_call}")
elif not response.from_cache:
logging.warning("HTTP %d looking up callsign %s using QRZ", lookup_call)
logging.warning(f"HTTP {response.status_code} looking up callsign {lookup_call} using QRZ")
except (KeyError, ValueError):
continue
except ConnectionError:
logging.warning(f"Connection error when looking up callsign %s using QRZ", lookup_call)
logging.warning(f"Connection error when looking up callsign {lookup_call} using QRZ")
continue
except (ConnectTimeout, ReadTimeout):
logging.warning(f"Timeout when looking up callsign %s using QRZ.", lookup_call)
logging.warning(f"Timeout when looking up callsign {lookup_call} using QRZ.")
continue
except Exception:
logging.exception("Exception when looking up callsign %s using QRZ", lookup_call)
logging.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
@@ -128,16 +134,19 @@ class QRZ(APIQueryCallsignDataProvider):
if "fname" in data:
name = data["fname"]
if "nick" in data:
name = f"{name} \"{data['nick']}\""
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:
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"])
@@ -146,16 +155,18 @@ class QRZ(APIQueryCallsignDataProvider):
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,
location_source="HOME QTH")
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,
location_source="HOME QTH",
)