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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
|
# 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.errors import IntentError, MatrixRequestError, MForbidden
from mautrix.types import ContentURI, EventID, UserID
from ... import bridge as br
from .handler import SECTION_ADMIN, CommandEvent, command_handler
@command_handler(
needs_admin=True,
needs_auth=False,
name="set-pl",
help_section=SECTION_ADMIN,
help_args="[_mxid_] <_level_>",
help_text="Set a temporary power level without affecting the remote platform.",
)
async def set_power_level(evt: CommandEvent) -> EventID:
try:
user_id = UserID(evt.args[0])
except IndexError:
return await evt.reply(f"**Usage:** `$cmdprefix+sp set-pl [mxid] <level>`")
if user_id.startswith("@"):
evt.args.pop(0)
else:
user_id = evt.sender.mxid
try:
level = int(evt.args[0])
except (KeyError, IndexError):
return await evt.reply("**Usage:** `$cmdprefix+sp set-pl [mxid] <level>`")
except ValueError:
return await evt.reply("The level must be an integer.")
levels = await evt.main_intent.get_power_levels(evt.room_id, ignore_cache=True)
levels.users[user_id] = level
try:
return await evt.main_intent.set_power_levels(evt.room_id, levels)
except MForbidden as e:
await evt.reply(f"I don't seem to have permission to update power levels: {e.message}")
except (MatrixRequestError, IntentError):
evt.log.exception("Failed to update power levels")
return await evt.reply("Failed to update power levels (see logs for more details)")
async def _get_mxid_param(
evt: CommandEvent, args: str
) -> tuple[br.BasePuppet | None, EventID | None]:
try:
user_id = UserID(evt.args[0])
except IndexError:
return None, await evt.reply(f"**Usage:** `$cmdprefix+sp {evt.command} {args}`")
if user_id.startswith("@") and ":" in user_id:
# TODO support parsing mention pills instead of requiring a plaintext mxid
puppet = await evt.bridge.get_puppet(user_id)
if not puppet:
return None, await evt.reply("The given user ID is not a valid ghost user.")
evt.args.pop(0)
return puppet, None
elif evt.is_portal and (puppet := await evt.portal.get_dm_puppet()):
return puppet, None
return None, await evt.reply(
"This is not a private chat portal, you must pass a user ID explicitly."
)
@command_handler(
needs_admin=True,
needs_auth=False,
name="set-avatar",
help_section=SECTION_ADMIN,
help_args="[_mxid_] <_mxc:// uri_>",
help_text="Set an avatar for a ghost user.",
)
async def set_ghost_avatar(evt: CommandEvent) -> EventID | None:
puppet, err = await _get_mxid_param(evt, "[mxid] <mxc:// URI>")
if err:
return err
try:
mxc_uri = ContentURI(evt.args[0])
except IndexError:
return await evt.reply("**Usage:** `$cmdprefix+sp set-avatar [mxid] <mxc_uri>`")
if not mxc_uri.startswith("mxc://"):
return await evt.reply("The avatar URL must start with `mxc://`")
try:
return await puppet.default_mxid_intent.set_avatar_url(mxc_uri)
except (MatrixRequestError, IntentError):
evt.log.exception("Failed to set avatar.")
return await evt.reply("Failed to set avatar (see logs for more details).")
@command_handler(
needs_admin=True,
needs_auth=False,
name="remove-avatar",
help_section=SECTION_ADMIN,
help_args="[_mxid_]",
help_text="Remove the avatar for a ghost user.",
)
async def remove_ghost_avatar(evt: CommandEvent) -> EventID | None:
puppet, err = await _get_mxid_param(evt, "[mxid]")
if err:
return err
try:
return await puppet.default_mxid_intent.set_avatar_url(ContentURI(""))
except (MatrixRequestError, IntentError):
evt.log.exception("Failed to remove avatar.")
return await evt.reply("Failed to remove avatar (see logs for more details).")
@command_handler(
needs_admin=True,
needs_auth=False,
name="set-displayname",
help_section=SECTION_ADMIN,
help_args="[_mxid_] <_displayname_>",
help_text="Set the display name for a ghost user.",
)
async def set_ghost_display_name(evt: CommandEvent) -> EventID | None:
puppet, err = await _get_mxid_param(evt, "[mxid] <displayname>")
if err:
return err
try:
return await puppet.default_mxid_intent.set_displayname(" ".join(evt.args))
except (MatrixRequestError, IntentError):
evt.log.exception("Failed to set display name.")
return await evt.reply("Failed to set display name (see logs for more details).")
@command_handler(
needs_admin=True,
needs_auth=False,
name="remove-displayname",
help_section=SECTION_ADMIN,
help_args="[_mxid_]",
help_text="Remove the display name for a ghost user.",
)
async def remove_ghost_display_name(evt: CommandEvent) -> EventID | None:
puppet, err = await _get_mxid_param(evt, "[mxid]")
if err:
return err
try:
return await puppet.default_mxid_intent.set_displayname("")
except (MatrixRequestError, IntentError):
evt.log.exception("Failed to remove display name.")
return await evt.reply("Failed to remove display name (see logs for more details).")
|
