#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ config.py — 把 server.py 原本硬编码在顶层的 DEFAULT_* 全部改成从环境变量读。 设计要点: - 静态项(上游/sitekey/工厂页)只在启动时读一次, 改了要重启进程。 - 动态项(worker_pages / shelf_size 等)只是给 TokenPool 的"初始值", 运行时仍可被管理面板热改(见 server.py 的 setter)。 - 敏感项(relay keys / admin key / proxy)只在私有环境变量里给, admin 面板里的敏感编辑默认禁用, 防公网误操作。 - HF_SPACE 由 Hugging Face 自动注入; 命中后 server.py 会强制走容器化 启动参数(--no-sandbox 等)。 - 本模块不抛异常: 任何缺失走默认值, 保证容器最小配置即可起服务。 """ import json import os from typing import Dict, Set # --------------------------- 工具: 解析环境变量 --------------------------- def _str(name: str, default: str) -> str: return os.environ.get(name, default) def _int(name: str, default: int) -> int: raw = os.environ.get(name) if raw is None or raw == "": return default try: return int(raw) except ValueError: return default def _bool(name: str, default: bool) -> bool: raw = os.environ.get(name) if raw is None: return default return raw.strip().lower() in ("1", "true", "yes", "on") def _relay_keys(name: str, default: Dict[str, str]) -> Set[str]: """relay key 集合: 既支持 JSON 对象 ({"key":"user"}) 或 JSON 数组 (["k1","k2"]), 也支持换行/逗号分隔的纯 key 列表(每项可带 =user 后缀)。""" raw = os.environ.get(name) if not raw: return set(default.keys()) raw = raw.strip() keys: Set[str] = set() # 先试 JSON: dict 或 list 都收 try: data = json.loads(raw) if isinstance(data, dict): keys.update(str(k).strip() for k in data if str(k).strip()) return keys if isinstance(data, list): keys.update(str(k).strip() for k in data if str(k).strip()) return keys except (json.JSONDecodeError, TypeError): pass # 否则当分隔列表(逗号或换行) for tok in raw.replace("\n", ",").split(","): tok = tok.strip() if "=" in tok: # 形如 "sk-relay-xxx=default-user" tok = tok.split("=", 1)[0].strip() if tok: keys.add(tok) return keys or set(default.keys()) # --------------------------- 默认值(本机开发兜底) --------------------------- _DEFAULT_KEYS: set = set() # 不内置任何默认 key; 未配 RELAY_KEYS 时所有 /v1 请求 401 class Config: """启动期一次性快照。运行时热改的动态项在 TokenPool 里自行持有。""" # --- 静态: 改需重启 --- UPSTREAM: str = _str("UPSTREAM", "https://api.deepinfra.com") SITEKEY: str = _str("SITEKEY", "0x4AAAAAADlBNBTRb73O02Vo") FACTORY_PAGE: str = _str("FACTORY_PAGE", "https://deepinfra.com/google/gemma-4-26B-A4B-it") # token 通用(任一可路由页皆可); 选 gemma-4-26B: # 激活参数最小(4B)+冷门, 签发不挤占热门模型的匿名配额 # 仅 real 模式使用; intercept/mock 模式只取其 host FACTORY_MODE: str = _str("FACTORY_MODE", "intercept") # token 工厂页承载方式(面板可热切): # intercept(默认/推荐): Playwright route 拦截, # 空壳页纯内存承载, 零基础设施, 不访问目标站 # mock: --host-resolver-rules 域名映射 + 容器内 # 本地 HTTPS 空壳页(自签证书), 绕开 CDP 拦截层 # real: 真实加载工厂页(旧行为, 保底) MOCK_PORT: int = _int("MOCK_PORT", 8443) # mock 模式本地 HTTPS 空壳页端口(仅容器内 loopback) UA: str = _str("UA", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/151.0.0.0 Safari/537.36 Edg/151.0.0.0") # --- 鉴权(私有) --- RELAY_KEYS: Set[str] = _relay_keys("RELAY_KEYS", _DEFAULT_KEYS) ADMIN_KEY: str = _str("ADMIN_KEY", "") # 空=禁用 /admin 控制台 ADMIN_ENABLED: bool = _bool("ADMIN_ENABLED", False) # HF 上默认关, 防 /admin 被公网发现 # --- 浏览器/容器 --- HF_SPACE: bool = bool(os.environ.get("HF_SPACE")) # HF 自动注入 HEADLESS: bool = _bool("HEADLESS", os.environ.get("HF_SPACE") is not None) # 默认: HF 上 true(省资源), 本机 false(POC 验证) # 若 Turnstile 在 HF 出口拒真 headless, 设 HEADLESS=false # 让 Dockerfile 的 Xvfb 兜住"headless=False"的指纹 PROFILE_DIR: str = _str("PROFILE_DIR", os.path.expanduser("~/.cache/relay-pw-profile")) PROXY: str = _str("PROXY", "") # 私有: 可信出口代理, 绕 HF 共享 IP # --- 动态初值(面板仍可热改) --- WORKER_PAGES: int = _int("WORKER_PAGES", 3) MAX_WORKER_PAGES: int = _int("MAX_WORKER_PAGES", 16) SHELF_SIZE: int = _int("SHELF_SIZE", 100) # 货架容量上限(自适应模式下为封顶值) TOKEN_MAX_AGE: int = _int("TOKEN_MAX_AGE", 250) IDLE_PAUSE: int = _int("IDLE_PAUSE", 300) MAX_RETRY: int = _int("MAX_RETRY", 5) # 上游错误重试次数(任意非负整数, 0=不重试) RETRY_STATUSES: str = _str("RETRY_STATUSES", "403,429,500,502,503,504") # 触发重试的上游状态码(逗号分隔); 400 类请求级错误不在其中 # --- 可靠性/配额治理 --- ADAPTIVE_SHELF: bool = _bool("ADAPTIVE_SHELF", True) # 货架目标随近 5min 实际 QPS 自动伸缩, # 避免固定大货架空转发白烧 Turnstile 签发 BROWSER_MAX_AGE: int = _int("BROWSER_MAX_AGE", 43200) # 浏览器上下文最大存活秒数, 到期由看门狗 # 在空闲窗口回收重启(防长跑内存膨胀) ISSUE_TIMEOUT: int = _int("ISSUE_TIMEOUT", 60) # 等 token 补货的最长秒数(超时 503) RATE_LIMIT_RPM: int = _int("RATE_LIMIT_RPM", 120) # 每个 relay key 的滑动窗口限速(次/分钟), 0=关闭 def as_dict(self) -> dict: """非敏感快照, 供 / 只读状态页回显。""" return { "upstream": self.UPSTREAM, "factory_page": self.FACTORY_PAGE, "factory_mode": self.FACTORY_MODE, "sitekey": self.SITEKEY, "hf_space": self.HF_SPACE, "headless": self.HEADLESS, "admin_enabled": self.ADMIN_ENABLED, "worker_pages": self.WORKER_PAGES, "max_worker_pages": self.MAX_WORKER_PAGES, "shelf_size": self.SHELF_SIZE, "token_max_age": self.TOKEN_MAX_AGE, "idle_pause": self.IDLE_PAUSE, "max_retry": self.MAX_RETRY, "retry_statuses": self.RETRY_STATUSES, "adaptive_shelf": self.ADAPTIVE_SHELF, "browser_max_age": self.BROWSER_MAX_AGE, "issue_timeout": self.ISSUE_TIMEOUT, "rate_limit_rpm": self.RATE_LIMIT_RPM, "proxy_configured": bool(self.PROXY), } cfg = Config()