Mercurial > libervia-backend
view libervia/backend/plugins/plugin_xep_0292.py @ 4294:a0ed5c976bf8 default tip @
component conferences, plugin XEP-0167, XEP-0298: add stream user metadata:
A/V conference now adds user metadata about the stream it is forwarding through XEP-0298.
This is parsed and added to metadata during confirmation on client side.
rel 448
author | Goffi <goffi@goffi.org> |
---|---|
date | Tue, 06 Aug 2024 23:43:11 +0200 |
parents | 0d7bb4df2343 |
children |
line wrap: on
line source
#!/usr/bin/env python3 # Libervia plugin for XEP-0292 # Copyright (C) 2009-2022 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 List, Dict, Union, Any, Optional from functools import partial from twisted.words.protocols.jabber.xmlstream import XMPPHandler from twisted.words.protocols.jabber import jid from twisted.words.xish import domish from zope.interface import implementer from wokkel import disco, iwokkel from libervia.backend.core.constants import Const as C from libervia.backend.core.i18n import _ from libervia.backend.core.log import getLogger from libervia.backend.core.core_types import SatXMPPEntity from libervia.backend.core import exceptions from libervia.backend.tools.common.async_utils import async_lru log = getLogger(__name__) IMPORT_NAME = "XEP-0292" PLUGIN_INFO = { C.PI_NAME: "vCard4 Over XMPP", C.PI_IMPORT_NAME: IMPORT_NAME, C.PI_TYPE: C.PLUG_TYPE_XEP, C.PI_MODES: C.PLUG_MODE_BOTH, C.PI_PROTOCOLS: ["XEP-0292"], C.PI_DEPENDENCIES: ["IDENTITY", "XEP-0060", "XEP-0163"], C.PI_MAIN: "XEP_0292", C.PI_HANDLER: "yes", C.PI_DESCRIPTION: _("""XEP-0292 (vCard4 Over XMPP) implementation"""), } NS_VCARD4 = "urn:ietf:params:xml:ns:vcard-4.0" VCARD4_NODE = "urn:xmpp:vcard4" text_fields = {"fn": "name", "nickname": "nicknames", "note": "description"} text_fields_inv = {v: k for k, v in text_fields.items()} class XEP_0292: namespace = NS_VCARD4 node = VCARD4_NODE def __init__(self, host): # XXX: as of XEP-0292 v0.11, there is a dedicated <IQ/> protocol in this XEP which # should be used according to the XEP. Hovewer it feels like an outdated behaviour # and other clients don't seem to use it. After discussing it on xsf@ MUC, it # seems that implemeting the dedicated <IQ/> protocol is a waste of time, and thus # it is not done here. It is expected that this dedicated protocol will be removed # from a future version of the XEP. log.info(_("vCard4 Over XMPP initialization")) host.register_namespace("vcard4", NS_VCARD4) self.host = host self._p = host.plugins["XEP-0060"] self._i = host.plugins["IDENTITY"] self._i.register( IMPORT_NAME, "nicknames", partial(self.getValue, field="nicknames"), partial(self.set_value, field="nicknames"), priority=1000, ) self._i.register( IMPORT_NAME, "description", partial(self.getValue, field="description"), partial(self.set_value, field="description"), priority=1000, ) def get_handler(self, client): return XEP_0292_Handler() def vcard_2_dict(self, vcard_elt: domish.Element) -> Dict[str, Any]: """Convert vcard element to equivalent identity metadata""" vcard: Dict[str, Any] = {} for metadata_elt in vcard_elt.elements(): # Text values for source_field, dest_field in text_fields.items(): if metadata_elt.name == source_field: if metadata_elt.text is not None: dest_type = self._i.get_field_type(dest_field) value = str(metadata_elt.text) if dest_type is str: if dest_field in vcard: vcard[dest_field] += value else: vcard[dest_field] = value elif dest_type is list: vcard.setdefault(dest_field, []).append(value) else: raise exceptions.InternalError( f"unexpected dest_type: {dest_type!r}" ) break else: log.debug( f"Following element is currently unmanaged: {metadata_elt.toXml()}" ) return vcard def dict_2_v_card(self, vcard: dict[str, Any]) -> domish.Element: """Convert vcard metadata to vCard4 element""" vcard_elt = domish.Element((NS_VCARD4, "vcard")) for field, elt_name in text_fields_inv.items(): value = vcard.get(field) if value: if isinstance(value, str): value = [value] if isinstance(value, list): for v in value: field_elt = vcard_elt.addElement(elt_name) field_elt.addElement("text", content=v) else: log.warning(f"ignoring unexpected value: {value!r}") return vcard_elt @async_lru(5) async def get_card(self, client: SatXMPPEntity, entity: jid.JID) -> dict: try: items, metadata = await self._p.get_items( client, entity, VCARD4_NODE, item_ids=["current"] ) except exceptions.NotFound: log.info(f"No vCard node found for {entity}") return {} item_elt = items[0] try: vcard_elt = next(item_elt.elements(NS_VCARD4, "vcard")) except StopIteration: log.info(f"vCard element is not present for {entity}") return {} return self.vcard_2_dict(vcard_elt) async def update_vcard_elt( self, client: SatXMPPEntity, vcard_elt: domish.Element, entity: Optional[jid.JID] = None, ) -> None: """Update VCard 4 of given entity, create node if doesn't already exist @param vcard_elt: whole vCard element to update @param entity: entity for which the vCard must be updated None to update profile's own vCard """ service = entity or client.jid.userhostJID() node_options = { self._p.OPT_ACCESS_MODEL: self._p.ACCESS_OPEN, self._p.OPT_PUBLISH_MODEL: self._p.PUBLISH_MODEL_PUBLISHERS, } await self._p.create_if_new_node(client, service, VCARD4_NODE, node_options) await self._p.send_item( client, service, VCARD4_NODE, vcard_elt, item_id=self._p.ID_SINGLETON ) async def update_v_card( self, client: SatXMPPEntity, vcard: Dict[str, Any], entity: Optional[jid.JID] = None, update: bool = True, ) -> None: """Update VCard 4 of given entity, create node if doesn't already exist @param vcard: identity metadata @param entity: entity for which the vCard must be updated None to update profile's own vCard @param update: if True, current vCard will be retrieved and updated with given vcard (thus if False, `vcard` data will fully replace previous one). """ service = entity or client.jid.userhostJID() if update: current_vcard = await self.get_card(client, service) current_vcard.update(vcard) vcard = current_vcard vcard_elt = self.dict_2_v_card(vcard) await self.update_vcard_elt(client, vcard_elt, service) async def getValue( self, client: SatXMPPEntity, entity: jid.JID, field: str, ) -> Optional[Union[str, List[str]]]: """Return generic value @param entity: entity from who the vCard comes @param field: name of the field to get This has to be a string field @return request value """ vcard_data = await self.get_card(client, entity) return vcard_data.get(field) async def set_value( self, client: SatXMPPEntity, value: Union[str, List[str]], entity: jid.JID, field: str, ) -> None: """Set generic value @param entity: entity from who the vCard comes @param field: name of the field to get This has to be a string field """ await self.update_v_card(client, {field: value}, entity) @implementer(iwokkel.IDisco) class XEP_0292_Handler(XMPPHandler): def getDiscoInfo(self, requestor, service, nodeIdentifier=""): return [disco.DiscoFeature(NS_VCARD4)] def getDiscoItems(self, requestor, service, nodeIdentifier=""): return []