southerntofu
/
simpleertube
1
0
Fork 0
Code Issues Pull Requests 3 Projects Releases Wiki Activity

Finally fix pagination

This commit is contained in:
southerntofu 2021-08-02 18:03:44 +02:00
parent cb2569c0c1
commit bcb1018256
6 changed files with 221 additions and 176 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

106
main.py
View File

@ -1,14 +1,15 @@
from quart import Quart, request, render_template, redirect
from math import ceil
import peertube
#from peertube import Cache, PaginatedResults, FailedRequest, API
from peertube import *
import sys
# Import everything from utils.py, to keep main.py focused
# on CLI arguments parsing, and web routes declarations
from utils import *
cache = peertube.Cache()
api = peertube.API(cache)
cache = Cache()
api = API(cache)
# Let's follow subscriptions from the current working dir
subscriptions = Subscriptions('.', api, cache)
@ -46,6 +47,12 @@ class Context():
def failed(self):
return self.failures
# Helper function to insert/log an error
def error(self, reason):
print("[ERROR] %s" % 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):
@ -58,20 +65,49 @@ class Context():
# 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" % key)
#print("[DEBUG] Considering future result %s" % repr(key))
if len(self.failures) > 0:
# Previous failures, don't run further requests
print("[DEBUG] Previous failures. Skipping future result %s" % key)
print("[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("Calling with arguments:\n")
#for arg in args: print(" - %s" % arg)
return self.insert_result(key, func(*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):
print("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
@ -89,7 +125,7 @@ class Context():
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]]
print("[DEBUG] Found previous entry %s at given route:%s" % (arg[0].key(), data))
print("[DEBUG] Found previous entry %s at given route: %s" % (arg[0].key(), data))
return data
else:
#print("[DEBUG] Found normal argument")
@ -106,14 +142,13 @@ class Context():
# Build pagination context. Since total page count is fetched from
def paginate(self, url, page, total_pages):
# TODO: Actually i think pagination URL takes no page number, as it's set in templates?
self.context["paginate_url"] = url.replace("${page}", str(page))
self.context["page"] = page
self.context["pages_total"] = 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_result(), so you don't have to store the API result in the route
# 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):
@ -121,20 +156,21 @@ class Context():
# There were previous errors, so we don't paginate, and skip silently
return self
if results_key not in self.context:
self.failures.append(FailedRequest("%s not found in pagination context. As no previous error happened, this is a programming error worth reporting." % results_key))
return self
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
# 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
print("[ERROR] Tried to paginate a failed result. You should insert results with context.insert_result, not context.insert to prevent this mistake.")
return self
# TODO: Pagination should not be done from the returned number of results,
# (because that's only `count` items), but from result["total"] which should be kept until here
return self.paginate(url, page, ceil(len(results) / 10))
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)
print("[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):
@ -143,11 +179,12 @@ class Context():
return self
# Build video context (captions, quality, embed, comments) from domain, id and request headers
def video(self, domain, id, args={}):
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"))) \
.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
@ -188,7 +225,7 @@ async def search_instance_redirect(domain):
@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) \
.paginate_results("results", "/search/" + query + "/", page) \
.insert("query", query)
return await render("simpleer_search_results.html", context.failed(), context.build)
@ -196,8 +233,8 @@ async def search(query, page):
@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)) \
.paginate_results("results", "/search" + query + "/", page) \
.insert_future_result("results", api.search_instance, (domain, query, (page - 1) * 10, 10)) \
.paginate_results("results", "/" + domain + "/search/" + query + "/", page) \
.insert("search_term", query) # TODO: rename query as in cross-instance search?
return await render("search_results.html", context.failed(), context.build)
# --- END SEARCH ROUTES ---
@ -223,10 +260,11 @@ async def instance_videos(domain, category, page):
# --- END INSTANCE ROUTES ---
# --- VIDEO ROUTE ---
@app.route("/<string:domain>/videos/watch/<string:id>/")
async def video(domain, id):
@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, request.args)
.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")
@ -253,7 +291,7 @@ async def accounts_redirect(domain, name):
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}", page) \
.paginate_results("video_channels", "/" + domain + "/accounts/" + name + "/video-channels/", page) \
.insert_future_result("account", api.account, (domain, name)) \
.insert("name", name)
return await render("accounts/channels.html", context.failed(), context.build)
@ -263,7 +301,7 @@ async def account_channels(domain, name, 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}", page) \
.paginate_results("videos", "/" + domain + "/accounts/" + name + "/videos/", page) \
.insert_future_result("account", api.account, (domain, name)) \
.insert("name", name)
return await render("accounts/videos.html", context.failed(), context.build)
@ -288,7 +326,7 @@ async def channel_videos(domain, name, page):
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}", page) \
.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("video_channels/videos.html", context.failed(), context.build)
@ -304,7 +342,7 @@ async def channel_playlists(domain, name, page):
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}", page) \
.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("video_channels/video_playlists.html", context.failed(), context.build)

