Source code for slixmpp.plugins.xep_0385.sims

import io
import logging
from collections.abc import Iterable
from datetime import datetime
from pathlib import Path
from typing import IO, ClassVar

from slixmpp.plugins import BasePlugin
from slixmpp.plugins.xep_0372.stanza import Reference
from slixmpp.stanza import Message
from slixmpp.xmlstream import register_stanza_plugin

from . import stanza

log = logging.getLogger(__name__)




class XEP_0385(BasePlugin):
    """
    XEP-0385: Stateless Inline Media Sharing (SIMS)

    Only support outgoing SIMS, incoming is not handled at all.
    """

    name = "xep_0385"
    description = "XEP-0385: Stateless Inline Media Sharing (SIMS)"
    dependencies: ClassVar[set[str]] = {"xep_0234", "xep_0300", "xep_0372"}
    stanza = stanza

    def plugin_init(self) -> None:
        register_stanza_plugin(
            self.xmpp.plugin["xep_0372"].stanza.Reference, stanza.Sims
        )
        register_stanza_plugin(Message, stanza.Sims)

        register_stanza_plugin(stanza.Sims, stanza.Sources)
        register_stanza_plugin(stanza.Sims, self.xmpp.plugin["xep_0234"].stanza.File)
        register_stanza_plugin(
            stanza.Sources,
            self.xmpp.plugin["xep_0372"].stanza.Reference,
            iterable=True,
        )

    def get_sims(
        self,
        path: Path | None = None,
        uris: Iterable[str] = (),
        media_type: str | None = None,
        desc: str | None = None,
        data: bytes | None = None,
        file: IO[bytes] | None = None,
    ) -> Reference:
        sims = stanza.Sims()
        for uri in uris:
            ref = self.xmpp.plugin["xep_0372"].stanza.Reference()
            ref["uri"] = uri
            ref["type"] = "data"
            sims["sources"].append(ref)
        if media_type:
            sims["file"]["media-type"] = media_type
        if desc:
            sims["file"]["desc"] = desc
        if path:
            sims["file"]["name"] = path.name
            stat = path.stat()
            sims["file"]["size"] = stat.st_size
            sims["file"]["date"] = datetime.fromtimestamp(stat.st_mtime)
        elif file:
            file.seek(0, io.SEEK_END)
            sims["file"]["size"] = file.tell()
        elif data:
            sims["file"]["size"] = len(data)

        h = self.xmpp.plugin["xep_0300"].compute_hash(
            filename=path, data=data, file=file
        )
        sims["file"].append(h)

        ref = self.xmpp.plugin["xep_0372"].stanza.Reference()
        ref.append(sims)
        ref["type"] = "data"
        return ref