codimd-git-sync/src/mattermost.py

183 lines
5.2 KiB
Python
Raw Normal View History

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import pprint as pp
from abc import ABC, abstractmethod
from datetime import datetime
from enum import Enum
from time import sleep
from typing import Dict, List
import toml
from colored import style
from mattermostdriver import Driver
from tabulate import tabulate
from mattermost_objects import MMUser, MMPost, MMChannelPosts
from utils import humanize_date_difference, timer
pp = pp.PrettyPrinter(indent=2)
class LogLevel(Enum):
INFO = "INFO"
ERROR = "ERROR"
class User(ABC):
@abstractmethod
def credentials_dict(self) -> dict:
pass
class NormalUser(User):
def __init__(self, login_id, password):
self.login_id = login_id
self.password = password
def credentials_dict(self) -> dict:
return {"login_id": self.login_id, "password": self.password}
def __repr__(self):
return "User<name: {}, password: ******>".format(self.login_id)
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():
with open("users.toml") as f:
usersstring = f.read()
usersdict = toml.loads(usersstring)
usr = None
for name, data in usersdict.items():
if "token" in data:
usr = TokenUser(token=data["token"])
elif "name" in data and "password" in data:
usr = NormalUser(login_id=data["name"], password=data["password"])
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["flynn"]):
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(f" = 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, user=None):
MMApi.__init__(self, user)
self.channel_id = self.get_channel_id(channel_name)
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}"')
# 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)