commit ccb3d06be599a350144e5f0ccd2ad6fde7691438 Author: Midgard Date: Sun Mar 14 01:55:32 2021 +0100 Initial commit diff --git a/mmcli.py b/mmcli.py new file mode 100755 index 0000000..d3c219f --- /dev/null +++ b/mmcli.py @@ -0,0 +1,230 @@ +#!/usr/bin/env python3 + +import sys +import argparse +import os +from collections import defaultdict +import datetime +from time import sleep +import time +import json +from typing import Dict, Set, Generator, Optional, NamedTuple +import mattermost +import mattermost.ws +import re + + +class NotFound(Exception): + def __init__(self, type_: str, name: str): + super().__init__(f"{type_} {name} not found") + self.type = type_ + self.name = name + + +def first(iterable, default=None): + for x in iterable: + return x + return default + + +def yes_no(x): + return "yes" if x else "no" + + +def get_posts_for_channel(self, channel_id: str, **kwargs) -> Generator[Dict, None, None]: + """ + Generator: Get a page of posts in a channel. Use the query parameters to modify the behaviour of this endpoint. + + @param channel_id: The channel ID to iterate over. + + Raises: + ApiException: Passed on from lower layers. + """ + page = 0 + while True: + data_page = self._get(f"/v4/channels/{channel_id}/posts", params={"page":str(page), "per_page":200, **kwargs}) + + if data_page["order"] == []: + break + page += 1 + + for order in data_page["order"]: + yield data_page["posts"][order] + + sleep(0.1) + + + +ID_PREFIX = "id:" + +def predicate_for_query(query: str): + """ + @return: a function that returns whether `query` matches its argument + """ + if query.startswith(ID_PREFIX): + id_ = query[len(ID_PREFIX):] + return lambda x: x["id"] == id_ + else: + return lambda x: x["name"] == query + + +def resolve_team(mm_api: mattermost.MMApi, query: str) -> Optional[Dict]: + return first(filter( + predicate_for_query(query), + mm_api.get_teams() + )) + + +def resolve_channel(mm_api: mattermost.MMApi, team_id: str, query: str) -> Optional[Dict]: + return first(filter( + predicate_for_query(query), + mm_api.get_team_channels(team_id) + )) + + +def resolve_team_channel(mm_api: mattermost.MMApi, query: str) -> Dict: + query_parts = query.split("/") + del query + if len(query_parts) != 2: + raise ValueError("Team/channel ID should be '/'") + + team = resolve_team(mm_api, query_parts[0]) + if not team: + raise NotFound("team", query_parts[0]) + + channel = resolve_channel(mm_api, team["id"], query_parts[1]) + if not channel: + return NotFound("channel", query_parts[1]) + + return team, channel + + +def login(mm_api, parsed): + print( + f"Logging in as {parsed.user}; password provided: {yes_no(parsed.password)}; " + f"TOTP token provided: {yes_no(parsed.totp)}", + file=sys.stderr) + mm_api.login(parsed.user, parsed.password, parsed.totp) + return mm_api._bearer + + +def cat(mm_api: mattermost.MMApi, parsed): + + + # FIXME Wrong order + + + channels = [ + resolve_team_channel(mm_api, query) + for query in parsed.channels + ] + + users = list(mm_api.get_users()) + + for team, channel in channels: + if not parsed.ids: + def attribute(key_value): + key, value = key_value + if key == "channel_id": + assert value == channel["id"] + return "channel", channel["name"] + if key == "user_id": + return "username", first(u["username"] for u in users if u["id"] == value) + return key_value + else: + def attribute(key_value): + return key_value + + for post in get_posts_for_channel(mm_api, channel["id"], after=parsed.after): + print(post_str(attribute, post, parsed)) + print(parsed.after) + + +def post_str(attribute, post, parsed): + obj = { + k: v + for k, v in map(attribute, post.items()) + if (v or k == "message") and (k != "update_at" or post["update_at"] != post["create_at"]) + } + + if parsed.format == "json": + return json.dumps(obj) + if parsed.format == "tsv": + msg = obj.get("message", "").replace("\\", "\\\\").replace("\t", r"\t").replace("\n", r"\n") + return f"{obj['id']}\t{obj['create_at']}\t{obj.get('username') or obj['user_id']}\t{msg}" + + +ACTIONS = { + "login": {"function": login, "accesstoken_required": False}, + "cat": {"function": cat}, +} + +FORMATTERS = { "json", "tsv", "csv" } + +ENVVAR_SERVER = "MM_SERVER" +ENVVAR_USERNAME = "MM_USERNAME" +ENVVAR_PASSWORD = "MM_PASSWORD" +ENVVAR_TOTP = "MM_TOTP" +ENVVAR_ACCESSTOKEN = "MM_ACCESSTOKEN" + +def main(): + prog_name = os.path.basename(sys.argv[0]) + description = "Interact with Mattermost on the CLI" + epilog = f""" +For further help, use `{prog_name} -h`. + +Where a "URL name" is required, "id:" plus an ID can also be used instead. So these could both be valid: + town-square + id:123abc456def789ghi012jkl34 + +Hint: JSON output can be filtered on the command line with jq(1). + """.strip() + + argparser = argparse.ArgumentParser( + prog_name, description=description, epilog=epilog, + formatter_class=argparse.RawTextHelpFormatter + ) + argparser.add_argument("-i", "--ids", help="use IDs instead of names", action="store_true") + argparser.add_argument( + "--format", help="output format; only json has all fields; default: json", choices=FORMATTERS, default="json") + + argparser.add_argument( + "--server", + help="e.g.: mattermost.example.org; example.org/mattermost; envvar: {ENVVAR_SERVER}", + default=os.getenv(ENVVAR_SERVER)) + + subparsers = argparser.add_subparsers(title="actions", dest="action", required=True) + + parser_login = subparsers.add_parser("login", help="retrieve an access token") + parser_login.add_argument("login_id", help="username or email", default=os.getenv(ENVVAR_USERNAME)) + parser_login.add_argument("--password", default=os.getenv(ENVVAR_PASSWORD)) + parser_login.add_argument("--totp", default=os.getenv(ENVVAR_TOTP)) + + parser_cat = subparsers.add_parser("cat", help="list messages in channel(s)") + parser_cat.add_argument( + "channels", nargs="+", help="URL names of team and channel: '/'") + parser_cat.add_argument("--after", help="all after post with ID") + parser_cat.add_argument("--since", help="all after timestamp") + parser_cat.add_argument("-f", "--follow", help="keep running, printing new posts as they come in") + + parsed = argparser.parse_args() + + if not parsed.server: + argparser.error( + f"server is required; use argument --server or environment variable {ENVVAR_SERVER}") + + access_token = os.getenv(ENVVAR_ACCESSTOKEN) + if ACTIONS[parsed.action].get("accesstoken_required", True) and not access_token: + argparser.error( + f"`{prog_name} {parsed.action}` requires access token; get one with `{prog_name} login` " + f"and set environment variable {ENVVAR_ACCESSTOKEN}") + + mm_api = mattermost.MMApi(f"https://{parsed.server}/api") + if access_token: + mm_api._headers.update({"Authorization": f"Bearer {access_token}"}) + + ACTIONS[parsed.action]["function"](mm_api, parsed) + + +if __name__ == "__main__": + main()