182 lines
5.2 KiB
Python
182 lines
5.2 KiB
Python
#!/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 MattermostObjects 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"]):
|
|
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, since)
|
|
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)
|