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.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_utils import get_ref_regex_for_sig from core.sig_lookup_helper import populate_sig_ref_info from core.utils import safe_json_dumps from data.lookup_credentials import extract_credentials from data.sig_ref import SIGRef from data.spot import Spot class APILookupCallHandler(tornado.web.RequestHandler): """API request handler for /api/v1/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.keys(): call = str(query_params.get("call")).upper() if re.match(r"^[A-Z0-9/\-]*$", call): # Take the callsign, make a "fake spot" so we can run infer_missing() on it, then repack the # resulting data in the correct way for the API response. credentials = extract_credentials(query_params) fake_spot = Spot(dx_call=call) fake_spot.infer_missing(credentials) data = { "call": call, "name": fake_spot.dx_name, "qth": fake_spot.dx_qth, "country": fake_spot.dx_country, "flag": fake_spot.dx_flag, "continent": fake_spot.dx_continent, "dxcc_id": fake_spot.dx_dxcc_id, "cq_zone": fake_spot.dx_cq_zone, "itu_zone": fake_spot.dx_itu_zone, "grid": fake_spot.dx_grid, "latitude": fake_spot.dx_latitude, "longitude": fake_spot.dx_longitude, "location_source": fake_spot.dx_location_source } self.write(safe_json_dumps(data)) else: self.write(safe_json_dumps("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 as e: logging.error("Exception when handling client request to call lookup API: %s", e, exc_info=True) 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/v1/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.keys() and "id" in query_params.keys(): sig = str(query_params.get("sig")).upper() ref_id = str(query_params.get("id")).upper() if sig in list(map(lambda p: p.name.upper(), SIGS)): if not get_ref_regex_for_sig(sig) or re.match(get_ref_regex_for_sig(sig), ref_id): data = populate_sig_ref_info(SIGRef(id=ref_id, sig=sig)) self.write(safe_json_dumps(data)) else: self.write(safe_json_dumps( "Error - '" + ref_id + "' does not look like a valid reference ID for " + sig + ".")) self.set_status(422) else: self.write(safe_json_dumps("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 as e: logging.error("Exception when handling client request to sig ref lookup API: %s", e, exc_info=True) 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/v1/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.keys(): 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 as e: logging.error("Exception when handling client request to grid ref lookup API: %s", e, exc_info=True) 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")