Mercurial > libervia-backend
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"