# Mattermost Bot module. # Copyright (c) 2016-2022 by Someone (aka. Jan Vales ) # published under MIT-License import json import pprint from inspect import cleandoc # pylint: disable=wrong-import-position from AbstractCommand import AbstractCommand class CommandFSInfAccess(AbstractCommand): TRIGGER = "fsinf-access" CONFIG = {"display_name": "somebot-command", "auto_complete": True, "auto_complete_hint": "'out-out'|'opt-in' []|'sync'", } CONFIG["auto_complete_desc"] = CONFIG["description"] = AbstractCommand.ICON_PUBLIC+"Manage fsinf-channel auto-joins. [FSINF]" def __init__(self, team_id, managed_teams, fsinf_intern_channelid, fsinf_teams_and_channels, datadir): super().__init__(team_id) self.managed_teams = managed_teams self.fsinf_intern_channelid = fsinf_intern_channelid self.fsinf_teams_and_channels = {x:{y:[] for y in fsinf_teams_and_channels[x]} for x in fsinf_teams_and_channels} self.datadir = datadir def on_POST(self, request, data): self._require_in_channel(self.fsinf_intern_channelid, data) # will throw an exception if not. (Dont try-except: Its handled up the stack.) # load and merge data on first use. self.fsinf_teams_and_channels = self._load_and_merge_optout_data(self.datadir+"./optout.json", self.fsinf_teams_and_channels) msg_text = data['text'].strip().split(" ", 1) if msg_text[0].strip() == 'opt-out': if data['team_id'] in self.fsinf_teams_and_channels and data['channel_id'] in self.fsinf_teams_and_channels[data['team_id']]: self.fsinf_teams_and_channels[data['team_id']][data['channel_id']] = sorted(list(set(self.fsinf_teams_and_channels[data['team_id']][data['channel_id']] + [data['user_id']]))) self._store_optout_data(self.datadir+"./optout.json", self.fsinf_teams_and_channels) request.respond_cmd_temp("## :white_check_mark: Success! :)\n#### Now leave this channel manually.") return else: request.respond_cmd_err("``/"+self.TRIGGER+"`` Trying to opt-out from an unmanaged channel.") return elif msg_text[0].strip() == 'opt-in' and len(msg_text) == 1: optout_list = [] for team, to_unpack in self.fsinf_teams_and_channels.items(): for channel, excluded in to_unpack.items(): if data['user_id'] in self.fsinf_teams_and_channels[team][channel]: optout_list += [channel] if optout_list: request.respond_cmd_temp("#### You have opted-out from these channels:\n + "+("\n + ".join([cid+" ("+self.bot.api.get_channel(cid)["name"]+")" for cid in optout_list]))) return else: request.respond_cmd_temp("#### You have not opted-out from any channels.") return elif msg_text[0].strip() == 'opt-in' and len(msg_text) > 1: managed = False chan_id = msg_text[1].strip() for team, to_unpack in self.fsinf_teams_and_channels.items(): if chan_id in self.fsinf_teams_and_channels[team]: managed = True if data['user_id'] in self.fsinf_teams_and_channels[team][chan_id]: newset = set(self.fsinf_teams_and_channels[team][chan_id]) newset.remove(data['user_id']) self.fsinf_teams_and_channels[team][chan_id] = sorted(list(newset)) self._store_optout_data(self.datadir+"./optout.json", self.fsinf_teams_and_channels) else: request.respond_cmd_err("``/"+self.TRIGGER+"`` You have not opted-out from this channel.") return if managed: request.respond_cmd_temp("## :white_check_mark: Success! :)\n#### Run ``/"+self.TRIGGER+" sync`` to apply the changes.") return else: request.respond_cmd_err("``/"+self.TRIGGER+"`` Trying to opt-in to an unmanaged or unknown channel.") return elif msg_text[0].strip() == 'sync': fsinf_intern_userids = set([u["user_id"] for u in self.bot.api.get_channel_members(self.fsinf_intern_channelid)]) all_users = {u["id"]:u["username"] for u in self.bot.api.get_users()} request.respond_cmd_temp("## :white_check_mark: Success! :)\n#### Sync started. See ~debug for details.") dbg_post = self.bot.debug_chan(cleandoc(""" ## ``/fsinf-access sync`` This will add people to various channels unless they are opted out already: + run **``/fsinf-access opt-out``** in the channel to opt out from + run **``/fsinf-access opt-in``** to list channels, you are opted out of. + run **``/fsinf-access opt-in ``** to opt-in again, then run **``/fsinf-access sync``** to apply this change. These accounts were detected to be fsinf-intern by being in ~off-topic: ``"""+str(sorted([all_users[u] for u in fsinf_intern_userids]))+"""`` Adding active fsinf accounts to these teams (and granting TEAM_ADMIN permissions) and channels unless opted-out (run ``/fsinf-access opt-out` in the channel to opt out from): ``` """)+"\n"+pprint.pformat({self.bot.api.get_team(tid)["name"]:{self.bot.api.get_channel(cid)["name"]:uids for cid,uids in self.fsinf_teams_and_channels[tid].items()} for tid in self.fsinf_teams_and_channels})+"\n"+cleandoc(""" ``` Applying now ... """)) # join into teams + make team admins for user_id in fsinf_intern_userids: for _, t in self.managed_teams.items(): self.bot.api.add_user_to_team(t[0], user_id, exc=False) self.bot.api.update_team_members_scheme_roles(t[0], user_id, {"scheme_user": True, "scheme_admin": True}, exc=False) # join into special channels for team_id, chan_ids in self.fsinf_teams_and_channels.items(): for chan_id, excluded_user_ids in chan_ids.items(): if not user_id in excluded_user_ids: self.bot.api.add_user_to_channel(chan_id, user_id, exc=False) self.bot.debug_chan("## :white_check_mark: Success! :)", root_id=dbg_post['id']) def _load_and_merge_optout_data(self, path, to_merge_with): try: merged = to_merge_with.copy() with open(path, "r") as f: stored_data = json.load(f) for team, to_unpack in to_merge_with.items(): for channel, excluded in to_unpack.items(): if team in stored_data and channel in stored_data[team]: merged[team][channel] = sorted(list(set(merged[team][channel]+stored_data[team][channel]))) return merged except: return to_merge_with def _store_optout_data(self, path, data): with open(path, "w") as f: json.dump(data, f, sort_keys=True, indent=2) self.bot.debug_chan("``/fsinf-access`` optout.json updated.")