1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89
|
# Copyright (c) 2022 Tulir Asokan
#
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
from __future__ import annotations
from mautrix.types import EventID
from .handler import (
SECTION_GENERAL,
CommandEvent,
HelpCacheKey,
HelpSection,
command_handler,
command_handlers,
)
@command_handler(
needs_auth=False, help_section=SECTION_GENERAL, help_text="Cancel an ongoing action."
)
async def cancel(evt: CommandEvent) -> EventID:
if evt.sender.command_status:
action = evt.sender.command_status["action"]
evt.sender.command_status = None
return await evt.reply(f"{action} cancelled.")
else:
return await evt.reply("No ongoing command.")
@command_handler(
needs_auth=False, help_section=SECTION_GENERAL, help_text="Get the bridge version."
)
async def version(evt: CommandEvent) -> None:
if not evt.processor.bridge:
await evt.reply("Bridge version unknown")
else:
await evt.reply(
f"[{evt.processor.bridge.name}]({evt.processor.bridge.repo_url}) "
f"{evt.processor.bridge.markdown_version or evt.processor.bridge.version}"
)
@command_handler(needs_auth=False)
async def unknown_command(evt: CommandEvent) -> EventID:
return await evt.reply("Unknown command. Try `$cmdprefix+sp help` for help.")
help_cache: dict[HelpCacheKey, str] = {}
async def _get_help_text(evt: CommandEvent) -> str:
cache_key = await evt.get_help_key()
if cache_key not in help_cache:
help_sections: dict[HelpSection, list[str]] = {}
for handler in command_handlers.values():
if (
handler.has_help
and handler.has_permission(cache_key)
and handler.is_enabled_for(evt)
):
help_sections.setdefault(handler.help_section, [])
help_sections[handler.help_section].append(handler.help + " ")
help_sorted = sorted(help_sections.items(), key=lambda item: item[0].order)
helps = ["#### {}\n{}\n".format(key.name, "\n".join(value)) for key, value in help_sorted]
help_cache[cache_key] = "\n".join(helps)
return help_cache[cache_key]
def _get_management_status(evt: CommandEvent) -> str:
if evt.is_management:
return "This is a management room: prefixing commands with `$cmdprefix` is not required."
elif evt.is_portal:
return (
"**This is a portal room**: you must always prefix commands with `$cmdprefix`.\n"
"Management commands will not be bridged."
)
return "**This is not a management room**: you must prefix commands with `$cmdprefix`."
@command_handler(
name="help",
needs_auth=False,
help_section=SECTION_GENERAL,
help_text="Show this help message.",
)
async def help_cmd(evt: CommandEvent) -> EventID:
return await evt.reply(_get_management_status(evt) + "\n" + await _get_help_text(evt))
|