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
146
147
148
149
150
151
152
153
154
155
156
|
# 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.
import logging
from typing import Optional
from slixmpp import JID
from slixmpp.stanza import Iq
from slixmpp.exceptions import XMPPError
from slixmpp.xmlstream import register_stanza_plugin
from slixmpp.xmlstream.handler import CoroutineCallback
from slixmpp.xmlstream.matcher import StanzaPath
from slixmpp.plugins import BasePlugin
from slixmpp.plugins.xep_0054 import VCardTemp, stanza
log = logging.getLogger(__name__)
class XEP_0054(BasePlugin):
"""
XEP-0054: vcard-temp
"""
name = 'xep_0054'
description = 'XEP-0054: vcard-temp'
dependencies = {'xep_0030', 'xep_0082'}
stanza = stanza
def plugin_init(self):
"""
Start the XEP-0054 plugin.
"""
register_stanza_plugin(Iq, VCardTemp)
self.api.register(self._set_vcard, 'set_vcard', default=True)
self.api.register(self._get_vcard, 'get_vcard', default=True)
self.api.register(self._del_vcard, 'del_vcard', default=True)
self._vcard_cache = {}
self.xmpp.register_handler(
CoroutineCallback('VCardTemp',
StanzaPath('iq/vcard_temp'),
self._handle_get_vcard))
def plugin_end(self):
self.xmpp.remove_handler('VCardTemp')
self.xmpp['xep_0030'].del_feature(feature='vcard-temp')
def session_bind(self, jid):
self.xmpp['xep_0030'].add_feature('vcard-temp')
def make_vcard(self) -> VCardTemp:
"""Return an empty vcard element."""
return VCardTemp()
async def get_vcard(self, jid: Optional[JID] = None, *,
local: Optional[bool] = None, cached: bool = False,
ifrom: Optional[JID] = None,
**iqkwargs) -> Iq:
"""Retrieve a VCard.
.. versionchanged:: 1.8.0
This function is now a coroutine.
:param jid: JID of the entity to fetch the VCard from.
:param local: Only check internally for a vcard.
:param cached: Whether to check in the local cache before
sending a query.
"""
if local is None:
if jid is not None and not isinstance(jid, JID):
jid = JID(jid)
if self.xmpp.is_component:
if jid.domain == self.xmpp.boundjid.domain:
local = True
else:
if str(jid) == str(self.xmpp.boundjid):
local = True
jid = jid.full
elif jid in (None, ''):
local = True
if local:
vcard = await self.api['get_vcard'](jid, None, ifrom)
if not isinstance(vcard, Iq):
iq = self.xmpp.Iq()
if vcard is None:
vcard = VCardTemp()
iq.append(vcard)
return iq
return vcard
if cached:
vcard = await self.api['get_vcard'](jid, None, ifrom)
if vcard is not None:
if not isinstance(vcard, Iq):
iq = self.xmpp.Iq()
iq.append(vcard)
return iq
return vcard
iq = self.xmpp.make_iq_get(ito=jid, ifrom=ifrom)
iq.enable('vcard_temp')
return await iq.send(**iqkwargs)
async def publish_vcard(self, vcard: Optional[VCardTemp] = None,
jid: Optional[JID] = None,
ifrom: Optional[JID] = None, **iqkwargs):
"""Publish a vcard.
.. versionchanged:: 1.8.0
This function is now a coroutine.
:param vcard: The VCard to publish.
:param jid: The JID to publish the VCard to.
"""
await self.api['set_vcard'](jid, None, ifrom, vcard)
if self.xmpp.is_component:
return
iq = self.xmpp.make_iq_set(ito=jid, ifrom=ifrom)
iq.append(vcard)
await iq.send(**iqkwargs)
async def _handle_get_vcard(self, iq: Iq):
if iq['type'] == 'result':
await self.api['set_vcard'](jid=iq['from'], args=iq['vcard_temp'])
return
elif iq['type'] == 'get' and self.xmpp.is_component:
vcard = await self.api['get_vcard'](iq['to'].bare, ifrom=iq['from'])
if isinstance(vcard, Iq):
vcard.send()
else:
iq = iq.reply()
iq.append(vcard)
iq.send()
elif iq['type'] == 'set':
raise XMPPError('service-unavailable')
# =================================================================
def _set_vcard(self, jid, node, ifrom, vcard):
self._vcard_cache[jid.bare] = vcard
def _get_vcard(self, jid, node, ifrom, vcard):
return self._vcard_cache.get(jid.bare, None)
def _del_vcard(self, jid, node, ifrom, vcard):
if jid.bare in self._vcard_cache:
del self._vcard_cache[jid.bare]
|
