chore: startup mattermost_comunication

config.py

@@ -0,0 +1,8 @@
+import tomllib
+from pprint import pprint
+
+with open("config.toml", mode="rb") as config_toml:
+    config = tomllib.load(config_toml)
+
+
+# pprint(config)

mattermost_client.py

@@ -0,0 +1,178 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+import pprint as pp
+from abc import ABC, abstractmethod
+from enum import Enum
+from typing import Dict
+from config import config
+
+from colored import Style
+from mattermostdriver import Driver
+
+from mattermost_objects import MMChannelPosts
+from utils import timer
+
+pp = pp.PrettyPrinter(indent=2)
+
+
+class LogLevel(Enum):
+    INFO = "INFO"
+    ERROR = "ERROR"
+
+
+class User(ABC):
+    @abstractmethod
+    def credentials_dict(self) -> dict:
+        pass
+
+class TokenUser(User):
+    def __init__(self, token):
+        self.token = token
+
+    def credentials_dict(self) -> dict:
+        return {"token": self.token}
+
+    def __repr__(self):
+        return "TokenUser<token: {}>".format(self.token)
+
+
+users: {str: [User]} = {}
+
+
+def loadusers():
+    usr = None
+    for name, data in config["mattermost"]["users"].items():
+        if "token" in data:
+            usr = TokenUser(token=data["token"])
+        else:
+            print("Invalid user '{}' in toml file".format(name))
+            exit(1)
+        users[name] = usr
+
+
+loadusers()
+
+
+def merge_dict(a: dict, b: dict) -> dict:
+    return {**a, **b}
+
+
+class MMApi(Driver):
+    def __init__(self, user: User = users["tyboro"]):
+        print(f"Initializing MMApi client for user {user}")
+        Driver.__init__(
+            self,
+            merge_dict(
+                {
+                    "url": "mattermost.zeus.gent",
+                    "port": 443,
+                    "debug": False,
+                },
+                user.credentials_dict(),
+            ),
+        )
+        self.login()
+        self.user_id = self.users.get_user(user_id="me")["id"]
+        self.team_id = self.teams.get_team_by_name("zeus")["id"]
+        print(" = Creating mattermost client")
+        print(f" =  - User: {self.user_id}")
+        print(f" =  - Team: {self.team_id}")
+
+    @staticmethod
+    def print_response(resp, title="Response"):
+        print("--------")
+        print(Style.BOLD + title + Style.RESET)
+        pp.pprint(resp)
+
+    def log(self, text: str, log_level: LogLevel = LogLevel.INFO):
+        print(f"{Style.BOLD}[{log_level.value}]{Style.RESET} {text}")
+
+    def get_channel_id(self, channel_name):
+        resp = self.channels.get_channel_by_name(self.team_id, channel_name)
+        id = resp["id"]
+        self.log(f"Fetching channel id for {channel_name}: {id}")
+        return id
+
+    @timer
+    def get_posts_for_channel(self, channel_id, since):
+        print(f"Fetching posts for {channel_id} since {since}")
+        page_size = 200
+        page_i = 0
+        data = {}
+        more = True
+        while more:
+            resp = self.posts.get_posts_for_channel(
+                channel_id,
+                params={"page": page_i, "per_page": page_size, "since": since},
+            )
+            page_i += 1
+            print(f"Fetching page {page_i}")
+            # print("-", end=" ")
+
+            paged_data = resp["posts"]
+            paged_count = len(paged_data)
+
+            if since != 0:
+                # The mattermost api is absolutely retarted
+                # If you add the since parameter and it's different then 0 it will give you 1000 posts max.
+                # It will not respect you page_index or page_size.
+                more = False
+            else:
+                if paged_count < page_size:
+                    more = False
+
+            # Transform the data into something more sensible or practical
+            if type(paged_data) is list:
+                paged_data = {item["id"]: item for item in paged_data}
+
+            # Append the paged_data to our global data variable
+            data = {**data, **paged_data}
+        print()
+
+        self.log(f"Post count: {len(data)}")
+        return data
+
+
+class ChannelApi(MMApi):
+    def __init__(self, channel_name=None, channel_id=None, user=None):
+        MMApi.__init__(self, user)
+        assert channel_name is not None or channel_id != None
+
+        if channel_name is not None:
+            self.channel_id = self.get_channel_id(channel_name)
+        if channel_id is not None:
+            self.channel_id = channel_id
+
+    def create_post(self, message: str, props: Dict = None) -> None:
+        resp = self.posts.create_post(
+            options={"channel_id": self.channel_id, "message": message, "props": props}
+        )
+        self.log(f'Message successfully created: "{message}"')
+
+    def create_threaded_post(
+        self, post_id: str, message: str, props: Dict = None
+    ) -> None:
+        resp = self.posts.create_post(
+            options={
+                "channel_id": self.channel_id,
+                "message": message,
+                "root_id": post_id,
+                "props": props,
+            }
+        )
+        self.log(f'Message successfully created: "{message}"')
+        # print_response("Create post", resp)
+
+
+if __name__ == "__main__":
+    foo = MMApi(user=users["flynn"])
+
+    # all_posts = foo.get_all_posts()
+
+    channel = foo.channels.get_channel_by_name(
+        foo.team_id,
+        "bestuur",
+    )
+    channel_id = channel["id"]
+    resp = foo.posts.get_posts_for_channel(channel_id, params={"per_page": 200})
+    channel_posts: MMChannelPosts = MMChannelPosts.load(resp)

