xmpp-discord-bridge/xmpp_discord_bridge/avatar.py

78 lines
2.9 KiB
Python
Raw Normal View History

2021-09-15 20:33:59 +00:00
import logging
2021-09-16 10:57:36 +00:00
import hashlib
import os
2021-09-15 20:33:59 +00:00
from slixmpp.exceptions import IqError
class AvatarManager:
2021-09-16 10:57:36 +00:00
"""
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.
"""
2021-09-15 20:33:59 +00:00
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")
2021-09-16 10:57:36 +00:00
def _save_avatar(self, jid, data, type_):
2021-09-15 20:33:59 +00:00
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"]
2021-09-16 10:57:36 +00:00
self._save_avatar(jid, data, type_)
2021-09-15 20:33:59 +00:00
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)
2021-09-16 10:57:36 +00:00
data = iq["pubsub"]["items"]["substanzas"][0]["data"]
self._save_avatar(jid, data, "image/png")
return True
2021-09-15 20:33:59 +00:00
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):
2021-09-16 10:57:36 +00:00
"""
Attempt to request the avatar of @jid
"""
2021-09-15 20:33:59 +00:00
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)
2021-09-16 10:57:36 +00:00
def get_avatar_url(self, jid):
"""
Returns either the URL to the avatar of @jid or None.
"""
2021-09-15 20:33:59 +00:00
return self._avatars.get(jid, None)