116
peertube.py
View File

@ -3,9 +3,14 @@ import requests
import json
import sys
from datetime import datetime
from utils import FailedRequest
# Helper Class for using caches, you can use any other cache that implements the same API
# Helper class to store paginated results
class PaginatedResults:
def __init__(self, total, data):
self.total = total
self.data = data
# Helper class for using caches, you can use any other cache that implements the same API
# Default TTL: 3600s (1h)
class Cache:
def __init__(self, ttl=3600):
@ -32,6 +37,7 @@ class Cache:
def set(self, key, value):
self.dict[key] = [ value, datetime.now() ]
# Takes a successful plaintext response and parses it as HTML to extract the title
def html_title(content):
soup = BeautifulSoup(content, "lxml")
title = soup.find('title')
@ -40,6 +46,14 @@ def html_title(content):
else:
return "PeerTube Instance"
# Takes a successfully-parsed JSON response and extracts data/total from it to build pagination
def paginator(response):
if "data" not in response or "total" not in response:
return FailedRequest("The API response provided to paginator appears not to be paginated")
# TODO: check that total is an integer and that it's greater than len(data)
#return response["total"], response["data"]
return PaginatedResults(response["total"], response["data"])
class API:
# The PeertubeAPI is initialized with a caching backend and a default TTL, that can be overriden in specific
# API request calls. The caching backend should implement a get(key, ttl) and set(key, value) API.
@ -102,7 +116,7 @@ class API:
# key: Key to extract from a successful response
# backend: the method to use for fetching URL (default: self.json_request), can be self.plaintext_request
# extractor: a lambda function to execute to extract stuff from a successful request, when key isn't set
def request(self, args, url, key=None, ttl=None, backend=None, extractor=None):
def request(self, args, url, key=None, backend=None, extractor=None, ttl=None):
# WTF python? '/'.join(("foo)) => "f/o/o", not "foo"?! Special case when only one arg
if isinstance(args, str): args = (args, )
cache_key = '/'.join(args)
@ -159,7 +173,7 @@ class API:
(str(start), str(count), self.search_source, query),
# self.search_source already has protocol pre-pended
"${2}/api/v1/search/videos?start=${0}&count=${1}&search=${3}",
key="data"
extractor=paginator
)
# Search a specific Peertube instance for `query`,
@ -168,7 +182,7 @@ class API:
return self.request(
(str(start), str(count), domain, term),
"https://${2}/api/v1/search/videos?start=${0}&count=${1}&search=${3}&sort=-match&searchTarget=local",
key="data"
extractor=paginator
)
# Default category is local, other categories are: trending, most-liked, recently-added, local
@ -182,19 +196,21 @@ class API:
return self.request(
(str(start), str(count), domain, category),
url,
key="data"
extractor=paginator
)
def video(self, domain, id, ttl=None):
return self.request(
(domain, id),
"https://${0}/api/v1/videos/${1}"
"https://${0}/api/v1/videos/${1}",
)
def video_captions(self, domain, id, ttl=None):
return self.request(
(domain, id),
"https://${0}/api/v1/videos/${1}/captions",
# NOTE: Captions look like paginated content because they have 'total' and 'data' field
# However they are, and that's good, not paginated.
key="data"
)
@ -208,11 +224,11 @@ class API:
backend=self.plaintext_request
)
def video_comments(self, domain, id, ttl=None):
def video_comments(self, domain, id, start=0, count=10, ttl=None):
return self.request(
(domain, id),
"https://${0}/api/v1/videos/${1}/comment-threads",
key="data"
(domain, id, str(start), str(count)),
"https://${0}/api/v1/videos/${1}/comment-threads?start=${2}&count=${3}",
extractor=paginator
)
def account(self, domain, name, ttl=None):
@ -225,14 +241,14 @@ class API:
return self.request(
(str(start), str(count), domain, name),
"https://${2}/api/v1/accounts/${3}/video-channels?start=${0}&count=${1}",
key="data"
extractor=paginator
)
def account_videos(self, domain, name, start=0, count=10, ttl=None):
return self.request(
(str(start), str(count), domain, name),
"https://${2}/api/v1/accounts/${3}/videos?start=${0}&count=${1}",
key="data"
extractor=paginator
)
def channel(self, domain, name, ttl=None):
@ -245,12 +261,82 @@ class API:
return self.request(
(str(start), str(count), domain, name),
"https://${2}/api/v1/video-channels/${3}/videos?start=${0}&count=${1}",
key="data"
extractor=paginator
)
def channel_playlists(self, domain, name, start=0, count=10, ttl=None):
return self.request(
(str(start), str(count), domain, name),
"https://${2}/api/v1/video-channels/${3}/video-playlists?start=${0}&count=${1}",
key="data"
extractor=paginator
)
# --- ERROR UTILS ---
# Semantic class to store remote errors
class FailedRequest:
def __init__(self, e):
self.exception = e
def format(self):
# If it's a rawtext error, print it
# Otherwise look for 'message' attribute
# Otherwise ask python to represent the exception
if isinstance(self.exception, str):
return self.exception
else: return getattr(self.exception, 'message', repr(self.exception))
# Strip < and > symbols to prevent content injection,
# and replace newlines with HTML line breaks <br>
def format_html(self):
return self.format().replace('<', "&lt;").replace('>', "&gt;").replace("\n", "<br>")
# Format a list of FailedRequest's
def format_errors(failures):
return list(map(lambda failure: failure.format(), failures))
def format_errors_html(failures):
return list(map(lambda failure: failure.format_html(), failures))
# --- END ERROR UTILS ---
# Extra information about video, not contained directly in API result
# a is video result
class VideoInfo:
def __init__(self, api, a, args):
# If the video is being built from a failed request, return that request instead
if isinstance(a, FailedRequest):
print("[ERROR] A video request failed, yet you called VideoInfo about it. You should probably not make useless requests.")
return FailedRequest("A previous video request failed, not attempting to fetch extra information about it.")
quality = args.get("quality")
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

