420 lines
21 KiB
Python
420 lines
21 KiB
Python
from quart import Quart, request, render_template, redirect
|
|
from math import ceil
|
|
#from peertube import Cache, PaginatedResults, FailedRequest, API
|
|
from peertube import *
|
|
import sys
|
|
import logging
|
|
|
|
# Import everything from utils.py, to keep main.py focused
|
|
# on CLI arguments parsing, and web routes declarations
|
|
from utils import *
|
|
|
|
cache = Cache()
|
|
api = API(cache)
|
|
|
|
from logging.config import dictConfig
|
|
|
|
dictConfig({
|
|
'version': 1,
|
|
'loggers': {
|
|
'quart.app': {
|
|
'level': 'DEBUG',
|
|
},
|
|
},
|
|
})
|
|
|
|
# --- WEBSERVER STUFF ---
|
|
# All utilities are above this point.
|
|
# Routes are defined below
|
|
app = Quart(__name__)
|
|
|
|
from werkzeug.routing import BaseConverter
|
|
class LangConverter(BaseConverter):
|
|
regex = r"[a-zA-Z]{2}"
|
|
app.url_map.converters['lang'] = LangConverter
|
|
|
|
# Simple class to designate a previous result, to be consumed in
|
|
# Context.insert_future_result()
|
|
class PreviousResult():
|
|
def __init__(self, key):
|
|
self.context_key = key
|
|
|
|
def key(self):
|
|
return self.context_key
|
|
|
|
# Class to build rendering context from, can be chained like
|
|
# context = Context().paginate(URL, start=0, count=10)
|
|
# .instance(domain)
|
|
# TODO: Since some context parsing will in fact produce fallible requests, how to error?
|
|
# Do we store self.failures(default None)? Then render can check for failures and render error.html instead
|
|
class Context():
|
|
def __init__(self, api):
|
|
self.context = {}
|
|
self.failures = []
|
|
self.api = api
|
|
|
|
# Helper function that's called by the lambda for passing context
|
|
def build(self, _ignored_old_api):
|
|
return self.context
|
|
|
|
# Helper function that's called during render for passing errors
|
|
def failed(self):
|
|
return self.failures
|
|
|
|
# Helper function to insert/log an error
|
|
def error(self, reason):
|
|
app.logger.error(reason)
|
|
self.failures.append(FailedRequest(reason))
|
|
return self
|
|
|
|
# Helper function to insert a successful result into context, or register failure otherwise
|
|
def insert_result(self, key, result):
|
|
if isinstance(result, FailedRequest):
|
|
self.failures.append(result)
|
|
else:
|
|
self.context[key] = result
|
|
return self
|
|
|
|
# Helper function to insert a raw value into context, even if it is a FailedRequest
|
|
def insert(self, key, value):
|
|
self.context[key] = value
|
|
return self
|
|
|
|
# Helper function to insert a successful result into context, from a function and tuple of arguments.
|
|
# If there were previous failures, no further request will be made. If you'd like to reuse a previous
|
|
# successful result, pass an PreviousResult("key") as argument, where key is where your previous result
|
|
# has been stored in the current context.
|
|
# Additionally, two more advanced features are supported:
|
|
# - if key is actually a tuple of strings, and the API responded with a tuple, the multiple values returned
|
|
# by the API will be inserted into those keys in context (no longer used at the moment, previously for pagination)
|
|
# - if the future result is in fact a PaginatedResults, automatically extract "data" from the entry, to insert into
|
|
# `key`, and extract "total" from the entry to insert into pagination["total"]
|
|
def insert_future_result(self, key, func, args):
|
|
#print("[DEBUG] Considering future result %s" % repr(key))
|
|
if len(self.failures) > 0:
|
|
# Previous failures, don't run further requests
|
|
app.logger.debug("Previous failures. Skipping future result %s" % repr(key))
|
|
return self
|
|
# Don't collapse (foo) into foo, keep it tupled
|
|
if not isinstance(args, tuple): args = (args, )
|
|
#print("[DEBUG] Arguments: %s" % args)
|
|
# If an argument is a previous result, extract it
|
|
args = tuple(map(lambda arg: self.unsafe_previous_result(arg), args))
|
|
#print("[DEBUG] Calling with arguments:\n")
|
|
#for arg in args: print("[DEBUG] - %s" % arg)
|
|
|
|
result = func(*args)
|
|
|
|
# Let's check key, if it's a tuple, we expect a tuple from running the API request,
|
|
# and they should be the same length
|
|
if isinstance(key, tuple):
|
|
# Here ("foo", "bar") tuple of context keys was passed as target, so we try to match
|
|
if not isinstance(result, tuple):
|
|
return self.error("PROGRAMMING ERROR: Requested insertion into several context keys, but API responded single response.")
|
|
if len(result) != len(key):
|
|
return self.error("PROGRAMMING ERROR: Requested insertion into %s context keys, but API returned %s arguments." % (str(len(result)), str(len(key))))
|
|
|
|
# Insert API results into every requested context key
|
|
for i in range(0, len(key)):
|
|
self.insert_result(key[i], result[i])
|
|
return self
|
|
|
|
# Simple key to insert_result result into, whether it succeeded or failed
|
|
# Special-case: if result is a PaginatedResults, then we insert the total results count into pagination["total"]
|
|
if isinstance(result, PaginatedResults):
|
|
app.logger.debug("Inserting pagination from API request!")
|
|
self.insert("pagination", { "total": result.total })
|
|
return self.insert_result(key, result.data)
|
|
# Very normal result
|
|
return self.insert_result(key, result)
|
|
|
|
# Extracts previous result from the current context
|
|
# Special-case, when an argument is a tuple of (PreviousResult, key, subkey...) we extract
|
|
# the key[subkey...] from the context's previous result
|
|
# NOTE: NEVER CALL THIS METHOD IF THERE WERE PREVIOUS FAILURES OR YOU'RE NOT SURE A KEY EXISTS
|
|
def unsafe_previous_result(self, arg):
|
|
if isinstance(arg, PreviousResult):
|
|
#print("[DEBUG] Found simple PreviousResult(%s)" % arg.key())
|
|
return self.context[arg.key()]
|
|
elif isinstance(arg, tuple):
|
|
if not isinstance(arg[0], PreviousResult):
|
|
#print("[DEBUG] Found normal tuple argument")
|
|
return arg # Normal tuple argument
|
|
data = self.context[arg[0].key()]
|
|
for entry in range(1, len(arg)):
|
|
# We take each part of the tuple as a "route" inside the previous result
|
|
data = data[arg[entry]]
|
|
app.logger.debug("Found previous entry %s at given route: %s" % (arg[0].key(), data))
|
|
return data
|
|
else:
|
|
#print("[DEBUG] Found normal argument")
|
|
return arg # Normal argument
|
|
|
|
# Build pagination context. Since total page count is fetched from
|
|
def paginate(self, url, page, total_pages):
|
|
self.context["pagination"]["pages"] = total_pages
|
|
self.context["pagination"]["url"] = url
|
|
self.context["pagination"]["current"] = page
|
|
return self
|
|
|
|
# Build pagination context from API results, that were previously inserted
|
|
# with self.insert_future_result(), so you don't have to store the API result in the route
|
|
# declaration, if you don't need to process it further
|
|
# DON'T CALL ME FROM INDEX ROUTE, SEE NOTE BELOW
|
|
def paginate_results(self, results_key, url, page):
|
|
if len(self.failures) > 0:
|
|
# There were previous errors, so we don't paginate, and skip silently
|
|
return self
|
|
if results_key not in self.context:
|
|
return self.error("PROGRAMMING ERROR: Attempting to paginate %s results, which are not found in current context despite no previous errors happening. Did you maybe call paginate_results before insert_future_result? Or maybe the results key %s is misspelled." % (results_key, results_key))
|
|
# TODO: Currently there is no handling of list of results, such as on the subscription page
|
|
# because the only page using it is index, and index currently doesn't support pagination
|
|
# because the merging algorithm is not smart (yet)
|
|
results = self.context[results_key]
|
|
if isinstance(results, FailedRequest):
|
|
# Request failed so we don't build the pagination
|
|
# Also this branch should never be reached because self.insert_result() should prevent it
|
|
return self.error("PROGRAMMING ERROR: Tried to paginate a failed result. You should insert results with context.insert_future_result to prevent this mistake.")
|
|
if "pagination" not in self.context or "total" not in self.context["pagination"]:
|
|
return self.error("PROGRAMMING ERROR: pagination.total count not found in current context. Did you really insert_future_result of an API request that produces paginated results?")
|
|
|
|
total_pages = ceil(self.context["pagination"]["total"] / 10)
|
|
app.logger.debug("Found %s pages when paginating %s" % (str(total_pages), results_key))
|
|
return self.paginate(url, page, total_pages)
|
|
|
|
# Build instance context (eg. domain and instance_name)
|
|
def instance(self, domain):
|
|
self.context["domain"] = domain
|
|
self.insert_future_result("instance", api.instance_name, (domain))
|
|
return self
|
|
|
|
# Build video context (captions, quality, embed, comments) from domain, id and request headers
|
|
def video(self, domain, id, page=1, args={}):
|
|
self.insert_future_result("video", self.api.video, (domain, id)) \
|
|
.insert_future_result("video_info", VideoInfo, (api, PreviousResult("video"), args)) \
|
|
.insert_future_result("captions", api.video_captions, (domain, id)) \
|
|
.insert_future_result("comments", comments, (api, domain, id, (PreviousResult("video"), "commentsEnabled"), (page - 1) * 10, 10)) \
|
|
.paginate_results("comments", "/" + domain + "/videos/watch/" + id + "/", page) \
|
|
.insert("quality", args.get("quality")) \
|
|
.insert("embed", args.get("embed"))
|
|
return self
|
|
|
|
# Build context for account/channel subscriptions. Failed requests are not considered critical, and are simply
|
|
# appended to the "failures" context entry, otherwise a single request would crash the whole homepage.
|
|
def subscriptions(self, page=1, lang=None):
|
|
# Let's follow subscriptions from the current working dir
|
|
# TODO: make it configurable, maybe by "playlist" defined in config?
|
|
subscriptions = Subscriptions('.', self.api, cache)
|
|
|
|
info = subscriptions.info()
|
|
# Lang-filtering is done deeper in the API, so if a channel has 3 differents videos per publication for different languages,
|
|
# we can still get the expected number of recent videos from there instead of only considering the eg. 5 latest videos in said lang
|
|
# TODO: Is it working well? Should write some tests
|
|
videos = subscriptions.videos(page=page, lang=lang)
|
|
failures = info.failures
|
|
failures.extend(videos.failures)
|
|
|
|
filtered_videos = videos.successes
|
|
|
|
self.insert("subscriptions", info.successes)
|
|
self.insert("videos", filtered_videos)
|
|
# Sanitize errors for HTML so we can have newlines in errors but not risk content injection
|
|
self.insert("failures", format_errors_html(failures))
|
|
return self
|
|
|
|
# --- INDEX ROUTE ---
|
|
@app.route("/", defaults = {"page": 1})
|
|
@app.route("/<int:page>")
|
|
async def main(page):
|
|
# TODO: Pagination
|
|
context = Context(api).subscriptions(page=page)
|
|
# Inside subscriptions variable in templates, you may find either an account info structure, or a channel info structure. Channels may be recognized due to `ownerAccount` property.
|
|
# Failed requests do not fail the index.html rendering, instead they are stored in "failures" context key
|
|
return await render("index.html", context.failed(), context.build)
|
|
|
|
@app.route("/<lang:lang>", defaults = {"page": 1})
|
|
@app.route("/<lang:lang>/<int:page>")
|
|
async def main_lang(lang, page):
|
|
# TODO: Pagination
|
|
context = Context(api).subscriptions(page=page, lang=lang)
|
|
return await render("index.html", context.failed(), context.build)
|
|
# --- END INDEX ROUTE ---
|
|
|
|
# --- SEARCH ROUTES ---
|
|
@app.route("/search", methods = ["POST"])
|
|
async def search_post_redirect():
|
|
query = (await request.form)["query"]
|
|
return redirect("/search/" + query)
|
|
|
|
@app.route("/search", methods = ["GET"])
|
|
async def search_get_redirect():
|
|
query = request.args.get("query")
|
|
return redirect("/search/" + query)
|
|
|
|
@app.route("/<string:domain>/search", methods=["POST"])
|
|
async def search_instance_redirect(domain):
|
|
query = (await request.form)["query"]
|
|
return redirect("/" + domain + "/search/" + query)
|
|
|
|
@app.route("/search/<string:query>", defaults = {"page": 1})
|
|
@app.route("/search/<string:query>/<int:page>")
|
|
async def search(query, page):
|
|
context = Context(api).insert_future_result("results", api.search, (query, (page - 1) * 10, 10)) \
|
|
.paginate_results("results", "/search/" + query + "/", page) \
|
|
.insert("query", query)
|
|
return await render("search.html", context.failed(), context.build)
|
|
|
|
@app.route("/<string:domain>/search/<string:query>", defaults = {"page": 1})
|
|
@app.route("/<string:domain>/search/<string:query>/<int:page>")
|
|
async def search_instance(domain, query, page):
|
|
context = Context(api).instance(domain) \
|
|
.insert_future_result("results", api.search_instance, (domain, query, (page - 1) * 10, 10)) \
|
|
.paginate_results("results", "/" + domain + "/search/" + query + "/", page) \
|
|
.insert("query", query)
|
|
return await render("instance/search.html", context.failed(), context.build)
|
|
# --- END SEARCH ROUTES ---
|
|
|
|
# --- INSTANCE ROUTES ---
|
|
@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/<string:category>", defaults = {"page": 1})
|
|
@app.route("/<string:domain>/videos/<string:category>/<int:page>")
|
|
async def instance_videos(domain, category, page):
|
|
if category not in [ "most-liked", "local", "trending", "recently-added" ]:
|
|
return await render_custom_error("No such video category: %s" % category, 404)
|
|
|
|
context = Context(api).instance(domain) \
|
|
.insert_future_result("videos", api.instance_videos, (domain, (page - 1) * 10, 10, category)) \
|
|
.paginate_results("videos", "/" + domain + "/videos/" + category + '/', page)
|
|
|
|
return await render("instance/%s.html" % category, context.failed(), context.build)
|
|
# --- END INSTANCE ROUTES ---
|
|
|
|
# --- VIDEO ROUTE ---
|
|
@app.route("/<string:domain>/videos/watch/<string:id>/", defaults = { "page": 1 })
|
|
@app.route("/<string:domain>/videos/watch/<string:id>/<int:page>")
|
|
async def video(domain, id, page):
|
|
context = Context(api).instance(domain) \
|
|
.video(domain, id, page, request.args)
|
|
return await render("video.html", context.failed(), context.build)
|
|
|
|
@app.route("/<string:domain>/videos/watch/<string:id>/<string:lang>.vtt")
|
|
async def captions(domain, id, lang):
|
|
captions = api.video_captions(domain, id)
|
|
if isinstance(captions, FailedRequest):
|
|
return await render_error(captions)
|
|
|
|
for entry in captions:
|
|
if entry["language"]["id"] == lang:
|
|
content = api.video_captions_proxy(domain, entry["captionPath"].split('/')[-1])
|
|
return await render_plaintext(content)
|
|
|
|
return await render_custom_error("This video has no subtitles/captions in requested language", code=404)
|
|
# --- END VIDEO ROUTE ---
|
|
|
|
# --- ACCOUNT ROUTES ---
|
|
@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_channels(domain, name, page):
|
|
context = Context(api).instance(domain) \
|
|
.insert_future_result("video_channels", api.account_channels, (domain, name, (page - 1) * 10, 10)) \
|
|
.paginate_results("video_channels", "/" + domain + "/accounts/" + name + "/video-channels/", page) \
|
|
.insert_future_result("account", api.account, (domain, name)) \
|
|
.insert("name", name)
|
|
return await render("account/channels.html", context.failed(), context.build)
|
|
|
|
@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):
|
|
context = Context(api).instance(domain) \
|
|
.insert_future_result("videos", api.account_videos, (domain, name, (page - 1) * 10, 10)) \
|
|
.paginate_results("videos", "/" + domain + "/accounts/" + name + "/videos/", page) \
|
|
.insert_future_result("account", api.account, (domain, name)) \
|
|
.insert("name", name)
|
|
return await render("account/videos.html", context.failed(), context.build)
|
|
|
|
@app.route("/<string:domain>/accounts/<string:name>/about")
|
|
async def account_about(domain, name):
|
|
context = Context(api).instance(domain) \
|
|
.insert_future_result("account", api.account, (domain, name)) \
|
|
.insert("name", name)
|
|
return await render("account/about.html", context.failed(), context.build)
|
|
# --- END ACCOUNT ROUTES ---
|
|
|
|
# --- CHANNEL ROUTES ---
|
|
@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 channel_videos(domain, name, page):
|
|
# TODO: Is it good to use ownerAccount.{host,name}?
|
|
context = Context(api).instance(domain) \
|
|
.insert_future_result("video_channel", api.channel, (domain, name)) \
|
|
.insert_future_result("videos", api.channel_videos, (domain, name, (page - 1) * 10, 10)) \
|
|
.paginate_results("videos", "/" + domain + "/video-channels/" + name + "/videos/", page) \
|
|
.insert_future_result("account", api.account, ((PreviousResult("video_channel"), "ownerAccount", "host"), (PreviousResult("video_channel"), "ownerAccount", "name"))) \
|
|
.insert("name", name)
|
|
return await render("channel/videos.html", context.failed(), context.build)
|
|
|
|
# TODO: Hope this route works, i don't know some account who has channel playlists to test with
|
|
@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 channel_playlists(domain, name, page):
|
|
channel = api.channel(domain, name)
|
|
if isinstance(channel, FailedRequest):
|
|
return await render_error(channel)
|
|
# TODO: Is it good to use ownerAccount.{host,name}?
|
|
context = Context(api).instance(domain) \
|
|
.insert_future_result("video_channel", api.channel, (domain, name)) \
|
|
.insert_future_result("video_playlists", api.channel_playlists, (domain, name, (page - 1) * 10, 10)) \
|
|
.paginate_results("video_playlists", "/" + domain + "/video-channels/" + name + "/video-playlists/", page) \
|
|
.insert_future_result("account", api.account, ((PreviousResult("video_channel"), "ownerAccount", "host"), (PreviousResult("video_channel"), "ownerAccount", "name"))) \
|
|
.insert("name", name)
|
|
return await render("channel/playlists.html", context.failed(), context.build)
|
|
|
|
@app.route("/<string:domain>/video-channels/<string:name>/about")
|
|
async def channel_about(domain, name):
|
|
context = Context(api).instance(domain) \
|
|
.insert_future_result("video_channel", api.channel, (domain, name)) \
|
|
.insert("name", name)
|
|
return await render("channel/about.html", context.failed(), context.build)
|
|
|
|
# --- DIVERSE ROUTES ---
|
|
@app.route("/favicon.ico")
|
|
async def favicon():
|
|
return await render_custom_error("We don't have a favicon yet. If you would like to contribute one, please send it to ~metalune/public-inbox@lists.sr.ht", code=404)
|
|
|
|
@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_error(FailedRequest(e))
|
|
# --- END DIVERSE ROUTES ---
|
|
# --- END WEBSERVER STUFF ---
|
|
|
|
# --- CLI STUFF ---
|
|
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)
|
|
# --- END CLI STUFF ---
|