Skip to content
Back to top

# Source code for slixmpp.plugins.xep_0084.avatar

    """
        Slixmpp: The Slick XMPP Library
        Copyright (C) 2012 Nathanael C. Fritz, Lance J.T. Stout
        This file is part of Slixmpp.

        See the file LICENSE for copying permission.
    """

    from __future__ import annotations

    import hashlib
    import logging

    from asyncio import Future
    from typing import (
        Dict,
        Iterable,
        List,
        Optional,
        Set,
        Union,
        TYPE_CHECKING,
    )

    from slixmpp.stanza import Iq
    from slixmpp.plugins import BasePlugin
    from slixmpp.xmlstream.handler import Callback
    from slixmpp.xmlstream.matcher import StanzaPath
    from slixmpp.xmlstream import register_stanza_plugin, JID
    from slixmpp.plugins.xep_0084.stanza import Data, MetaData, Pointer
    from slixmpp.plugins.xep_0084 import stanza

    try:
        from typing import TypedDict
    except ImportError:
        from typing_extensions import TypedDict

    class AvatarMetadataItem(TypedDict, total=False):
        bytes: int
        id: str
        type: str
        height: int
        width: int
        url: str

    MetadataItems = Union[
        AvatarMetadataItem,
        List[AvatarMetadataItem],
        Set[AvatarMetadataItem]
    ]

    log = logging.getLogger(__name__)

    class XEP_0084(BasePlugin):

        name = 'xep_0084'
        description = 'XEP-0084: User Avatar'
        dependencies = {'xep_0163', 'xep_0060'}
        stanza = stanza

        def plugin_init(self):
            pubsub_stanza = self.xmpp['xep_0060'].stanza
            register_stanza_plugin(pubsub_stanza.Item, Data)
            register_stanza_plugin(pubsub_stanza.EventItem, Data)

            self.xmpp['xep_0060'].map_node_event(Data.namespace, 'avatar_data')

        def plugin_end(self):
            self.xmpp['xep_0030'].del_feature(feature=MetaData.namespace)
            self.xmpp['xep_0163'].remove_interest(MetaData.namespace)

        def session_bind(self, jid):
            self.xmpp['xep_0163'].register_pep('avatar_metadata', MetaData)

        def generate_id(self, data) -> str:
            return hashlib.sha1(data).hexdigest()

        def retrieve_avatar(self, jid: JID, id: str, **pubsubkwargs) -> Future:
            """Retrieve an avatar.

            :param jid: JID of the entity to get the avatar from.
            :param id: Identifier of the item containing the avatar.
            """
            return self.xmpp['xep_0060'].get_item(
                jid,
                Data.namespace,
                id,
                **pubsubkwargs
            )

        def publish_avatar(self, data: bytes, **pubsubkwargs) -> Future:
            """Publish an avatar.

            :param data: The avatar, in bytes representation.
            """
            payload = Data()
            payload['value'] = data
            return self.xmpp['xep_0163'].publish(
                payload,
                id=self.generate_id(data),
                **pubsubkwargs
            )

        def publish_avatar_metadata(self, items: Optional[MetadataItems] = None,
                                    pointers: Optional[Iterable[Pointer]] = None,
                                    **pubsubkwargs) -> Future:
            """Publish avatar metadata.

            :param items: Metadata items to store
            :param pointers: Optional pointers
            """
            metadata = MetaData()
            if items is None:
                items = []
            if not isinstance(items, (list, set)):
                items = [items]
            for info in items:
                metadata.add_info(info['id'], info['type'], info['bytes'],
                        height=info.get('height', ''),
                        width=info.get('width', ''),
                        url=info.get('url', ''))

            if pointers is not None:
                for pointer in pointers:
                    metadata.add_pointer(pointer)

            return self.xmpp['xep_0163'].publish(
                metadata,
                id=info['id'],
                **pubsubkwargs
            )

        def stop(self, **pubsubkwargs) -> Future:
            """
            Clear existing avatar metadata information to stop notifications.
            """
            metadata = MetaData()
            return self.xmpp['xep_0163'].publish(
                metadata,
                node=MetaData.namespace,
                **pubsubkwargs
            )