initial commit (STAN v3)

.github/dependabot.yml

@@ -0,0 +1,12 @@
+# To get started with Dependabot version updates, you'll need to specify which
+# package ecosystems to update and where the package manifests are located.
+# Please see the documentation for more information:
+# https://docs.github.com/github/administering-a-repository/configuration-options-for-dependency-updates
+# https://containers.dev/guide/dependabot
+
+version: 2
+updates:
+ - package-ecosystem: "devcontainers"
+   directory: "/"
+   schedule:
+     interval: weekly

.github/workflows/mypy.yml

@@ -0,0 +1,29 @@
+name: mypy
+
+on:
+  push:
+    paths:
+      ".github/workflows/mypy.yml"
+      "**/*.py"
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v2
+
+    - name: Set up Python 3.12
+      uses: actions/setup-python@v2
+      with:
+        python-version: 3.12
+
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install mypy
+        pip install -r requirements.txt
+
+    - name: Run mypy
+      run: |
+        python -m mypy .

.github/workflows/pylint.yml

@@ -0,0 +1,29 @@
+name: pylint
+
+on:
+  push:
+    paths:
+      ".github/workflows/pylint.yml"
+      "**/*.py"
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v4
+
+    - name: Set up Python 3.12
+      uses: actions/setup-python@v5
+      with:
+        python-version: 3.12
+
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install pylint
+        pip install -r requirements.txt
+
+    - name: Run pylint
+      run: |
+        python -m pylint *.py */ --rcfile=.pylintrc --recursive=y --disable=fixme

.gitignore

@@ -0,0 +1,4 @@
+__pycache__
+.ruff_cache
+.env
+token

.pylintrc

@@ -0,0 +1,2 @@
+[MESSAGES CONTROL]
+disable=too-few-public-methods # R0903

README.md

@@ -0,0 +1,3 @@
+# Stan Discord Bot
+
+This repository contains version 3 of the STAIR Discord bot, Stan.

common/singleton.py

@@ -0,0 +1,18 @@
+# -*- coding: utf-8 -*-
+"""Singleton metaclass"""
+
+__copyright__ = "Copyright (c) 2024 STAIR. All Rights Reserved."
+__email__ = "[email protected]"
+
+from typing import Any
+
+
+class Singleton(type):
+    """Metaclass for Singletons"""
+
+    _instances: dict[type, Any] = {}
+
+    def __call__(cls, *args: Any, **kwds: Any) -> Any:
+        if cls not in cls._instances:
+            cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwds)
+        return cls._instances[cls]

db/datamodels/hslu_student.py

@@ -0,0 +1,68 @@
+# -*- coding: utf-8 -*-
+"""Data model for a student at the HSLU."""
+
+from __future__ import annotations
+
+__copyright__ = "Copyright (c) 2024 STAIR. All Rights Reserved."
+__email__ = "[email protected]"
+
+from typing import Any
+from dataclasses import dataclass
+from enum import StrEnum
+import csv
+from pyaddict import JDict
+
+
+class StudyModel(StrEnum):
+    """The study model of the student"""
+
+    FULL_TIME = "Vollzeitstudium"
+    PART_TIME = "Teilzeitstudium"
+    UNKNOWN = "Unknown"
+
+    @classmethod
+    def get(cls, value: str) -> StudyModel:
+        """Get the study model from a string"""
+        try:
+            return cls(value)
+        except ValueError:
+            return cls.UNKNOWN
+
+
+@dataclass
+class HsluStudent:  # pylint: disable=too-many-instance-attributes
+    """Data model for a student at the HSLU, as provided by HSLU."""
+
+    id: int
+    first_name: str
+    last_name: str
+    course_id: str  # e.g., I.BSCAIML.2001
+    course_name: (
+        str  # e.g., I.Bachelor of Science in Artificial Intelligence & Machine Learning
+    )
+    status: str  # eg., jA.Immatrikuliert (whatever that's supposed to mean)
+    start_date: str  # e.g., 14.09.2020
+    study_model: StudyModel
+    email: str  # e.g., [email protected]
+
+    @classmethod
+    def _from_csv_dict(cls, value: dict[str, Any]) -> HsluStudent:
+        data = JDict(value)
+        return cls(
+            data.ensureCast("ID Person", int),
+            data.ensure("Vornamen", str),
+            data.ensure("Nachname", str),
+            data.ensure("Anlassnummer", str),
+            data.ensure("Anlassbezeichnung", str),
+            data.ensure("Status (Anmeldung)", str),
+            data.ensure("Eintritt per", str),
+            StudyModel.get(data.ensure("Ausbildungsform", str)),
+            data.ensure("email", str),
+        )
+
+    @classmethod
+    def from_csv(cls, plain: str) -> list[HsluStudent]:
+        """Create a list of students from a CSV document"""
+        plain = plain.replace("E-Mail", "email")
+        students = csv.DictReader(plain.splitlines())
+        return [cls._from_csv_dict(student) for student in students]

