changeset 4391:c2228563bf0f

plugin XEP-0060: Add a model to handle node metadata: rel 463
author Goffi <goffi@goffi.org>
date Fri, 29 Aug 2025 18:53:03 +0200
parents 5e48634ccada
children dcda916f16f6
files libervia/backend/plugins/plugin_xep_0060.py
diffstat 1 files changed, 143 insertions(+), 4 deletions(-) [+]
line wrap: on
line diff
--- a/libervia/backend/plugins/plugin_xep_0060.py	Fri Aug 29 18:52:58 2025 +0200
+++ b/libervia/backend/plugins/plugin_xep_0060.py	Fri Aug 29 18:53:03 2025 +0200
@@ -19,11 +19,10 @@
 
 from collections import namedtuple
 from functools import reduce
-from typing import Any, Callable, Dict, Iterable, List, Optional, Set, Tuple, Union
-import urllib.error
+from typing import Any, Callable, ClassVar, Dict, Iterable, List, Optional, Self, Set, Union
 import urllib.parse
-import urllib.request
 
+from pydantic import BaseModel, Field
 from twisted.internet import defer, reactor
 from twisted.words.protocols.jabber import error, jid
 from twisted.words.xish import domish
@@ -40,6 +39,7 @@
 from libervia.backend.core.i18n import _
 from libervia.backend.core.log import getLogger
 from libervia.backend.core.core_types import SatXMPPClient
+from libervia.backend.models.types import JIDType
 from libervia.backend.plugins.plugin_xep_0059 import RSMRequest
 from libervia.backend.tools import utils
 from libervia.backend.tools import sat_defer
@@ -77,7 +77,146 @@
 }
 
 
-class XEP_0060(object):
+class NodeMetadata(BaseModel):
+    """Node metadata model."""
+    model_config = {
+        "extra": "allow"
+    }
+
+    type: str | None = None
+    creator: JIDType | None = None
+    creation_date: float | None = None
+    title: str | None = None
+    description: str | None = None
+    language: str | None = None
+    contact: list[JIDType] | None = None
+    access_model: str | None = None
+    publish_model: str | None = None
+    max_items: int | None = None
+    owner: list[JIDType] | None = None
+    publisher: list[JIDType] | None = None
+    num_subscribers: int | None = None
+
+    _fields_defs: ClassVar[dict] = {
+        "pubsub#type": {"type": "text-single"},
+        "pubsub#creator": {"type": "jid-single"},
+        "pubsub#creation_date": {"type": "text-single"},
+        "pubsub#title": {"type": "text-single"},
+        "pubsub#description": {"type": "text-single"},
+        "pubsub#language": {"type": "list-single"},
+        "pubsub#contact": {"type": "jid-multi"},
+        "pubsub#access_model": {"type": "list-single"},
+        "pubsub#publish_model": {"type": "list-single"},
+        "pubsub#max_items": {"type": "text-single"},
+        "pubsub#owner": {"type": "jid-multi"},
+        "pubsub#publisher": {"type": "jid-multi"},
+        "pubsub#num_subscribers": {"type": "text-single"},
+    }
+
+    @classmethod
+    def from_data_form(cls, form: data_form.Form) -> Self:
+        """Create a NodeMetadata instance from a data form.
+
+        @param form: Data form containing node metadata.
+        @return: Filled instance of this class.
+        @raise TypeError: Type of the form do not correspond to what is expected according
+            to specifications.
+        """
+        fields = {}
+        form.typeCheck(cls._fields_defs)
+
+        for field in form.fields.values():
+            if field.var == "pubsub#type":
+                fields["type"] = field.value
+            elif field.var == "pubsub#creator":
+                fields["creator"] = field.value
+            elif field.var == "pubsub#creation_date":
+                fields["creation_date"] = field.value
+            elif field.var == "pubsub#title":
+                fields["title"] = field.value
+            elif field.var == "pubsub#description":
+                fields["description"] = field.value
+            elif field.var == "pubsub#language":
+                fields["language"] = field.value
+            elif field.var == "pubsub#contact":
+                fields["contact"] = field.values
+            elif field.var == "pubsub#access_model":
+                fields["access_model"] = field.value
+            elif field.var == "pubsub#publish_model":
+                fields["publish_model"] = field.value
+            elif field.var == "pubsub#max_items":
+                if field.value == "max":
+                    fields["max_items"] = -1
+                else:
+                    try:
+                        fields["max_items"] = int(field.value)
+                    except (ValueError, TypeError):
+                        log.warning(f"Invalid max_items found: {field.value!r}.")
+                        fields["max_items"] = None
+            elif field.var == "pubsub#owner":
+                fields["owner"] = field.values
+            elif field.var == "pubsub#publisher":
+                fields["publisher"] = field.values
+            elif field.var == "pubsub#num_subscribers":
+                try:
+                    fields["num_subscribers"] = int(field.value)
+                except (ValueError, TypeError):
+                    log.warning(f"Invalid num_subscribers found: {field.value!r}.")
+                    fields["num_subscribers"] = None
+            elif field.var is None:
+                continue
+            else:
+                # We use debug and not warning here because fields can be handled by
+                # subclasses.
+                log.debug(f"Ignored field: {field.var!r} (values: {field.values!r}).")
+
+        return cls(**fields)
+
+    def to_data_form(self) -> data_form.Form:
+        """Convert this instance to a data form.
+
+        @return: Data form representation of this instance.
+        """
+        form = data_form.Form(formType="result", formNamespace=pubsub.NS_PUBSUB_META_DATA)
+
+        # Handle max_items conversion from -1 to "max" for the spec.
+        if self.max_items is not None:
+            max_items_value = "max" if self.max_items == -1 else str(self.max_items)
+        else:
+            max_items_value = None
+
+        fields_map = {
+            "pubsub#type": self.type,
+            "pubsub#creator": self.creator,
+            "pubsub#creation_date": self.creation_date,
+            "pubsub#title": self.title,
+            "pubsub#description": self.description,
+            "pubsub#language": self.language,
+            "pubsub#contact": self.contact,
+            "pubsub#access_model": self.access_model,
+            "pubsub#publish_model": self.publish_model,
+            "pubsub#max_items": max_items_value,
+            "pubsub#owner": self.owner,
+            "pubsub#publisher": self.publisher,
+            "pubsub#num_subscribers": (
+                str(self.num_subscribers)
+                if self.num_subscribers is not None
+                else None
+            ),
+        }
+        form.makeFields(
+            {k: v for k, v in fields_map.items() if v},
+            fieldDefs=self._fields_defs,
+        )
+
+        return form
+
+    def to_element(self) -> domish.Element:
+        """Generate the <x> element corresponding to this form."""
+        return self.to_data_form().toElement()
+
+
+class XEP_0060:
     OPT_ACCESS_MODEL = "pubsub#access_model"
     OPT_PERSIST_ITEMS = "pubsub#persist_items"
     OPT_MAX_ITEMS = "pubsub#max_items"