Prefer curl impersonation for protected media

This commit is contained in:
Dymas
2026-09-07 21:30:16 +02:00
parent 5f75fef4de
commit baf5298631
6 changed files with 90 additions and 15 deletions
+51 -11
View File
@@ -215,32 +215,72 @@ def request_headers(headers=None, referer=None):
def curl_candidates():
def candidate_score(name):
text = Path(str(name or "")).name
numbers = [int(value) for value in re.findall(r"\d+", text)]
version = numbers[0] if numbers else 0
family_score = 4 if text.startswith("curl_chrome") else 3 if text.startswith("curl_firefox") else 2 if text.startswith("curl_safari") else 1
return (family_score, version, text)
configured = str(os.environ.get("KAIZOKU_CURL_BIN") or "").strip()
names = [
configured,
"curl_chrome120",
"curl_chrome116",
"curl_chrome110",
"curl-impersonate",
"curl",
]
names = [configured]
wrapper_prefixes = ("curl_chrome", "curl_firefox", "curl_safari", "curl_edge")
path_dirs = [Path(part) for part in os.environ.get("PATH", "").split(os.pathsep) if part]
discovered = []
for directory in path_dirs:
try:
for path in directory.iterdir():
if path.name.startswith(wrapper_prefixes) and os.access(path, os.X_OK):
discovered.append(path.name)
except OSError:
continue
discovered.sort(key=candidate_score, reverse=True)
names.extend(discovered)
names.extend(
[
"curl_chrome142",
"curl_chrome136",
"curl_chrome133a",
"curl_chrome133",
"curl_chrome131",
"curl_chrome124",
"curl_chrome123",
"curl_chrome120",
"curl_chrome116",
"curl_chrome110",
"curl_firefox135",
"curl_firefox133",
"curl-impersonate",
"curl",
]
)
candidates = []
for name in names:
if not name or name in candidates:
if not name:
continue
resolved = shutil.which(name)
if resolved:
if resolved and resolved not in candidates:
candidates.append(resolved)
return candidates
def curl_uses_browser_wrapper(curl_bin):
return Path(str(curl_bin or "")).name.startswith(("curl_chrome", "curl_firefox", "curl_safari", "curl_edge"))
def curl_request_headers(curl_bin, headers=None):
if not curl_uses_browser_wrapper(curl_bin):
return request_headers(headers)
return {key: value for key, value in (headers or {}).items() if value}
def curl_fetch_bytes(url, headers=None, timeout=30):
candidates = curl_candidates()
if not candidates:
raise RuntimeError("curl is not available for media fetch fallback.")
clean_headers = request_headers(headers)
last_error = None
for curl_bin in candidates:
clean_headers = curl_request_headers(curl_bin, headers)
cmd = [
curl_bin,
"--location",