mattermost_comunication.py

@@ -0,0 +1,60 @@
+import mattermostdriver.exceptions
+
+import mattermost_client
+from config import config
+from mattermost_client import ChannelApi, MMApi
+
+
+def send_message(file_info, message):
+    channel_id = file_info["originating_mm_post_channel_id"]
+    post_id = file_info["originating_mm_post_id"]
+
+    # TODO Comment below line, this is for testing purposes
+    # channel_id = MMApi().get_channel_id("bestuur-dev")
+    channel = ChannelApi(
+        channel_id=channel_id,
+        user=mattermost_client.users[config["mattermost"]["selected_user"]],
+    )
+
+    try:
+        channel.create_threaded_post(
+            post_id,
+            f"{message}",
+        )
+    except mattermostdriver.exceptions.InvalidOrMissingParameters as e:
+        # This will occur when we try to react to a file in a channel that is not the same as the originating channel.
+        unique_post_url = f"{config['mattermost']['server_url']}/pl/{post_id}"
+        channel.create_post(
+            f"{unique_post_url}\n\n{message}",
+        )
+
+
+def report_newly_found_file(file_info):
+    git_url = f"https://{config['gitea']['server_url']}/{config['gitea']['remote_org']}/{config['gitea']['remote_repo']}"
+    message = f"I found a new CodiMD file in this post! Making work of putting it on git :)\n - Requested location in the [drive]({git_url}): {file_info['metadata']['sync-to']}"
+    send_message(file_info, message)
+
+
+def report_newly_found_but_invalid_file(file_info):
+    message = """Hi there! :wave: 
+I'm your friendly neighbourhood document sync bot.
+I could synchronize this CodiMD file automatically to our Git DRIVE for safekeeping, but the necessary metadata block is not present.
+You can easily add the correct info and I will do the rest of the work for you!
+
+Just add the following lines to your file, the location in your file is not important but at the top would be my recommendation.
+
+```
+:::spoiler git drive sync
+- sync-to: <a valid path on the DRIVE, for ex.: verslagen/21-22/2022-05-13.md>
+:::
+```"""
+    send_message(file_info, message)
+
+
+send_message(
+    {
+        "originating_mm_post_channel_id": "dm1abp4wfidezmig1yqyu53mmy",
+        "originating_mm_post_id": "dm1abp4wfidezmig1yqyu53mmy"
+    },
+    "this is a test message"
+)

mattermost_objects.py

