mirror of
https://git.ianrenton.com/ian/spothole.git
synced 2026-09-20 14:27:42 +00:00
Add telnet server
This commit is contained in:
Ian Renton
@@ -0,0 +1,202 @@
|
||||
import logging
|
||||
import re
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
|
||||
import pytz
|
||||
import tornado
|
||||
from tornado import httputil
|
||||
from tornado.web import Application
|
||||
|
||||
from core.call_lookup_helper import get_call_info
|
||||
from core.constants import SIGS
|
||||
from core.geo_utils import (
|
||||
lat_lon_for_grid_sw_corner_plus_size,
|
||||
lat_lon_to_cq_zone,
|
||||
lat_lon_to_itu_zone,
|
||||
)
|
||||
from core.prometheus_metrics_handler import api_requests_counter
|
||||
from core.sig_lookup_helper import populate_missing_sig_ref_info
|
||||
from core.sig_utils import get_ref_regex_for_sig
|
||||
from core.utils import safe_json_dumps
|
||||
from data.lookup_credentials import extract_credentials
|
||||
from data.sig_ref import SIGRef
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class APILookupCallHandler(tornado.web.RequestHandler):
|
||||
"""API request handler for /api/v2/lookup/call"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
application: "Application",
|
||||
request: httputil.HTTPServerRequest,
|
||||
**kwargs: Any,
|
||||
):
|
||||
self._web_server_metrics = None
|
||||
super().__init__(application, request, **kwargs)
|
||||
|
||||
def initialize(self, web_server_metrics):
|
||||
self._web_server_metrics = web_server_metrics
|
||||
|
||||
def get(self):
|
||||
try:
|
||||
# Metrics
|
||||
self._web_server_metrics["last_api_access_time"] = datetime.now(pytz.UTC)
|
||||
self._web_server_metrics["api_access_counter"] += 1
|
||||
self._web_server_metrics["status"] = "OK"
|
||||
api_requests_counter.inc()
|
||||
|
||||
# request.arguments contains lists for each param key because technically the client can supply multiple,
|
||||
# reduce that to just the first entry, and convert bytes to string
|
||||
query_params = {k: v[0].decode("utf-8") for k, v in self.request.arguments.items()}
|
||||
|
||||
# The "call" query param must exist and look like a callsign
|
||||
if "call" in query_params:
|
||||
call = str(query_params.get("call")).upper()
|
||||
if re.match(r"^[A-Z0-9/\-]*$", call):
|
||||
credentials = extract_credentials(self.request.headers)
|
||||
callsign_data = get_call_info(call, credentials)
|
||||
self.write(safe_json_dumps(callsign_data))
|
||||
|
||||
else:
|
||||
self.write(safe_json_dumps(f"Error - '{call}' does not look like a valid callsign."))
|
||||
self.set_status(422)
|
||||
else:
|
||||
self.write(safe_json_dumps("Error - call must be provided"))
|
||||
self.set_status(422)
|
||||
|
||||
except Exception:
|
||||
logger.exception("Exception when handling client request to call lookup API")
|
||||
self.write(safe_json_dumps("Error - an internal server error occurred."))
|
||||
self.set_status(500)
|
||||
|
||||
self.set_header("Cache-Control", "no-store")
|
||||
self.set_header("Content-Type", "application/json")
|
||||
|
||||
|
||||
class APILookupSIGRefHandler(tornado.web.RequestHandler):
|
||||
"""API request handler for /api/v2/lookup/sigref"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
application: "Application",
|
||||
request: httputil.HTTPServerRequest,
|
||||
**kwargs: Any,
|
||||
):
|
||||
self._web_server_metrics = None
|
||||
super().__init__(application, request, **kwargs)
|
||||
|
||||
def initialize(self, web_server_metrics):
|
||||
self._web_server_metrics = web_server_metrics
|
||||
|
||||
def get(self):
|
||||
try:
|
||||
# Metrics
|
||||
self._web_server_metrics["last_api_access_time"] = datetime.now(pytz.UTC)
|
||||
self._web_server_metrics["api_access_counter"] += 1
|
||||
self._web_server_metrics["status"] = "OK"
|
||||
api_requests_counter.inc()
|
||||
|
||||
# request.arguments contains lists for each param key because technically the client can supply multiple,
|
||||
# reduce that to just the first entry, and convert bytes to string
|
||||
query_params = {k: v[0].decode("utf-8") for k, v in self.request.arguments.items()}
|
||||
|
||||
# "sig" and "id" query params must exist, SIG must be known, and if we have a reference regex for that SIG,
|
||||
# the provided id must match it.
|
||||
if "sig" in query_params and "id" in query_params:
|
||||
sig = str(query_params.get("sig")).upper()
|
||||
ref_id = str(query_params.get("id")).upper()
|
||||
if sig in [p.name.upper() for p in SIGS]:
|
||||
if not get_ref_regex_for_sig(sig) or re.match(get_ref_regex_for_sig(sig), ref_id):
|
||||
data = populate_missing_sig_ref_info(SIGRef(id=ref_id, sig=sig))
|
||||
self.write(safe_json_dumps(data))
|
||||
|
||||
else:
|
||||
self.write(
|
||||
safe_json_dumps(f"Error - '{ref_id}' does not look like a valid reference ID for {sig}.")
|
||||
)
|
||||
self.set_status(422)
|
||||
else:
|
||||
self.write(safe_json_dumps(f"Error - sig '{sig}' is not known."))
|
||||
self.set_status(422)
|
||||
else:
|
||||
self.write(safe_json_dumps("Error - sig and id must be provided"))
|
||||
self.set_status(422)
|
||||
|
||||
except Exception:
|
||||
logger.exception("Exception when handling client request to sig ref lookup API")
|
||||
self.write(safe_json_dumps("Error - an internal server error occurred."))
|
||||
self.set_status(500)
|
||||
|
||||
self.set_header("Cache-Control", "no-store")
|
||||
self.set_header("Content-Type", "application/json")
|
||||
|
||||
|
||||
class APILookupGridHandler(tornado.web.RequestHandler):
|
||||
"""API request handler for /api/v2/lookup/grid"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
application: "Application",
|
||||
request: httputil.HTTPServerRequest,
|
||||
**kwargs: Any,
|
||||
):
|
||||
self._web_server_metrics = None
|
||||
super().__init__(application, request, **kwargs)
|
||||
|
||||
def initialize(self, web_server_metrics):
|
||||
self._web_server_metrics = web_server_metrics
|
||||
|
||||
def get(self):
|
||||
try:
|
||||
# Metrics
|
||||
self._web_server_metrics["last_api_access_time"] = datetime.now(pytz.UTC)
|
||||
self._web_server_metrics["api_access_counter"] += 1
|
||||
self._web_server_metrics["status"] = "OK"
|
||||
api_requests_counter.inc()
|
||||
|
||||
# request.arguments contains lists for each param key because technically the client can supply multiple,
|
||||
# reduce that to just the first entry, and convert bytes to string
|
||||
query_params = {k: v[0].decode("utf-8") for k, v in self.request.arguments.items()}
|
||||
|
||||
# "grid" query param must exist.
|
||||
if "grid" in query_params:
|
||||
grid = str(query_params.get("grid")).upper()
|
||||
lat, lon, lat_cell_size, lon_cell_size = lat_lon_for_grid_sw_corner_plus_size(grid)
|
||||
if lat is not None and lon is not None and lat_cell_size is not None and lon_cell_size is not None:
|
||||
center_lat = lat + lat_cell_size / 2.0
|
||||
center_lon = lon + lon_cell_size / 2.0
|
||||
center_cq_zone = lat_lon_to_cq_zone(center_lat, center_lon)
|
||||
center_itu_zone = lat_lon_to_itu_zone(center_lat, center_lon)
|
||||
|
||||
response = {
|
||||
"center": {
|
||||
"latitude": center_lat,
|
||||
"longitude": center_lon,
|
||||
"cq_zone": center_cq_zone,
|
||||
"itu_zone": center_itu_zone,
|
||||
},
|
||||
"southwest": {
|
||||
"latitude": lat,
|
||||
"longitude": lon,
|
||||
},
|
||||
"northeast": {
|
||||
"latitude": lat + lat_cell_size,
|
||||
"longitude": lon + lon_cell_size,
|
||||
},
|
||||
}
|
||||
self.write(safe_json_dumps(response))
|
||||
|
||||
else:
|
||||
self.write(safe_json_dumps("Error - grid must be provided"))
|
||||
self.set_status(422)
|
||||
|
||||
except Exception:
|
||||
logger.exception("Exception when handling client request to grid ref lookup API")
|
||||
self.write(safe_json_dumps("Error - an internal server error occurred."))
|
||||
self.set_status(500)
|
||||
|
||||
self.set_header("Cache-Control", "no-store")
|
||||
self.set_header("Content-Type", "application/json")
|
||||
Reference in New Issue
Block a user