SimpleerTube/main.py

609 lines
21 KiB
Python

from quart import Quart, request, render_template, redirect
from datetime import datetime
from dateutil import parser as dateutil
from math import ceil
import peertube
import html2text
import sys
h2t = html2text.HTML2Text()
h2t.ignore_links = True
# Wrapper, only containing information that's important for us, and in some cases provides simplified ways to get information
class VideoWrapper:
def __init__(self, a, quality):
self.name = a["name"]
self.uuid = a["uuid"]
self.channel = a["channel"]
self.description = a["description"]
self.thumbnailPath = a["thumbnailPath"]
self.category = a["category"]
self.licence = a["licence"]
self.language = a["language"]
self.captions = a["captions"]
self.privacy = a["privacy"]
self.tags = a["tags"]
self.views = a["views"]
self.likes = a["likes"]
self.dislikes = a["dislikes"]
self.embedPath = a["embedPath"]
self.commentsEnabled = a["commentsEnabled"]
self.resolutions = []
self.video = None
self.files = a["files"]
if len(self.files) == 0:
self.files = ((a["streamingPlaylists"])[0])["files"]
self.default_res = None
for entry in self.files:
resolution = (entry["resolution"])["id"]
self.resolutions.append(entry["resolution"])
# chose the default quality
if resolution != 0 and quality == None:
if self.default_res == None:
self.default_res = resolution
self.video = entry["fileUrl"]
elif abs(720 - resolution) < abs(720 - self.default_res):
self.default_res = resolution
self.video = entry["fileUrl"]
if str(resolution) == str(quality):
self.video = entry["fileUrl"]
if quality == None:
self.quality = self.default_res
else:
self.quality = quality
self.no_quality_selected = not self.video
# Helper Class for using caches
class Cache:
def __init__(self, criteria = lambda diff: diff.days > 0):
self.dict = {}
self.criteria = criteria
def get(self, arg, func):
if arg in self.dict:
last_time_updated = (self.dict[arg])[1]
time_diff = datetime.now() - last_time_updated
if self.criteria(time_diff):
self.dict[arg] = [
func(arg),
datetime.now()
]
else:
self.dict[arg] = [
func(arg),
datetime.now()
]
return (self.dict[arg])[0]
cached_instance_names = Cache()
cached_account_infos = Cache()
cached_video_channel_infos = Cache()
cached_subscriptions = Cache(criteria = lambda diff: diff.total_seconds() > 60)
cached_account_videos = Cache(criteria = lambda diff: diff.total_seconds() > 1800)
cached_channel_videos = Cache(criteria = lambda diff: diff.total_seconds() > 1800)
# cache the instance names so we don't have to send a request to the domain every time someone
# loads any site
def get_instance_name(domain):
return cached_instance_names.get(domain, peertube.get_instance_name)
# simple wrapper that is used inside the cached_account_infos
def get_account(info):
info = info.split("@")
return peertube.account(info[1], info[0])
def get_account_info(name):
return cached_account_infos.get(name, get_account)
# simple wrapper that is used inside the cached_video_channel_infos
def get_video_channel(info):
info = info.split("@")
return peertube.video_channel(info[1], info[0])
def get_video_channel_info(name):
return cached_video_channel_infos.get(name, get_video_channel)
# Get latest remote videos from channel name
def get_latest_channel_videos(name):
return cached_channel_videos.get(name, latest_channel_videos)
# Refresh latest remote videos from channel name
def latest_channel_videos(name):
print("[CACHE] Refreshing channel videos for %s" % name)
(name, domain) = name.split('@')
return peertube.video_channel_videos(domain, name, 0)
# Get latest remote videos from account name
def get_latest_account_videos(name):
return cached_account_videos.get(name, latest_account_videos)
# Refresh latest remote videos from account name
def latest_account_videos(name):
print("[CACHE] Refreshing account videos for %s" % name)
(name, domain) = name.split('@')
return peertube.account_videos(domain, name, 0)
# Get local accounts subscriptions, as specified in accounts.list
def get_subscriptions_accounts():
return cached_subscriptions.get("accounts", load_subscriptions_accounts)
# Refresh local accounts subscriptions
def load_subscriptions_accounts(_):
return load_subscriptions("accounts")
# Get the latest videos from local accounts subscriptions, ordered by most recent; only return `limit` number of videos
def get_subscriptions_accounts_videos(limit=12):
latest = []
for sub in get_subscriptions_accounts():
result = get_latest_account_videos(sub)
if "error" not in result:
account_latest = get_latest_account_videos(sub)["data"]
latest.extend(account_latest)
else:
print("[WARN] Unable to get content from account " + sub)
latest.sort(key = lambda vid: dateutil.isoparse(vid["createdAt"]), reverse=True)
return latest[0:limit]
# Get local channels subscriptions, as specified in channel.list
def get_subscriptions_channels():
return cached_subscriptions.get("channels", load_subscriptions_channels)
# Refresh local channels subscriptions
def load_subscriptions_channels(_):
return load_subscriptions("channels")
# Load subscriptions from a file called `kind`.list (60s cache)
def load_subscriptions(kind):
print("[CACHE] Refreshing subscriptions %s from %s.list" % (kind, kind))
try:
with open(kind + '.list', 'r') as f:
subscriptions = map(find_subscription, f.read().splitlines())
except Exception as e:
print("No `channels.list` file to load for local subscriptions")
subscriptions = []
# Remove comment entries and empty lines
return list(filter(lambda entry: entry != '', subscriptions))
# Builds a unified id@server from one of those syntaxes, additionally stripping extra whitespace and ignoring `#` as comments:
# - id@server
# - @id@server
# - http(s)://server/c/id
# - http(s)://server/a/id
def find_subscription(request):
identifier = request
identifier = identifier.split('#')[0].strip()
# Comment line is returned as empty string
if identifier == '': return ''
if identifier.startswith('@'):
# Strip @ from identifier
return identifier[1:]
if identifier.startswith('http'):
identifier = identifier[4:]
# HTTPS?
if identifier.startswith('s'): identifier = identifier[1:]
# Remove ://
identifier = identifier[3:]
parts = identifier.split('/')
domain = parts[0]
if parts[1] == 'a' or parts[1] == 'c':
# Account or channel found, take the next part
return parts[2] + '@' + domain
else:
# Just check there's an @ in there and it should be fine
if '@' in identifier:
return identifier
# No match was found, we don't understand this URL
print("[WARN] Identifier not understood from local subscriptions:\n%s" % request)
return ''
# Get the latest videos from local channels subscriptions, ordered by most recent; only return `limit` number of videos
def get_subscriptions_channels_videos(limit=12):
latest = []
for sub in get_subscriptions_channels():
result = get_latest_channel_videos(sub)
if "error" not in result:
channel_latest = get_latest_channel_videos(sub)["data"]
latest.extend(channel_latest)
else:
print("[WARN] Unable to get content from channel " + sub)
latest.sort(key = lambda vid: dateutil.isoparse(vid["createdAt"]), reverse=True)
return latest[0:limit]
# Get the latest videos from local channels and accounts subscriptions combined, ordered by most recent; only return `limit` number of videos; NOTE: duplicates are not handled, why would you add both an account and the corresponding channel?
def get_subscriptions_videos(limit=12):
latest = get_subscriptions_channels_videos(limit=limit)
latest.extend(get_subscriptions_accounts_videos(limit=limit))
# TODO: maybe refactor so we don't have to reorder twice? Or maybe the get_ functions can take a ordered=True argument? In this case here, it would be false, because we sort after
latest.sort(key = lambda vid: dateutil.isoparse(vid["createdAt"]), reverse=True)
return latest[0:limit]
# Get the info about local accounts subscriptions
def get_subscriptions_accounts_info():
return map(lambda sub: get_account_info(sub), get_subscriptions_accounts())
# Get the info about local channels subscriptions
def get_subscriptions_channels_info():
return map(lambda sub: get_video_channel_info(sub), get_subscriptions_channels())
# Get the info about local subscriptions for accounts and channels, as a tuple of lists
def get_subscriptions_info():
list = []
list.extend(get_subscriptions_accounts_info())
list.extend(get_subscriptions_channels_info())
return list
app = Quart(__name__)
@app.route("/")
async def main():
videos = get_subscriptions_videos(limit=12)
# Inside subscriptions variable, you may find either an account info structure, or a channel info structure. Channels may be recognized due to `ownerAccount` property.
subscriptions = get_subscriptions_info()
return await render_template(
"index.html",
videos=videos,
subscriptions=subscriptions,
)
@app.route("/instance", methods=["POST"])
async def jump_to_instance():
domain = (await request.form)["domain"]
return redirect("/" + domain)
@app.route("/search", methods = ["POST"])
async def simpleer_search_redirect():
query = (await request.form)["query"]
return redirect("/search/" + query)
@app.route("/search", methods = ["GET"])
async def simpleer_search_get_redirect():
query = request.args.get("query")
return redirect("/search/" + query)
@app.route("/search/<string:query>", defaults = {"page": 1})
@app.route("/search/<string:query>/<int:page>")
async def simpleer_search(query, page):
results = peertube.sepia_search(query, (page - 1) * 10)
return await render_template(
"simpleer_search_results.html",
results = results,
query = query,
# details for pagination
page=page,
pages_total=ceil(results["total"] / 10),
)
@app.route("/<string:domain>/")
async def instance(domain):
# favicon.ico is not a domain name
if domain == "favicon.ico": return await favicon()
return redirect("/" + domain + "/videos/trending")
@app.route("/<string:domain>/videos/local", defaults = {"page": 1})
@app.route("/<string:domain>/videos/local/<int:page>")
async def instance_videos_local(domain, page):
vids = peertube.get_videos_local(domain, (page - 1) * 10)
return await render_template(
"instance/local.html",
domain=domain,
instance_name=get_instance_name(domain),
videos = vids,
# details for pagination
page=page,
pagination_url="/" + domain + "/videos/local/",
pages_total=ceil(vids["total"] / 10),
)
@app.route("/<string:domain>/videos/trending", defaults = {"page": 1})
@app.route("/<string:domain>/videos/trending/<int:page>")
async def instance_videos_trending(domain, page):
vids = peertube.get_videos_trending(domain, (page - 1) * 10)
return await render_template(
"instance/trending.html",
domain=domain,
instance_name=get_instance_name(domain),
videos = vids,
# details for pagination
page=page,
pagination_url="/" + domain + "/videos/trending/",
pages_total=ceil(vids["total"] / 10),
)
@app.route("/<string:domain>/videos/most-liked", defaults = {"page": 1})
@app.route("/<string:domain>/videos/most-liked/<int:page>")
async def instance_videos_most_liked(domain, page):
vids = peertube.get_videos_most_liked(domain, (page - 1) * 10)
return await render_template(
"instance/most-liked.html",
domain=domain,
instance_name=get_instance_name(domain),
videos = vids,
# details for pagination
page=page,
pagination_url="/" + domain + "/videos/most-liked/",
pages_total=ceil(vids["total"] / 10),
)
@app.route("/<string:domain>/videos/recently-added", defaults = {"page": 1})
@app.route("/<string:domain>/videos/recently-added/<int:page>")
async def instance_videos_recently_added(domain, page):
vids = peertube.get_videos_recently_added(domain, (page - 1) * 10)
return await render_template(
"instance/recently-added.html",
domain=domain,
instance_name=get_instance_name(domain),
videos = vids,
# details for pagination
page=page,
pagination_url="/" + domain + "/videos/recently-added/",
pages_total=ceil(vids["total"] / 10),
)
@app.route("/<string:domain>/search", methods=["POST"])
async def search_redirect(domain):
query = (await request.form)["query"]
return redirect("/" + domain + "/search/" + query)
@app.route("/<string:domain>/search/<string:term>", defaults = {"page": 1})
@app.route("/<string:domain>/search/<string:term>/<int:page>")
async def search(domain, term, page):
results = peertube.search(domain, term, (page - 1) * 10)
return await render_template(
"search_results.html",
domain=domain,
instance_name=get_instance_name(domain),
results=results,
search_term=term,
# details for pagination
page=page,
pagination_url="/" + domain + "/search/" + term + "/",
pages_total=(results["total"] / 10)
)
@app.route("/<string:domain>/w/<string:id>/")
@app.route("/<string:domain>/videos/watch/<string:id>/")
async def video(domain, id):
data = peertube.video(domain, id)
data["captions"] = peertube.video_captions(domain, id)
quality = request.args.get("quality")
embed = request.args.get("embed")
vid = VideoWrapper(data, quality)
quality = int(vid.quality)
# only make a request for the comments if commentsEnabled
comments = ""
if data["commentsEnabled"]:
comments = peertube.get_comments(domain, id)
# Strip the HTML from the comments and convert them to plain text
new_comments = {"total": comments["total"], "data": []}
for comment in comments["data"]:
text = h2t.handle(comment["text"]).strip().strip("\n")
comment["text"] = text
new_comments["data"].append(comment)
comments = new_comments
return await render_template(
"video.html",
domain=domain,
instance_name=get_instance_name(domain),
video=vid,
comments=comments,
quality=quality,
embed=embed,
)
def build_channel_or_account_name(domain, name):
if '@' in name:
return name
return name + "@" + domain
# --- Accounts ---
@app.route("/<string:domain>/accounts/<string:name>")
async def accounts_redirect(domain, name):
return redirect("/" + domain + "/accounts/" + name + "/video-channels")
@app.route("/<string:domain>/accounts/<string:name>/video-channels", defaults = {"page": 1})
@app.route("/<string:domain>/accounts/<string:name>/video-channels/<int:page>")
async def account__video_channels(domain, name, page):
video_channels = peertube.account_video_channels(domain, name, (page - 1) * 10)
return await render_template(
"accounts/video_channels.html",
domain=domain,
instance_name=get_instance_name(domain),
name = name,
account = get_account_info(build_channel_or_account_name(domain, name)),
video_channels = video_channels,
# details for pagination
page=page,
pagination_url="/" + domain + "/accounts/" + name + "/video-channels/",
pages_total=ceil(video_channels["total"] / 10)
)
@app.route("/<string:domain>/accounts/<string:name>/videos", defaults = {"page": 1})
@app.route("/<string:domain>/accounts/<string:name>/videos/<int:page>")
async def account__videos(domain, name, page):
vids = peertube.account_videos(domain, name, (page - 1) * 10)
return await render_template(
"accounts/videos.html",
domain=domain,
instance_name=get_instance_name(domain),
name = name,
account = get_account_info(build_channel_or_account_name(domain, name)),
videos = vids,
# details for pagination
page=page,
pagination_url="/" + domain + "/accounts/" + name + "/videos/",
pages_total=ceil(vids["total"] / 10)
)
@app.route("/<string:domain>/accounts/<string:name>/about")
async def account__about(domain, name):
return await render_template(
"accounts/about.html",
domain=domain,
instance_name=get_instance_name(domain),
name = name,
account = get_account_info(build_channel_or_account_name(domain, name)),
about = peertube.account(domain, name)
)
# --- Video-Channels ---
@app.route("/<string:domain>/video-channels/<string:name>")
async def video_channels_redirect(domain, name):
return redirect("/" + domain + "/video-channels/" + name + "/videos")
@app.route("/<string:domain>/video-channels/<string:name>/videos", defaults = {"page": 1})
@app.route("/<string:domain>/video-channels/<string:name>/videos/<int:page>")
async def video_channels__videos(domain, name, page):
vids = peertube.video_channel_videos(domain, name, (page - 1) * 10)
return await render_template(
"video_channels/videos.html",
domain=domain,
instance_name=get_instance_name(domain),
name = name,
video_channel = get_video_channel_info(build_channel_or_account_name(domain, name)),
page=page,
pagination_url="/" + domain + "/video-channels/" + name + "/videos/",
pages_total=ceil(vids["total"] / 10),
videos = vids,
)
@app.route("/<string:domain>/video-channels/<string:name>/video-playlists", defaults = {"page": 1})
@app.route("/<string:domain>/video-channels/<string:name>/video-playlists/<int:page>")
async def video_channels__video_playlists(domain, name, page):
video_playlists = peertube.video_channel_video_playlists(domain, name, (page - 1) * 10)
return await render_template(
"video_channels/video_playlists.html",
domain=domain,
instance_name=get_instance_name(domain),
name = name,
video_channel = get_video_channel_info(build_channel_or_account_name(domain, name)),
video_playlists = video_playlists,
page=page,
pagination_url="/" + domain + "/video-channels/" + name + "/video-playlists/",
pages_total=ceil(video_playlists["total"] / 10)
)
@app.route("/<string:domain>/video-channels/<string:name>/about")
async def video_channels__about(domain, name):
return await render_template(
"video_channels/about.html",
domain=domain,
instance_name=get_instance_name(domain),
name = name,
video_channel = get_video_channel_info(build_channel_or_account_name(domain, name)),
about = peertube.video_channel(domain, name)
)
# --- Subtitles/captions proxying ---
@app.route("/<string:domain>/videos/watch/<string:id>/<string:lang>.vtt")
async def subtitles(domain, id, lang):
try:
captions = peertube.video_captions(domain, id)
for entry in captions["data"]:
if entry["language"]["id"] == lang: return peertube.video_captions_download(domain, entry["captionPath"].split('/')[-1])
return await render_template(
"error.html",
error_number = "404",
error_reason = "This video has no subtitles/captions inthe requested language"
), 404
except Exception as e:
return await render_template(
"error.html",
error_number = "500",
error_reason = e
), 500
# --- Favicon ---
@app.route("/favicon.ico")
async def favicon():
return await render_template(
"error.html",
error_number = "404",
error_reason = "We don't have a favicon yet. If you would like to contribute one, please send it to ~metalune/public-inbox@lists.sr.ht"
), 404
# --- OpenSearch ---
@app.route("/opensearch.xml")
async def opensearch():
try:
with open('opensearch.xml', 'r') as f:
return f.read().replace('$BASEURL', request.headers["Host"])
except Exception as e:
return await render_template(
"error.html",
error_number = "500",
error_reason = e
), 500
if __name__ == "__main__":
if len(sys.argv) == 3:
interface = sys.argv[1]
port = sys.argv[2]
elif len(sys.argv) == 2:
interface = "127.0.0.1"
port = sys.argv[1]
else:
interface = "127.0.0.1"
port = "5000"
app.run(host=interface, port=port)