kick_users.py 3.2 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192
  1. #!/usr/bin/env python
  2. from __future__ import print_function
  3. from argparse import ArgumentParser
  4. import json
  5. import requests
  6. import sys
  7. import urllib
  8. try:
  9. raw_input
  10. except NameError: # Python 3
  11. raw_input = input
  12. def _mkurl(template, kws):
  13. for key in kws:
  14. template = template.replace(key, kws[key])
  15. return template
  16. def main(hs, room_id, access_token, user_id_prefix, why):
  17. if not why:
  18. why = "Automated kick."
  19. print(
  20. "Kicking members on %s in room %s matching %s" % (hs, room_id, user_id_prefix)
  21. )
  22. room_state_url = _mkurl(
  23. "$HS/_matrix/client/api/v1/rooms/$ROOM/state?access_token=$TOKEN",
  24. {"$HS": hs, "$ROOM": room_id, "$TOKEN": access_token},
  25. )
  26. print("Getting room state => %s" % room_state_url)
  27. res = requests.get(room_state_url)
  28. print("HTTP %s" % res.status_code)
  29. state_events = res.json()
  30. if "error" in state_events:
  31. print("FATAL")
  32. print(state_events)
  33. return
  34. kick_list = []
  35. room_name = room_id
  36. for event in state_events:
  37. if not event["type"] == "m.room.member":
  38. if event["type"] == "m.room.name":
  39. room_name = event["content"].get("name")
  40. continue
  41. if not event["content"].get("membership") == "join":
  42. continue
  43. if event["state_key"].startswith(user_id_prefix):
  44. kick_list.append(event["state_key"])
  45. if len(kick_list) == 0:
  46. print("No user IDs match the prefix '%s'" % user_id_prefix)
  47. return
  48. print("The following user IDs will be kicked from %s" % room_name)
  49. for uid in kick_list:
  50. print(uid)
  51. doit = raw_input("Continue? [Y]es\n")
  52. if len(doit) > 0 and doit.lower() == "y":
  53. print("Kicking members...")
  54. # encode them all
  55. kick_list = [urllib.quote(uid) for uid in kick_list]
  56. for uid in kick_list:
  57. kick_url = _mkurl(
  58. "$HS/_matrix/client/api/v1/rooms/$ROOM/state/m.room.member/$UID?access_token=$TOKEN",
  59. {"$HS": hs, "$UID": uid, "$ROOM": room_id, "$TOKEN": access_token},
  60. )
  61. kick_body = {"membership": "leave", "reason": why}
  62. print("Kicking %s" % uid)
  63. res = requests.put(kick_url, data=json.dumps(kick_body))
  64. if res.status_code != 200:
  65. print("ERROR: HTTP %s" % res.status_code)
  66. if res.json().get("error"):
  67. print("ERROR: JSON %s" % res.json())
  68. if __name__ == "__main__":
  69. parser = ArgumentParser("Kick members in a room matching a certain user ID prefix.")
  70. parser.add_argument("-u", "--user-id", help="The user ID prefix e.g. '@irc_'")
  71. parser.add_argument("-t", "--token", help="Your access_token")
  72. parser.add_argument("-r", "--room", help="The room ID to kick members in")
  73. parser.add_argument(
  74. "-s", "--homeserver", help="The base HS url e.g. http://matrix.org"
  75. )
  76. parser.add_argument("-w", "--why", help="Reason for the kick. Optional.")
  77. args = parser.parse_args()
  78. if not args.room or not args.token or not args.user_id or not args.homeserver:
  79. parser.print_help()
  80. sys.exit(1)
  81. else:
  82. main(args.homeserver, args.room, args.token, args.user_id, args.why)