1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
|
"""
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
)
|
