Source code for matrixctl.addons.delroom.addon

#!/usr/bin/env python
# matrixctl
# Copyright (c) 2020  Michael Sasser <Michael@MichaelSasser.org>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""Use this module to add the ``delroom`` subcommand to ``matrixctl``."""

from __future__ import annotations

import json
import logging

from argparse import Namespace
from time import sleep

from matrixctl.errors import InternalResponseError
from matrixctl.handlers.api import RequestBuilder
from matrixctl.handlers.api import Response
from matrixctl.handlers.api import request
from matrixctl.handlers.yaml import YAML
from matrixctl.typehints import JsonDict


__author__: str = "Michael Sasser"
__email__: str = "Michael@MichaelSasser.org"


logger = logging.getLogger(__name__)


[docs]def addon(arg: Namespace, yaml: YAML) -> int: """Delete an empty room from the database. Parameters ---------- arg : argparse.Namespace The ``Namespace`` object of argparse's ``parse_args()`` yaml : matrixctl.handlers.yaml.YAML The configuration file handler. Returns ------- err_code : int Non-zero value indicates error code, or zero on success. """ body: JsonDict = handle_arguments(arg) req: RequestBuilder = RequestBuilder( token=yaml.get("server", "api", "token"), domain=yaml.get("server", "api", "domain"), path=f"rooms/{arg.room}", method="DELETE", api_version="v2", json=body, timeout=1200, ) try: response: Response = request(req) except InternalResponseError: logger.error("Could not delete room.") return 1 try: json_response: JsonDict = response.json() except json.decoder.JSONDecodeError as e: logger.fatal("The JSON response could not be loaded by MatrixCtl.") raise InternalResponseError(f"The response was: {response = }") from e try: json_response = handle_status(yaml, json_response["delete_id"]) except InternalResponseError as e: if e.message: logger.fatal(e.message) logger.fatal( "MatrixCtl was not able to verify the status of the request." ) return 1 print(json.dumps(json_response, indent=4)) return 0
[docs]def handle_status(yaml: YAML, delete_id: str) -> JsonDict: # noqa: C901 """Handle the status of a delete room request. Parameters ---------- yaml : matrixctl.handlers.yaml.YAML The configuration file handler. delete_id: str The delete id of a delete room request. Returns ------- response: matrixctl.typehints.JsonDict, optional The response as dict, containing the status. """ req: RequestBuilder = RequestBuilder( token=yaml.get("server", "api", "token"), domain=yaml.get("server", "api", "domain"), path=f"rooms/delete_status/{delete_id}", method="GET", api_version="v2", timeout=1200.0, ) # Lock messages to only print them once msglock_shutting_down: bool = False msglock_purging: bool = False while True: sleep(1) try: response: Response = request(req) except InternalResponseError as e: raise InternalResponseError( "The delete room request was probably successful but the" " status request failed. You just have to wait a bit." ) from e try: json_response: JsonDict = response.json() except json.decoder.JSONDecodeError as e: logger.fatal( "The JSON status response could not be loaded by MatrixCtl." ) raise InternalResponseError( f"The response was: {response = }" ) from e if response is not None: logger.debug("response: %s", response) # complete if json_response["status"] == "complete": print( "Status: Complete (the room has been deleted successfully)" ) break # shutting_down if json_response["status"] == "shutting_down": if not msglock_shutting_down: print( "Status: Shutting Down (removing users from the room)" ) msglock_shutting_down = True logger.info( "The server is still shutting_down the room. " "Please wait..." ) sleep(5) continue # purging if json_response["status"] == "purging": if not msglock_purging: print( "Status: Purging (purging the room and event data from" " database)" ) msglock_purging = True logger.info( "The server is still purging the room. Please wait..." ) sleep(5) continue # failed if json_response["status"] == "failed": logger.critical( ( "The server returned, that the approach failed with " "the following message: %s." ), json_response["status"], ) break break return json_response
[docs]def handle_arguments(arg: Namespace) -> JsonDict: """Build the parameters used for the delroom request. Parameters ---------- arg : argparse.Namespace The ``Namespace`` object of argparse's ``parse_args()`` Returns ------- body : matrixctl.typehints.JsonDict The params. """ if arg.force_purge and arg.no_purge: arg.no_purge = False body: JsonDict = { "block": arg.block, "purge": arg.no_purge, "force_purge": arg.force_purge, } if arg.new_room_admin is not None: body["new_room_user_id"] = arg.new_room_admin body["room_name"] = arg.new_room_name if arg.message is None: body["message"] = ( f"{arg.room} has been shutdown due to content violations " "on this server. Please review our Terms of Service." ) else: body["message"] = arg.message logger.debug("Body: %s", body) return body
# vim: set ft=python :