@@ -0,0 +1,137 @@
+from typing import Dict, List, NamedTuple
+
+
+class MMUser(NamedTuple):
+    id: str
+    create_at: int
+    update_at: int
+    delete_at: int
+    username: str
+    first_name: str
+    last_name: str
+    nickname: str
+    email: str
+    auth_data: str
+    auth_service: str
+    roles: str
+    locale: str
+    timezone: dict
+    position: any
+
+    is_bot: bool = None
+    bot_description: str = None
+    email_verified: bool = None
+    notify_props: dict = None
+    last_password_update: int = None
+    failed_attempts: int = None
+    mfa_active: bool = False
+    terms_of_service_id: str = None
+    terms_of_service_create_at: int = None
+    props: dict = {}
+    last_picture_update: int = None
+
+    @staticmethod
+    def load(data):
+        try:
+            return MMUser(**data)
+        except TypeError as e:
+            print("[ERROR] Could not load dict into MMUser namedtuple")
+            print(str(e))
+
+
+class MMPostProps(NamedTuple):
+    from_webhook: str = False
+    override_icon_url: str = None
+    override_username: str = None
+    webhook_display_name: str = None
+
+    channel_mentions: Dict = None
+    matterircd_krcggydky38kdcuubsc7fddc7w: str = None
+    matterircd_s4ptwhx7wfnx7qwexp1khorh7e: str = None
+    username: str = None
+    userId: str = None
+    old_header: str = None
+    new_header: str = None
+    old_purpose: str = None
+    new_purpose: str = None
+    old_displayname: str = None
+    new_displayname: str = None
+    remove_link_preview: str = None
+    removedUserId: str = None
+    addedUserId: str = None
+    removedUsername: str = None
+    addedUsername: str = None
+    message: str = None
+    attachments: str = None
+    from_bot: str = False
+    disable_group_highlight: str = None
+
+
+class MMPost(NamedTuple):
+    channel_id: str
+    create_at: int
+    delete_at: int
+    edit_at: int
+    hashtags: str
+    id: str
+    is_pinned: bool
+    message: str
+    metadata: Dict
+    original_id: str
+    pending_post_id: str
+    root_id: str
+    type: str
+    update_at: int
+    user_id: str
+    parent_id: str = None
+    message_source: str = None
+    has_reactions: bool = None
+    file_ids: List[str] = None
+    props: MMPostProps = None
+    reply_count: int = None
+    last_reply_at: str = None
+    participants: any = None
+
+    def from_human(self):
+        return self.props is None or (
+            self.props.from_webhook is False and self.props.from_bot is False
+        )
+
+    @staticmethod
+    def load(data):
+        try:
+            props = None
+            if "props" in data:
+                try:
+                    props: MMPostProps = MMPostProps(**data["props"])
+                except TypeError as e:
+                    print("[ERROR] Could not load dict into MMPostProps namedtuple")
+                    print(str(e))
+            del data["props"]
+            return MMPost(props=props, **data)
+        except TypeError as e:
+            print("[ERROR] Could not load dict into MMPost namedtuple")
+            print(str(e))
+
+
+class MMChannelPosts(NamedTuple):
+    prev_post_id: str
+    next_post_id: str
+    order: List[str]
+    posts: Dict[str, MMPost]
+    has_next: any
+    first_inaccessible_post_time: any
+    reply_count: any = None
+    disable_group_highlight: any = None
+
+    @staticmethod
+    def load(data):
+        try:
+            posts: Dict[str, MMPost] = {
+                k: MMPost.load(v) for (k, v) in data["posts"].items()
+            }
+            del data["posts"]
+            return MMChannelPosts(posts=posts, **data)
+        except TypeError as e:
+            print("[ERROR] Could not load dict into MMUser namedtuple")
+            print(str(e))

requirements.txt

@@ -6,4 +6,5 @@ PyPDF2==3.0.1
 pdfplumber==0.11.4
 requests==2.32.3
 selenium==4.27.1
-seleniumbase==4.33.11
+seleniumbase==4.33.11
+mattermostdriver