1
0

kick_users.py 3.1 KB

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