mirror of
https://git.ianrenton.com/ian/spothole.git
synced 2026-09-21 14:57:42 +00:00
59 lines
2.1 KiB
Python
59 lines
2.1 KiB
Python
import re
|
|
from typing import Any
|
|
|
|
from tornado.web import RequestHandler
|
|
|
|
from webserver.handlers.api.spots import APISpotsHandler, APISpotsStreamHandler
|
|
|
|
_GRID_SOURCE_RE = re.compile(r'"dx_location_source":\s*"GRID"')
|
|
_LEGACY_PARAM_TO_HEADER_MAP = {
|
|
"qrz_username": "X-QRZ-Username",
|
|
"qrz_password": "X-QRZ-Password",
|
|
"qrz_session_key": "X-QRZ-Session-Key",
|
|
"hamqth_username": "X-HamQTH-Username",
|
|
"hamqth_password": "X-HamQTH-Password",
|
|
"hamqth_session_id": "X-HamQTH-Session-ID",
|
|
}
|
|
|
|
|
|
def _handle_legacy_params(handler: RequestHandler) -> None:
|
|
"""Copy v1 query-string QRZ/HamQTH credentials into the v2 headers, so the v2 handler can see them"""
|
|
|
|
for param, header in _LEGACY_PARAM_TO_HEADER_MAP.items():
|
|
if header in handler.request.headers:
|
|
continue
|
|
value = handler.get_query_argument(param, default=None)
|
|
if value:
|
|
handler.request.headers[header] = value
|
|
|
|
|
|
class V1APISpotsHandler(APISpotsHandler):
|
|
"""API request handler for /api/v1/spots (GET). Included in early Spothole v2 for backwards compatibility."""
|
|
|
|
def prepare(self) -> None:
|
|
_handle_legacy_params(self)
|
|
super().prepare()
|
|
|
|
def write(self, chunk: str | bytes | dict[str, Any]) -> None:
|
|
if isinstance(chunk, str):
|
|
chunk = _GRID_SOURCE_RE.sub('"dx_location_source": "SPOT"', chunk)
|
|
super().write(chunk)
|
|
|
|
|
|
class V1APISpotsStreamHandler(APISpotsStreamHandler):
|
|
"""API request handler for /api/v1/spots/stream (SSE). Included in early Spothole v2 for backwards compatibility."""
|
|
|
|
def prepare(self) -> None:
|
|
_handle_legacy_params(self)
|
|
super().prepare()
|
|
|
|
def write_message(self, *args: Any, **kwargs: Any) -> None:
|
|
args = list(args)
|
|
for i, a in enumerate(args):
|
|
if isinstance(a, str) and '"dx_location_source"' in a:
|
|
args[i] = _GRID_SOURCE_RE.sub('"dx_location_source": "SPOT"', a)
|
|
for k, v in kwargs.items():
|
|
if isinstance(v, str) and '"dx_location_source"' in v:
|
|
kwargs[k] = _GRID_SOURCE_RE.sub('"dx_location_source": "SPOT"', v)
|
|
super().write_message(*args, **kwargs)
|