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("/") 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("/", defaults = {"page": 1}) @app.route("//") 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("//search", methods=["POST"]) async def search_instance_redirect(domain): query = (await request.form)["query"] return redirect("/" + domain + "/search/" + query) @app.route("/search/", defaults = {"page": 1}) @app.route("/search//") 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("//search/", defaults = {"page": 1}) @app.route("//search//") 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("//") 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("//videos/", defaults = {"page": 1}) @app.route("//videos//") 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("//videos/watch//", defaults = { "page": 1 }) @app.route("//videos/watch//") 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("//videos/watch//.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("//accounts/") async def accounts_redirect(domain, name): return redirect("/" + domain + "/accounts/" + name + "/video-channels") @app.route("//accounts//video-channels", defaults = {"page": 1}) @app.route("//accounts//video-channels/") 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("//accounts//videos", defaults = {"page": 1}) @app.route("//accounts//videos/") 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("//accounts//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("//video-channels/") async def video_channels_redirect(domain, name): return redirect("/" + domain + "/video-channels/" + name + "/videos") @app.route("//video-channels//videos", defaults = {"page": 1}) @app.route("//video-channels//videos/") 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("//video-channels//video-playlists", defaults = {"page": 1}) @app.route("//video-channels//video-playlists/") 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("//video-channels//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 ---