Alerting integration

.gitignore

@@ -95,6 +95,9 @@ ENV/
 # database file
 *.db
 
+# Environment variables file
+.env
+
 # Spyder project settings
 .spyderproject
 

config.cfg

@@ -25,3 +25,24 @@ DATABASE=sqlite:////<path to your project>/dashboard.db
 TIMEZONE=Europe/Amsterdam
 COLORS={'main':'[0,97,255]',
         'static':'[255,153,0]'}
+
+[alerting]
+ENABLED=False
+TYPE=email
+# TYPE=email,issue,chat
+
+# email
+SMTP_HOST=<example_host>
+SMTP_PORT=<example_port>
+SMTP_USER=<example_user>
+SMTP_PASSWORD=<example_password>
+SMTP_TO=<example_email1,example_email2,example_email3>
+
+# issue
+GITHUB_TOKEN=<example_github_token>
+REPOSITORY_OWNER=<example_repo_owner>
+REPOSITORY_NAME=<example_repo_name>
+
+# chat
+CHAT_PLATFORM=TEAMS
+CHAT_WEBHOOK_URL=<example_url>

flask_monitoringdashboard/core/alert/README.md

@@ -0,0 +1,13 @@
+# Alerts
+
+## Issue Creation
+
+At the moment the secrets and variables for the issue are saved in a .env file
+
+There is a template with file name `.env_template` that can be used to fill in the info needed for the issue creation alert to work.
+
+
+
+
+
+

flask_monitoringdashboard/core/alert/alert_content.py

@@ -0,0 +1,60 @@
+import importlib.resources
+import os
+import traceback
+from datetime import datetime
+
+from jinja2 import Environment, FileSystemLoader
+
+import flask_monitoringdashboard
+from flask_monitoringdashboard.core.config import Config
+
+template_root = os.path.join(importlib.resources.files(flask_monitoringdashboard).__str__(), 'templates', 'alert')
+template_env = Environment(loader=FileSystemLoader(template_root))
+template = template_env.get_template('report.html')
+
+
+class AlertContent:
+
+    def __init__(self, exception: BaseException, config: Config):
+        self._exception = exception
+
+        self.created_at = datetime.now(config.timezone)
+        self.created_at_str = self.created_at.strftime('%Y-%m-%d %H:%M:%S')
+        self.send_at = None  # When was the alert sent, None if not sent yet
+
+        self.exception_type = exception.__class__.__name__
+
+        self.title = self._create_title()
+        self.stack_trace = ''.join(traceback.format_exception(type(exception), exception, exception.__traceback__))
+
+    def _create_title(self) -> str:
+        return f"[{self.exception_type}] Uncaught exception at {self.created_at_str}"
+
+    def get_limited_stack_trace(self, char_limit: int | None) -> str:
+        if not char_limit:
+            return self.stack_trace
+        return self.stack_trace[:char_limit] + ('...' if len(self.stack_trace) > char_limit else '')
+
+    def create_body_text(self, char_limit: int | None) -> str:
+        return f"An exception of type {self.exception_type} \n\nStack Trace: {self.get_limited_stack_trace(char_limit)}"
+
+    def create_body_markdown(self, char_limit: int | None) -> str:
+        return (
+            f"**Type:** `{self.exception_type}`\n"
+            f"**Timestamp:** `{self.created_at_str}`\n"
+            f"**Stack Trace:**\n```\n{self.get_limited_stack_trace(char_limit)}\n```"
+        )
+
+    def create_body_mrkdwn(self, char_limit: int | None) -> str:
+        return (
+            f"*Type:* `{self.exception_type}`\n"
+            f"*Timestamp:* `{self.created_at_str}`\n"
+            f"*Stack Trace:*\n```\n{self.get_limited_stack_trace(char_limit)}\n```"
+        )
+
+    def create_body_html(self, char_limit: int | None) -> str:
+        return template.render(
+            exc_type=self.exception_type,
+            timestamp=self.created_at_str,
+            stacktrace=self.get_limited_stack_trace(char_limit)
+        )

flask_monitoringdashboard/core/alert/chat.py

