79 lines
3.0 KiB
Python
79 lines
3.0 KiB
Python
import logging
|
|
import hashlib
|
|
import os
|
|
|
|
from slixmpp.exceptions import IqError
|
|
|
|
class AvatarManager:
|
|
"""
|
|
The AvatarManager class is responsible for aquiring the XMPP avatars
|
|
using the means of either XEP-0153 (XEP-0054), XEP-0084 or (in the future)
|
|
XEP-0292, storing and providing a publicly available URL to those avatars.
|
|
"""
|
|
def __init__(self, xmpp, config):
|
|
self._xmpp = xmpp
|
|
self._path = config["avatars"]["path"]
|
|
self._public = config["avatars"]["url"]
|
|
|
|
self._avatars = {}
|
|
|
|
self._logger = logging.getLogger("xmpp.avatar")
|
|
|
|
def _save_avatar(self, jid, data, type_):
|
|
filename = hashlib.sha1(data).hexdigest() + "." + type_.split("/")[1]
|
|
path = os.path.join(self._path, filename)
|
|
|
|
if os.path.exists(path):
|
|
self._logger.debug("Avatar for %s already exists, not saving it again",
|
|
jid)
|
|
else:
|
|
with open(path, "wb") as f:
|
|
f.write(data)
|
|
|
|
self._avatars[jid] = filename
|
|
|
|
async def try_xep_0153(self, jid):
|
|
try:
|
|
iq = await self._xmpp.plugin["xep_0054"].get_vcard(jid=jid,
|
|
ifrom=self._xmpp._bot_jid_full)
|
|
type_ = iq["vcard_temp"]["PHOTO"]["TYPE"]
|
|
data = iq["vcard_temp"]["PHOTO"]["BINVAL"]
|
|
self._save_avatar(jid, data, type_)
|
|
return True
|
|
except IqError:
|
|
self._logger.debug("Avatar retrieval via XEP-0054/XEP-0153 failed. Probably no vCard for XEP-0054 published")
|
|
return False
|
|
|
|
async def try_xep_0084(self, jid):
|
|
try:
|
|
iq = await self._xmpp.plugin["xep_0060"].get_items(jid=jid,
|
|
node="urn:xmpp:avatar:data",
|
|
max_items=1,
|
|
ifrom=self._xmpp._bot_jid_full)
|
|
data = iq["pubsub"]["items"]["substanzas"][0]["data"]
|
|
self._save_avatar(jid, data, "image/png")
|
|
return True
|
|
except IqError:
|
|
self._logger.debug("Avatar retrieval via XEP-0084 failed. Probably no avatar published or subscription model not fulfilled.")
|
|
return False
|
|
|
|
async def aquire_avatar(self, jid):
|
|
"""
|
|
Attempt to request the avatar of @jid
|
|
"""
|
|
for f in [self.try_xep_0153, self.try_xep_0084]:
|
|
if await f(jid):
|
|
self._logger.debug("Avatar retrieval successful for %s",
|
|
jid)
|
|
return
|
|
|
|
self._logger.debug("Avatar retrieval failed for %s. Giving up.",
|
|
jid)
|
|
|
|
def get_avatar_url(self, jid):
|
|
"""
|
|
Returns either the URL to the avatar of @jid or None.
|
|
"""
|
|
filename = self._avatars.get(jid, None)
|
|
return self._public + filename if filename else None
|