db/datamodels/verified_user.py

@@ -0,0 +1,28 @@
+# -*- coding: utf-8 -*-
+"""Data model for a verified user."""
+
+from __future__ import annotations
+
+__copyright__ = "Copyright (c) 2024 STAIR. All Rights Reserved."
+__email__ = "[email protected]"
+
+from enum import StrEnum
+from dataclasses import dataclass, field
+
+
+class UserState(StrEnum):
+    """The state of the user / server member"""
+
+    GUEST = "Unverified"
+    STUDENT = "Student"
+    GRADUATE = "Graduate"
+
+
+@dataclass
+class VerifiedUser:
+    """Data model for a verified user"""
+
+    discord_id: int
+    email: str
+    state: UserState
+    id: int = field(default=0)

db/db.py

@@ -0,0 +1,112 @@
+# -*- coding: utf-8 -*-
+"""Database module"""
+
+__copyright__ = "Copyright (c) 2024 STAIR. All Rights Reserved."
+__email__ = "[email protected]"
+
+import logging
+from dataclasses import asdict
+import dataset  # type: ignore
+from common.singleton import Singleton
+from .datamodels.hslu_student import HsluStudent
+from .datamodels.verified_user import VerifiedUser, UserState
+
+VERIFIED_USERS_TABLE = "Verified_Users"
+HSLU_STUDENTS_TABLE = "HSLU_Students"
+
+DB_USERNAME = "postgres"
+DB_PASSWORD = "postgres"
+DB_HOST = "localhost:5432"
+DB_PROTOCOL = "postgresql"
+
+
+class Database(metaclass=Singleton):
+    """Wraps the database using dataset & our custom data models"""
+
+    def __init__(self) -> None:
+        self._db = dataset.connect(
+            f"{DB_PROTOCOL}://{DB_USERNAME}:{DB_PASSWORD}@{DB_HOST}/"
+        )
+        self._logger = logging.getLogger("Database")
+
+        self._users_table: dataset.Table = self._db[VERIFIED_USERS_TABLE]
+        self._hslu_students_table: dataset.Table = self._db[HSLU_STUDENTS_TABLE]
+
+        # self._users_table.delete() # for development
+
+    def all_students(self) -> list[HsluStudent]:
+        """Get all students from the database."""
+        return [HsluStudent(**x) for x in self._hslu_students_table.all()]
+
+    def update_students(
+        self, students: list[HsluStudent]
+    ) -> tuple[list[VerifiedUser], list[VerifiedUser]]:
+        """
+        Update the students in the database. This will also update the state of
+        the users in the database.
+
+        Returns a list of new graduates.
+        """
+        current_members = self.all_verified()
+        previous_students = [x for x in current_members if x.state == UserState.STUDENT]
+        previous_graduates = [
+            x for x in current_members if x.state == UserState.GRADUATE
+        ]
+
+        self._hslu_students_table.drop()
+        # make unique
+        students = list({student.email: student for student in students}.values())
+        self._hslu_students_table.insert_many([asdict(student) for student in students])
+        # all that were previously students should now be graduates
+        new_graduates = []
+        new_students = []
+
+        for student in previous_students:
+            if any(x.email == student.email for x in students):
+                continue
+            self._logger.debug("%s is now a graduate", student)
+            self._users_table.update(
+                {"state": UserState.GRADUATE, "email": student.email},
+                ["email"],
+            )
+            new_graduates.append(student)
+
+        for student in previous_graduates:
+            if not any(x.email == student.email for x in students):
+                continue
+            self._logger.debug("%s is now a student again", student)
+            self._users_table.update(
+                {"state": UserState.STUDENT, "email": student.email},
+                ["email"],
+            )
+            new_students.append(student)
+
+        return new_graduates, new_students
+
+    def student_by_email(self, email: str) -> HsluStudent | None:
+        """Search for a student by email. Returns None if not found."""
+        result = self._hslu_students_table.find_one(email=email)
+        if result:
+            return HsluStudent(**result)
+        return None
+
+    def get_member(self, discord_id) -> VerifiedUser | None:
+        """Get a verified user by their Discord ID, returning None if not yet verified."""
+        result = self._users_table.find_one(discord_id=discord_id)
+        if result:
+            return VerifiedUser(**result)
+        return None
+
+    def all_verified(self) -> list[VerifiedUser]:
+        """Get all verified users from the database."""
+        return [VerifiedUser(**x) for x in self._users_table.all()]
+
+    def verify_member(self, discord_id: int, email: str) -> None:
+        """Verify a member by their Discord ID and email."""
+        self._users_table.insert(
+            asdict(
+                VerifiedUser(
+                    discord_id=discord_id, email=email, state=UserState.STUDENT
+                )
+            )
+        )

