spothole/webserver/handlers/api/alerts.py

import copy
import logging
from datetime import datetime
from typing import Any

import pytz
import tornado
import tornado_eventsource.handler
from tornado import httputil
from tornado.web import Application

from core.utils import safe_json_dumps
from data.lookup_credentials import extract_credentials

logger = logging.getLogger(__name__)


class APIAlertsHandler(tornado.web.RequestHandler):
    """API request handler for /api/v2/alerts"""

    def __init__(
        self,
        application: "Application",
        request: httputil.HTTPServerRequest,
        **kwargs: Any,
    ):
        self._alerts = None
        super().__init__(application, request, **kwargs)

    def initialize(self, alerts):
        self._alerts = alerts

    @staticmethod
    def _enrich(alerts, credentials):
        enriched = []
        for alert in alerts:
            alert_copy = copy.deepcopy(alert)
            alert_copy.infer_missing(credentials)
            enriched.append(alert_copy)
        return enriched

    def get(self):
        try:
            # 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()}

            # Fetch all alerts matching the query, then optionally enrich with online data
            credentials = extract_credentials(self.request.headers)
            data = get_alert_list_with_filters(self._alerts, query_params)
            fields = [f.strip() for f in query_params["fields"].split(",")] if "fields" in query_params else []
            if credentials:
                data = self._enrich(data, credentials)
            # Filter for only the required fields, if necessary
            if fields:
                data = filter_fields(data, fields)
            self.write(safe_json_dumps(data))
            self.set_status(200)
        except ValueError as e:
            self.write(safe_json_dumps(f"Bad request - {e!s}"))
            self.set_status(400)
        except Exception:
            logger.exception("Exception when handling client request to alerts 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 APIAlertsStreamHandler(tornado_eventsource.handler.EventSourceHandler):
    """API request handler for /api/v2/alerts/stream"""

    def __init__(self, application, request, **kwargs: Any):
        self._sse_alert_broadcaster = None
        self._query_params = None
        self._credentials = None
        self._fields = None
        super().__init__(application, request, **kwargs)

    def initialize(self, sse_alert_broadcaster):
        self._sse_alert_broadcaster = sse_alert_broadcaster

    def custom_headers(self):
        """Custom headers to avoid e.g. nginx reverse proxy from buffering SSE data"""

        return {"Cache-Control": "no-store", "X-Accel-Buffering": "no"}

    def open(self):
        try:
            # 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
            self._query_params = {k: v[0].decode("utf-8") for k, v in self.request.arguments.items()}
            self._credentials = extract_credentials(self.request.headers)
            self._fields = (
                [f.strip() for f in self._query_params["fields"].split(",")] if "fields" in self._query_params else []
            )

            # Flush headers immediately so nginx doesn't time out waiting for a response
            self.write_message("keepalive", "")

            # Register to handle new alerts arriving. The callback() method will get called with the new alert as an
            # argument.
            self._sse_alert_broadcaster.register(self)

        except Exception:
            logger.exception("Exception when serving SSE socket")
            self.close()

    def close(self):
        """When the user closes the socket, deregister ourselves from the alert broadcaster"""

        self._sse_alert_broadcaster.unregister(self)
        super().close()

    def callback(self, alert):
        """Callback when a new alert arrives"""

        try:
            # If the new alert matches our param filters, send it to the client. If not, ignore it.
            if alert_allowed_by_query(alert, self._query_params):
                # Add lookup data if we have credentials
                if self._credentials:
                    alert = copy.deepcopy(alert)
                    alert.infer_missing(self._credentials)
                # Filter fields returned if necessary
                if self._fields:
                    alert = filter_fields([alert], self._fields)[0]
                self.write_message(msg=safe_json_dumps(alert))
        except Exception:
            logger.exception("Exception in SSE callback, connection will be closed")
            self.close()


def get_alert_list_with_filters(all_alerts, query):
    """Utility method to apply filters to the overall alert list and return only a subset. Enables query parameters in
    the main "alerts" GET call."""

    # Create a shallow copy of the alert list ordered by start time, then filter the list to reduce it only to alerts
    # that match the filter parameters in the query string. Finally, apply a limit to the number of alerts returned.
    # The list of query string filters is defined in the API docs.
    alert_ids = all_alerts.keys()
    alerts = []
    for k in alert_ids:
        a = all_alerts.get(k)
        if a is not None:
            alerts.append(a)
    alerts = sorted(alerts, key=lambda alert: alert.start_time if alert and alert.start_time else 0)
    alerts = list(filter(lambda alert: alert_allowed_by_query(alert, query), alerts))
    if "limit" in query:
        alerts = alerts[: int(query.get("limit"))]
    return alerts


def alert_allowed_by_query(alert, query):
    """Given URL query params and an alert, figure out if the alert "passes" the requested filters or is rejected. The list
    of query parameters and their function is defined in the API docs."""

    for k in query:
        match k:
            case "received_since":
                since = datetime.fromtimestamp(float(query.get(k)), pytz.UTC)
                if not alert.received_time or alert.received_time <= since:
                    return False
            case "max_duration":
                max_duration = int(query.get(k))
                # Check the duration if end_time is provided. If end_time is not provided, assume the activation is
                # "short", i.e. it always passes this check. If dxpeditions_skip_max_duration_check is true and
                # the alert is a dxpedition, or contests_skip_max_duration_check and the alert is a contest, it also
                # always passes the check.
                if (
                    alert.sig == "DXpedition"
                    and "dxpeditions_skip_max_duration_check" in query
                    and query.get("dxpeditions_skip_max_duration_check").upper() == "TRUE"
                ):
                    continue
                if (
                    alert.sig == "Contest"
                    and "contests_skip_max_duration_check" in query
                    and query.get("contests_skip_max_duration_check").upper() == "TRUE"
                ):
                    continue
                if alert.end_time and alert.start_time and alert.end_time - alert.start_time > max_duration:
                    return False
            case "source":
                sources = query.get(k).split(",")
                if not alert.source or alert.source not in sources:
                    return False
            case "sig":
                # If a list of activities is provided, the alert must have an activity and it must match one of them.
                # The special activity "NO_SIG", when supplied in the list, matches alerts with no activity.
                activities = query.get(k).split(",")
                include_no_activity = "NO_SIG" in activities
                if not alert.sig and not include_no_activity:
                    return False
                if alert.sig and alert.sig not in activities:
                    return False
            case "dx_continent":
                dxconts = query.get(k).split(",")
                if not alert.dx_continent or alert.dx_continent not in dxconts:
                    return False
            case "dx_call_includes":
                dx_call_includes = query.get(k).strip()
                if not alert.dx_call or dx_call_includes.upper() not in alert.dx_call.upper():
                    return False
            case "text_includes":
                text_includes = query.get(k).strip()
                if (
                    (not alert.dx_call or text_includes.upper() not in alert.dx_call.upper())
                    and (not alert.comment or text_includes.upper() not in alert.comment.upper())
                    and (not alert.freqs_modes or text_includes.upper() not in alert.freqs_modes.upper())
                ):
                    return False
    return True


def filter_fields(alerts, fields):
    """Given a list of alert objects, return copies containing only the named fields."""

    return [{k: v for k, v in alert.__dict__.items() if k in fields} for alert in alerts]