diff --git a/libervia/backend/plugins/plugin_xep_0084.py b/libervia/backend/plugins/plugin_xep_0084.py
index 53fffcf8..6465bdc2 100644
--- a/libervia/backend/plugins/plugin_xep_0084.py
+++ b/libervia/backend/plugins/plugin_xep_0084.py
@@ -16,23 +16,22 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see .
-from typing import Any, Dict
-
from base64 import b64decode
from pathlib import Path
+from typing import Any
from twisted.internet import defer
+from twisted.words.protocols.jabber import error, jid
from twisted.words.protocols.jabber.xmlstream import XMPPHandler
-from twisted.words.protocols.jabber import jid, error
from twisted.words.xish import domish
from zope.interface import implementer
-from tx_xmpp import disco, itx_xmpp, pubsub
+from libervia.backend.core import exceptions
from libervia.backend.core.constants import Const as C
+from libervia.backend.core.core_types import SatXMPPEntity
from libervia.backend.core.i18n import _
from libervia.backend.core.log import getLogger
-from libervia.backend.core.core_types import SatXMPPEntity
-from libervia.backend.core import exceptions
+from tx_xmpp import disco, itx_xmpp, pubsub
log = getLogger(__name__)
@@ -80,55 +79,65 @@ class XEP_0084:
client = self.host.get_client(profile)
defer.ensureDeferred(self.on_metadata_update(client, itemsEvent))
+ @staticmethod
+ def _parse_metadata_info(
+ item_elt: domish.Element,
+ ) -> tuple[str, str, int] | None:
+ """Parse avatar metadata item and extract ID, media type, and size.
+
+ @param item_elt: pubsub item element containing metadata.
+ @return: (avatar_id, media_type, size) if a valid info element is found,
+ None if avatar is disabled or metadata is invalid.
+ """
+ try:
+ metadata_elt = next(item_elt.elements(NS_AVATAR_METADATA, "metadata"))
+ except StopIteration:
+ log.warning(f"missing metadata element: {item_elt.toXml()}")
+ return None
+
+ for info_elt in metadata_elt.elements(NS_AVATAR_METADATA, "info"):
+ try:
+ avatar_id = str(info_elt["id"])
+ media_type = str(info_elt["type"])
+ size = int(info_elt["bytes"])
+ except (KeyError, ValueError):
+ log.warning(f"invalid element: {item_elt.toXml()}")
+ continue
+ if not avatar_id:
+ continue
+ # FIXME: to simplify, we only handle image/png for now
+ if media_type == "image/png":
+ return (avatar_id, media_type, size)
+
+ # mandatory image/png is missing, or avatar is disabled
+ # (https://xmpp.org/extensions/xep-0084.html#pub-disable)
+ return None
+
async def on_metadata_update(
self, client: SatXMPPEntity, itemsEvent: pubsub.ItemsEvent
) -> None:
"""Handle PEP avatar metadata update event.
- Parse metadata, fetch data if not cached, and update identity cache.
+ @param client: Client instance.
+ @param itemsEvent: pubsub items event with avatar metadata.
"""
entity = itemsEvent.sender
service = entity.userhostJID()
-
items = itemsEvent.items
- if not items:
- return
- if items[0].name == "retract":
+ if not items or items[0].name == "retract":
log.debug(f"Avatar retracted for {entity.full()}")
await self._i.update(client, IMPORT_NAME, "avatar", None, entity)
return
- item_elt = items[0]
- try:
- metadata_elt = next(item_elt.elements(NS_AVATAR_METADATA, "metadata"))
- except StopIteration:
- log.warning(f"missing metadata element: {item_elt.toXml()}")
- return
-
- for info_elt in metadata_elt.elements(NS_AVATAR_METADATA, "info"):
- try:
- metadata = {
- "id": str(info_elt["id"]),
- "size": int(info_elt["bytes"]),
- "media_type": str(info_elt["type"]),
- }
- avatar_id = metadata["id"]
- if not avatar_id:
- raise ValueError
- except (KeyError, ValueError):
- log.warning(f"invalid element: {item_elt.toXml()}")
- return None
- # FIXME: to simplify, we only handle image/png for now
- if metadata["media_type"] == "image/png":
- break
- else:
- # mandatory image/png is missing, or avatar is disabled
- # (https://xmpp.org/extensions/xep-0084.html#pub-disable)
+ parsed = self._parse_metadata_info(items[0])
+ if parsed is None:
log.debug(f"Avatar disabled for {entity.full()}")
await self._i.update(client, IMPORT_NAME, "avatar", None, entity)
return
+ avatar_id, media_type, _size = parsed
+
# Check if avatar is already cached
try:
cached = self.host.memory.get_entity_datum(client, entity, "avatar")
@@ -148,7 +157,7 @@ class XEP_0084:
cache_data = self.host.common_cache.get_metadata(avatar_id)
if not cache_data:
cache_data = await self._fetch_avatar_data(
- client, service, avatar_id, metadata["media_type"]
+ client, service, avatar_id, media_type
)
if cache_data is None:
log.warning(f"Failed to fetch avatar data for {entity.full()}")
@@ -199,28 +208,40 @@ class XEP_0084:
return {"path": Path(f.name), "mime_type": media_type}
async def get_avatar(self, client: SatXMPPEntity, entity_jid: jid.JID) -> dict | None:
- """Get avatar data from cache."""
+ """Get avatar data by fetching from PEP metadata node.
+
+ @param client: Client instance.
+ @param entity_jid: entity to get avatar for.
+ @return: avatar metadata dict if found, None otherwise.
+ """
+ service = entity_jid.userhostJID()
try:
- cached = self.host.memory.get_entity_datum(client, entity_jid, "avatar")
- except (KeyError, exceptions.UnknownEntityError):
+ data_items, __ = await self._p.get_items(client, service, NS_AVATAR_METADATA)
+ except error.StanzaError:
return None
- if not cached:
+ if not data_items:
return None
- cache_uid = cached.get("cache_uid")
- filename = cached.get("filename")
- if not cache_uid or not filename:
+ parsed = self._parse_metadata_info(data_items[0])
+ if parsed is None:
return None
- try:
- self.host.common_cache.get_metadata(cache_uid)
- except Exception:
- return None
+ avatar_id, media_type, _size = parsed
- return cached
+ cache_data = self.host.common_cache.get_metadata(avatar_id)
+ if not cache_data:
+ cache_data = await self._fetch_avatar_data(
+ client, service, avatar_id, media_type
+ )
+ if cache_data is None:
+ return None
- def build_item_data_elt(self, avatar_data: Dict[str, Any]) -> domish.Element:
+ return self._i.avatar_build_metadata(
+ cache_data["path"], cache_data["mime_type"], avatar_id
+ )
+
+ def build_item_data_elt(self, avatar_data: dict[str, Any]) -> domish.Element:
"""Generate the item for the data node
@param avatar_data: data as build by identity plugin (need to be filled with
@@ -230,7 +251,7 @@ class XEP_0084:
data_elt.addContent(avatar_data["base64"])
return pubsub.Item(id=avatar_data["cache_uid"], payload=data_elt)
- def build_item_metadata_elt(self, avatar_data: Dict[str, Any]) -> domish.Element:
+ def build_item_metadata_elt(self, avatar_data: dict[str, Any]) -> domish.Element:
"""Generate the item for the metadata node
@param avatar_data: data as build by identity plugin (need to be filled with
@@ -246,7 +267,7 @@ class XEP_0084:
return pubsub.Item(id=self._p.ID_SINGLETON, payload=metadata_elt)
async def set_avatar(
- self, client: SatXMPPEntity, avatar_data: Dict[str, Any], entity: jid.JID
+ self, client: SatXMPPEntity, avatar_data: dict[str, Any], entity: jid.JID
) -> None:
"""Set avatar of the profile