email_client.py

@@ -0,0 +1,64 @@
+# -*- coding: utf-8 -*-
+"""Email client module"""
+
+from __future__ import annotations
+
+__copyright__ = "Copyright (c) 2024 STAIR. All Rights Reserved."
+__email__ = "[email protected]"
+
+import os
+import logging
+from msal import PublicClientApplication  # type: ignore
+from dotenv import load_dotenv
+import aiohttp
+
+load_dotenv()
+
+AD_APP_ID = os.getenv("AD_APP_ID")
+EMAIL_ADDRESS = os.getenv("EMAIL_ADDRESS")
+EMAIL_NAME = os.getenv("EMAIL_NAME")
+SCOPES = ["Mail.Send"]
+
+
+class EmailClient:
+    """Client for sending emails via Microsoft Graph API."""
+
+    def __init__(self) -> None:
+        self._logger = logging.getLogger(__name__)
+        self._app = PublicClientApplication(
+            AD_APP_ID, authority="https://login.microsoftonline.com/common"
+        )
+        flow = self._app.initiate_device_flow(SCOPES)
+        self._logger.info(flow["message"])
+        self._app.acquire_token_by_device_flow(flow)
+        accounts = self._app.get_accounts()
+        assert len(accounts) > 0
+        self._account = accounts[0]
+
+    def _get_token(self) -> str:
+        result = self._app.acquire_token_silent(SCOPES, account=self._account)
+        return result["access_token"]
+
+    async def send_email(self, subject: str, content: str, to: str) -> bool:
+        """
+        Send an email
+
+        :param subject: The subject of the email (e.g. "STAIR Discord Verification")
+        :param content: The content of the email (e.g. "Your verification code is 123456")
+        :param to: The recipient's email address
+        """
+        body = {
+            "Message": {
+                "Subject": subject,
+                "Body": {"ContentType": "Text", "Content": content},
+                "ToRecipients": [{"EmailAddress": {"Address": to}}],
+            },
+            "SaveToSentItems": "true",
+        }
+        async with aiohttp.ClientSession() as session:
+            async with session.post(
+                "https://graph.microsoft.com/v1.0/me/sendMail",
+                headers={"Authorization": "Bearer " + self._get_token()},
+                json=body,
+            ) as r:
+                return r.status == 202