Spaces:
Paused
Paused
File size: 2,020 Bytes
99a7ebb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 | """Global outbound proxy helpers for upstream ChatGPT and CPA requests."""
from __future__ import annotations
import time
from urllib.parse import urlparse
from curl_cffi.requests import Session
from services.config import config
class ProxySettingsStore:
def build_session_kwargs(self, **session_kwargs) -> dict[str, object]:
proxy = config.get_proxy_settings()
if proxy:
session_kwargs["proxy"] = proxy
return session_kwargs
def _clean(value: object) -> str:
return str(value or "").strip()
def _is_valid_proxy_url(url: str) -> bool:
parsed = urlparse(url)
return parsed.scheme in {"http", "https", "socks5", "socks5h"} and bool(parsed.netloc)
def test_proxy(url: str, *, timeout: float = 15.0) -> dict:
candidate = _clean(url)
if not candidate:
return {"ok": False, "status": 0, "latency_ms": 0, "error": "proxy url is required"}
if not _is_valid_proxy_url(candidate):
return {"ok": False, "status": 0, "latency_ms": 0, "error": "invalid proxy url"}
session = Session(impersonate="edge101", verify=True, proxy=candidate)
started = time.perf_counter()
try:
response = session.get(
"https://chatgpt.com/api/auth/csrf",
headers={"user-agent": "Mozilla/5.0 (chatgpt2api proxy test)"},
timeout=timeout,
)
latency_ms = int((time.perf_counter() - started) * 1000)
return {
"ok": response.status_code < 500,
"status": int(response.status_code),
"latency_ms": latency_ms,
"error": None if response.status_code < 500 else f"HTTP {response.status_code}",
}
except Exception as exc:
latency_ms = int((time.perf_counter() - started) * 1000)
return {
"ok": False,
"status": 0,
"latency_ms": latency_ms,
"error": str(exc) or exc.__class__.__name__,
}
finally:
session.close()
proxy_settings = ProxySettingsStore()
|