PC-Admin/matrix-moderation-tool
1
0
Fork 0
mirror of https://github.com/PC-Admin/matrix-moderation-tool.git synced 2024-12-19 23:20:26 -05:00
Code Issues Packages Projects Releases Wiki Activity

finish re-write of rooms functions

Browse Source
This commit is contained in:
PC-Admin 2023-07-15 04:36:40 +08:00
parent 07ec96334b
commit 664c8611d4
1 changed files with 87 additions and 78 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

165
room_commands.py
View File

@ -53,6 +53,7 @@ def export_room_state(preset_internal_ID):
f.write(response.text) f.write(response.text)
print(response.text) print(response.text)
return(response.text)
# Example # Example
# $ curl -kXGET 'https://matrix.perthchat.org/_synapse/admin/v1/rooms/!OeqILBxiHahidSQQoC:matrix.org/state?access_token=ACCESS_TOKEN' # $ curl -kXGET 'https://matrix.perthchat.org/_synapse/admin/v1/rooms/!OeqILBxiHahidSQQoC:matrix.org/state?access_token=ACCESS_TOKEN'
@ -274,42 +275,51 @@ def shutdown_room(preset_internal_ID,preset_user_ID,preset_new_room_name,preset_
print("Input invalid! exiting.") print("Input invalid! exiting.")
return return
# First export the state events of the room to examine them later or import them to rdlist # First export the state events of the room to examine them later or import them to rdlist
export_room_state(internal_ID) room_status = export_room_state(internal_ID)
print ("Exported room state events to file, this data can be useful for profiling a room after you've blocked/purged it: ./state_events" + internal_ID + "_state.json")
command_string = 'curl -H "Authorization: Bearer ' + hardcoded_variables.access_token + "\" --data '{ \"new_room_user_id\": \"@" + username + ":" + hardcoded_variables.base_url + "\" , \"room_name\": \"" + new_room_name + "\", \"message\": \"" + message + "\", \"block\": " + block_choice + ", \"purge\": " + purge_choice + " }' -X DELETE 'https://" + hardcoded_variables.homeserver_url + "/_synapse/admin/v2/rooms/" + internal_ID + "'" # Convert the string to a dictionary
#print("\n" + command_string + "\n") room_status_dict = json.loads(room_status)
process = subprocess.run([command_string], shell=True, stdout=subprocess.PIPE, universal_newlines=True)
output = process.stdout
#print(output)
status = "null" if "Room not found" not in room_status_dict.get('error', ''):
count = 0 print(f"Exported room state events to file, this data can be useful for profiling a room after you've blocked/purged it: ./state_events{internal_ID}_state.json")
sleep_time = 1
while status != "complete" and count < 8: headers = {"Authorization": f"Bearer {hardcoded_variables.access_token}"}
time.sleep(sleep_time) data = {
count = count + 1 "new_room_user_id": f"@{username}:{hardcoded_variables.base_url}",
sleep_time = sleep_time * 2 "room_name": new_room_name,
command_string = 'curl -H "Authorization: Bearer ' + hardcoded_variables.access_token + "\" -kX GET 'https://" + hardcoded_variables.homeserver_url + '/_synapse/admin/v2/rooms/' + internal_ID + "/delete_status'" "message": message,
#print("\n" + command_string + "\n") "block": block_choice,
process = subprocess.run([command_string], shell=True, stdout=subprocess.PIPE, universal_newlines=True) "purge": purge_choice
output_json = json.loads(process.stdout) }
#print(output_json) delete_room_url = f"https://{hardcoded_variables.homeserver_url}/_synapse/admin/v2/rooms/{internal_ID}"
status = output_json["results"][0]["status"] response = requests.delete(delete_room_url, headers=headers, json=data, verify=True)
print("status: " + status)
#print("count: " + str(count))
if status != "complete":
print("Sleeping for " + str(sleep_time) + " seconds...")
if status == "complete": status = "null"
print(internal_ID + " has been successfully shutdown!") count = 0
if str(output_json["results"][0]["shutdown_room"]["kicked_users"]) != '[]': sleep_time = 1
print("List of kicked users:")
for entry in output_json["results"][0]["shutdown_room"]["kicked_users"]: while status != "complete" and count < 8:
print(entry) time.sleep(sleep_time)
print("") count += 1
sleep_time *= 2
check_status_url = f"https://{hardcoded_variables.homeserver_url}/_synapse/admin/v2/rooms/{internal_ID}/delete_status"
status_response = requests.get(check_status_url, headers=headers, verify=True)
output_json = status_response.json()
status = output_json["results"][0]["status"]
print(f"status: {status}")
if status != "complete":
print(f"Sleeping for {sleep_time} seconds...")
if status == "complete":
print(f"{internal_ID} has been successfully shutdown!")
if str(output_json["results"][0]["shutdown_room"]["kicked_users"]) != '[]':
print("List of kicked users:")
for entry in output_json["results"][0]["shutdown_room"]["kicked_users"]:
print(entry)
print("")
else:
print("The room was not found.")
# Example: # Example:
#$ curl -H "Authorization: Bearer ACCESS_TOKEN" --data '{ "new_room_user_id": "@PC-Admin:perthchat.org", "room_name": "VIOLATION ROOM", "message": "YOU HAVE BEEN NAUGHTY!", "block": true, "purge": true }' -X DELETE 'https://matrix.perthchat.org/_synapse/admin/v2/rooms/!yUykDcYIEtrbSxOyPD:perthchat.org' #$ curl -H "Authorization: Bearer ACCESS_TOKEN" --data '{ "new_room_user_id": "@PC-Admin:perthchat.org", "room_name": "VIOLATION ROOM", "message": "YOU HAVE BEEN NAUGHTY!", "block": true, "purge": true }' -X DELETE 'https://matrix.perthchat.org/_synapse/admin/v2/rooms/!yUykDcYIEtrbSxOyPD:perthchat.org'
@ -368,14 +378,13 @@ def shutdown_multiple_rooms():
def delete_room(preset_internal_ID): def delete_room(preset_internal_ID):
if preset_internal_ID == '': if preset_internal_ID == '':
internal_ID = input("\nEnter the internal id of the room you want to delete (Example: !OLkDvaYjpNrvmwnwdj:matrix.org): ") internal_ID = input("\nEnter the internal id of the room you want to delete (Example: !OLkDvaYjpNrvmwnwdj:matrix.org): ")
elif preset_internal_ID != '': else:
internal_ID = preset_internal_ID internal_ID = preset_internal_ID
headers = {"Authorization": "Bearer " + hardcoded_variables.access_token}
command_string = 'curl -H "Authorization: Bearer ' + hardcoded_variables.access_token + "\" --data '{ \"block\": false, \"purge\": true }' -X DELETE 'https://" + hardcoded_variables.homeserver_url + "/_synapse/admin/v2/rooms/" + internal_ID + "'" data = {"block": False, "purge": True}
print("\n" + command_string + "\n") url = f'https://{hardcoded_variables.homeserver_url}/_synapse/admin/v2/rooms/{internal_ID}'
process = subprocess.run([command_string], shell=True, stdout=subprocess.PIPE, universal_newlines=True) response = requests.delete(url, headers=headers, data=json.dumps(data))
output = process.stdout print("\n", response.text, "\n")
print(output)
status = "null" status = "null"
count = 0 count = 0
@ -383,27 +392,23 @@ def delete_room(preset_internal_ID):
while status != "complete" and count < 8: while status != "complete" and count < 8:
time.sleep(sleep_time) time.sleep(sleep_time)
count = count + 1 count += 1
sleep_time = sleep_time * 2 sleep_time *= 2
command_string = 'curl -H "Authorization: Bearer ' + hardcoded_variables.access_token + "\" -kX GET 'https://" + hardcoded_variables.homeserver_url + '/_synapse/admin/v2/rooms/' + internal_ID + "/delete_status'" url_status = f'https://{hardcoded_variables.homeserver_url}/_synapse/admin/v2/rooms/{internal_ID}/delete_status'
#print("\n" + command_string + "\n") response = requests.get(url_status, headers=headers, verify=True)
process = subprocess.run([command_string], shell=True, stdout=subprocess.PIPE, universal_newlines=True) response_json = response.json()
#print("\nOutput type: " + str(type(process.stdout))) status = response_json["results"][0]["status"]
#print("Output value: " + str(process.stdout) + "\n")
output_json = json.loads(process.stdout)
#print(output_json)
status = output_json["results"][0]["status"]
print("status: " + status) print("status: " + status)
#print("count: " + str(count))
if status != "complete": if status != "complete":
print("Sleeping for " + str(sleep_time) + " seconds...") print(f"Sleeping for {sleep_time} seconds...")
if status == "complete": if status == "complete":
print(internal_ID + " has been successfully deleted!") print(f"{internal_ID} has been successfully deleted!")
if str(output_json["results"][0]["shutdown_room"]["kicked_users"]) != '[]': kicked_users = response_json["results"][0]["shutdown_room"]["kicked_users"]
if kicked_users:
print("List of kicked users:") print("List of kicked users:")
for entry in output_json["results"][0]["shutdown_room"]["kicked_users"]: for user in kicked_users:
print(entry) print(user)
print("") print("")
# Example: # Example:
@ -441,44 +446,48 @@ def delete_multiple_rooms():
def purge_room_to_timestamp(preset_internal_ID, preset_timestamp): def purge_room_to_timestamp(preset_internal_ID, preset_timestamp):
if preset_internal_ID == '': if preset_internal_ID == '':
internal_ID = input("\nEnter the internal id of the room you want to delete (Example: !OLkDvaYjpNrvmwnwdj:matrix.org): ") internal_ID = input("\nEnter the internal id of the room you want to delete (Example: !OLkDvaYjpNrvmwnwdj:matrix.org): ")
elif preset_internal_ID != '': else:
internal_ID = preset_internal_ID internal_ID = preset_internal_ID
if preset_timestamp == '': if preset_timestamp == '':
timestamp = input("\nEnter the epoche timestamp in microseconds (Example: 1661058683000): ") timestamp = input("\nEnter the epoch timestamp in microseconds (Example: 1661058683000): ")
elif preset_timestamp != '': else:
timestamp = preset_timestamp timestamp = preset_timestamp
command_string = 'curl --header "Authorization: Bearer ' + hardcoded_variables.access_token + "\" -X POST -H \"Content-Type: application/json\" -d '{ \"delete_local_events\": false, \"purge_up_to_ts\": " + timestamp + " }' 'https://" + hardcoded_variables.homeserver_url + "/_synapse/admin/v1/purge_history/" + internal_ID + "'" headers = {"Authorization": "Bearer " + hardcoded_variables.access_token, "Content-Type": "application/json"}
print("\n" + command_string + "\n") data = {"delete_local_events": False, "purge_up_to_ts": int(timestamp)}
process = subprocess.run([command_string], shell=True, stdout=subprocess.PIPE, universal_newlines=True) url = f'https://{hardcoded_variables.homeserver_url}/_synapse/admin/v1/purge_history/{internal_ID}'
output = process.stdout
print(output)
output_json = json.loads(process.stdout)
purge_id = output_json["purge_id"]
response = requests.post(url, headers=headers, data=json.dumps(data))
print("\n", response.text, "\n")
response_json = response.json()
if "errcode" in response_json:
print(f"Error: {response_json['error']}")
return
purge_id = response_json.get("purge_id")
status = "null" status = "null"
count = 0 count = 0
sleep_time = 0.5 sleep_time = 0.5
while status != "complete" and count < 8: while status != "complete" and count < 8:
time.sleep(sleep_time) time.sleep(sleep_time)
count = count + 1 count += 1
sleep_time = sleep_time * 2 sleep_time *= 2
command_string = 'curl -H "Authorization: Bearer ' + hardcoded_variables.access_token + "\" -kX GET 'https://" + hardcoded_variables.homeserver_url + '/_synapse/admin/v1/purge_history_status/' + purge_id + "'"
print("\n" + command_string + "\n") url_status = f'https://{hardcoded_variables.homeserver_url}/_synapse/admin/v1/purge_history_status/{purge_id}'
process = subprocess.run([command_string], shell=True, stdout=subprocess.PIPE, universal_newlines=True) response = requests.get(url_status, headers=headers)
#print("\nOutput type: " + str(type(process.stdout))) response_json = response.json()
#print("Output value: " + str(process.stdout) + "\n")
output_json = json.loads(process.stdout) status = response_json.get("status")
#print(output_json)
status = output_json["status"]
print("status: " + status) print("status: " + status)
#print("count: " + str(count))
if status != "complete": if status != "complete":
print("Sleeping for " + str(sleep_time) + " seconds...") print(f"Sleeping for {sleep_time} seconds...")
if status == "complete": if status == "complete":
print(internal_ID + " has successfully had its history purged!") print(f"{internal_ID} has successfully had its history purged!")
print("") print("")
# Example: # Example: