Tidying up a bit database and remote media from a synapse server using the admin API

synapse-tidying.py

#!/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)