29
templates/base.html
View File

@ -1,3 +1,4 @@
{%- import "macros.html" as macros %}
<!doctype html>
<html>
@ -37,36 +38,12 @@
{% block head_content %}{% endblock %}
{% if pagination_url %}
{% if pages_total > 1 %}
{% if page > 1 %}
<a href="{{ pagination_url }}{{ page - 1 }}">Previous</a>
<b> | </b>
{% endif %}
Page {{ page }} of {{ pages_total }}
{% if page < pages_total %}
<b> | </b>
<a href="{{ pagination_url }}{{ page + 1}}">Next</a>
{% endif %}
{% endif %}
{% endif %}
{{ macros.paginate(pagination) }}
<br>
{% block content %}{% endblock %}
{% if pagination_url %}
{% if pages_total > 1 %}
{% if page > 1 %}
<a href="{{ pagination_url }}{{ page - 1 }}">Previous</a>
<b> | </b>
{% endif %}
Page {{ page }} of {{ pages_total }}
{% if page < pages_total %}
<b> | </b>
<a href="{{ pagination_url }}{{ page + 1}}">Next</a>
{% endif %}
{% endif %}
{% endif %}
{{ macros.paginate(pagination) }}
</center>
</body>
</html>

15
templates/macros.html Normal file
View File

@ -0,0 +1,15 @@
{% macro paginate(pagination) %}
{% if pagination %}
{% if pagination.pages > 1 %}
{% if pagination.current > 1 %}
<a href="{{ pagination.url }}{{ pagination.current - 1 }}">Previous</a>
<b> | </b>
{% endif %}
Page {{ pagination.current }} of {{ pagination.pages }}
{% if pagination.current < pagination.pages %}
<b> | </b>
<a href="{{ pagination.url }}{{ pagination.current + 1}}">Next</a>
{% endif %}
{% endif %}
{% endif %}
{% endmacro %}

25
templates/simpleer_search_results.html
View File

