Source code for slidge.core.muc.bookmarks

import logging
from typing import TYPE_CHECKING, Generic, Type

from slixmpp import JID
from slixmpp.jid import _unescape_node

from ...util import SubclassableOnce
from ...util.types import LegacyGroupIdType, LegacyMUCType
from import ESCAPE_TABLE
from ..mixins.lock import NamedLockMixin
from .room import LegacyMUC

    from ..session import BaseSession

[docs]class LegacyBookmarks( Generic[LegacyGroupIdType, LegacyMUCType], NamedLockMixin, metaclass=SubclassableOnce, ): def __init__(self, session: "BaseSession"): self.session = session self.xmpp = session.xmpp self.user = session.user self._mucs_by_legacy_id = dict[LegacyGroupIdType, LegacyMUCType]() self._mucs_by_bare_jid = dict[str, LegacyMUCType]() self._muc_class: Type[LegacyMUCType] = LegacyMUC.get_self_or_unique_subclass() self._user_nick: str = self.session.user.jid.node super().__init__() self.log = logging.getLogger(f"{self.user.bare_jid}:bookmarks") self.ready = self.session.xmpp.loop.create_future() if not self.xmpp.GROUPS: self.ready.set_result(True) @property
[docs] def user_nick(self): return self._user_nick
@user_nick.setter def user_nick(self, nick: str): self._user_nick = nick
[docs] def __iter__(self): return iter(self._mucs_by_legacy_id.values())
[docs] def __repr__(self): return f"<Bookmarks of {self.user}>"
[docs] async def __finish_init_muc(self, legacy_id: LegacyGroupIdType, jid: JID): muc = self._muc_class(self.session, legacy_id=legacy_id, jid=jid) await muc.update_info() if not muc.user_nick: muc.user_nick = self._user_nick self.log.debug("MUC created: %r", muc) self._mucs_by_legacy_id[muc.legacy_id] = muc self._mucs_by_bare_jid[muc.jid.bare] = muc return muc
[docs] async def legacy_id_to_jid_local_part(self, legacy_id: LegacyGroupIdType): return str(legacy_id).translate(ESCAPE_TABLE)
[docs] async def jid_local_part_to_legacy_id(self, local_part: str): return _unescape_node(local_part)
[docs] async def by_jid(self, jid: JID) -> LegacyMUCType: bare = jid.bare async with self.lock(("bare", bare)): muc = self._mucs_by_bare_jid.get(bare) if muc is None: self.log.debug("Attempting to instantiate a new MUC for JID %s", jid) local_part = jid.node legacy_id = await self.jid_local_part_to_legacy_id(local_part) if self.get_lock(("legacy_id", legacy_id)): self.log.debug("Not instantiating %s after all", jid) return await self.by_legacy_id(legacy_id) self.log.debug("%r is group %r", local_part, legacy_id) muc = await self.__finish_init_muc(legacy_id, JID(bare)) else: self.log.debug("Found an existing MUC instance: %s", muc) return muc
[docs] async def by_legacy_id(self, legacy_id: LegacyGroupIdType) -> LegacyMUCType: async with self.lock(("legacy_id", legacy_id)): muc = self._mucs_by_legacy_id.get(legacy_id) if muc is None: self.log.debug("Create new MUC instance for legacy ID %s", legacy_id) local = await self.legacy_id_to_jid_local_part(legacy_id) bare = f"{local}@{self.xmpp.boundjid}" jid = JID(bare) if self.get_lock(("bare", bare)): self.log.debug("Not instantiating %s after all", legacy_id) return await self.by_jid(jid) muc = await self.__finish_init_muc(legacy_id, jid) else: self.log.debug("Found an existing MUC instance: %s", muc) return muc
[docs] async def fill(self): """ Establish a user's known groups. This has to be overridden in plugins with group support and at the minimum, this should ``await self.by_legacy_id(group_id)`` for all the groups a user is part of. Ideally, set the group subject, friendly, number, etc. in this method Slidge internals will call this on successful ``BaseSession.login()`` """ if self.xmpp.GROUPS: raise NotImplementedError( "The plugin advertised support for groups but" " LegacyBookmarks.fill() was not overridden." )