simon/jacuzzi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

default impersonation on yt-dlp

Browse Source
This commit is contained in:
Simon
2026-06-23 06:19:50 +00:00
parent b5b3e13dd0
commit 3d4b90b0e1
1 changed files with 50 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

72
backend/main.py
View File

@@ -7,9 +7,30 @@ import urllib.parse
from requests.adapters import HTTPAdapter
from urllib3.util import Retry
import yt_dlp
from yt_dlp.networking.impersonate import ImpersonateTarget
from curl_cffi import requests as impersonate_requests
import threading
import io
from urllib.parse import urljoin
# Browser to impersonate at the TLS/HTTP layer. Some origins (e.g. the
# "animeidhentai" hottub channel) fingerprint clients and reset/403 anything
# that isn't a real browser, so impersonation must be on by default.
IMPERSONATE_TARGET = os.getenv('STREAM_IMPERSONATE', 'chrome').strip() or 'chrome'
# curl_cffi sessions wrap a single libcurl handle and are not safe to share
# across threads; keep one per worker thread so the Flask `threaded=True`
# server can proxy concurrent segments without corrupting state.
_thread_local = threading.local()
def get_impersonate_session():
sess = getattr(_thread_local, 'session', None)
if sess is None:
sess = impersonate_requests.Session(impersonate=IMPERSONATE_TARGET)
_thread_local.session = sess
return sess
# Stream params that have dedicated meaning and must never be treated as headers.
STREAM_RESERVED_PARAMS = {'url'}
# Headers that affect the transport layer rather than the resource itself; allowing
@@ -264,23 +285,17 @@ def stream_video():
return chunk[4:8] == b'ftyp'
def build_upstream_headers(referer):
# We fetch upstream through curl_cffi with browser impersonation, which
# supplies a coherent User-Agent / Accept / Sec-CH-UA / Accept-Encoding
# set matching the impersonated browser. Forwarding the client's own
# values (the player may be Firefox while we impersonate Chrome) would
# contradict the TLS fingerprint and defeat impersonation, so we only
# pass headers the origin genuinely needs for authorization.
headers = {
'User-Agent': request.headers.get('User-Agent'),
'Accept': request.headers.get('Accept'),
'Accept-Language': request.headers.get('Accept-Language'),
'Accept-Encoding': request.headers.get('Accept-Encoding'),
'Referer': referer,
'Origin': referer,
}
# Pass through fetch metadata and client hints when present
for key in (
'Sec-Fetch-Mode', 'Sec-Fetch-Site', 'Sec-Fetch-Dest', 'Sec-Fetch-User',
'Sec-CH-UA', 'Sec-CH-UA-Mobile', 'Sec-CH-UA-Platform', 'DNT'
):
if key in request.headers:
headers[key] = request.headers[key]
if forward_cookies and 'Cookie' in request.headers:
headers['Cookie'] = request.headers['Cookie']
dbg("forwarding cookies")
@@ -340,7 +355,16 @@ def stream_video():
if 'Range' in request.headers:
safe_request_headers['Range'] = request.headers['Range']
resp = session.get(target_url, headers=safe_request_headers, stream=True, timeout=30, allow_redirects=True)
resp = get_impersonate_session().get(target_url, headers=safe_request_headers, stream=True, timeout=30, allow_redirects=True)
# Some channel proxies (e.g. the "animeidhentai" hottub proxy) use
# inverted hotlink protection: they 403 any request that carries a
# Referer/Origin and only serve referer-less ones. Other CDNs require
# the spoofed referer. Satisfy both by retrying once without it.
if resp.status_code == 403 and ('Referer' in safe_request_headers or 'Origin' in safe_request_headers):
dbg("upstream 403 with referer; retrying without referer/origin")
resp.close()
referer_less = {k: v for k, v in safe_request_headers.items() if k not in ('Referer', 'Origin')}
resp = get_impersonate_session().get(target_url, headers=referer_less, stream=True, timeout=30, allow_redirects=True)
if debug_enabled:
dbg(f"upstream status={resp.status_code} content_type={resp.headers.get('Content-Type')} content_length={resp.headers.get('Content-Length')}")
@@ -436,11 +460,14 @@ def stream_video():
for key, value in upstream_headers.items():
if value:
headers[key] = value
if 'User-Agent' not in headers:
headers['User-Agent'] = 'Mozilla/5.0'
if 'Accept' not in headers:
headers['Accept'] = '*/*'
resp = session.get(playlist_url, headers=headers, stream=True, timeout=30)
resp = get_impersonate_session().get(playlist_url, headers=headers, stream=True, timeout=30)
# See proxy_response: retry without referer for inverted hotlink
# protection that 403s any refered request.
if resp.status_code == 403 and ('Referer' in headers or 'Origin' in headers):
dbg("playlist upstream 403 with referer; retrying without referer/origin")
resp.close()
referer_less = {k: v for k, v in headers.items() if k not in ('Referer', 'Origin')}
resp = get_impersonate_session().get(playlist_url, headers=referer_less, stream=True, timeout=30)
base_url = resp.url
if resp.status_code >= 400:
@@ -624,15 +651,16 @@ def stream_video():
'format_sort': ['res', 'fps', 'vcodec:avc1', 'acodec:aac'],
'quiet': False,
'no_warnings': False,
'http_headers': {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
},
# Impersonate a real browser by default so origins that fingerprint
# clients (e.g. the "animeidhentai" hottub channel) don't 403.
'impersonate': ImpersonateTarget.from_str(IMPERSONATE_TARGET),
}
passthrough_source = request.json if request.method == 'POST' else request.args
passthrough_headers = collect_passthrough_headers(passthrough_source)
dbg(f"passthrough_headers={list(passthrough_headers.keys())}")
ydl_opts['http_headers'].update(passthrough_headers)
if passthrough_headers:
ydl_opts.setdefault('http_headers', {}).update(passthrough_headers)
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
# Extract the info