Coverage for slidge/command/user.py: 54%
215 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-10 12:26 +0000
1# Commands available to users
2import contextlib
3from copy import deepcopy
4from typing import Any, cast
6from slixmpp import JID
7from slixmpp.exceptions import XMPPError
9from ..db.meta import JSONSerializable
10from ..db.models import Space
11from ..util.types import AnyRoom, AnySession, RoomType, UserPreferences
12from .base import (
13 Command,
14 CommandAccess,
15 Confirmation,
16 ConfirmationSession,
17 Form,
18 FormField,
19 FormSession,
20 FormValues,
21 SearchResult,
22 TableResult,
23)
24from .categories import CONTACTS, ROOMS, SPACES
27class Search(Command[AnySession]):
28 NAME = "🔎 Search for contacts"
29 HELP = "Search for contacts via this gateway"
30 CHAT_COMMAND = "find"
31 NODE = CONTACTS.node + "/" + CHAT_COMMAND
32 ACCESS = CommandAccess.USER_LOGGED
33 CATEGORY = CONTACTS
35 async def run(
36 self, session: "AnySession | None", _ifrom: JID, *args: str
37 ) -> FormSession[AnySession] | SearchResult | None:
38 assert session is not None
39 await session.ready
40 if args:
41 return await session.on_search(
42 {self.xmpp.SEARCH_FIELDS[0].var: " ".join(args)}
43 )
44 return FormSession[AnySession](
45 title=self.xmpp.SEARCH_TITLE,
46 instructions=self.xmpp.SEARCH_INSTRUCTIONS,
47 fields=self.xmpp.SEARCH_FIELDS,
48 handler=self.search,
49 )
51 @staticmethod
52 async def search(
53 form_values: FormValues, session: "AnySession | None", _ifrom: JID
54 ) -> SearchResult:
55 assert session is not None
56 results = await session.on_search(form_values) # type: ignore
57 if results is None:
58 raise XMPPError("item-not-found", "No contact was found")
60 return results
63class SyncContacts(Command[AnySession]):
64 NAME = "🔄 Sync XMPP roster"
65 HELP = (
66 "Synchronize your XMPP roster with your legacy contacts. "
67 "Slidge will only add/remove/modify contacts in its dedicated roster group"
68 )
69 CHAT_COMMAND = "sync-contacts"
70 NODE = CONTACTS.node + "/" + CHAT_COMMAND
71 ACCESS = CommandAccess.USER_LOGGED
72 CATEGORY = CONTACTS
74 async def run(
75 self,
76 session: "AnySession | None",
77 _ifrom: JID,
78 *args: str,
79 ) -> ConfirmationSession[AnySession]:
80 assert session is not None
81 await session.ready
82 return ConfirmationSession[AnySession](
83 prompt="Are you sure you want to sync your roster?",
84 success=None,
85 handler=self.sync,
86 )
88 async def sync(self, session: "AnySession | None", _ifrom: JID) -> str:
89 if session is None:
90 raise RuntimeError
91 added, removed, updated = await session.contacts._sync()
92 return f"{added} added, {removed} removed, {updated} updated"
95class ListContacts(Command[AnySession]):
96 NAME = HELP = "👤 List your legacy contacts"
97 CHAT_COMMAND = "contacts"
98 NODE = CONTACTS.node + "/" + CHAT_COMMAND
99 ACCESS = CommandAccess.USER_LOGGED
100 CATEGORY = CONTACTS
102 async def run(
103 self,
104 session: "AnySession | None",
105 _ifrom: JID,
106 *_: str,
107 ) -> TableResult:
108 assert session is not None
109 await session.ready
110 contacts = sorted(
111 session.contacts, key=lambda c: c.name.casefold() if c.name else ""
112 )
113 return TableResult(
114 description="Your buddies",
115 fields=[FormField("name"), FormField("jid", type="jid-single")],
116 items=[{"name": c.name, "jid": c.jid.bare} for c in contacts],
117 )
120class ListRooms(Command[AnySession]):
121 NAME = HELP = "👥 List your legacy rooms"
122 CHAT_COMMAND = "rooms"
123 NODE = ROOMS.node + "/" + CHAT_COMMAND
124 ACCESS = CommandAccess.USER_LOGGED
125 CATEGORY = ROOMS
127 async def run(
128 self, session: "AnySession | None", _ifrom: JID, *_: str
129 ) -> TableResult:
130 assert session is not None
131 await session.ready
132 rooms: list[AnyRoom] = sorted(
133 session.rooms, key=lambda g: (g.name or g.jid.node).casefold()
134 )
135 return TableResult(
136 description="Your rooms",
137 fields=[FormField("name"), FormField("jid", type="jid-single")],
138 items=[
139 {"name": g.name or str(g.legacy_id), "jid": g.jid.bare} for g in rooms
140 ],
141 jids_are_mucs=True,
142 )
145class ListSpaces(Command[AnySession]):
146 NAME = "🌐 List my spaces"
147 HELP = "List the spaces you are part of. Spaces are collections of rooms."
148 CHAT_COMMAND = "spaces"
149 NODE = ROOMS.node + "/" + CHAT_COMMAND
150 ACCESS = CommandAccess.USER_LOGGED
151 CATEGORY = SPACES
153 related_to_spaces = True
155 async def run(
156 self, session: "AnySession | None", _ifrom: JID, *_: str
157 ) -> TableResult:
158 assert session is not None
159 spaces = await _get_updated_spaces(session)
160 return TableResult(
161 description="Your spaces. If your client does not support spaces, use the 'space-rooms' command.",
162 fields=[FormField("name"), FormField("iri")],
163 items=[
164 {
165 "name": s.name or str(s.legacy_id),
166 "iri": f"xmpp:{self.xmpp.boundjid.bare}?;node={session.spaces.legacy_id_to_node(s.legacy_id)}",
167 }
168 for s in spaces
169 ],
170 jids_are_mucs=True,
171 )
174class ListRoomsInSpace(Command[AnySession]):
175 NAME = "🌐 List the rooms in a space"
176 HELP = "List the rooms of a space you are part of. Spaces are collections of rooms."
177 CHAT_COMMAND = "space-rooms"
178 NODE = ROOMS.node + "/" + CHAT_COMMAND
179 ACCESS = CommandAccess.USER_LOGGED
180 CATEGORY = SPACES
182 async def run(
183 self, session: "AnySession | None", _ifrom: JID, *_: str
184 ) -> "FormSession[AnySession]":
185 assert session is not None
186 spaces = await _get_updated_spaces(session)
187 return FormSession(
188 title="Your spaces",
189 instructions="Select a space to view its rooms",
190 fields=[
191 FormField(
192 var="space_legacy_id",
193 label="Space",
194 required=True,
195 type="list-single",
196 options=[
197 {
198 "label": s.name or str(s.legacy_id),
199 "value": str(s.legacy_id),
200 }
201 for s in spaces
202 ],
203 )
204 ],
205 handler=self.list_rooms,
206 handler_args=({str(s.legacy_id): s.name for s in spaces},),
207 )
209 async def list_rooms(
210 self,
211 form_values: FormValues,
212 session: "AnySession | None",
213 _ifrom: JID,
214 space_names: dict[str, str],
215 ) -> TableResult:
216 assert session is not None
217 space_legacy_id = form_values.get("space_legacy_id")
218 if space_legacy_id is None:
219 raise XMPPError("bad-request", "You need to specify a space")
220 assert isinstance(space_legacy_id, str)
221 await session.ready
222 with self.xmpp.store.session() as orm:
223 rooms = sorted(
224 self.xmpp.store.spaces.get_rooms(
225 orm,
226 session.user_pk,
227 session.spaces.legacy_id_to_node(space_legacy_id),
228 ),
229 key=lambda r: (r.name or str(r.jid.node)).casefold(),
230 )
231 name = space_names.get(space_legacy_id)
232 return TableResult(
233 fields=[
234 FormField("name", "Name"),
235 FormField("jid", "JID", type="jid-single"),
236 ],
237 description=f"Rooms of '{name or space_legacy_id}'",
238 items=[{"name": r.name or str(r.legacy_id), "jid": r.jid} for r in rooms],
239 jids_are_mucs=True,
240 )
243class Login(Command[AnySession]):
244 NAME = "🔐 Re-login to the legacy network"
245 HELP = "Login to the legacy service"
246 CHAT_COMMAND = "re-login"
247 NODE = "https://slidge.im/command/core/" + CHAT_COMMAND
249 ACCESS = CommandAccess.USER_NON_LOGGED
251 async def run(
252 self,
253 session: "AnySession | None",
254 _ifrom: JID,
255 *_: str,
256 ) -> str:
257 assert session is not None
258 if session.is_logging_in:
259 raise XMPPError("bad-request", "You are already logging in.")
260 session.is_logging_in = True
261 try:
262 msg = await self.xmpp.login_wrap(session)
263 except Exception as e: # noqa: BLE001
264 session.send_gateway_status(f"Re-login failed: {e}", show="dnd")
265 raise XMPPError(
266 "internal-server-error", etype="wait", text=f"Could not login: {e}"
267 )
268 finally:
269 session.is_logging_in = False
270 session.logged = True
271 session.send_gateway_status(msg or "Re-connected", show="chat")
272 session.send_gateway_message(msg or "Re-connected")
273 return msg
276class CreateGroup(Command[AnySession]):
277 NAME = "🆕 New legacy group"
278 HELP = "Create a group on the legacy service"
279 CHAT_COMMAND = "create-group"
280 NODE = ROOMS.node + "/" + CHAT_COMMAND
281 CATEGORY = ROOMS
283 ACCESS = CommandAccess.USER_LOGGED
285 async def run(
286 self,
287 session: "AnySession | None",
288 _ifrom: JID,
289 *_: str,
290 ) -> FormSession[AnySession]:
291 assert session is not None
292 await session.ready
293 contacts = session.contacts.known_contacts(only_friends=True)
294 return FormSession(
295 title="Create a new group",
296 instructions="Pick contacts that should be part of this new group",
297 fields=[
298 FormField(var="group_name", label="Name of the group", required=True),
299 FormField(
300 var="contacts",
301 label="Contacts to add to the new group",
302 type="list-multi",
303 options=[
304 {"value": contact.legacy_id, "label": contact.name}
305 for contact in sorted(contacts.values(), key=lambda c: c.name)
306 ],
307 required=False,
308 ),
309 ],
310 handler=self.finish,
311 )
313 @staticmethod
314 async def finish(
315 form_values: FormValues,
316 session: "AnySession | None",
317 *_: Any, # noqa:ANN401
318 ) -> TableResult:
319 assert session is not None
320 legacy_id = await session.rooms.on_creation_request(
321 cast(str, form_values["group_name"]),
322 cast(list[str], form_values.get("contacts", [])),
323 )
324 muc = await session.rooms.get_or_create(legacy_id)
325 return TableResult(
326 description=f"Your new group: xmpp:{muc.jid}?join",
327 fields=[FormField("name"), FormField("jid", type="jid-single")],
328 items=[{"name": muc.name, "jid": muc.jid}],
329 jids_are_mucs=True,
330 )
333class Preferences(Command[AnySession]):
334 NAME = "⚙️ Preferences"
335 HELP = "Customize the gateway behaviour to your liking"
336 CHAT_COMMAND = "preferences"
337 NODE = "https://slidge.im/command/core/preferences"
338 ACCESS = CommandAccess.USER
340 async def run(
341 self,
342 session: "AnySession | None",
343 _ifrom: JID,
344 *_: str,
345 ) -> FormSession[AnySession]:
346 fields = deepcopy(self.xmpp.PREFERENCES)
347 assert session is not None
348 current = session.user.preferences
349 for field in fields:
350 field.value = current.get(field.var, field.value) # type:ignore
351 return Form(
352 title="Preferences",
353 instructions=self.HELP,
354 fields=fields,
355 handler=self.finish, # type:ignore
356 handler_kwargs={"previous": current},
357 )
359 async def finish(
360 self,
361 form_values: UserPreferences,
362 session: "AnySession | None",
363 *_: Any, # noqa:ANN401
364 previous: JSONSerializable,
365 ) -> str:
366 assert session is not None
367 if previous == form_values:
368 return "No preference was changed"
370 previous = previous.copy()
371 user = session.user
372 user.preferences.update(form_values) # type:ignore
373 self.xmpp.store.users.update(user)
375 with contextlib.suppress(NotImplementedError):
376 await session.on_preferences(previous, form_values)
378 if not previous["sync_avatar"] and form_values["sync_avatar"]:
379 await self.xmpp.fetch_user_avatar(session)
380 else:
381 user.avatar_hash = None
383 return "Your preferences have been updated."
386class Unregister(Command[AnySession]):
387 NAME = "❌ Unregister from the gateway"
388 HELP = "Unregister from the gateway"
389 CHAT_COMMAND = "unregister"
390 NODE = "https://slidge.im/command/core/unregister"
391 ACCESS = CommandAccess.USER
393 async def run(
394 self, session: "AnySession | None", _ifrom: JID, *_: str
395 ) -> ConfirmationSession[AnySession]:
396 return ConfirmationSession[AnySession](
397 prompt=f"Are you sure you want to unregister from '{self.xmpp.boundjid}'?",
398 success=f"You are not registered to '{self.xmpp.boundjid}' anymore.",
399 handler=self.unregister,
400 )
402 async def unregister(self, session: "AnySession | None", _ifrom: JID) -> str:
403 assert session is not None
404 await self.xmpp.unregister_user(session.user)
405 return "You are not registered anymore. Bye!"
408class LeaveGroup(Command[AnySession]):
409 NAME = HELP = "❌ Leave a legacy group"
410 CHAT_COMMAND = "leave-group"
411 NODE = ROOMS.node + "/" + CHAT_COMMAND
412 ACCESS = CommandAccess.USER_LOGGED
413 CATEGORY = ROOMS
415 async def run(
416 self,
417 session: "AnySession | None",
418 ifrom: JID,
419 *_: str,
420 ) -> FormSession[AnySession]:
421 assert session is not None
422 await session.ready
423 rooms = sorted(session.rooms, key=lambda g: g.DISCO_NAME.casefold())
424 return FormSession(
425 title="Leave a group",
426 instructions="Select the group you want to leave",
427 fields=[
428 FormField(
429 "room",
430 "Group name",
431 type="list-single",
432 options=[
433 {"label": g.name or str(g.legacy_id), "value": str(i)}
434 for i, g in enumerate(rooms)
435 ],
436 )
437 ],
438 handler=self.confirm,
439 handler_args=(rooms,),
440 )
442 async def confirm(
443 self,
444 form_values: FormValues,
445 _session: "AnySession | None",
446 _ifrom: JID,
447 rooms: list[AnyRoom],
448 ) -> Confirmation:
449 group = rooms[int(form_values["room"])] # type:ignore
450 return Confirmation(
451 prompt=f"Are you sure you want to leave the group '{group.name}'?",
452 handler=self.finish,
453 handler_args=(group,),
454 )
456 @staticmethod
457 async def finish(session: AnySession, _ifrom: JID, group: AnyRoom) -> None:
458 await group.on_leave()
459 await session.rooms.remove(group, reason="You left this group via slidge.")
462class InviteInRooms(Command[AnySession]):
463 NAME = "💌 Re-invite me in my rooms"
464 HELP = "Ask the gateway to send invitations for all your private rooms"
465 CHAT_COMMAND = "re-invite"
466 NODE = ROOMS.node + "/" + CHAT_COMMAND
467 ACCESS = CommandAccess.USER_LOGGED
468 CATEGORY = ROOMS
470 async def run(self, session: "AnySession | None", _ifrom: JID, *_: str) -> None:
471 assert session is not None
472 await session.ready
473 for muc in session.rooms:
474 if muc.type == RoomType.GROUP:
475 session.send_gateway_invite(
476 muc, reason="You asked to be re-invited in all rooms."
477 )
480async def _get_updated_spaces(session: AnySession) -> list[Space]:
481 await session.ready
482 await session.rooms.ready
483 return sorted(session.spaces._db_rows, key=lambda s: s.name or str(s.legacy_id))