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

flynt pass to provide consistency to string formatters and concatenation

Browse Source
This commit is contained in:
Ian Renton
2026-08-15 07:55:32 +01:00
parent bff5b79f8f
commit 7391c28cd0
72 changed files with 184 additions and 193 deletions
Download Patch File Download Diff File Expand all files Collapse all files
providers/callsigndata/clublogxml.py
+1 -1
View File
@@ -26,7 +26,7 @@ class ClublogXML(FileDownloadCallsignDataProvider):
logging.warning(
"Clublog XML callsign data provider configured but no api key was provided, this has been disabled.")
super().__init__("Clublog XML", provider_config, self.DATA_URL + "?api=" + self._api_key,
super().__init__("Clublog XML", provider_config, f"{self.DATA_URL}?api={self._api_key}",
self.CACHE_PATH_ZIPPED, self.POLL_INTERVAL_DAYS, DATA_STORE.callsign_data_clublogxml)
def _handle_file(self, path):
providers/callsigndata/file_download_callsign_data_provider.py
+5 -5
View File
@@ -22,7 +22,7 @@ class FileDownloadCallsignDataProvider(CallsignDataProvider):
self._poll_interval = poll_interval
self._thread = None
self._stop_event = Event()
self._url_data_cache = URLDataCache("callsigndata_" + name)
self._url_data_cache = URLDataCache(f"callsigndata_{name}")
if self.enabled:
self.status = "Ready"
@@ -31,7 +31,7 @@ class FileDownloadCallsignDataProvider(CallsignDataProvider):
# Fire off the polling thread. It will poll immediately on startup, then sleep for poll_interval between
# subsequent polls, so start() returns immediately and the application can continue starting.
logging.info(
"Set up query of " + self.name + " callsign reference data every " + str(self._poll_interval) + " days.")
f"Set up query of {self.name} callsign reference data every {self._poll_interval!s} days.")
self._thread = Thread(target=self._run, name=f"FileDownloadCallsignDataProvider-{self.name}")
self._thread.start()
@@ -48,7 +48,7 @@ class FileDownloadCallsignDataProvider(CallsignDataProvider):
try:
# Request the file. Use the data cache (with a TTL of 1 day) here, not as the main mechanism for
# caching, but just so continual restarts of the software during testing don't hammer the servers.
logging.debug("Downloading " + self.name + " callsign reference data...")
logging.debug(f"Downloading {self.name} callsign reference data...")
http_response = self._url_data_cache.get(self._url, headers=HTTP_HEADERS)
# Check response code was good
if http_response.ok:
@@ -61,7 +61,7 @@ class FileDownloadCallsignDataProvider(CallsignDataProvider):
if ok:
self.status = "OK"
self.last_update_time = datetime.now(pytz.UTC)
logging.info("Updated callsign reference data from " + self.name)
logging.info(f"Updated callsign reference data from {self.name}")
else:
self.status = "Error"
logging.warning(f"Error updating callsign reference data from {self.name}.")
@@ -78,7 +78,7 @@ class FileDownloadCallsignDataProvider(CallsignDataProvider):
logging.warning(f"Timeout when downloading callsign reference data from {self.name}.")
except Exception:
self.status = "Error"
logging.exception("Exception in callsign reference data provider (" + self.name + ")")
logging.exception(f"Exception in callsign reference data provider ({self.name})")
self._stop_event.wait(timeout=1)
def _handle_file(self, path):
providers/callsigndata/hamqth.py
+4 -6
View File
@@ -22,11 +22,11 @@ class HamQTH(APIQueryCallsignDataProvider):
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._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(CACHE_DIR + "/urls/hamqth-creds",
self._CREDENTIALS_CACHE = CachedSession(f"{CACHE_DIR}/urls/hamqth-creds",
expire_after=timedelta(minutes=55))
def _perform_new_lookup(self, callsign, lookup_credentials):
@@ -44,8 +44,7 @@ class HamQTH(APIQueryCallsignDataProvider):
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),
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"]:
@@ -74,8 +73,7 @@ class HamQTH(APIQueryCallsignDataProvider):
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)
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"]
providers/callsigndata/qrz.py
+5 -6
View File
@@ -24,7 +24,7 @@ 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(CACHE_DIR + "/urls/qrz-creds",
self._CREDENTIALS_CACHE = CachedSession(f"{CACHE_DIR}/urls/qrz-creds",
expire_after=timedelta(minutes=55))
def _perform_new_lookup(self, callsign, lookup_credentials):
@@ -42,8 +42,7 @@ class QRZ(APIQueryCallsignDataProvider):
elif lookup_credentials.qrz_username and lookup_credentials.qrz_password:
try:
login_response = self._CREDENTIALS_CACHE.get(
self._QRZ_BASE_URL + "?username=" + urllib.parse.quote_plus(lookup_credentials.qrz_username) +
"&password=" + urllib.parse.quote_plus(lookup_credentials.qrz_password) + "&agent=spothole",
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", {})
@@ -73,7 +72,7 @@ class QRZ(APIQueryCallsignDataProvider):
for lookup_call in calls_to_try:
try:
response = self._URL_DATA_CACHE.get(
self._QRZ_BASE_URL + "?s=" + session_key + "&callsign=" + urllib.parse.quote_plus(lookup_call),
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", {})
@@ -129,9 +128,9 @@ class QRZ(APIQueryCallsignDataProvider):
if "fname" in data:
name = data["fname"]
if "nick" in data:
name = name + " \"" + data["nick"] + "\""
name = f"{name} \"{data['nick']}\""
if "name" in data:
name = name + " " + data["name"]
name = f"{name} {data['name']}"
# Check for sensible latitudes
lat = None