libervia-backend: libervia/backend/plugins/plugin_xep

comparison libervia/backend/plugins/plugin_xep_0277.py @ 4175:30f7513e5590

plugin XEP-0277: generate and parse altertate links with the new `alt_links` data in `extra`

author	Goffi <goffi@goffi.org>
date	Tue, 05 Dec 2023 13:14:03 +0100
parents	0e48181d50ab
children	cf0ea77f9537

comparison

equal deleted inserted replaced

-:6929dabf3a7e
+:30f7513e5590
 # 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/>.
 import time
+from urllib.parse import quote, urlparse
 import dateutil
 import calendar
 from mimetypes import guess_type
 from secrets import token_urlsafe
 from typing import List, Optional, Dict, Tuple, Any, Dict
 """
 if service is None:
 service = client.jid.userhostJID()
 extra: Dict[str, Any] = {}
-microblog_data: Dict[str, Any] = {
+mb_data: Dict[str, Any] = {
 "service": service.full(),
 "extra": extra
 }
 def check_conflict(key, increment=False):
 @param increment(bool): if suffix the key with an increment
 instead of raising an exception
 @raise exceptions.DataError: the key already exists
 (not raised if increment is True)
 """
-if key in microblog_data:
+if key in mb_data:
 if not increment:
 raise failure.Failure(
 exceptions.DataError(
 "key {} is already present for item {}"
 ).format(key, item_elt["id"])
 )
 else:
 idx = 1  # the idx 0 is the key without suffix
 fmt = "{}#{}"
 new_key = fmt.format(key, idx)
-while new_key in microblog_data:
+while new_key in mb_data:
 idx += 1
 new_key = fmt.format(key, idx)
 key = new_key
 return key
 _("Content of type XHTML must declare its namespace!")
 )
 )
 key = check_conflict("{}_xhtml".format(elem.name))
 data = data_elt.toXml()
-microblog_data[key] = yield self.host.plugins["TEXT_SYNTAXES"].clean_xhtml(
+mb_data[key] = yield self.host.plugins["TEXT_SYNTAXES"].clean_xhtml(
 data
 )
 else:
 key = check_conflict(elem.name)
-microblog_data[key] = str(elem)
+mb_data[key] = str(elem)
 id_ = item_elt.getAttribute("id", "")  # there can be no id for transient nodes
-microblog_data["id"] = id_
+mb_data["id"] = id_
 if item_elt.uri not in (pubsub.NS_PUBSUB, NS_PUBSUB_EVENT):
 msg = "Unsupported namespace {ns} in pubsub item {id_}".format(
 ns=item_elt.uri, id_=id_
 )
 log.warning(msg)
 raise failure.Failure(exceptions.DataError(msg))
 # uri
 # FIXME: node should alway be set in the future, check FIXME in method signature
 if node is not None:
-microblog_data["node"] = node
+mb_data["node"] = node
-microblog_data['uri'] = xmpp_uri.build_xmpp_uri(
+mb_data['uri'] = xmpp_uri.build_xmpp_uri(
 "pubsub",
 path=service.full(),
 node=node,
 item=id_,
 )
 # language
 try:
-microblog_data["language"] = entry_elt[(C.NS_XML, "lang")].strip()
+mb_data["language"] = entry_elt[(C.NS_XML, "lang")].strip()
 except KeyError:
 pass
 # atom:id
 try:
 id_elt = next(entry_elt.elements(NS_ATOM, "id"))
 except StopIteration:
 msg = ("No atom id found in the pubsub item {}, this is not standard !"
 .format(id_))
 log.warning(msg)
-microblog_data["atom_id"] = ""
+mb_data["atom_id"] = ""
 else:
-microblog_data["atom_id"] = str(id_elt)
+mb_data["atom_id"] = str(id_elt)
 # title/content(s)
 # FIXME: ATOM and XEP-0277 only allow 1 <title/> element
 #        but in the wild we have some blogs with several ones
 for content_elt in entry_elt.elements(NS_ATOM, "content"):
 yield parseElement(content_elt)
 # we check that text content is present
 for key in ("title", "content"):
-if key not in microblog_data and ("{}_xhtml".format(key)) in microblog_data:
+if key not in mb_data and ("{}_xhtml".format(key)) in mb_data:
 log.warning(
 "item {id_} provide a {key}_xhtml data but not a text one".format(
 id_=id_, key=key
 )
 )
 # ... and do the conversion if it's not
-microblog_data[key] = yield self.host.plugins["TEXT_SYNTAXES"].convert(
+mb_data[key] = yield self.host.plugins["TEXT_SYNTAXES"].convert(
-microblog_data["{}_xhtml".format(key)],
+mb_data["{}_xhtml".format(key)],
 self.host.plugins["TEXT_SYNTAXES"].SYNTAX_XHTML,
 self.host.plugins["TEXT_SYNTAXES"].SYNTAX_TEXT,
 False,
 )
-if "content" not in microblog_data:
+if "content" not in mb_data:
 # use the atom title data as the microblog body content
-microblog_data["content"] = microblog_data["title"]
+mb_data["content"] = mb_data["title"]
-del microblog_data["title"]
+del mb_data["title"]
-if "title_xhtml" in microblog_data:
+if "title_xhtml" in mb_data:
-microblog_data["content_xhtml"] = microblog_data["title_xhtml"]
+mb_data["content_xhtml"] = mb_data["title_xhtml"]
-del microblog_data["title_xhtml"]
+del mb_data["title_xhtml"]
 # published/updated dates
 try:
 updated_elt = next(entry_elt.elements(NS_ATOM, "updated"))
 except StopIteration:
 msg = "No atom updated element found in the pubsub item {}".format(id_)
 raise failure.Failure(exceptions.DataError(msg))
-microblog_data["updated"] = calendar.timegm(
+mb_data["updated"] = calendar.timegm(
 dateutil.parser.parse(str(updated_elt)).utctimetuple()
 )
 try:
 published_elt = next(entry_elt.elements(NS_ATOM, "published"))
 except StopIteration:
-microblog_data["published"] = microblog_data["updated"]
+mb_data["published"] = mb_data["updated"]
 else:
-microblog_data["published"] = calendar.timegm(
+mb_data["published"] = calendar.timegm(
 dateutil.parser.parse(str(published_elt)).utctimetuple()
 )
 # links
-comments = microblog_data['comments'] = []
+comments = mb_data['comments'] = []
 for link_elt in entry_elt.elements(NS_ATOM, "link"):
 href = link_elt.getAttribute("href")
 if not href:
 log.warning(
 f'missing href in <link> element: {link_elt.toXml()}'
 if media_type is not None:
 attachment["media_type"] = media_type
 attachments = extra.setdefault("attachments", [])
 attachments.append(attachment)
+elif rel == "alternate":
+link_data = {"url": href}
+media_type = link_elt.getAttribute("type") or guess_type(href)[0]
+if media_type:
+link_data["media_type"] = media_type
+else:
+log.warning(
+f"Invalid or missing media type for alternate link: {href}"
+)
+extra.setdefault("alt_links", []).append(link_data)
 else:
 log.warning(
 f"Unmanaged link element: {link_elt.toXml()}"
 )
 log.warning(
 "No name element found in author element of item {}".format(id_)
 )
 author = None
 else:
-author = microblog_data["author"] = str(name_elt).strip()
+author = mb_data["author"] = str(name_elt).strip()
 # uri
 try:
 uri_elt = next(author_elt.elements(NS_ATOM, "uri"))
 except StopIteration:
 log.debug(
 "No uri element found in author element of item {}".format(id_)
 )
 if publisher:
-microblog_data["author_jid"] = publisher
+mb_data["author_jid"] = publisher
 else:
 uri = str(uri_elt)
 if uri.startswith("xmpp:"):
 uri = uri[5:]
-microblog_data["author_jid"] = uri
+mb_data["author_jid"] = uri
 else:
-microblog_data["author_jid"] = (
+mb_data["author_jid"] = (
 item_elt.getAttribute("publisher") or ""
 )
-if not author and microblog_data["author_jid"]:
+if not author and mb_data["author_jid"]:
 # FIXME: temporary workaround for missing author name, would be
 #   better to use directly JID's identity (to be done from frontends?)
 try:
-microblog_data["author"] = jid.JID(microblog_data["author_jid"]).user
+mb_data["author"] = jid.JID(mb_data["author_jid"]).user
 except Exception as e:
 log.warning(f"No author name found, and can't parse author jid: {e}")
 if not publisher:
 log.debug("No publisher attribute, we can't verify author jid")
-microblog_data["author_jid_verified"] = False
+mb_data["author_jid_verified"] = False
 elif jid.JID(publisher).userhostJID() == jid.JID(uri).userhostJID():
-microblog_data["author_jid_verified"] = True
+mb_data["author_jid_verified"] = True
 else:
 if "repeated" not in extra:
 log.warning(
 "item atom:uri differ from publisher attribute, spoofing "
 "attempt ? atom:uri = {} publisher = {}".format(
 uri, item_elt.getAttribute("publisher")
 )
 )
-microblog_data["author_jid_verified"] = False
+mb_data["author_jid_verified"] = False
 # email
 try:
 email_elt = next(author_elt.elements(NS_ATOM, "email"))
 except StopIteration:
 pass
 else:
-microblog_data["author_email"] = str(email_elt)
+mb_data["author_email"] = str(email_elt)
-if not microblog_data.get("author_jid"):
+if not mb_data.get("author_jid"):
 if publisher:
-microblog_data["author_jid"] = publisher
+mb_data["author_jid"] = publisher
-microblog_data["author_jid_verified"] = True
+mb_data["author_jid_verified"] = True
 else:
 iq_elt = xml_tools.find_ancestor(item_elt, "iq", C.NS_STREAM)
-microblog_data["author_jid"] = iq_elt["from"]
+mb_data["author_jid"] = iq_elt["from"]
-microblog_data["author_jid_verified"] = False
+mb_data["author_jid_verified"] = False
 # categories
 categories = [
 category_elt.getAttribute("term", "")
 for category_elt in entry_elt.elements(NS_ATOM, "category")
 ]
-microblog_data["tags"] = categories
+mb_data["tags"] = categories
 ## the trigger ##
 # if other plugins have things to add or change
 yield self.host.trigger.point(
-"XEP-0277_item2data", item_elt, entry_elt, microblog_data
+"XEP-0277_item2data", item_elt, entry_elt, mb_data
 )
-defer.returnValue(microblog_data)
+defer.returnValue(mb_data)
 async def mb_data_2_entry_elt(self, client, mb_data, item_id, service, node):
 """Convert a data dict to en entry usable to create an item
 @param mb_data: data dict as given by bridge method.
 ("size", "lenght")
 ):
 value = attachment.get(key)
 if value:
 link_elt[attr]  = str(value)
+## alternate links ##
+alt_links = extra.get("alt_links")
+if alt_links:
+for link_data in alt_links:
+url_template = link_data["url"]
+url = url_template.format(
+service=quote(service.full(), safe=""),
+node=quote(node, safe=""),
+item=quote(item_id, safe="")
+)
+link_elt = entry_elt.addElement("link")
+link_elt["href"] = url
+link_elt["rel"] = "alternate"
+media_type = link_data.get("media_type")
+if not media_type:
+parsed_url = urlparse(url)
+if parsed_url.scheme in ["http", "https"]:
+media_type = "text/html"
+else:
+media_type = guess_type(url)[0] or "application/octet-stream"
+link_elt["type"] = media_type
 ## author ##
 author_elt = entry_elt.addElement("author")
 try:
 author_name = mb_data["author"]

Mercurial > libervia-backend

comparison libervia/backend/plugins/plugin_xep_0277.py @ 4175:30f7513e5590