ythdd/views.py

#!/usr/bin/python3
from flask import redirect, render_template, request, Response
from flask_sqlalchemy import SQLAlchemy
from markupsafe import escape
import hashlib, json, re, requests
import ythdd_globals

def homepage():
	return "homepage"

def home():
	return "welcome home!"

def index():
	return "index"

def thumbnailProxy(received_request):

	# apparently, this can be set to
	# https://img.youtube.com/ as well
	prefix = "https://i.ytimg.com/"

	if received_request.count("/") < 1 or received_request.index("/") != 11:
		return Response(json.dumps({
			'status': 400, 
			'error_msg': 'invalid request. pretend this is a thumbnail :D'
			}), mimetype='application/json', status=400)

	quality_urls = ['maxresdefault', 'sddefault', 'hqdefault', 'mqdefault', 'default', '1', '2', '3']
	video_id, requested_quality = received_request.split('/')
	
	thumbnail = requests.get(prefix + "vi/" + video_id + "/" + requested_quality, headers=ythdd_globals.getHeaders(caller='proxy'), stream=True)
	thumbnail.raw.decode_content = True

	quality_id = 0
	if requested_quality == "maxres.jpg":
		# if requested quality is maxres,
		# provide the best quality possible
		while thumbnail.status_code != 200:
			thumbnail = requests.get(prefix + "vi/" + video_id + "/" + quality_urls[quality_id] + ".jpg", headers=ythdd_globals.getHeaders(caller='proxy'), stream=True)
			thumbnail.raw.decode_content = True
			quality_id += 1

	response = Response(thumbnail.raw, mimetype=thumbnail.headers['content-type'], status=thumbnail.status_code)

	return response

def ggphtProxy(received_request):

	prefix = "https://yt3.ggpht.com/"

	# fix for how materialious fetches avatars
	if received_request.startswith("guc/"):
		return gucProxy(received_request.removeprefix("guc/"))

	ggpht = requests.get(prefix + received_request, headers=ythdd_globals.getHeaders(caller='proxy'), stream=True)
	ggpht.raw.decode_content = True
	response = Response(ggpht.raw, mimetype=ggpht.headers['content-type'], status=ggpht.status_code)

	return response

def gucProxy(received_request):

	prefix = "https://yt3.googleusercontent.com/"

	guc = requests.get(prefix + received_request, headers=ythdd_globals.getHeaders(caller='proxy'), stream=True)
	guc.raw.decode_content = True
	response = Response(guc.raw, mimetype=guc.headers['content-type'], status=guc.status_code)

	return response

def imgProxy(received_request):

	# will proxy /img/no_thumbnail.jpg
	prefix = "https://i.ytimg.com/"

	thumbnail = requests.get(prefix + "img/" + received_request, headers=ythdd_globals.getHeaders(caller='proxy'), stream=True)
	thumbnail.raw.decode_content = True
	response = Response(thumbnail.raw, mimetype=thumbnail.headers['content-type'], status=thumbnail.status_code)

	return response

def videoplaybackProxy():
	# inspired by Yotter's video proxy
	# https://github.com/ytorg/Yotter/blob/b43a72ab7bfa5a59916fa3259cbc39165717c6bb/app/routes.py#L527

	if not ythdd_globals.config['proxy']['allow_proxying_videos']:
		return Response(json.dumps({"error": "Administrator has disabled this endpoint"}), mimetype="application/json", status=403)

	headers = dict(request.headers)
	proxy_headers = ythdd_globals.getHeaders(caller='proxy')
	if "Range" in headers:
		proxy_headers["Range"] = headers["Range"]

	params = dict(request.args)
	# support md5 videoplayback url shortening
	if "md5" in params:
		if params["md5"] not in ythdd_globals.general_cache["hashed_videoplayback"]:
			return Response(json.dumps({"error": "Videoplayback request not cached or expired."}), mimetype="application/json", status=404)
		return redirect(ythdd_globals.general_cache["hashed_videoplayback"][params["md5"]]["original_url"])
	else:
		md5sum = hashlib.md5(request.url.encode("utf-8")).hexdigest()
		ythdd_globals.general_cache["hashed_videoplayback"][md5sum] = {"original_url": request.url}


	# reconstruct the url
	# first attempt: from host param
	host = params.get('host')
	# failed? then try to get it from the rest of the params
	if host is None:
		# second attempt: reconstruct url from mn and mvi?
		# the host schema seems to be as follows:
		# rr{mvi[any]/fvip[any]?}---{mn[any]}.googlevideo.com
		# regarding mvi/fvip, it seems that any value smaller than 5 passes
		try:
			mvi = params.get('mvi').split(',')[-1]
			mn  = params.get('mn').split(',')[-1]
			if int(mvi) > 5:
				mvi = 3 # invidious uses this as fallback
			host = f"rr{mvi}---{mn}.googlevideo.com"
		except (AttributeError, ValueError):
			return Response(json.dumps({"error": "Couldn't extract crucial parameters for hostname reconstruction"}, mimetype="application/json", status=400))
	else:
		# don't echo host "hint" back to the googlevideo server
		del params['host']
	# run a regex sanity check
	if re.fullmatch(r"[\w-]+\.googlevideo\.com", host) is None:
		# fallback behavior for unexpected hostnames
		return Response(json.dumps({"error": "Please either pass a valid host, or don't pass any"}), mimetype="application/json", status=400)

	try:
		# request the proxied data
		remote_response = requests.get(f"https://{host}/videoplayback", headers=proxy_headers, params=params, stream=True)
	except:
		return Response(json.dumps({"error": "Couldn't connect to googlevideo host"}), mimetype="application/json", status=500)

	# determine the chunk size
	chunk_size = 10 * 1024 # by default it's 10 MB (as this is the most youtube is willing to send without ratelimiting)
	# or the one in initcwndbps (if user enabled the config flag to match chunk_size with initcwndbps)
	if ythdd_globals.config['proxy']['match_initcwndbps']:
		try:
			chunk_size = int(params.get('initcwndbps') / 1024)
		except:
			pass
	# return a chunked response
	resp = Response(remote_response.iter_content(chunk_size=chunk_size), content_type=remote_response.headers['Content-Type'], status=remote_response.status_code, headers=remote_response.headers, direct_passthrough=True)
	resp.cache_control.public = True
	resp.cache_control.max_age = int(60_000)

	return resp