Skip to content

Instantly share code, notes, and snippets.

@Eskuero
Last active August 26, 2023 20:33
Show Gist options
  • Save Eskuero/7808a27d08472e1c50811494d12d407d to your computer and use it in GitHub Desktop.
Save Eskuero/7808a27d08472e1c50811494d12d407d to your computer and use it in GitHub Desktop.
Tidying up a bit database and remote media from a synapse server using the admin API
#!/usr/bin/python
import requests
import time
import sys
import json
# Login and get a token
endpoint = "https://fromshado.ws"
# Admin credentials
ADMINUSER = ""
PASSWORD = ""
# Events older than this seconds will be purged
MAXAGE = 604800
# Login
data = {
"type": "m.login.password",
"identifier": {
"type": "m.id.user",
"user": ADMINUSER
},
"password": PASSWORD
}
try:
r = requests.post(url = f"{endpoint}/_matrix/client/r0/login", json = data)
token = r.json()["access_token"]
except:
print(r.text)
sys.exit(0)
datadelete = {
"room_name": "Clearing abandoned",
"message": "Clearing abandoned",
"block": False,
"purge": True
}
print("\nPurging abandoned rooms")
# Get roomlist
r = requests.get(url = f"{endpoint}/_synapse/admin/v1/rooms?limit=1000", headers = {"Authorization": "Bearer " + token})
roomlist = r.json()["rooms"]
# Purge all rooms where no local user remains
for room in roomlist:
if room["joined_local_members"] == 0:
print("Purging room " + room['room_id'])
r = requests.delete(url = f"{endpoint}/_synapse/admin/v1/rooms/" + room['room_id'], data = json.dumps(datadelete), headers = {"Authorization": "Bearer " + token})
print(r.text)
print("\nClearing everything older than one week")
# Get roomlist after the purge
r = requests.get(url = f"{endpoint}/_synapse/admin/v1/rooms?limit=1000", headers = {"Authorization": "Bearer " + token})
roomlist = r.json()["rooms"]
# Clear all events older than one week
epoch = int(time.time() - MAXAGE ) * 1000
for room in roomlist:
r = requests.post(url = f"{endpoint}/_synapse/admin/v1/purge_history/" + room["room_id"], json = {"delete_local_events": "true", "purge_up_to_ts": epoch}, headers = {"Authorization": "Bearer " + token})
try:
purgeid = r.json()["purge_id"]
except KeyError:
print("Cleaning " + room["room_id"] + ": " + r.json()["error"])
else:
status = requests.get(url = f"{endpoint}/_synapse/admin/v1/purge_history_status/" + purgeid, headers = {"Authorization": "Bearer " + token}).json()["status"]
while status == "active":
time.sleep(1)
status = requests.get(url = f"{endpoint}/_synapse/admin/v1/purge_history_status/" + purgeid, headers = {"Authorization": "Bearer " + token}).json()["status"]
print("Cleaning " + room["room_id"] + " " + status)
# Purging remote media older than one week
r = requests.post(url = f"{endpoint}/_synapse/admin/v1/purge_media_cache?before_ts=" + str(epoch), headers = {"Authorization": "Bearer " + token})
print(r)
print(r.json())
# Logout and disable the token
print("\nLogging out")
r = requests.post(url = f"{endpoint}/_matrix/client/r0/logout", headers = {"Authorization": "Bearer " + token})
print(r)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment