502 lines
18 KiB
Python
502 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
|
|
|
|
# Libervia plugin to handle events
|
|
# Copyright (C) 2009-2026 Jérôme Poisson (goffi@goffi.org)
|
|
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
|
|
# 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 dataclasses import dataclass
|
|
import json
|
|
from textwrap import dedent
|
|
from typing import Callable, Dict, List, Optional, Union
|
|
from urllib import parse
|
|
import fnmatch
|
|
|
|
from lxml import etree
|
|
import treq
|
|
from twisted.internet import defer
|
|
|
|
from libervia.backend.core.constants import Const as C
|
|
from libervia.backend.core.core_types import SatXMPPEntity
|
|
from libervia.backend.core.exceptions import ConflictError
|
|
from libervia.backend.core.i18n import _
|
|
from libervia.backend.core.log import getLogger
|
|
from libervia.backend.tools.common import data_format
|
|
from libervia.backend.tools.common.async_utils import async_lru
|
|
|
|
log = getLogger(__name__)
|
|
|
|
PLUGIN_INFO = {
|
|
C.PI_NAME: "Preview",
|
|
C.PI_IMPORT_NAME: "Preview",
|
|
C.PI_TYPE: C.PLUG_TYPE_MISC,
|
|
C.PI_PROTOCOLS: ["Open Graph", "oEmbed"],
|
|
C.PI_DEPENDENCIES: ["TEXT_SYNTAXES"],
|
|
C.PI_MAIN: "Preview",
|
|
C.PI_HANDLER: "no",
|
|
C.PI_DESCRIPTION: dedent(_("""\
|
|
Retrieves and provides a preview of URLs using various protocols. Initially, it
|
|
uses the Open Graph protocol for most web pages. Specialized handlers are
|
|
implemented for YouTube using the oEmbed protocol.
|
|
""")),
|
|
}
|
|
|
|
OG_TAGS = [
|
|
"title",
|
|
"type",
|
|
"image",
|
|
"url",
|
|
"audio",
|
|
"description",
|
|
"determiner",
|
|
"locale",
|
|
"locale:alternate",
|
|
"site_name",
|
|
"video",
|
|
]
|
|
|
|
|
|
class PreviewFetchError(Exception):
|
|
pass
|
|
|
|
|
|
@dataclass
|
|
class Protocol:
|
|
name: str
|
|
callback: Callable
|
|
priority: int
|
|
|
|
|
|
class Preview:
|
|
protocols: Dict[str, Protocol] = {}
|
|
domain_protocols: Dict[str, str] = {}
|
|
|
|
def __init__(self, host):
|
|
log.info(_("Preview plugin initialization"))
|
|
self.host = host
|
|
|
|
# generic protocols
|
|
|
|
self.register("open_graph", self.fetch_open_graph_data, priority=100)
|
|
self.register("oembed", self.fetch_generic_oembed_data, priority=50)
|
|
self.register("generic", self.fetch_generic_data, priority=0)
|
|
|
|
# domain specific protocols
|
|
|
|
self.register("oembed-youtube", self.fetch_youtube_oembed_data, priority=-100)
|
|
self.register_domain_protocol(
|
|
["www.youtube.com", "youtu.be", "m.youtube.com"], "oembed-youtube"
|
|
)
|
|
|
|
self.register("wikipedia", self.fetch_wikipedia_data, priority=-80)
|
|
self.register_domain_protocol(["*.wikipedia.org"], "wikipedia")
|
|
|
|
self.register("invidious", self.fetch_invidious_data, priority=-90)
|
|
self.register_domain_protocol(
|
|
["yewtu.be", "www.yewtu.be", "invidious.fdn.fr"], "invidious"
|
|
)
|
|
|
|
# bridge methods
|
|
|
|
host.bridge.add_method(
|
|
"url_preview_get",
|
|
".plugin",
|
|
in_sign="sss",
|
|
out_sign="s",
|
|
method=self._url_preview_get,
|
|
async_=True,
|
|
)
|
|
|
|
# API
|
|
|
|
def _url_preview_get(
|
|
self, url: str, options: str, profile_key: str
|
|
) -> defer.Deferred:
|
|
client = self.host.get_client(profile_key)
|
|
d = defer.ensureDeferred(
|
|
self.get_preview_data(client, url, data_format.deserialise(options))
|
|
)
|
|
d.addCallback(data_format.serialise)
|
|
return d
|
|
|
|
@async_lru()
|
|
async def get_preview_data(
|
|
self, client: SatXMPPEntity, url: str, options: dict
|
|
) -> Optional[dict]:
|
|
"""Fetch preview data from a url using registered protocols
|
|
|
|
@param url: The url to fetch the preview data from
|
|
@param options: Additional options that may be used while fetching preview data
|
|
@return: A dictionary containing the preview data or None if no data could be
|
|
fetched
|
|
"""
|
|
parsed_url = parse.urlparse(url)
|
|
domain = parsed_url.netloc
|
|
|
|
preview_data: Optional[dict] = None
|
|
matched_protocol = None
|
|
for registered_domain, registered_protocol in self.domain_protocols.items():
|
|
if fnmatch.fnmatch(domain, registered_domain):
|
|
matched_protocol = registered_protocol
|
|
break
|
|
|
|
if matched_protocol is not None:
|
|
callback = self.protocols[matched_protocol].callback
|
|
preview_data = await callback(client, url, options)
|
|
else:
|
|
for name, protocol in sorted(
|
|
self.protocols.items(), key=lambda item: item[1].priority, reverse=True
|
|
):
|
|
try:
|
|
preview_data = await protocol.callback(client, url, options)
|
|
except Exception as e:
|
|
log.warning(f"Can't run protocol {name} for {url}: {e}")
|
|
else:
|
|
if preview_data is not None:
|
|
matched_protocol = protocol.name
|
|
break
|
|
|
|
if preview_data is not None:
|
|
preview_data["protocol"] = matched_protocol
|
|
# we don't clean html for youtube as we need Javascript to make it work, and
|
|
# for invidious as we generate it ourself
|
|
if "html" in preview_data:
|
|
if matched_protocol in ("oembed-youtube", "invidious"):
|
|
# this flag indicate that we know the source of HTML and we should be
|
|
# able to trust it. This will add `allow-scripts` and
|
|
# `allow-same-origin` in the preview <iframe> "sandbox" attribute
|
|
preview_data["html_known"] = True
|
|
else:
|
|
preview_data["html_known"] = False
|
|
clean_xhtml = self.host.plugins["TEXT_SYNTAXES"].clean_xhtml
|
|
try:
|
|
preview_data["html"] = clean_xhtml(preview_data["html"])
|
|
except Exception as e:
|
|
log.warning(f"Can't clean html data: {e}\n{preview_data}")
|
|
del preview_data["html"]
|
|
|
|
return preview_data
|
|
|
|
@classmethod
|
|
def register(cls, name: str, callback: Callable, priority: int = 0):
|
|
"""Register a protocol to retrieve preview data
|
|
|
|
The registered callback should return a dictionary of preview data if available,
|
|
or None otherwise.
|
|
|
|
@param name: Unique name of the protocol
|
|
@param callback: Async callback function to fetch preview data
|
|
@param priority: Priority of the protocol, with higher numbers indicating higher
|
|
priority
|
|
@return: None
|
|
"""
|
|
if name in cls.protocols:
|
|
raise ConflictError(f"Protocol with the name {name} is already registered.")
|
|
|
|
cls.protocols[name] = Protocol(name=name, callback=callback, priority=priority)
|
|
|
|
@classmethod
|
|
def register_domain_protocol(cls, domains: Union[str, List[str]], protocol_name: str):
|
|
"""Register a protocol for a specific domain or list of domains
|
|
|
|
@param domains: The domain name or list of domain names
|
|
@param protocol_name: The name of the protocol to be associated with the domain(s)
|
|
@return: None
|
|
"""
|
|
protocol_name = protocol_name.replace(" ", "").lower()
|
|
if protocol_name not in cls.protocols:
|
|
raise ConflictError(
|
|
f"Protocol with the name {protocol_name} is not registered."
|
|
)
|
|
|
|
if isinstance(domains, str):
|
|
domains = [domains]
|
|
|
|
for domain in domains:
|
|
domain = domain.strip()
|
|
if not domain:
|
|
log.warning("empty string used as domain, ignoring")
|
|
continue
|
|
cls.domain_protocols[domain] = protocol_name
|
|
|
|
# Open Graph
|
|
|
|
async def fetch_open_graph_data(
|
|
self, client: SatXMPPEntity, url: str, options: dict
|
|
) -> Optional[dict]:
|
|
"""Fetch Open Graph data from a url
|
|
|
|
This method implements the Open Graph protocol, details of which can be found at:
|
|
http://ogp.me/
|
|
|
|
@param url: The url to fetch the Open Graph data from
|
|
@param options: Additional options that may be used while fetching data
|
|
@return: A dictionary containing the Open Graph data or None if no data could be
|
|
fetched
|
|
"""
|
|
resp = await treq.get(url)
|
|
|
|
if resp.code == 200:
|
|
html = await resp.text()
|
|
parser = etree.HTMLParser()
|
|
tree = etree.fromstring(html, parser)
|
|
|
|
# Extract Open Graph data
|
|
metadata = {}
|
|
for tag in OG_TAGS:
|
|
og_el = tree.find('.//meta[@property="og:{tag}"]'.format(tag=tag))
|
|
if og_el is not None:
|
|
metadata[tag] = og_el.get("content")
|
|
|
|
if metadata:
|
|
if "site_name" in metadata and not "provider_name" in metadata:
|
|
metadata["provider_name"] = metadata["site_name"]
|
|
return metadata
|
|
|
|
return None
|
|
else:
|
|
raise PreviewFetchError(
|
|
f"Failed to fetch preview for {url}, status code: {resp.code}"
|
|
)
|
|
|
|
# oEmbed
|
|
|
|
async def _fetch_oembed_data(self, oembed_url: str) -> Optional[dict]:
|
|
"""Fetch oEmbed data from a given oEmbed URL
|
|
|
|
@param oembed_url: The url to fetch the oEmbed data from
|
|
@return: A dictionary containing the oEmbed data or None if no data could be
|
|
fetched
|
|
"""
|
|
resp = await treq.get(oembed_url)
|
|
if resp.code == 200:
|
|
return json.loads(await resp.text())
|
|
else:
|
|
raise PreviewFetchError(
|
|
f"Failed to fetch oEmbed preview for {oembed_url}, status code: "
|
|
f"{resp.code}"
|
|
)
|
|
|
|
async def fetch_youtube_oembed_data(
|
|
self, client: SatXMPPEntity, url: str, options: dict
|
|
) -> Optional[dict]:
|
|
"""Fetch YouTube oEmbed data from a url
|
|
|
|
@param url: The url to fetch the YouTube oEmbed data from
|
|
@param options: Additional options that may be used while fetching data
|
|
@return: A dictionary containing the YouTube oEmbed data or None if no data could
|
|
be fetched
|
|
"""
|
|
oembed_url = f"https://www.youtube.com/oembed?url={parse.quote(url)}&format=json"
|
|
data = await self._fetch_oembed_data(oembed_url)
|
|
if data is not None and "html" in data:
|
|
html = data["html"]
|
|
root = etree.HTML(html)
|
|
iframe_elt = root.xpath("//iframe")
|
|
if iframe_elt:
|
|
iframe_elt[0].attrib[
|
|
"style"
|
|
] = "position: absolute; top: 0; left: 0; width: 100%; height: 100%;"
|
|
data["html"] = etree.tostring(root, method="html", encoding="unicode")
|
|
else:
|
|
log.warning("No <iframe> found in the YouTube oEmbed response")
|
|
|
|
return data
|
|
|
|
async def fetch_generic_oembed_data(
|
|
self, client: SatXMPPEntity, url: str, options: dict
|
|
) -> Optional[dict]:
|
|
"""Fetch generic oEmbed data from a url
|
|
|
|
@param url: The url to fetch the oEmbed data from
|
|
@param options: Additional options that may be used while fetching data
|
|
@return: A dictionary containing the oEmbed data or None if no data could be
|
|
fetched
|
|
"""
|
|
resp = await treq.get(url)
|
|
if resp.code == 200:
|
|
html = await resp.text()
|
|
parser = etree.HTMLParser()
|
|
tree = etree.fromstring(html, parser)
|
|
|
|
# Find oEmbed URL
|
|
oembed_link = tree.find('.//link[@type="application/json+oembed"]')
|
|
if oembed_link is not None:
|
|
oembed_url = oembed_link.get("href")
|
|
return await self._fetch_oembed_data(oembed_url)
|
|
else:
|
|
return None
|
|
else:
|
|
raise PreviewFetchError(
|
|
f"Failed to fetch preview for {url}, status code: {resp.code}"
|
|
)
|
|
|
|
async def fetch_generic_data(
|
|
self, client: SatXMPPEntity, url: str, options: dict
|
|
) -> Optional[dict]:
|
|
"""Fetch generic data from a url
|
|
|
|
This method attempts to extract the title, description, and author metadata from
|
|
the HTML of the page. If these data cannot be found, the method will return None.
|
|
|
|
@param url: The url to fetch the generic data from
|
|
@param options: Additional options that may be used while fetching data
|
|
@return: A dictionary containing the generic data or None if no data could be
|
|
fetched
|
|
"""
|
|
resp = await treq.get(url)
|
|
if resp.code == 200:
|
|
html = await resp.text()
|
|
parser = etree.HTMLParser()
|
|
tree = etree.fromstring(html, parser)
|
|
|
|
# Find title, description, and author metadata
|
|
title_el = tree.find(".//title")
|
|
desc_el = tree.find('.//meta[@name="description"]')
|
|
author_el = tree.find('.//meta[@name="author"]')
|
|
|
|
metadata = {
|
|
"title": title_el.text if title_el is not None else "",
|
|
"description": desc_el.get("content") if desc_el is not None else "",
|
|
"author_name": author_el.get("content") if author_el is not None else "",
|
|
"url": url,
|
|
"provider_name": parse.urlparse(url).netloc,
|
|
"provider_url": f"{parse.urlparse(url).scheme}://{parse.urlparse(url).netloc}",
|
|
}
|
|
|
|
return metadata if any(metadata.values()) else None
|
|
else:
|
|
raise PreviewFetchError(
|
|
f"Failed to fetch generic preview for {url}, status code: {resp.code}"
|
|
)
|
|
|
|
# Wikipedia
|
|
|
|
async def fetch_wikipedia_data(
|
|
self, client: SatXMPPEntity, url: str, options: dict
|
|
) -> Optional[dict]:
|
|
"""Fetch Wikipedia data from a url
|
|
|
|
This method implements the Wikipedia API, details of which can be found at:
|
|
https://www.mediawiki.org/wiki/API:Main_page
|
|
|
|
@param url: The url to fetch the Wikipedia data from
|
|
@param options: Additional options that may be used while fetching data
|
|
@return: A dictionary containing the Wikipedia data or None if no data could be
|
|
fetched
|
|
"""
|
|
parsed_url = parse.urlparse(url)
|
|
page_name = parsed_url.path.split("/")[-1]
|
|
|
|
# Use the Wikipedia API to get a summary of the page and a preview image
|
|
api_url = (
|
|
f"https://{parsed_url.netloc}/w/api.php?format=json&action=query&"
|
|
f"prop=extracts|pageimages&exintro&explaintext&redirects=1&piprop=thumbnail"
|
|
f"&pithumbsize=300&titles={page_name}"
|
|
)
|
|
|
|
resp = await treq.get(api_url)
|
|
if resp.code == 200:
|
|
data = json.loads(await resp.text())
|
|
# The page ID is the first key in the "pages" dictionary
|
|
page_id = next(iter(data["query"]["pages"].keys()))
|
|
page = data["query"]["pages"][page_id]
|
|
|
|
# The API may return a page with a missing title or extract if the page does
|
|
# not exist
|
|
if "missing" in page:
|
|
return None
|
|
|
|
return {
|
|
"provider_name": "Wikipedia",
|
|
"provider_url": "https://www.wikipedia.org",
|
|
"title": page.get("title"),
|
|
"description": page.get("extract"),
|
|
"url": url,
|
|
"image": (
|
|
page.get("thumbnail", {}).get("source")
|
|
if "thumbnail" in page
|
|
else None
|
|
),
|
|
}
|
|
else:
|
|
raise PreviewFetchError(
|
|
f"Failed to fetch Wikipedia preview for {url}, status code: {resp.code}"
|
|
)
|
|
|
|
# Invidious
|
|
|
|
async def fetch_invidious_data(
|
|
self, client: SatXMPPEntity, url: str, options: dict
|
|
) -> Optional[dict]:
|
|
"""
|
|
Fetch Invidious data from a url and generate HTML iframe.
|
|
|
|
@param url: The url to fetch the Invidious data from.
|
|
@param options: Additional options that may be used while fetching data.
|
|
@return: A dictionary containing the Invidious data or None if no data could be fetched.
|
|
"""
|
|
parsed_url = parse.urlparse(url)
|
|
if "watch" in parsed_url.path:
|
|
video_id = parse.parse_qs(parsed_url.query).get("v", [None])[0]
|
|
else:
|
|
video_id = parsed_url.path.strip("/")
|
|
if not video_id:
|
|
log.warning(f"Can't extract video ID from {url}")
|
|
return None
|
|
|
|
invidious_api_url = f"https://{parsed_url.netloc}/api/v1/videos/{video_id}"
|
|
|
|
resp = await treq.get(invidious_api_url)
|
|
if resp.code == 200:
|
|
video_data = await resp.json()
|
|
# construct the iframe html code
|
|
html = (
|
|
f"<iframe"
|
|
f' width="100%"'
|
|
f' height="auto"'
|
|
f' src="https://{parsed_url.netloc}/embed/{video_id}"'
|
|
f' frameborder="0" '
|
|
f' allow="'
|
|
f" accelerometer;"
|
|
f" autoplay;"
|
|
f" clipboard-write;"
|
|
f" encrypted-media;"
|
|
f" gyroscope;"
|
|
f' picture-in-picture"'
|
|
f' style="'
|
|
f" position: absolute;"
|
|
f" top: 0;"
|
|
f" left: 0;"
|
|
f" width: 100%;"
|
|
f' height: 100%;"'
|
|
f" allowfullscreen></iframe>"
|
|
)
|
|
# structure the data to be returned
|
|
data = {
|
|
"title": video_data.get("title"),
|
|
"description": video_data.get("description"),
|
|
"url": url,
|
|
"image": video_data.get("videoThumbnails", [{}])[0].get("url"),
|
|
"provider_name": "Invidious",
|
|
"provider_url": f"https://{parsed_url.netloc}",
|
|
"html": html,
|
|
"author_name": video_data.get("author"),
|
|
"author_url": f"https://{parsed_url.netloc}/channel/{video_data.get('authorId')}",
|
|
}
|
|
return data
|
|
else:
|
|
log.warning(f"Unable to fetch video data from Invidious API for {video_id}")
|
|
return None
|