209 lines
6.5 KiB
Python
209 lines
6.5 KiB
Python
"""YouTube push proxy resolution (from d2ypp2 douyin_youtube_ffplay, simplified)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import shutil
|
|
import socket
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
|
|
_PROXY_ENV_KEYS = (
|
|
"http_proxy",
|
|
"https_proxy",
|
|
"HTTP_PROXY",
|
|
"HTTPS_PROXY",
|
|
"ALL_PROXY",
|
|
"all_proxy",
|
|
)
|
|
|
|
|
|
def _is_disabled_value(value: str) -> bool:
|
|
return (value or "").strip().lower() in {"0", "false", "no", "off", "none", "direct"}
|
|
|
|
|
|
def _is_local_tcp_open(host: str, port: int, timeout: float = 0.35) -> bool:
|
|
try:
|
|
with socket.create_connection((host, port), timeout=timeout):
|
|
return True
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def normalize_push_proxy_mode(configured: str | None = None) -> str:
|
|
value = configured or os.environ.get("YOUTUBE_PUSH_PROXY_MODE") or os.environ.get("LIVE_YOUTUBE_PUSH_PROXY_MODE")
|
|
if not value:
|
|
value = "auto"
|
|
low = (value or "").strip().lower()
|
|
if low in {"0", "false", "no", "off", "none", "direct"}:
|
|
return "off"
|
|
if low in {"env", "environment", "http_proxy", "https_proxy"}:
|
|
return "env"
|
|
if low in {"proxychains", "proxy", "chain", "proxychains4"}:
|
|
return "proxychains"
|
|
return "auto"
|
|
|
|
|
|
def resolve_youtube_push_proxy(configured: str | None = None) -> str:
|
|
for value in (
|
|
configured,
|
|
os.environ.get("YOUTUBE_PUSH_PROXY"),
|
|
os.environ.get("LIVE_YOUTUBE_PUSH_PROXY"),
|
|
):
|
|
value = (value or "").strip()
|
|
if _is_disabled_value(value):
|
|
return ""
|
|
if value:
|
|
return value
|
|
|
|
auto = (os.environ.get("YOUTUBE_PUSH_PROXY_AUTO") or "1").strip().lower()
|
|
port_raw = os.environ.get("YOUTUBE_PUSH_PROXY_PORT") or os.environ.get("LIVE_YOUTUBE_PUSH_PROXY_PORT") or "7890"
|
|
try:
|
|
proxy_port = int(str(port_raw).strip())
|
|
except (TypeError, ValueError):
|
|
proxy_port = 7890
|
|
if auto not in {"0", "false", "no", "off"} and _is_local_tcp_open("127.0.0.1", proxy_port):
|
|
return f"socks5://127.0.0.1:{proxy_port}"
|
|
|
|
for key in ("https_proxy", "http_proxy", "HTTPS_PROXY", "HTTP_PROXY", "ALL_PROXY", "all_proxy"):
|
|
value = (os.environ.get(key) or "").strip()
|
|
if _is_disabled_value(value):
|
|
return ""
|
|
if value:
|
|
return value
|
|
return ""
|
|
|
|
|
|
def redact_proxy_url(value: str) -> str:
|
|
return re.sub(r"(://[^:/@\s]+:)[^@/\s]+(@)", r"\1<redacted>\2", value or "")
|
|
|
|
|
|
def should_use_proxychains(push_proxy: str, proxy_mode: str = "auto") -> bool:
|
|
if not push_proxy:
|
|
return False
|
|
mode = normalize_push_proxy_mode(proxy_mode)
|
|
if mode in {"off", "env"}:
|
|
return False
|
|
if mode == "proxychains":
|
|
return bool(shutil.which("proxychains4"))
|
|
scheme = (urlparse(push_proxy if "://" in push_proxy else f"socks5://{push_proxy}").scheme or "socks5").lower()
|
|
if scheme.startswith("socks"):
|
|
return bool(shutil.which("proxychains4"))
|
|
return False
|
|
|
|
|
|
def proxychains_proxy_line(push_proxy: str) -> str | None:
|
|
value = (push_proxy or "").strip()
|
|
if not value:
|
|
return None
|
|
parsed = urlparse(value if "://" in value else f"socks5://{value}")
|
|
host = parsed.hostname
|
|
port = parsed.port
|
|
if not host or not port:
|
|
return None
|
|
scheme = (parsed.scheme or "socks5").lower()
|
|
if scheme in {"socks", "socks5", "socks5h"}:
|
|
proxy_type = "socks5"
|
|
elif scheme == "socks4":
|
|
proxy_type = "socks4"
|
|
elif scheme in {"http", "https"}:
|
|
proxy_type = "http"
|
|
else:
|
|
return None
|
|
parts = [proxy_type, host, str(port)]
|
|
if parsed.username:
|
|
parts.append(parsed.username)
|
|
if parsed.password:
|
|
parts.append(parsed.password)
|
|
return " ".join(parts)
|
|
|
|
|
|
def proxychains_config_for_push(push_proxy: str, *, base_dir: str | Path | None = None) -> str | None:
|
|
explicit = (
|
|
os.environ.get("YOUTUBE_PROXYCHAINS_CONF")
|
|
or os.environ.get("LIVE_YOUTUBE_PROXYCHAINS_CONF")
|
|
or ""
|
|
).strip()
|
|
if explicit and os.path.isfile(explicit):
|
|
return explicit
|
|
|
|
proxy_line = proxychains_proxy_line(push_proxy)
|
|
root = Path(base_dir or os.path.dirname(os.path.abspath(__file__)))
|
|
if proxy_line:
|
|
runtime_dir = root / ".process_runner"
|
|
try:
|
|
runtime_dir.mkdir(parents=True, exist_ok=True)
|
|
conf = runtime_dir / "proxychains-youtube.conf"
|
|
conf.write_text(
|
|
"strict_chain\n"
|
|
"proxy_dns\n"
|
|
"tcp_read_time_out 30000\n"
|
|
"tcp_connect_time_out 8000\n\n"
|
|
"[ProxyList]\n"
|
|
f"{proxy_line}\n",
|
|
encoding="utf-8",
|
|
)
|
|
try:
|
|
os.chmod(conf, 0o600)
|
|
except OSError:
|
|
pass
|
|
return str(conf)
|
|
except OSError:
|
|
pass
|
|
|
|
bundled = root / "config" / "proxychains-youtube.conf"
|
|
if bundled.is_file():
|
|
return str(bundled)
|
|
return None
|
|
|
|
|
|
def ffmpeg_env_with_push_proxy(
|
|
push_proxy: str,
|
|
proxy_mode: str = "auto",
|
|
*,
|
|
use_proxychains: bool = False,
|
|
base_env: dict[str, str] | None = None,
|
|
) -> dict[str, str]:
|
|
env = dict(base_env or os.environ)
|
|
for key in _PROXY_ENV_KEYS:
|
|
env.pop(key, None)
|
|
push_proxy = (push_proxy or "").strip()
|
|
if not push_proxy:
|
|
return env
|
|
mode = normalize_push_proxy_mode(proxy_mode)
|
|
if mode == "off" or use_proxychains:
|
|
return env
|
|
for key in _PROXY_ENV_KEYS:
|
|
env[key] = push_proxy
|
|
return env
|
|
|
|
|
|
def proxychains_command(ffmpeg_command: list[str], push_proxy: str, proxy_mode: str = "auto", *, base_dir: str | Path | None = None) -> list[str] | None:
|
|
if not should_use_proxychains(push_proxy, proxy_mode):
|
|
return None
|
|
conf = proxychains_config_for_push(push_proxy, base_dir=base_dir)
|
|
if not conf:
|
|
return None
|
|
return ["proxychains4", "-q", "-f", conf, *ffmpeg_command]
|
|
|
|
|
|
def apply_push_proxy_to_ffmpeg_cmd(
|
|
ffmpeg_command: list[str],
|
|
push_proxy: str,
|
|
proxy_mode: str = "auto",
|
|
*,
|
|
base_dir: str | Path | None = None,
|
|
) -> tuple[list[str], bool]:
|
|
"""Return (command, used_proxychains). Falls back to -http_proxy when proxychains unavailable."""
|
|
wrapped = proxychains_command(ffmpeg_command, push_proxy, proxy_mode, base_dir=base_dir)
|
|
if wrapped:
|
|
return wrapped, True
|
|
mode = normalize_push_proxy_mode(proxy_mode)
|
|
if push_proxy and mode != "off":
|
|
cmd = list(ffmpeg_command)
|
|
cmd.insert(1, "-http_proxy")
|
|
cmd.insert(2, push_proxy)
|
|
return cmd, False
|
|
return ffmpeg_command, False
|