1
0
mirror of https://git.sr.ht/~cadence/NewLeaf synced 2024-10-10 03:17:30 +00:00
NewLeaf/index.py
Cadence Ember 714f1030fb
Channel path fixes I'm pretty sure I already did?
- use "channels" as default path, not "user"
- cache based on the combination of the path and the id
- fix channel latest
2022-09-12 23:13:37 +12:00

157 lines
4.7 KiB
Python

import cherrypy
import json
import pathlib
import requests
import yt_dlp
from extractors.video import extract_video
from extractors.channel import extract_channel, extract_channel_videos, extract_channel_latest
from extractors.manifest import extract_manifest
from extractors.search import extract_search
from extractors.suggestions import extract_search_suggestions
from extractors.captions import extract_captions
from extractors.comments import extract_comments
import configuration
@cherrypy.tools.register("before_finalize", priority=60)
def custom_headers():
cherrypy.response.headers["access-control-allow-origin"] = "*"
class NewLeaf(object):
def _cp_dispatch(self, vpath):
if vpath[:4] == ["api", "manifest", "dash", "id"]:
vpath[:4] = ["manifest"]
return self
if vpath[:2] == ["api", "v1"]:
endpoints = [
["channels", 1, 2],
["videos", 1, 1],
["search", 0, 1],
["captions", 1, 1],
["comments", 1, 1]
]
for e in endpoints:
if vpath[2] == e[0] and len(vpath) >= e[1]+3 and len(vpath) <= e[2]+3:
vpath[:3] = [e[0]]
return self
return vpath
@cherrypy.expose
@cherrypy.tools.json_out()
def videos(self, id, **kwargs):
return extract_video(id)
@cherrypy.expose
@cherrypy.tools.encode()
def manifest(self, id, **kwargs):
result = extract_manifest(id)
if type(result) is dict:
cherrypy.response.headers["content-type"] = "application/json"
return bytes(json.dumps(result), "utf8")
elif type(result) is requests.models.Response:
cherrypy.response.headers["content-type"] = result.headers["content-type"]
return result
else:
cherrypy.response.headers["content-type"] = "application/dash+xml"
return result
@cherrypy.expose
@cherrypy.tools.json_out()
def channels(self, *suffix, second__path="channel", **kwargs):
ucid = ""
part = ""
possible_parts = ("videos", "latest", "playlists")
if len(suffix) == 1:
ucid = suffix[0]
else: # len(suffix) >= 2
if suffix[0] in possible_parts:
[part, ucid] = suffix
elif suffix[1] in possible_parts:
[ucid, part] = suffix
else:
return {
"error": "Two components specified in URL, but neither component was recognised as a part keyword.",
"identifier": "PART_KEYWORD_NOT_RECOGNISED"
}
possible_paths = ("channel",) if part == "latest" else ("channel", "c", "user")
if second__path not in possible_paths:
return {
"error": "second__path parameter must be one of: " + str(possible_paths),
"identifier": "PATH_PARAMETER_NOT_RECOGNISED"
}
if part == "playlists":
return []
elif part == "latest":
return extract_channel_latest(ucid)
elif part == "videos":
return extract_channel_videos(ucid, second__path)
else: # part == "", so extract whole channel
return extract_channel(ucid, second__path)
@cherrypy.expose
@cherrypy.tools.json_out()
def search(self, *suffix, q, **kwargs):
if suffix == ("suggestions",):
return self.suggestions(q=q)
return extract_search(q)
@cherrypy.expose
@cherrypy.tools.json_out()
def suggestions(self, *, q, **kwargs):
return extract_search_suggestions(q)
@cherrypy.expose
def captions(self, id, **kwargs):
try:
result = extract_captions(id, **kwargs)
if type(result) is dict:
cherrypy.response.headers["content-type"] = "application/json"
return bytes(json.dumps(result), "utf8")
else:
cherrypy.response.headers["content-type"] = "text/vtt; charset=UTF-8"
return result
except StopIteration:
cherrypy.response.status = "400"
cherrypy.response.headers["content-type"] = "application/json"
return bytes(json.dumps({
"error": "No captions matching that language or label",
"identifier": "NO_MATCHING_CAPTIONS"
}), "utf8")
@cherrypy.expose
@cherrypy.tools.json_out()
def comments(self, id, **kwargs):
return extract_comments(id)
@cherrypy.expose
def vi(self, id, file):
r = requests.get("https://i.ytimg.com/vi/{}/{}".format(id, file), stream=True)
r.raise_for_status()
cherrypy.response.headers["content-type"] = r.headers["content-type"]
return next(r.iter_content(chunk_size=None))
@cherrypy.expose
def ggpht(self, *path):
r = requests.get("https://yt3.ggpht.com/{}".format("/".join(path)), stream=True)
r.raise_for_status()
cherrypy.response.headers["content-type"] = r.headers["content-type"]
return next(r.iter_content(chunk_size=None))
bind_port = getattr(configuration, "bind_port", 3000)
bind_host = getattr(configuration, "bind_host", "0.0.0.0")
server_root = pathlib.Path(__file__).parent.joinpath("root")
cherrypy.config.update({"server.socket_port": bind_port, "server.socket_host": bind_host})
cherrypy.quickstart(NewLeaf(), "/", {
"/": {
"tools.custom_headers.on": True,
"tools.staticdir.on": True,
"tools.staticdir.dir": str(server_root.absolute()),
"tools.staticdir.index": "index.html"
}
})