from __future__ import annotations import os from typing import Any from urllib.parse import urlsplit def _is_truthy(value: str | None) -> bool: return (value or "").strip().lower() in {"1", "true", "yes", "on"} def _normalize_proxy_scheme(scheme: str) -> str: normalized = (scheme or "").strip().lower() if normalized == "socks4": return "socks4a" if normalized == "socks5": return "socks5h" return normalized def build_proxy_url_from_proxychains_env(*, require_enabled: bool = True) -> str | None: if require_enabled and not _is_truthy(os.environ.get("ENABLE_PROXYCHAINS")): return None scheme = _normalize_proxy_scheme(os.environ.get("PROXYCHAINS_PROXY_TYPE", "socks4")) host = (os.environ.get("PROXYCHAINS_PROXY_HOST") or "").strip() port = (os.environ.get("PROXYCHAINS_PROXY_PORT") or "").strip() if not host or not port: return None return f"{scheme}://{host}:{port}" def _same_proxy_endpoint(left: str, right: str) -> bool: left_parts = urlsplit(left) right_parts = urlsplit(right) return ( left_parts.hostname == right_parts.hostname and left_parts.port == right_parts.port ) def resolve_requests_proxy_url(proxy_url: str | None) -> str: value = (proxy_url or "").strip() if not value: return "" proxychains_url = build_proxy_url_from_proxychains_env() if proxychains_url and _same_proxy_endpoint(value, proxychains_url): # Avoid double-proxying requests when the whole process is already wrapped by proxychains. return "" return value def build_requests_proxies(proxy_url: str | None) -> dict[str, str] | None: value = resolve_requests_proxy_url(proxy_url) if not value: return None return {"http": value, "https": value} def apply_requests_proxy(session: Any, proxy_url: str | None) -> Any: proxies = build_requests_proxies(proxy_url) if not proxies: return session session.proxies.update(proxies) session.trust_env = False return session