Coverage for slidge/command/user.py: 50%
243 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-28 18:29 +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 slidge.db.meta import JSONSerializable
10from slidge.db.models import Space
12from ..util.types import (
13 AnyMUC,
14 AnySession,
15 MucType,
16 UserPreferences,
17)
18from .base import (
19 Command,
20 CommandAccess,
21 Confirmation,
22 ConfirmationSession,
23 Form,
24 FormField,
25 FormSession,
26 FormValues,
27 SearchResult,
28 TableResult,
29)
30from .categories import CONTACTS, GROUPS, SPACES
33class Search(Command[AnySession]):
34 NAME = "🔎 Search for contacts"
35 HELP = "Search for contacts via this gateway"
36 CHAT_COMMAND = "find"
37 NODE = CONTACTS.node + "/" + CHAT_COMMAND
38 ACCESS = CommandAccess.USER_LOGGED
39 CATEGORY = CONTACTS
41 async def run(
42 self, session: "AnySession | None", _ifrom: JID, *args: str
43 ) -> FormSession[AnySession] | SearchResult | None:
44 assert session is not None
45 await session.ready
46 if args:
47 return await session.on_search(
48 {self.xmpp.SEARCH_FIELDS[0].var: " ".join(args)}
49 )
50 return FormSession[AnySession](
51 title=self.xmpp.SEARCH_TITLE,
52 instructions=self.xmpp.SEARCH_INSTRUCTIONS,
53 fields=self.xmpp.SEARCH_FIELDS,
54 handler=self.search,
55 )
57 @staticmethod
58 async def search(
59 form_values: FormValues, session: "AnySession | None", _ifrom: JID
60 ) -> SearchResult:
61 assert session is not None
62 results = await session.on_search(form_values) # type: ignore
63 if results is None:
64 raise XMPPError("item-not-found", "No contact was found")
66 return results
69class SyncContacts(Command[AnySession]):
70 NAME = "🔄 Sync XMPP roster"
71 HELP = (
72 "Synchronize your XMPP roster with your legacy contacts. "
73 "Slidge will only add/remove/modify contacts in its dedicated roster group"
74 )
75 CHAT_COMMAND = "sync-contacts"
76 NODE = CONTACTS.node + "/" + CHAT_COMMAND
77 ACCESS = CommandAccess.USER_LOGGED
78 CATEGORY = CONTACTS
80 async def run(
81 self,
82 session: "AnySession | None",
83 _ifrom: JID,
84 *args: str,
85 ) -> ConfirmationSession[AnySession]:
86 assert session is not None
87 await session.ready
88 return ConfirmationSession[AnySession](
89 prompt="Are you sure you want to sync your roster?",
90 success=None,
91 handler=self.sync,
92 )
94 async def sync(self, session: "AnySession | None", _ifrom: JID) -> str:
95 if session is None:
96 raise RuntimeError
97 roster_iq = await self.xmpp["xep_0356"].get_roster(session.user_jid.bare)
99 contacts = session.contacts.known_contacts()
101 added = 0
102 removed = 0
103 updated = 0
104 for item in roster_iq["roster"]:
105 groups = set(item["groups"])
106 if self.xmpp.ROSTER_GROUP in groups:
107 contact = contacts.pop(item["jid"], None)
108 if contact is None:
109 if len(groups) == 1:
110 await self.xmpp["xep_0356"].set_roster(
111 session.user_jid, {item["jid"]: {"subscription": "remove"}}
112 )
113 removed += 1
114 else:
115 groups.remove(self.xmpp.ROSTER_GROUP)
116 await self.xmpp["xep_0356"].set_roster(
117 session.user_jid,
118 {
119 item["jid"]: {
120 "subscription": item["subscription"],
121 "name": item["name"],
122 "groups": groups,
123 }
124 },
125 )
126 updated += 1
127 else:
128 if contact.name != item["name"]:
129 await contact.add_to_roster(force=True)
130 updated += 1
132 # we popped before so this only acts on slidge contacts not in the xmpp roster
133 for contact in contacts.values():
134 added += 1
135 await contact.add_to_roster()
137 return f"{added} added, {removed} removed, {updated} updated"
140class ListContacts(Command[AnySession]):
141 NAME = HELP = "👤 List your legacy contacts"
142 CHAT_COMMAND = "contacts"
143 NODE = CONTACTS.node + "/" + CHAT_COMMAND
144 ACCESS = CommandAccess.USER_LOGGED
145 CATEGORY = CONTACTS
147 async def run(
148 self,
149 session: "AnySession | None",
150 _ifrom: JID,
151 *_: str,
152 ) -> TableResult:
153 assert session is not None
154 await session.ready
155 contacts = sorted(
156 session.contacts, key=lambda c: c.name.casefold() if c.name else ""
157 )
158 return TableResult(
159 description="Your buddies",
160 fields=[FormField("name"), FormField("jid", type="jid-single")],
161 items=[{"name": c.name, "jid": c.jid.bare} for c in contacts],
162 )
165class ListGroups(Command[AnySession]):
166 NAME = HELP = "👥 List your legacy groups"
167 CHAT_COMMAND = "groups"
168 NODE = GROUPS.node + "/" + CHAT_COMMAND
169 ACCESS = CommandAccess.USER_LOGGED
170 CATEGORY = GROUPS
172 async def run(
173 self, session: "AnySession | None", _ifrom: JID, *_: str
174 ) -> TableResult:
175 assert session is not None
176 await session.ready
177 groups: list[AnyMUC] = sorted(
178 session.bookmarks, key=lambda g: (g.name or g.jid.node).casefold()
179 )
180 return TableResult(
181 description="Your groups",
182 fields=[FormField("name"), FormField("jid", type="jid-single")],
183 items=[
184 {"name": g.name or str(g.legacy_id), "jid": g.jid.bare} for g in groups
185 ],
186 jids_are_mucs=True,
187 )
190class ListSpaces(Command[AnySession]):
191 NAME = "🌐 List my spaces"
192 HELP = "List the spaces you are part of. Spaces are collections of rooms."
193 CHAT_COMMAND = "spaces"
194 NODE = GROUPS.node + "/" + CHAT_COMMAND
195 ACCESS = CommandAccess.USER_LOGGED
196 CATEGORY = SPACES
198 related_to_spaces = True
200 async def run(
201 self, session: "AnySession | None", _ifrom: JID, *_: str
202 ) -> TableResult:
203 assert session is not None
204 spaces = await _get_updated_spaces(session)
205 return TableResult(
206 description="Your spaces. If your client does not support spaces, use the 'space-rooms' command.",
207 fields=[FormField("name"), FormField("iri")],
208 items=[
209 {
210 "name": s.name or str(s.legacy_id),
211 "iri": f"xmpp:{self.xmpp.boundjid.bare}?;node={await session.bookmarks.space_legacy_id_to_node(s.legacy_id)}",
212 }
213 for s in spaces
214 ],
215 jids_are_mucs=True,
216 )
219class ListRoomsInSpace(Command[AnySession]):
220 NAME = "🌐 List the rooms in a space"
221 HELP = "List the rooms of a space you are part of. Spaces are collections of rooms."
222 CHAT_COMMAND = "space-rooms"
223 NODE = GROUPS.node + "/" + CHAT_COMMAND
224 ACCESS = CommandAccess.USER_LOGGED
225 CATEGORY = SPACES
227 async def run(
228 self, session: "AnySession | None", _ifrom: JID, *_: str
229 ) -> "FormSession[AnySession]":
230 assert session is not None
231 spaces = await _get_updated_spaces(session)
232 return FormSession(
233 title="Your spaces",
234 instructions="Select a space to view its rooms",
235 fields=[
236 FormField(
237 var="space_legacy_id",
238 label="Space",
239 required=True,
240 type="list-single",
241 options=[
242 {
243 "label": s.name or str(s.legacy_id),
244 "value": str(s.legacy_id),
245 }
246 for s in spaces
247 ],
248 )
249 ],
250 handler=self.list_rooms,
251 handler_args=({str(s.legacy_id): s.name for s in spaces},),
252 )
254 async def list_rooms(
255 self,
256 form_values: FormValues,
257 session: "AnySession | None",
258 _ifrom: JID,
259 space_names: dict[str, str],
260 ) -> TableResult:
261 assert session is not None
262 space_legacy_id = form_values.get("space_legacy_id")
263 if space_legacy_id is None:
264 raise XMPPError("bad-request", "You need to specify a space")
265 assert isinstance(space_legacy_id, str)
266 await session.ready
267 with self.xmpp.store.session() as orm:
268 rooms = sorted(
269 self.xmpp.store.spaces.get_rooms(
270 orm,
271 session.user_pk,
272 await session.bookmarks.space_legacy_id_to_node(space_legacy_id),
273 ),
274 key=lambda r: (r.name or str(r.jid.node)).casefold(),
275 )
276 name = space_names.get(space_legacy_id)
277 return TableResult(
278 fields=[
279 FormField("name", "Name"),
280 FormField("jid", "JID", type="jid-single"),
281 ],
282 description=f"Rooms of '{name or space_legacy_id}'",
283 items=[{"name": r.name or str(r.legacy_id), "jid": r.jid} for r in rooms],
284 jids_are_mucs=True,
285 )
288class Login(Command[AnySession]):
289 NAME = "🔐 Re-login to the legacy network"
290 HELP = "Login to the legacy service"
291 CHAT_COMMAND = "re-login"
292 NODE = "https://slidge.im/command/core/" + CHAT_COMMAND
294 ACCESS = CommandAccess.USER_NON_LOGGED
296 async def run(
297 self,
298 session: "AnySession | None",
299 _ifrom: JID,
300 *_: str,
301 ) -> str:
302 assert session is not None
303 if session.is_logging_in:
304 raise XMPPError("bad-request", "You are already logging in.")
305 session.is_logging_in = True
306 try:
307 msg = await self.xmpp.login_wrap(session)
308 except Exception as e: # noqa: BLE001
309 session.send_gateway_status(f"Re-login failed: {e}", show="dnd")
310 raise XMPPError(
311 "internal-server-error", etype="wait", text=f"Could not login: {e}"
312 )
313 finally:
314 session.is_logging_in = False
315 session.logged = True
316 session.send_gateway_status(msg or "Re-connected", show="chat")
317 session.send_gateway_message(msg or "Re-connected")
318 return msg
321class CreateGroup(Command[AnySession]):
322 NAME = "🆕 New legacy group"
323 HELP = "Create a group on the legacy service"
324 CHAT_COMMAND = "create-group"
325 NODE = GROUPS.node + "/" + CHAT_COMMAND
326 CATEGORY = GROUPS
328 ACCESS = CommandAccess.USER_LOGGED
330 async def run(
331 self,
332 session: "AnySession | None",
333 _ifrom: JID,
334 *_: str,
335 ) -> FormSession[AnySession]:
336 assert session is not None
337 await session.ready
338 contacts = session.contacts.known_contacts(only_friends=True)
339 return FormSession(
340 title="Create a new group",
341 instructions="Pick contacts that should be part of this new group",
342 fields=[
343 FormField(var="group_name", label="Name of the group", required=True),
344 FormField(
345 var="contacts",
346 label="Contacts to add to the new group",
347 type="list-multi",
348 options=[
349 {"value": str(contact.jid), "label": contact.name}
350 for contact in sorted(contacts.values(), key=lambda c: c.name)
351 ],
352 required=False,
353 ),
354 ],
355 handler=self.finish,
356 )
358 @staticmethod
359 async def finish(
360 form_values: FormValues,
361 session: "AnySession | None",
362 *_: Any, # noqa:ANN401
363 ) -> TableResult:
364 assert session is not None
365 legacy_id: str = await session.on_create_group(
366 cast(str, form_values["group_name"]),
367 [
368 await session.contacts.by_jid(JID(j))
369 for j in form_values.get("contacts", []) # type:ignore
370 ],
371 )
372 muc = await session.bookmarks.by_legacy_id(legacy_id)
373 return TableResult(
374 description=f"Your new group: xmpp:{muc.jid}?join",
375 fields=[FormField("name"), FormField("jid", type="jid-single")],
376 items=[{"name": muc.name, "jid": muc.jid}],
377 jids_are_mucs=True,
378 )
381class Preferences(Command[AnySession]):
382 NAME = "⚙️ Preferences"
383 HELP = "Customize the gateway behaviour to your liking"
384 CHAT_COMMAND = "preferences"
385 NODE = "https://slidge.im/command/core/preferences"
386 ACCESS = CommandAccess.USER
388 async def run(
389 self,
390 session: "AnySession | None",
391 _ifrom: JID,
392 *_: str,
393 ) -> FormSession[AnySession]:
394 fields = deepcopy(self.xmpp.PREFERENCES)
395 assert session is not None
396 current = session.user.preferences
397 for field in fields:
398 field.value = current.get(field.var, field.value) # type:ignore
399 return Form(
400 title="Preferences",
401 instructions=self.HELP,
402 fields=fields,
403 handler=self.finish, # type:ignore
404 handler_kwargs={"previous": current},
405 )
407 async def finish(
408 self,
409 form_values: UserPreferences,
410 session: "AnySession | None",
411 *_: Any, # noqa:ANN401
412 previous: JSONSerializable,
413 ) -> str:
414 assert session is not None
415 if previous == form_values:
416 return "No preference was changed"
418 previous = previous.copy()
419 user = session.user
420 user.preferences.update(form_values) # type:ignore
421 self.xmpp.store.users.update(user)
423 with contextlib.suppress(NotImplementedError):
424 await session.on_preferences(previous, form_values) # type:ignore[arg-type]
426 if not previous["sync_avatar"] and form_values["sync_avatar"]:
427 await self.xmpp.fetch_user_avatar(session)
428 else:
429 user.avatar_hash = None
431 return "Your preferences have been updated."
434class Unregister(Command[AnySession]):
435 NAME = "❌ Unregister from the gateway"
436 HELP = "Unregister from the gateway"
437 CHAT_COMMAND = "unregister"
438 NODE = "https://slidge.im/command/core/unregister"
439 ACCESS = CommandAccess.USER
441 async def run(
442 self, session: "AnySession | None", _ifrom: JID, *_: str
443 ) -> ConfirmationSession[AnySession]:
444 return ConfirmationSession[AnySession](
445 prompt=f"Are you sure you want to unregister from '{self.xmpp.boundjid}'?",
446 success=f"You are not registered to '{self.xmpp.boundjid}' anymore.",
447 handler=self.unregister,
448 )
450 async def unregister(self, session: "AnySession | None", _ifrom: JID) -> str:
451 assert session is not None
452 await self.xmpp.unregister_user(session.user)
453 return "You are not registered anymore. Bye!"
456class LeaveGroup(Command[AnySession]):
457 NAME = HELP = "❌ Leave a legacy group"
458 CHAT_COMMAND = "leave-group"
459 NODE = GROUPS.node + "/" + CHAT_COMMAND
460 ACCESS = CommandAccess.USER_LOGGED
461 CATEGORY = GROUPS
463 async def run(
464 self,
465 session: "AnySession | None",
466 ifrom: JID,
467 *_: str,
468 ) -> FormSession[AnySession]:
469 assert session is not None
470 await session.ready
471 groups = sorted(session.bookmarks, key=lambda g: g.DISCO_NAME.casefold())
472 return FormSession(
473 title="Leave a group",
474 instructions="Select the group you want to leave",
475 fields=[
476 FormField(
477 "group",
478 "Group name",
479 type="list-single",
480 options=[
481 {"label": g.name or str(g.legacy_id), "value": str(i)}
482 for i, g in enumerate(groups)
483 ],
484 )
485 ],
486 handler=self.confirm,
487 handler_args=(groups,),
488 )
490 async def confirm(
491 self,
492 form_values: FormValues,
493 _session: "AnySession | None",
494 _ifrom: JID,
495 groups: list[AnyMUC],
496 ) -> Confirmation:
497 group = groups[int(form_values["group"])] # type:ignore
498 return Confirmation(
499 prompt=f"Are you sure you want to leave the group '{group.name}'?",
500 handler=self.finish,
501 handler_args=(group,),
502 )
504 @staticmethod
505 async def finish(session: AnySession, _ifrom: JID, group: AnyMUC) -> None:
506 await group.on_leave()
507 await session.bookmarks.remove(group, reason="You left this group via slidge.")
510class InviteInGroups(Command[AnySession]):
511 NAME = "💌 Re-invite me in my groups"
512 HELP = "Ask the gateway to send invitations for all your private groups"
513 CHAT_COMMAND = "re-invite"
514 NODE = GROUPS.node + "/" + CHAT_COMMAND
515 ACCESS = CommandAccess.USER_LOGGED
516 CATEGORY = GROUPS
518 async def run(self, session: "AnySession | None", _ifrom: JID, *_: str) -> None:
519 assert session is not None
520 await session.ready
521 for muc in session.bookmarks:
522 if muc.type == MucType.GROUP:
523 session.send_gateway_invite(
524 muc, reason="You asked to be re-invited in all groups."
525 )
528async def _get_updated_spaces(session: AnySession) -> list[Space]:
529 await session.ready
530 with session.xmpp.store.session() as orm:
531 spaces = list(session.xmpp.store.spaces.get_all(orm, session.user_pk))
532 updated_spaces: list[Space] = []
533 for space in spaces:
534 try:
535 updated_spaces.append(await session.bookmarks.update_space_if_needed(space))
536 except Exception:
537 session.log.exception(
538 "Something went wrong trying to update space '%r'", space
539 )
541 return sorted(updated_spaces, key=lambda s: s.name or str(s.legacy_id))