import re import requests, requests_cache from flask import Flask, Blueprint, request, Response from flask_login import current_user frontend = Blueprint('proxy', __name__) def app(): app = Flask(__name__) app.register_blueprint(frontend) return app # https://github.com/psf/requests/issues/5612 def silence_errors(generator): try: yield from generator except requests.exceptions.RequestException: return None @frontend.route("/videoplayback/") @frontend.route("/videoplayback") def videoplayback(path=None): if path is not None: # seg.ts URL args = path.split("/")[0:-1] host = dict(zip(args[::2], args[1::2])).get('hls_chunk_host') path = "/" + path.replace(';','%3B').replace('=','%3D') #host = request.args.get('__host__') else: fvip = request.args.get("fvip", "3") mn = request.args.get("mn", "").split(",")[-1] host = f"r{fvip}---{mn}.googlevideo.com" path = "" REQUEST_HEADERS_WHITELIST = { "accept", "accept-encoding", "cache-control", "range", } RESPONSE_HEADERS_WHITELIST = { "accept-ranges", "cache-control", "content-length", "content-range", "content-type", "expires", "x-content-type-options", } req_headers = { k:v for k,v in request.headers if k.lower() in REQUEST_HEADERS_WHITELIST } with requests_cache.disabled(): r = requests.get(f"https://{host}/videoplayback{path}", request.args.to_dict(), headers=req_headers, stream=True) resp_headers = { k:v for k,v in r.headers.items() if k.lower() in RESPONSE_HEADERS_WHITELIST } response = Response(silence_errors(r.iter_content(chunk_size=8192)), status=r.status_code, headers=resp_headers) response.call_on_close(lambda: r.close()) response.headers['Access-Control-Allow-Origin'] = '*' return response @frontend.route("/api/manifest/hls_playlist/") @frontend.route("/api/manifest/hls_variant/") def hls_manifest(path): with requests_cache.disabled(): path_fixed = request.path.replace(';','%3B').replace('=','%3D') r = requests.get(f"https://manifest.googlevideo.com{path_fixed}", request.args.to_dict()) if not r.ok: return "", r.status_code lines = [] for line in r.text.splitlines(): if line.startswith('#'): lines.append(line) elif line.startswith('https://'): scheme, _, host, _path = line.split('/', 3) assert '.googlevideo.com' in host #lines.append("/" + _path + "?__host__=" + host) lines.append("/" + _path ) else: raise StopIteration #if line.startswith('https://manifest.googlevideo.com/'): rv = "\n".join(lines) #rv = r.text #rv = rv.replace("https://manifest.googlevideo.com/", f"https://{request.host}/") #rv = re.sub(r"https://([a-z0-9-]+\.googlevideo\.com)", r"https://subscriptions.gir.st/ts-proxy/\1", rv) return rv, {'content-type': 'application/x-mpegURL', 'Access-Control-Allow-Origin': '*'} @frontend.route("/x-hls/") @frontend.route("/x-hls/") def xxx(vid="5qap5aO4i9A"): from ..common.common import get_video_info _, map, _, _, _ = get_video_info(vid, -1) url = map['hlsManifestUrl'].replace("https://manifest.googlevideo.com", "") return f"""

""" if __name__ == '__main__': app().run(debug=True)