187 lines
6.9 KiB
Python
187 lines
6.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Idempotently merge WARP/FlareSolverr proxy_runtime defaults into config.json."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
DEFAULT_USER_AGENT = (
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
|
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
|
"Chrome/145.0.0.0 Safari/537.36"
|
|
)
|
|
|
|
DEFAULT_PROXY_RUNTIME: dict[str, Any] = {
|
|
"enabled": False,
|
|
"egress_mode": "direct",
|
|
"proxy_url": "",
|
|
"resource_proxy_url": "",
|
|
"skip_ssl_verify": False,
|
|
"reset_session_status_codes": [403],
|
|
"clearance": {
|
|
"enabled": False,
|
|
"mode": "none",
|
|
"cf_cookies": "",
|
|
"cf_clearance": "",
|
|
"user_agent": DEFAULT_USER_AGENT,
|
|
"browser": "chrome",
|
|
"flaresolverr_url": "",
|
|
"timeout_sec": 60,
|
|
"refresh_interval": 3600,
|
|
"warm_up_on_start": False,
|
|
},
|
|
}
|
|
|
|
|
|
def _env_bool(name: str, default: bool) -> bool:
|
|
value = os.getenv(name)
|
|
if value is None:
|
|
return default
|
|
lowered = value.strip().lower()
|
|
if lowered in {"1", "true", "yes", "on"}:
|
|
return True
|
|
if lowered in {"0", "false", "no", "off"}:
|
|
return False
|
|
return default
|
|
|
|
|
|
def _env_int(name: str, default: int, minimum: int = 0) -> int:
|
|
try:
|
|
return max(minimum, int(os.getenv(name, str(default))))
|
|
except (TypeError, ValueError, OverflowError):
|
|
return default
|
|
|
|
|
|
def _warp_runtime_defaults() -> dict[str, Any]:
|
|
clearance_enabled = _env_bool("CHATGPT2API_PROXY_RUNTIME_CLEARANCE_ENABLED", True)
|
|
clearance_mode = os.getenv("CHATGPT2API_PROXY_RUNTIME_CLEARANCE_MODE", "flaresolverr").strip().lower()
|
|
if clearance_mode not in {"none", "manual", "flaresolverr"}:
|
|
clearance_mode = "flaresolverr" if clearance_enabled else "none"
|
|
if not clearance_enabled:
|
|
clearance_mode = "none"
|
|
|
|
egress_mode = os.getenv("CHATGPT2API_PROXY_RUNTIME_EGRESS_MODE", "single_proxy").strip().lower()
|
|
if egress_mode not in {"direct", "single_proxy"}:
|
|
egress_mode = "single_proxy"
|
|
|
|
runtime = copy.deepcopy(DEFAULT_PROXY_RUNTIME)
|
|
runtime.update(
|
|
{
|
|
"enabled": _env_bool("CHATGPT2API_PROXY_RUNTIME_ENABLED", True),
|
|
"egress_mode": egress_mode,
|
|
"proxy_url": os.getenv("CHATGPT2API_PROXY_RUNTIME_PROXY_URL", "http://privoxy:8118").strip(),
|
|
"resource_proxy_url": os.getenv("CHATGPT2API_PROXY_RUNTIME_RESOURCE_PROXY_URL", "").strip(),
|
|
"skip_ssl_verify": _env_bool("CHATGPT2API_PROXY_RUNTIME_SKIP_SSL_VERIFY", False),
|
|
"reset_session_status_codes": [
|
|
int(part.strip())
|
|
for part in os.getenv("CHATGPT2API_PROXY_RUNTIME_RESET_STATUS_CODES", "403").split(",")
|
|
if part.strip().isdigit() and 100 <= int(part.strip()) <= 599
|
|
] or [403],
|
|
}
|
|
)
|
|
runtime["clearance"].update(
|
|
{
|
|
"enabled": clearance_enabled,
|
|
"mode": clearance_mode,
|
|
"user_agent": os.getenv("CHATGPT2API_PROXY_RUNTIME_USER_AGENT", DEFAULT_USER_AGENT).strip() or DEFAULT_USER_AGENT,
|
|
"browser": os.getenv("CHATGPT2API_PROXY_RUNTIME_BROWSER", "chrome").strip() or "chrome",
|
|
"flaresolverr_url": os.getenv("CHATGPT2API_FLARESOLVERR_URL", "http://flaresolverr:8191").strip(),
|
|
"timeout_sec": _env_int("CHATGPT2API_PROXY_RUNTIME_CLEARANCE_TIMEOUT_SEC", 60, 1),
|
|
"refresh_interval": _env_int("CHATGPT2API_PROXY_RUNTIME_CLEARANCE_REFRESH_INTERVAL", 3600, 60),
|
|
"warm_up_on_start": _env_bool("CHATGPT2API_PROXY_RUNTIME_WARM_UP_ON_START", False),
|
|
}
|
|
)
|
|
return runtime
|
|
|
|
|
|
def _deep_fill_missing(target: dict[str, Any], defaults: dict[str, Any]) -> bool:
|
|
changed = False
|
|
for key, value in defaults.items():
|
|
if key not in target:
|
|
target[key] = copy.deepcopy(value)
|
|
changed = True
|
|
continue
|
|
if isinstance(target[key], dict) and isinstance(value, dict):
|
|
changed = _deep_fill_missing(target[key], value) or changed
|
|
return changed
|
|
|
|
|
|
def _looks_like_repository_default(runtime: Any) -> bool:
|
|
if not isinstance(runtime, dict):
|
|
return True
|
|
candidate = copy.deepcopy(runtime)
|
|
_deep_fill_missing(candidate, DEFAULT_PROXY_RUNTIME)
|
|
return candidate == DEFAULT_PROXY_RUNTIME
|
|
|
|
|
|
def _mask_url(value: str) -> str:
|
|
return re.sub(r"(https?://)([^\s/@:]+):([^\s/@]+)@", r"\1[REDACTED]@", value or "", flags=re.I)
|
|
|
|
|
|
def main() -> int:
|
|
config_path = Path(os.getenv("CHATGPT2API_CONFIG_FILE", "/app/config.json"))
|
|
if not config_path.exists():
|
|
print(f"Config file not found, creating {config_path}")
|
|
data: dict[str, Any] = {}
|
|
else:
|
|
try:
|
|
data = json.loads(config_path.read_text(encoding="utf-8"))
|
|
except json.JSONDecodeError as exc:
|
|
print(f"Invalid JSON in {config_path}: {exc}", file=sys.stderr)
|
|
return 1
|
|
if not isinstance(data, dict):
|
|
print(f"Config root must be an object: {config_path}", file=sys.stderr)
|
|
return 1
|
|
|
|
desired = _warp_runtime_defaults()
|
|
existing = data.get("proxy_runtime")
|
|
changed = False
|
|
|
|
if _looks_like_repository_default(existing) or _env_bool("CHATGPT2API_PROXY_RUNTIME_FORCE", False):
|
|
data["proxy_runtime"] = desired
|
|
changed = True
|
|
print("Created proxy_runtime defaults")
|
|
else:
|
|
runtime = existing if isinstance(existing, dict) else {}
|
|
changed = _deep_fill_missing(runtime, DEFAULT_PROXY_RUNTIME)
|
|
data["proxy_runtime"] = runtime
|
|
print("Proxy runtime already configured")
|
|
|
|
if changed:
|
|
config_path.parent.mkdir(parents=True, exist_ok=True)
|
|
payload = json.dumps(data, ensure_ascii=False, indent=2) + "\n"
|
|
tmp_path = config_path.with_suffix(config_path.suffix + ".tmp")
|
|
try:
|
|
tmp_path.write_text(payload, encoding="utf-8")
|
|
tmp_path.replace(config_path)
|
|
except OSError as exc:
|
|
# Docker bind-mounted single files can reject atomic rename with EBUSY.
|
|
# Fall back to in-place write so the init job works with both file and
|
|
# directory mounts.
|
|
if getattr(exc, "errno", None) != 16:
|
|
raise
|
|
config_path.write_text(payload, encoding="utf-8")
|
|
tmp_path.unlink(missing_ok=True)
|
|
|
|
runtime = data.get("proxy_runtime") if isinstance(data.get("proxy_runtime"), dict) else {}
|
|
clearance = runtime.get("clearance") if isinstance(runtime.get("clearance"), dict) else {}
|
|
print(
|
|
"Proxy runtime summary: "
|
|
f"enabled={bool(runtime.get('enabled'))}, "
|
|
f"egress_mode={runtime.get('egress_mode')}, "
|
|
f"proxy_url={_mask_url(str(runtime.get('proxy_url') or ''))}, "
|
|
f"clearance_mode={clearance.get('mode')}, "
|
|
f"flaresolverr_url={_mask_url(str(clearance.get('flaresolverr_url') or ''))}"
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|