@ -1,3 +1,4 @@
{%- import "macros.html" as macros -%}{# TODO: This template should extend base.html #}
<!doctype html>
<html>
<head>
@ -15,17 +16,7 @@
<br>
<br>
{% if pages_total > 1 %}
{% if page > 1 %}
<a href="/search/{{ query }}/{{ page - 1 }}">Previous</a>
<b> | </b>
{% endif %}
Page {{ page }} of {{ pages_total }}
{% if page < pages_total %}
<b> | </b>
<a href="/search/{{ query }}/{{ page + 1}}">Next</a>
{% endif %}
{% endif %}
{{ macros.paginate(pagination) }}
<br>
<br>
@ -54,17 +45,7 @@
</div>
<hr>
{% if pages_total > 1 %}
{% if page > 1 %}
<a href="/search/{{ query }}/{{ page - 1 }}">Previous</a>
<b> | </b>
{% endif %}
Page {{ page }} of {{ pages_total }}
{% if page < pages_total %}
<b> | </b>
<a href="/search/{{ query }}/{{ page + 1}}">Next</a>
{% endif %}
{% endif %}
{{ macros.paginate(pagination) }}
</center>
</body>
</html>

106
utils.py
View File

@ -1,53 +1,11 @@
from quart import render_template
from datetime import datetime
from dateutil import parser as dateutil
import peertube
#import peertube
from peertube import PaginatedResults, FailedRequest
import html2text
import sys
# Extra information about video, not contained directly in API result
# a is video result
class VideoInfo:
def __init__(self, api, a, args):
# If the video is being built from a failed request, return that request instead
if isinstance(a, FailedRequest):
print("[ERROR] A video request failed, yet you called VideoInfo about it. You should probably not make useless requests.")
return FailedRequest("A previous video request failed, not attempting to fetch extra information about it.")
quality = args.get("quality")
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
# --- IDENTIFIERS UTILS ---
# Builds a unified id@server from one of those syntaxes, additionally stripping extra whitespace and ignoring `#` as comments:
# - id@server
@ -127,7 +85,7 @@ async def render(template, result, context):
# Feed me a single peertube.Api that failed (FailedRequest class), i will render an error
async def render_error(result):
if not isinstance(result, peertube.FailedRequest):
if not isinstance(result, FailedRequest):
print("[ERROR] Called render_error with a non-failed result:\n%s" % result)
return
return await render_template(
@ -146,7 +104,7 @@ async def render_custom_error(reason, code=500):
# Feed me a peertube.Api result and i will output plaintext, or render an error
async def render_plaintext(result):
if isinstance(result, peertube.FailedRequest):
if isinstance(result, FailedRequest):
return await render_template(
"error.html",
error_number = "500",
@ -160,19 +118,30 @@ h2t = html2text.HTML2Text()
h2t.ignore_links = True
# NOTE: Python makes it really hard to accept both boolean and stringy `enabled` values
def comments(api, domain, id, enabled=True):
# This function MUST be called with successful results, don't pass it anything that hasn't been fail-proofed
def comments(api, domain, id, enabled=True, start=0, count=10):
# Skip comments querying/formatting when API previously returned comments are disabled
if enabled != True and str(enabled).lower() != "true": return ""
comments = api.video_comments(domain, id)
comments = api.video_comments(domain, id, start=start, count=count)
if isinstance(comments, FailedRequest):
return comments
# Successful request, get comment count from PaginatedResults
if not isinstance(comments, PaginatedResults):
return FailedRequest("WTF. comments() util was called with something that is not PaginatedResults")
# TODO: do some cleanup
#total_comments = comments.total
#comments = comments.data
# Strip the HTML from the comments and convert them to plain text
new_comments = []
for comment in comments:
for comment in comments.data:
text = h2t.handle(comment["text"]).strip().strip("\n")
comment["text"] = text
new_comments.append(comment)
return new_comments
comments.data = new_comments
#print(total_comments)
#print(len(new_comments))
#return total_comments, new_comments
return comments
# --- END COMMENT UTILS ---
# --- SUBSCRIPTION UTILS ---
@ -294,9 +263,12 @@ class Subscriptions:
if isinstance(result, FailedRequest):
# Single failure from one API request
failures.append(result)
elif not isinstance(result, PaginatedResults):
# We are expecting paginated results here!
failures.append(FailedRequest("accounts_videos expecting paginated results. WTF?"))
else:
# Multiple successes from one API request
latest.extend(result)
latest.extend(result.data)
latest.sort(key = lambda vid: dateutil.isoparse(vid["createdAt"]), reverse=True)
return (latest[0:limit], failures)
@ -337,37 +309,13 @@ class Subscriptions:
if isinstance(result, FailedRequest):
# Single failure from one API request
failures.append(result)
elif not isinstance(result, PaginatedResults):
# We are expecting paginated results here!
failures.append(FailedRequest("channels_videos expecting paginated results. WTF?"))
else:
# Multiple successes from one API request
latest.extend(result)
latest.extend(result.data)
latest.sort(key = lambda vid: dateutil.isoparse(vid["createdAt"]), reverse=True)
return latest[0:limit], failures
# --- END SUBSCRIPTION UTILS ---
# --- ERROR UTILS ---
# Semantic class to store remote errors
class FailedRequest:
def __init__(self, e):
self.exception = e
def format(self):
# If it's a rawtext error, print it
# Otherwise look for 'message' attribute
# Otherwise ask python to represent the exception
if isinstance(self.exception, str):
return self.exception
else: return getattr(self.exception, 'message', repr(self.exception))
# Strip < and > symbols to prevent content injection,
# and replace newlines with HTML line breaks <br>
def format_html(self):
return self.format().replace('<', "&lt;").replace('>', "&gt;").replace("\n", "<br>")
# Format a list of FailedRequest's
def format_errors(failures):
return list(map(lambda failure: failure.format(), failures))
def format_errors_html(failures):
return list(map(lambda failure: failure.format_html(), failures))
# --- END ERROR UTILS ---