Mercurial > libervia-backend
diff libervia/backend/plugins/plugin_xep_0446.py @ 4334:111dce64dcb5
plugins XEP-0300, XEP-0446, XEP-0447, XEP0448 and others: Refactoring to use Pydantic:
Pydantic models are used more and more in Libervia, for the bridge API, and also to
convert `domish.Element` to internal representation.
Type hints have also been added in many places.
rel 453
author | Goffi <goffi@goffi.org> |
---|---|
date | Tue, 03 Dec 2024 00:12:38 +0100 |
parents | 0d7bb4df2343 |
children |
line wrap: on
line diff
--- a/libervia/backend/plugins/plugin_xep_0446.py Tue Dec 03 00:11:00 2024 +0100 +++ b/libervia/backend/plugins/plugin_xep_0446.py Tue Dec 03 00:12:38 2024 +0100 @@ -15,16 +15,17 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. -from logging import exception -from typing import Optional, Union, Tuple, Dict, Any from pathlib import Path +from typing import Self, cast +from pydantic import BaseModel, Field from twisted.words.xish import domish +from libervia.backend.core import exceptions from libervia.backend.core.constants import Const as C from libervia.backend.core.i18n import _ from libervia.backend.core.log import getLogger -from libervia.backend.core import exceptions +from libervia.backend.plugins.plugin_xep_0300 import XEP_0300 from libervia.backend.tools import utils log = getLogger(__name__) @@ -45,26 +46,171 @@ NS_FILE_METADATA = "urn:xmpp:file:metadata:0" +class FileMetadata(BaseModel): + """ + Model for file metadata. + """ + + name: str | None = Field(None, description="Name of the file.") + media_type: str | None = Field(None, description="Media type of the file.") + desc: str | None = Field(None, description="Description of the file.") + size: int | None = Field(None, description="Size of the file in bytes.") + file_hash: tuple[str, str] | None = Field( + None, description="File hash as a tuple of (algo, hash)." + ) + date: float | int | None = Field( + None, description="Timestamp of the last modification datetime." + ) + width: int | None = Field(None, description="Image or video width in pixels.") + height: int | None = Field(None, description="Image or video height in pixels.") + length: int | None = Field(None, description="Video or audio length in milliseconds.") + thumbnail: str | None = Field(None, description="URL to a thumbnail.") + + _hash: XEP_0300 | None = None + + @classmethod + def from_element(cls, file_metadata_elt: domish.Element) -> Self: + """Create a FileMetadata instance from a <file> element or its parent. + + @param file_metadata_elt: The <file> element or a parent element. + @return: FileMetadata instance. + @raise exceptions.NotFound: If the <file> element is not found. + """ + assert cls._hash is not None, "_hash attribute is not set" + + if file_metadata_elt.uri != NS_FILE_METADATA or file_metadata_elt.name != "file": + child_file_metadata_elt = next( + file_metadata_elt.elements(NS_FILE_METADATA, "file"), None + ) + if child_file_metadata_elt is None: + raise exceptions.NotFound("<file> element not found") + else: + file_metadata_elt = child_file_metadata_elt + + kwargs = {} + for key in ( + "name", + "media_type", + "desc", + "size", + "date", + "width", + "height", + "length", + ): + elt = next(file_metadata_elt.elements(NS_FILE_METADATA, key), None) + if elt is not None: + if key in ("name", "media_type", "desc"): + content = str(elt) + if key == "name": + content = Path(content).name + kwargs[key] = content + elif key in ("size", "width", "height", "length"): + kwargs[key] = int(str(elt)) + elif key == "date": + kwargs[key] = utils.parse_xmpp_date(str(elt)) + + hash_elt = next(file_metadata_elt.elements(NS_FILE_METADATA, "hash"), None) + if hash_elt: + try: + algo, hash_ = cls._hash.parse_hash_elt(hash_elt) + except exceptions.NotFound: + pass + except exceptions.DataError: + from libervia.backend.tools.xml_tools import p_fmt_elt + + log.warning(f"invalid <hash/> element:\n{p_fmt_elt(file_metadata_elt)}") + else: + kwargs["file_hash"] = (algo, hash_) + + return cls(**kwargs) + + def to_element(self) -> domish.Element: + """Build the <file> element from this instance's data. + + @return: <file> element. + """ + assert self._hash is not None, "_hash attribute is not set" + + file_elt = domish.Element((NS_FILE_METADATA, "file")) + + for key, value in ( + ("name", self.name), + ("media-type", self.media_type), + ("desc", self.desc), + ("size", self.size), + ("width", self.width), + ("height", self.height), + ("length", self.length), + ): + if value is not None: + file_elt.addElement(key, content=str(value)) + + if self.file_hash is not None: + hash_algo, hash_ = self.file_hash + file_elt.addChild(self._hash.build_hash_elt(hash_, hash_algo)) + + if self.date is not None: + file_elt.addElement("date", content=utils.xmpp_date(self.date)) + + if self.thumbnail is not None: + log.warning("thumbnail is not implemented yet") + + return file_elt + + @classmethod + def from_filedata_dict(cls, file_data: dict) -> Self: + """Create an instance of FileMetadata from data dict as returned by ``memory``. + + @param data: A filedata dict as returned by ``memory.get_files`` + @return: An instance of FileMetadata. + """ + + # Extracting relevant fields + name = file_data["name"] + media_type = f'{file_data["media_type"]}/{file_data["media_subtype"]}' + desc = None # TODO + size = file_data["size"] + file_hash = (file_data["hash_algo"], file_data["file_hash"]) + date = file_data.get("modified") or file_data["created"] + width = None # TODO + height = None # TODO + length = None # TODO + thumbnail = None # TODO + + return cls( + name=name, + media_type=media_type, + desc=desc, + size=size, + file_hash=file_hash, + date=date, + width=width, + height=height, + length=length, + thumbnail=thumbnail, + ) + + class XEP_0446: - def __init__(self, host): log.info(_("XEP-0446 (File Metadata Element) plugin initialization")) host.register_namespace("file-metadata", NS_FILE_METADATA) - self._hash = host.plugins["XEP-0300"] + FileMetadata._hash = cast(XEP_0300, host.plugins["XEP-0300"]) - def get_file_metadata_elt( + def generate_file_metadata( self, - name: Optional[str] = None, - media_type: Optional[str] = None, - desc: Optional[str] = None, - size: Optional[int] = None, - file_hash: Optional[Tuple[str, str]] = None, - date: Optional[Union[float, int]] = None, - width: Optional[int] = None, - height: Optional[int] = None, - length: Optional[int] = None, - thumbnail: Optional[str] = None, - ) -> domish.Element: + name: str | None = None, + media_type: str | None = None, + desc: str | None = None, + size: int | None = None, + file_hash: tuple[str, str] | None = None, + date: float | int | None = None, + width: int | None = None, + height: int | None = None, + length: int | None = None, + thumbnail: str | None = None, + ) -> FileMetadata: """Generate the element describing a file @param name: name of the file @@ -79,87 +225,25 @@ @param thumbnail: URL to a thumbnail @return: ``<file/>`` element """ - if name: - name = Path(name).name - file_elt = domish.Element((NS_FILE_METADATA, "file")) - for name, value in ( - ("name", name), - ("media-type", media_type), - ("desc", desc), - ("size", size), - ("width", width), - ("height", height), - ("length", length), - ): - if value is not None: - file_elt.addElement(name, content=str(value)) - if file_hash is not None: - hash_algo, hash_ = file_hash - file_elt.addChild(self._hash.build_hash_elt(hash_, hash_algo)) - if date is not None: - file_elt.addElement("date", utils.xmpp_date(date)) - if thumbnail is not None: - # TODO: implement thumbnails - log.warning("thumbnail is not implemented yet") - return file_elt + return FileMetadata( + name=name, + media_type=media_type, + desc=desc, + size=size, + file_hash=file_hash, + date=date, + width=width, + height=height, + length=length, + thumbnail=thumbnail, + ) - def parse_file_metadata_elt( - self, file_metadata_elt: domish.Element - ) -> Dict[str, Any]: + def parse_file_metadata_elt(self, file_metadata_elt: domish.Element) -> FileMetadata: """Parse <file/> element @param file_metadata_elt: <file/> element a parent element can also be used - @return: file metadata. It's a dict whose keys correspond to - [get_file_metadata_elt] parameters + @return: FileMetadata instance. @raise exceptions.NotFound: no <file/> element has been found """ - - if file_metadata_elt.name != "file": - try: - file_metadata_elt = next( - file_metadata_elt.elements(NS_FILE_METADATA, "file") - ) - except StopIteration: - raise exceptions.NotFound - data: Dict[str, Any] = {} - - for key, type_ in ( - ("name", str), - ("media-type", str), - ("desc", str), - ("size", int), - ("date", "timestamp"), - ("width", int), - ("height", int), - ("length", int), - ): - elt = next(file_metadata_elt.elements(NS_FILE_METADATA, key), None) - if elt is not None: - if type_ in (str, int): - content = str(elt) - if key == "name": - # we avoid malformed names or names containing path elements - content = Path(content).name - elif key == "media-type": - key = "media_type" - data[key] = type_(content) - elif type == "timestamp": - data[key] = utils.parse_xmpp_date(str(elt)) - else: - raise exceptions.InternalError - - try: - algo, hash_ = self._hash.parse_hash_elt(file_metadata_elt) - except exceptions.NotFound: - pass - except exceptions.DataError: - from libervia.backend.tools.xml_tools import p_fmt_elt - - log.warning("invalid <hash/> element:\n{p_fmt_elt(file_metadata_elt)}") - else: - data["file_hash"] = (algo, hash_) - - # TODO: thumbnails - - return data + return FileMetadata.from_element(file_metadata_elt)