mirror of
https://git.ianrenton.com/ian/spothole.git
synced 2026-02-04 01:04:33 +00:00
58 lines
2.7 KiB
Python
58 lines
2.7 KiB
Python
import tornado
|
|
|
|
# API request handler for /api/v1/alerts
|
|
class APIAlertsHandler(tornado.web.RequestHandler):
|
|
def get(self):
|
|
# todo
|
|
self.write("Hello, world")
|
|
|
|
# API request handler for /api/v1/alerts/stream
|
|
class APIAlertsStreamHandler(tornado.web.RequestHandler):
|
|
def get(self):
|
|
# todo
|
|
self.write("Hello, world")
|
|
|
|
|
|
|
|
# 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.
|
|
def alert_allowed_by_query(alert, query):
|
|
for k in query.keys():
|
|
match k:
|
|
case "received_since":
|
|
since = datetime.fromtimestamp(int(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, it also always passes the check.
|
|
if alert.is_dxpedition and (bool(query.get(
|
|
"dxpeditions_skip_max_duration_check")) if "dxpeditions_skip_max_duration_check" in query.keys() else False):
|
|
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 sigs is provided, the alert must have a sig and it must match one of them.
|
|
# The special "sig" "NO_SIG", when supplied in the list, mathches alerts with no sig.
|
|
sigs = query.get(k).split(",")
|
|
include_no_sig = "NO_SIG" in sigs
|
|
if not alert.sig and not include_no_sig:
|
|
return False
|
|
if alert.sig and alert.sig not in sigs:
|
|
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
|
|
return True
|