File: meta.py

package info (click to toggle)
mautrix-python 0.20.7-1
  • links: PTS, VCS
  • area: main
  • in suites: sid, trixie
  • size: 1,812 kB
  • sloc: python: 19,103; makefile: 16
file content (89 lines) | stat: -rw-r--r-- 3,090 bytes parent folder | download
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))