Add core app

core/lib/elastic.py

@@ -0,0 +1,32 @@
+from datetime import datetime
+
+from django.conf import settings
+from elasticsearch import Elasticsearch
+
+from core.util import logs
+
+log = logs.get_logger(__name__)
+
+client = None
+
+
+def initialise_elasticsearch():
+    """
+    Initialise the Elasticsearch client.
+    """
+    auth = (settings.ELASTICSEARCH_USERNAME, settings.ELASTICSEARCH_PASSWORD)
+    client = Elasticsearch(
+        settings.ELASTICSEARCH_HOST, http_auth=auth, verify_certs=False
+    )
+    return client
+
+
+def store_msg(index, msg):
+    global client
+    if not client:
+        client = initialise_elasticsearch()
+    if "ts" not in msg:
+        msg["ts"] = datetime.utcnow().isoformat()
+    result = client.index(index=index, body=msg)
+    if not result["result"] == "created":
+        log.error(f"Indexing of '{msg}' failed: {result}")

core/lib/notify.py

@@ -0,0 +1,38 @@
+import requests
+
+from core.util import logs
+
+NTFY_URL = "https://ntfy.sh"
+
+log = logs.get_logger(__name__)
+
+
+# Actual function to send a message to a topic
+def raw_sendmsg(msg, title=None, priority=None, tags=None, url=None, topic=None):
+    if url is None:
+        url = NTFY_URL
+    headers = {"Title": "Pluto"}
+    if title:
+        headers["Title"] = title
+    if priority:
+        headers["Priority"] = priority
+    if tags:
+        headers["Tags"] = tags
+    requests.post(
+        f"{url}/{topic}",
+        data=msg,
+        headers=headers,
+    )
+
+
+# Sendmsg helper to send a message to a user's notification settings
+def sendmsg(user, *args, **kwargs):
+    notification_settings = user.get_notification_settings()
+
+    if notification_settings.ntfy_topic is None:
+        # No topic set, so don't send
+        return
+    else:
+        topic = notification_settings.ntfy_topic
+
+    raw_sendmsg(*args, **kwargs, url=notification_settings.ntfy_url, topic=topic)