mirrors/SimpleerTube
1
0
Fork 0
mirror of https://codeberg.org/SimpleWeb/SimpleerTube.git synced 2024-11-22 00:21:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

Specify 5 second timeout time for any request done to a peertube instance

Browse source
This commit is contained in:
metalune 2021-10-05 20:57:09 +02:00
parent 04c20bb8bc
commit 6debfcd8f9
1 changed files with 22 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

39
peertube.py
View file

@ -1,17 +1,22 @@
from bs4 import BeautifulSoup from bs4 import BeautifulSoup
import requests import requests
import functools
import json import json
s = requests.Session()
s.request = functools.partial(s.request, timeout=5)
# --- Sepiasearch --- # --- Sepiasearch ---
def sepia_search(query, start=0, count=10): def sepia_search(query, start=0, count=10):
url = "https://search.joinpeertube.org/api/v1/search/videos?search=" + query + "&start=" + str(start) + "&count=" + str(count) url = "https://search.joinpeertube.org/api/v1/search/videos?search=" + query + "&start=" + str(start) + "&count=" + str(count)
return json.loads(requests.get(url).text) return json.loads(s.get(url).text)
# --- ---- # --- ----
def get_instance_name(domain): def get_instance_name(domain):
soup = BeautifulSoup(requests.get("https://" + domain).text, "lxml") soup = BeautifulSoup(s.get("https://" + domain).text, "lxml")
title = soup.find('title') title = soup.find('title')
if title: if title:
return title.text return title.text
@ -20,70 +25,70 @@ def get_instance_name(domain):
def video(domain, id): def video(domain, id):
url = "https://" + domain + "/api/v1/videos/" + id url = "https://" + domain + "/api/v1/videos/" + id
return json.loads(requests.get(url).text) return json.loads(s.get(url).text)
def video_captions(domain, id): def video_captions(domain, id):
url = "https://" + domain + "/api/v1/videos/" + id + "/captions" url = "https://" + domain + "/api/v1/videos/" + id + "/captions"
return json.loads(requests.get(url).text) return json.loads(s.get(url).text)
def video_captions_download(domain, caption_id): def video_captions_download(domain, caption_id):
# URL is hardcoded to prevent further proxying. URL may change with updates, see captions API # URL is hardcoded to prevent further proxying. URL may change with updates, see captions API
# eg. https://kolektiva.media/api/v1/videos/9c9de5e8-0a1e-484a-b099-e80766180a6d/captions # eg. https://kolektiva.media/api/v1/videos/9c9de5e8-0a1e-484a-b099-e80766180a6d/captions
# TODO: What if the captionPath doesn't follow this format on an instance? Should we really proxy ANYTHING returned by API? # TODO: What if the captionPath doesn't follow this format on an instance? Should we really proxy ANYTHING returned by API?
url = "https://" + domain + "/lazy-static/video-captions/" + caption_id url = "https://" + domain + "/lazy-static/video-captions/" + caption_id
return requests.get(url).text return s.get(url).text
def search(domain, term, start=0, count=10): def search(domain, term, start=0, count=10):
url = "https://" + domain + "/api/v1/search/videos?start=" + str(start) + "&count=" + str(count) + "&search=" + term + "&sort=-match&searchTarget=local" url = "https://" + domain + "/api/v1/search/videos?start=" + str(start) + "&count=" + str(count) + "&search=" + term + "&sort=-match&searchTarget=local"
return json.loads(requests.get(url).text) return json.loads(s.get(url).text)
def get_comments(domain, id): def get_comments(domain, id):
url = "https://" + domain + "/api/v1/videos/" + id + "/comment-threads" url = "https://" + domain + "/api/v1/videos/" + id + "/comment-threads"
return json.loads(requests.get(url).text) return json.loads(s.get(url).text)
def get_videos_trending(domain, start=0, count=10): def get_videos_trending(domain, start=0, count=10):
url = "https://" + domain + "/api/v1/videos?sort=-trending&start=" + str(start) + "&count=" + str(count) url = "https://" + domain + "/api/v1/videos?sort=-trending&start=" + str(start) + "&count=" + str(count)
return json.loads(requests.get(url).text) return json.loads(s.get(url).text)
def get_videos_most_liked(domain, start=0, count=10): def get_videos_most_liked(domain, start=0, count=10):
url = "https://" + domain + "/api/v1/videos?sort=-likes&start=" + str(start) + "&count=" + str(count) url = "https://" + domain + "/api/v1/videos?sort=-likes&start=" + str(start) + "&count=" + str(count)
return json.loads(requests.get(url).text) return json.loads(s.get(url).text)
def get_videos_recently_added(domain, start=0, count=10): def get_videos_recently_added(domain, start=0, count=10):
url = "https://" + domain + "/api/v1/videos?sort=-publishedAt&start=" + str(start) + "&count=" + str(count) url = "https://" + domain + "/api/v1/videos?sort=-publishedAt&start=" + str(start) + "&count=" + str(count)
return json.loads(requests.get(url).text) return json.loads(s.get(url).text)
def get_videos_local(domain, start=0, count=10): def get_videos_local(domain, start=0, count=10):
url = "https://" + domain + "/api/v1/videos?sort=-publishedAt&filter=local&start=" + str(start) + "&count=" + str(count) url = "https://" + domain + "/api/v1/videos?sort=-publishedAt&filter=local&start=" + str(start) + "&count=" + str(count)
return json.loads(requests.get(url).text) return json.loads(s.get(url).text)
# --- Accounts --- # --- Accounts ---
def account_video_channels(domain, name, start=0, count=10): def account_video_channels(domain, name, start=0, count=10):
url = "https://" + domain + "/api/v1/accounts/" + name + "/video-channels?start=" + str(start) + "&count=" + str(count) url = "https://" + domain + "/api/v1/accounts/" + name + "/video-channels?start=" + str(start) + "&count=" + str(count)
return json.loads(requests.get(url).text) return json.loads(s.get(url).text)
def account_videos(domain, name, start=0, count=10): def account_videos(domain, name, start=0, count=10):
url = "https://" + domain + "/api/v1/accounts/" + name + "/videos?start=" + str(start) + "&count=" + str(count) url = "https://" + domain + "/api/v1/accounts/" + name + "/videos?start=" + str(start) + "&count=" + str(count)
return json.loads(requests.get(url).text) return json.loads(s.get(url).text)
def account(domain, name): def account(domain, name):
url = "https://" + domain + "/api/v1/accounts/" + name url = "https://" + domain + "/api/v1/accounts/" + name
return json.loads(requests.get(url).text) return json.loads(s.get(url).text)
# --- Video Channels --- # --- Video Channels ---
def video_channel_videos(domain, name, start=0, count=10): def video_channel_videos(domain, name, start=0, count=10):
url = "https://" + domain + "/api/v1/video-channels/" + name + "/videos?start=" + str(start) + "&count=" + str(count) url = "https://" + domain + "/api/v1/video-channels/" + name + "/videos?start=" + str(start) + "&count=" + str(count)
return json.loads(requests.get(url).text) return json.loads(s.get(url).text)
def video_channel_video_playlists(domain, name, start=0, count=10): def video_channel_video_playlists(domain, name, start=0, count=10):
url = "https://" + domain + "/api/v1/video-channels/" + name + "/video-playlists?start=" + str(start) + "&count=" + str(count) url = "https://" + domain + "/api/v1/video-channels/" + name + "/video-playlists?start=" + str(start) + "&count=" + str(count)
return json.loads(requests.get(url).text) return json.loads(s.get(url).text)
def video_channel(domain, name): def video_channel(domain, name):
url = "https://" + domain + "/api/v1/video-channels/" + name url = "https://" + domain + "/api/v1/video-channels/" + name
return json.loads(requests.get(url).text) return json.loads(s.get(url).text)
if __name__ == "__main__": if __name__ == "__main__":
#name = get_instance_name("videos.lukesmith.xyz") #name = get_instance_name("videos.lukesmith.xyz")