Mercurial > libervia-backend
view libervia/backend/plugins/plugin_exp_invitation_pubsub.py @ 4334:111dce64dcb5
plugins XEP-0300, XEP-0446, XEP-0447, XEP0448 and others: Refactoring to use Pydantic:
Pydantic models are used more and more in Libervia, for the bridge API, and also to
convert `domish.Element` to internal representation.
Type hints have also been added in many places.
rel 453
author | Goffi <goffi@goffi.org> |
---|---|
date | Tue, 03 Dec 2024 00:12:38 +0100 |
parents | 0d7bb4df2343 |
children |
line wrap: on
line source
#!/usr/bin/env python3 # SàT plugin to send invitations for Pubsub # Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org) # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. from typing import Optional from twisted.internet import defer from twisted.words.protocols.jabber import jid from twisted.words.xish import domish from libervia.backend.core.i18n import _ from libervia.backend.core.constants import Const as C from libervia.backend.core.log import getLogger from libervia.backend.core.xmpp import SatXMPPEntity from libervia.backend.tools import utils from libervia.backend.tools.common import data_format log = getLogger(__name__) PLUGIN_INFO = { C.PI_NAME: "Pubsub Invitation", C.PI_IMPORT_NAME: "PUBSUB_INVITATION", C.PI_TYPE: "EXP", C.PI_PROTOCOLS: [], C.PI_DEPENDENCIES: ["XEP-0060", "INVITATION"], C.PI_RECOMMENDATIONS: [], C.PI_MAIN: "PubsubInvitation", C.PI_HANDLER: "no", C.PI_DESCRIPTION: _("Invitations for pubsub based features"), } class PubsubInvitation: def __init__(self, host): log.info(_("Pubsub Invitation plugin initialization")) self.host = host self._p = host.plugins["XEP-0060"] # namespace to handler map self._ns_handler = {} host.bridge.add_method( "ps_invite", ".plugin", in_sign="sssssss", out_sign="", method=self._send_pubsub_invitation, async_=True, ) def register(self, namespace: str, handler) -> None: self._ns_handler[namespace] = handler self.host.plugins["INVITATION"].register_namespace(namespace, self.on_invitation) def _send_pubsub_invitation( self, invitee_jid_s, service_s, node, item_id=None, name=None, extra_s="", profile_key=C.PROF_KEY_NONE, ): client = self.host.get_client(profile_key) invitee_jid = jid.JID(invitee_jid_s) service = jid.JID(service_s) extra = data_format.deserialise(extra_s) return defer.ensureDeferred( self.invite( client, invitee_jid, service, node, item_id or None, name=name or None, extra=extra, ) ) async def invite( self, client: SatXMPPEntity, invitee_jid: jid.JID, service: jid.JID, node: str, item_id: Optional[str] = None, name: str = "", extra: Optional[dict] = None, ) -> None: if extra is None: extra = {} else: namespace = extra.get("namespace") if namespace: try: handler = self._ns_handler[namespace] preflight = handler.invite_preflight except KeyError: pass except AttributeError: log.debug(f"no invite_preflight method found for {namespace!r}") else: await utils.as_deferred( preflight, client, invitee_jid, service, node, item_id, name, extra, ) if item_id is None: item_id = extra.pop("default_item_id", None) # we authorize our invitee to see the nodes of interest await self._p.set_node_affiliations( client, service, node, {invitee_jid: "member"} ) log.debug(f"affiliation set on {service}'s {node!r} node") # now we send the invitation self.host.plugins["INVITATION"].send_pubsub_invitation( client, invitee_jid, service, node, item_id, name=name or None, extra=extra ) async def on_invitation( self, client: SatXMPPEntity, namespace: str, name: str, extra: dict, service: jid.JID, node: str, item_id: Optional[str], item_elt: domish.Element, ) -> None: if extra is None: extra = {} try: handler = self._ns_handler[namespace] preflight = handler.on_invitation_preflight except KeyError: pass except AttributeError: log.debug(f"no on_invitation_preflight method found for {namespace!r}") else: await utils.as_deferred( preflight, client, namespace, name, extra, service, node, item_id, item_elt, ) if item_id is None: item_id = extra.pop("default_item_id", None) creator = extra.pop("creator", False) element = extra.pop("element", None) if not name: name = extra.pop("name", "") return await self.host.plugins["LIST_INTEREST"].register_pubsub( client, namespace, service, node, item_id, creator, name, element, extra )