modules/TACommandSyncFSInf.py

   1 # Mattermost Bot module.
   2 #  Copyright (c) 2016-2022 by Someone <someone@somenet.org> (aka. Jan Vales <jan@jvales.net>)
   3 #  published under MIT-License
   4
   5 import json
   6 import pprint
   7 from inspect import cleandoc
   8
   9
  10 # pylint: disable=wrong-import-position
  11 from AbstractCommand import AbstractCommand
  12 class TACommandSyncFSInf(AbstractCommand):
  13     TRIGGER = "fsinf-access"
  14     CONFIG = {"display_name": "somebot-command", "auto_complete": True,
  15               "auto_complete_hint": "'out-out'|'opt-in' [<channel_id>]|'sync'",
  16              }
  17     CONFIG["auto_complete_desc"] = CONFIG["description"] = AbstractCommand.ICON_PUBLIC+"Manage fsinf-channel auto-joins. [FSINF]"
  18
  19
  20     def __init__(self, team_id, managed_teams, fsinf_intern_channelid, ignore_accounts_channelid, fsinf_teams_and_channels, datadir):
  21         super().__init__(team_id)
  22
  23         self.managed_teams = managed_teams
  24         self.fsinf_intern_channelid = fsinf_intern_channelid
  25         self.ignore_accounts_channelid = ignore_accounts_channelid
  26         self.fsinf_teams_and_channels = fsinf_teams_and_channels
  27         self.datadir = datadir
  28
  29
  30     def on_POST(self, request, data):
  31         self._require_in_channel(self.fsinf_intern_channelid, data) # will throw an exception if not. (Dont try-except: Its handled up the stack.)
  32
  33         # load and merge data on first use.
  34         self.fsinf_teams_and_channels = self._load_and_merge_optout_data(self.datadir+"./optout.json", self.fsinf_teams_and_channels)
  35
  36         msg_text = data['text'].strip().split(" ", 1)
  37         if msg_text[0].strip() == 'opt-out':
  38             if data['team_id'] in self.fsinf_teams_and_channels and data['channel_id'] in self.fsinf_teams_and_channels[data['team_id']]:
  39                 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']])))
  40                 self._store_optout_data(self.datadir+"./optout.json", self.fsinf_teams_and_channels)
  41                 request.respond_cmd_temp("## :white_check_mark: Success! :)\n#### Now leave this channel manually.")
  42                 return
  43             else:
  44                 request.respond_cmd_err("``/"+self.TRIGGER+"`` Trying to opt-out from an unmanaged channel.")
  45                 return
  46
  47         elif msg_text[0].strip() == 'opt-in' and len(msg_text) == 1:
  48             optout_list = []
  49             for team, to_unpack in self.fsinf_teams_and_channels.items():
  50                 for channel, excluded in to_unpack.items():
  51                     if data['user_id'] in self.fsinf_teams_and_channels[team][channel]:
  52                         optout_list += [channel]
  53             if optout_list:
  54                 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]))
  55                 return
  56             else:
  57                 request.respond_cmd_temp("#### You have not opted-out from any channels.")
  58                 return
  59
  60         elif msg_text[0].strip() == 'opt-in' and len(msg_text) > 1:
  61             managed = False
  62             chan_id = msg_text[1].strip()
  63             for team, to_unpack in self.fsinf_teams_and_channels.items():
  64                 if chan_id in self.fsinf_teams_and_channels[team]:
  65                     managed = True
  66                     if data['user_id'] in self.fsinf_teams_and_channels[team][chan_id]:
  67                         newset = set(self.fsinf_teams_and_channels[team][chan_id])
  68                         newset.remove(data['user_id'])
  69                         self.fsinf_teams_and_channels[team][chan_id] = sorted(list(newset))
  70                         self._store_optout_data(self.datadir+"./optout.json", self.fsinf_teams_and_channels)
  71                     else:
  72                         request.respond_cmd_err("``/"+self.TRIGGER+"`` You have not opted-out from this channel.")
  73                         return
  74
  75             if managed:
  76                 request.respond_cmd_temp("## :white_check_mark: Success! :)\n#### Run ``/"+self.TRIGGER+" sync`` to apply the changes.")
  77                 return
  78             else:
  79                 request.respond_cmd_err("``/"+self.TRIGGER+"`` Trying to opt-in to an unmanaged or unknown channel.")
  80                 return
  81
  82         elif msg_text[0].strip() == 'sync':
  83             fsinf_intern_userids = set([u["user_id"] for u in self.bot.api.get_channel_members(self.fsinf_intern_channelid)])
  84             fsinf_ignored_userids = set([u["user_id"] for u in self.bot.api.get_channel_members(self.ignore_accounts_channelid)])
  85             all_users = {u["id"]:u["username"] for u in self.bot.api.get_users()}
  86
  87             fsinf_aktive_userids = fsinf_intern_userids-fsinf_ignored_userids
  88             request.respond_cmd_temp("## :white_check_mark: Success! :)\n#### Sync started. See ~debug for details.")
  89
  90             dbg_post = self.bot.debug_chan(cleandoc("""
  91                 ## ``/fsinf-access sync``
  92                 These accounts were detected to be fsinf-intern by being in ~off-topic:
  93                 ``"""+str(sorted([all_users[u] for u in fsinf_intern_userids]))+"""``
  94
  95                 These accounts were detected to be ignored (in ~sync-fsinf-ignore -> excluded):
  96                 ``"""+str(sorted([all_users[u] for u in fsinf_ignored_userids]))+"""``
  97
  98                 Adding active fsinf accounts to these teams (and granting TEAM_ADMIN permissions) and channels unless opted-out:
  99                 ```
 100                 """)+"\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("""
 101                 ```
 102                 Applying now ...
 103                 """))
 104
 105             # join into teams + make team admins
 106             for user_id in fsinf_aktive_userids:
 107                 for _, t in self.managed_teams.items():
 108                     self.bot.api.add_user_to_team(t[0], user_id, exc=False)
 109                     self.bot.api.update_team_members_scheme_roles(t[0], user_id, {"scheme_user": True, "scheme_admin": True}, exc=False)
 110
 111                 # join into special channels
 112                 for team_id, chan_ids in self.fsinf_teams_and_channels.items():
 113                     for chan_id, excluded_user_ids in chan_ids.items():
 114                         if not user_id in excluded_user_ids:
 115                             self.bot.api.add_user_to_channel(chan_id, user_id, exc=False)
 116
 117             self.bot.debug_chan("## :white_check_mark: Success! :)", root_id=dbg_post['id'])
 118
 119
 120     def _load_and_merge_optout_data(self, path, to_merge_with):
 121         try:
 122             merged = to_merge_with.copy()
 123             with open(path, "r") as f:
 124                 stored_data = json.load(f)
 125                 for team, to_unpack in to_merge_with.items():
 126                     for channel, excluded in to_unpack.items():
 127                         if team in stored_data and channel in stored_data[team]:
 128                             merged[team][channel] = sorted(list(set(merged[team][channel]+stored_data[team][channel])))
 129             return merged
 130         except:
 131             return to_merge_with
 132
 133
 134     def _store_optout_data(self, path, data):
 135         with open(path, "w") as f:
 136            json.dump(data, f, sort_keys=True, indent=2)
 137         self.bot.debug_chan("``/fsinf-access`` optout.json updated.")