@@ -0,0 +1,112 @@
+import requests
+
+from flask_monitoringdashboard.core.alert.alert_content import AlertContent
+
+SLACK_CHAR_LIMIT = 2750
+ROCKET_CHAT_CHAR_LIMIT = 4500
+TEAMS_CHAR_LIMIT = 5000
+
+
+def send_message(alert_content: AlertContent):
+    from flask_monitoringdashboard import config
+
+    payload_creators = {
+        "SLACK": create_slack_payload,
+        "ROCKET_CHAT": create_rocket_chat_payload,
+        "TEAMS": create_teams_payload,
+    }
+
+    creator = payload_creators.get(config.chat_platform)
+    if not creator:
+        print("Invalid chat platform.")
+        return
+
+    payload = creator(alert_content)
+
+    try:
+        resp = requests.post(config.chat_webhook_url, json=payload, timeout=5)
+        resp.raise_for_status()
+    except Exception as e:
+        print("Alert delivery failed:", e)
+
+
+def create_slack_payload(alert_content: AlertContent):
+    return {
+        "blocks": [
+            {
+                "type": "section",
+                "text": {"type": "mrkdwn", "text": alert_content.create_body_mrkdwn(SLACK_CHAR_LIMIT)}
+            }
+        ]
+    }
+
+
+def create_rocket_chat_payload(alert_content: AlertContent):
+    return {
+        "text": alert_content.create_body_markdown(ROCKET_CHAT_CHAR_LIMIT),
+    }
+
+
+def create_teams_payload(alert_content: AlertContent):
+    return {
+        "type": "message",
+        "attachments": [
+            {
+                "contentType": "application/vnd.microsoft.card.adaptive",
+                "content": {
+                    "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
+                    "type": "AdaptiveCard",
+                    "version": "1.4",
+                    "body": [
+                        {
+                            "type": "TextBlock",
+                            "text": "Exception Alert",
+                            "weight": "Bolder",
+                            "size": "Large",
+                            "color": "Attention"
+                        },
+                        {
+                            "type": "FactSet",
+                            "facts": [
+                                {
+                                    "title": "Type:",
+                                    "value": f"{alert_content.exception_type}"
+                                },
+                                {
+                                    "title": "Timestamp:",
+                                    "value": f"{alert_content.created_at_str}"
+                                }
+                            ]
+                        },
+                        {
+                            "type": "TextBlock",
+                            "text": "Stack Trace:",
+                            "weight": "Bolder",
+                            "wrap": True
+                        },
+                        {
+                            "type": "Container",
+                            "id": "stackContainer",
+                            "isVisible": True,
+                            "items": [
+                                {
+                                    "type": "CodeBlock",
+                                    "codeSnippet": f"{alert_content.get_limited_stack_trace(TEAMS_CHAR_LIMIT)}",
+                                    "language": "bash",
+                                    "wrap": True,
+                                    "fontType": "Monospace",
+                                    "targetWidth": "Standard",
+                                    "fallback": {
+                                        "type": "TextBlock",
+                                        "text": f"{alert_content.get_limited_stack_trace(TEAMS_CHAR_LIMIT)}",
+                                        "wrap": True,
+                                        "fontType": "Monospace"
+                                    }
+                                }
+                            ]
+                        }
+                    ]
+                }
+            }
+        ]
+    }

flask_monitoringdashboard/core/alert/email.py

@@ -0,0 +1,29 @@
+import smtplib
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
+
+from flask_monitoringdashboard.core.alert.alert_content import AlertContent
+
+
+def send_email(alert_content: AlertContent):
+    from flask_monitoringdashboard import config
+
+    message = MIMEMultipart('alternative')
+    message.set_charset('utf-8')
+    message['Subject'] = alert_content.title
+    message['From'] = config.smtp_user
+    message['To'] = ', '.join(config.smtp_to)
+
+    message.attach(MIMEText(alert_content.create_body_text(None), 'plain', 'utf-8'))
+    message.attach(MIMEText(alert_content.create_body_html(None), 'html', 'utf-8'))
+
+    try:
+        with smtplib.SMTP(config.smtp_host, int(config.smtp_port)) as smtp:
+            smtp.starttls()
+
+            if config.smtp_password:
+                smtp.login(config.smtp_user, config.smtp_password)
+
+            smtp.sendmail(config.smtp_user, config.smtp_to, message.as_string())
+    except Exception as e:
+        print("Error sending email alert:", e)

flask_monitoringdashboard/core/alert/github_request_info.py

@@ -0,0 +1,16 @@
+class GitHubRequestInfo:
+    """
+    RequestInfo is a class that represents the information of a request.
+    """
+
+    def __init__(self, github_token: str, repo_owner: str, repo_name: str):
+        """
+        Initializes the RequestInfo object with the given parameters.
+
+        :param github_token: the PAT token that has access to the repository
+        :param repo_owner: The owner/organisation that contains the repository
+        :param repo_name: The name of the repository
+        """
+        self.repo_owner = repo_owner
+        self.repo_name = repo_name
+        self.github_token = github_token

flask_monitoringdashboard/core/alert/issue.py

@@ -0,0 +1,53 @@
+import requests
+
+from flask_monitoringdashboard.core.alert.alert_content import AlertContent
+from .github_request_info import GitHubRequestInfo
+
+GITHUB_CHAR_LIMIT = 60000
+
+
+def get_base_repo_url(repo_owner: str, repo_name: str):
+    return f"https://api.github.com/repos/{repo_owner}/{repo_name}/"
+
+
+def get_issue_url(repo_owner: str, repo_name: str):
+    return get_base_repo_url(repo_owner, repo_name) + "issues"
+
+
+def get_endpoint_url(repo_owner: str, repo_name: str, endpoint: str):
+    return f"https://api.github.com/repos/{repo_owner}/{repo_name}/{endpoint}"
+
+
+def make_post_request(request_info: GitHubRequestInfo, endpoint: str, data):
+    url = get_endpoint_url(request_info.repo_owner, request_info.repo_name, endpoint)
+    headers = _post_headers(request_info.github_token)
+
+    return requests.post(url, headers=headers, json=data)
+
+
+def create_issue(
+        request_info: GitHubRequestInfo,
+        alert_content: AlertContent) -> requests.Response:
+    data = {
+        "title": alert_content.title,
+        "body": alert_content.create_body_markdown(GITHUB_CHAR_LIMIT),
+        "labels": ["automated-issue", "exception"],
+    }
+
+    return make_post_request(request_info, "issues", data)
+
+
+def _post_headers(github_token: str):
+    headers = {
+        "Authorization": f"Bearer {github_token}",
+        "Accept": "application/vnd.github.v3+json"
+    }
+    return headers
+
+
+def main():
+    print("This is a utility file with helper functions not to be run directly.")
+
+
+if __name__ == "__main__":
+    main()