pp01

Running

pp01 / config.py

feat: factory_mode 三模式(intercept/mock/real) — 空壳工厂页, 签发与目标站页面解耦

8088875 about 2 months ago

8.12 kB

#!/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()

	#!/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()