Source code for slixmpp.plugins.xep_0447.sfs
import logging
from datetime import datetime
from pathlib import Path
from typing import Iterable, Optional
from slixmpp.plugins import BasePlugin
from slixmpp.stanza import Message
from slixmpp.xmlstream import register_stanza_plugin
from . import stanza
log = logging.getLogger(__name__)
[docs]class XEP_0447(BasePlugin):
"""
XEP-0447: Stateless File Sharing
Only support outgoing SFS, incoming is not handled at all.
"""
name = "xep_0447"
description = "XEP-0447: Stateless File Sharing"
dependencies = {"xep_0300", "xep_0446"}
stanza = stanza
def plugin_init(self):
register_stanza_plugin(Message, stanza.StatelessFileSharing)
register_stanza_plugin(stanza.StatelessFileSharing, stanza.Sources)
register_stanza_plugin(
stanza.StatelessFileSharing, self.xmpp["xep_0446"].stanza.File
)
register_stanza_plugin(stanza.Sources, stanza.UrlData, iterable=True)
def get_sfs(
self,
path: Path,
uris: Iterable[str],
media_type: Optional[str],
desc: Optional[str],
):
sfs = stanza.StatelessFileSharing()
sfs["disposition"] = "inline"
for uri in uris:
ref = stanza.UrlData()
ref["target"] = uri
sfs["sources"].append(ref)
if media_type:
sfs["file"]["media-type"] = media_type
if desc:
sfs["file"]["desc"] = desc
sfs["file"]["name"] = path.name
stat = path.stat()
sfs["file"]["size"] = stat.st_size
sfs["file"]["date"] = datetime.fromtimestamp(stat.st_mtime)
h = self.xmpp.plugin["xep_0300"].compute_hash(path)
h["value"] = h["value"].decode()
sfs["file"].append(h)
return sfs