erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

plugin URL preview: URL preview first draft

Browse source
This commit is contained in:
Goffi 2023-06-27 15:48:15 +02:00
commit 53ee4011e3
1 changed files with 503 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

503
libervia/backend/plugins/plugin_misc_url_preview.py Normal file
View file

@ -0,0 +1,503 @@
#!/usr/bin/env python3
# Libervia plugin to handle events
# Copyright (C) 2009-2022 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