mirror of
https://github.com/val1813/kwcode.git
synced 2026-09-03 06:34:30 +08:00
c6197d0316
P1: KWCODE.md规则注入、/plan风险评估、Checkpoint快照、DocReader P2: 模型能力自适应、飞轮通知、价值量化仪表盘 搜索: 四级提取管道、并行搜索+BM25重排、意图感知、ChatExpert门控 UI: spinner动画、结果摘要、静默日志、重影大字Header 新增: kwcode setup-search 一键安装SearXNG 测试: 282/282 PASS (含17个E2E真实模型测试) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
125 lines
3.6 KiB
Python
125 lines
3.6 KiB
Python
"""
|
||
网络环境探测 + 代理配置。
|
||
所有 httpx 调用统一从这里获取 client 配置。
|
||
|
||
探测逻辑：
|
||
1. 读环境变量 KAIWU_PROXY > HTTPS_PROXY > HTTP_PROXY
|
||
2. 读 ~/.kaiwu/config.yaml 的 proxy 字段
|
||
3. 探测 DDG 是否可达（3s timeout）→ 判断是否国内网络
|
||
4. 缓存结果，整个 session 只探测一次
|
||
"""
|
||
|
||
import logging
|
||
import os
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
|
||
import httpx
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Session-level cache
|
||
_network_cache: Optional[dict] = None
|
||
|
||
|
||
def get_proxy() -> Optional[str]:
|
||
"""
|
||
从环境变量或 config 读取代理地址。
|
||
优先级：KAIWU_PROXY > HTTPS_PROXY > HTTP_PROXY > ~/.kaiwu/config.yaml
|
||
"""
|
||
for var in ("KWCODE_PROXY", "KAIWU_PROXY", "HTTPS_PROXY", "HTTP_PROXY",
|
||
"kwcode_proxy", "kaiwu_proxy", "https_proxy", "http_proxy"):
|
||
val = os.environ.get(var)
|
||
if val:
|
||
return val
|
||
|
||
# Try ~/.kwcode/config.yaml first, then legacy ~/.kaiwu/config.yaml
|
||
for dirname in (".kwcode", ".kaiwu"):
|
||
config_path = os.path.join(Path.home(), dirname, "config.yaml")
|
||
if os.path.exists(config_path):
|
||
try:
|
||
import yaml
|
||
with open(config_path, "r", encoding="utf-8") as f:
|
||
cfg = yaml.safe_load(f) or {}
|
||
proxy = cfg.get("proxy")
|
||
if not proxy:
|
||
# Also check nested default.proxy
|
||
default = cfg.get("default", {})
|
||
if isinstance(default, dict):
|
||
proxy = default.get("proxy")
|
||
if proxy:
|
||
return proxy
|
||
except Exception:
|
||
pass
|
||
|
||
return None
|
||
|
||
|
||
def _probe_url(url: str, timeout: float = 3.0) -> bool:
|
||
"""测试 URL 是否可达（HEAD 请求，不下载内容）。"""
|
||
try:
|
||
proxy = get_proxy()
|
||
kwargs = {"timeout": timeout, "follow_redirects": True}
|
||
if proxy:
|
||
kwargs["proxy"] = proxy
|
||
resp = httpx.head(url, **kwargs)
|
||
return resp.status_code < 500
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def detect_network(force: bool = False) -> dict:
|
||
"""
|
||
探测网络环境，返回：
|
||
{
|
||
"china": bool, # 是否国内网络（DDG不通）
|
||
"proxy": str|None, # 代理地址
|
||
"ddg_ok": bool, # DuckDuckGo 可达
|
||
"hf_ok": bool, # HuggingFace 可达
|
||
}
|
||
结果缓存，整个 session 只探测一次（除非 force=True）。
|
||
"""
|
||
global _network_cache
|
||
if _network_cache is not None and not force:
|
||
return _network_cache
|
||
|
||
proxy = get_proxy()
|
||
ddg_ok = _probe_url("https://html.duckduckgo.com/html/")
|
||
hf_ok = _probe_url("https://huggingface.co")
|
||
china = not ddg_ok
|
||
|
||
_network_cache = {
|
||
"china": china,
|
||
"proxy": proxy,
|
||
"ddg_ok": ddg_ok,
|
||
"hf_ok": hf_ok,
|
||
}
|
||
|
||
if china:
|
||
logger.info("[network] 检测到国内网络 (DDG=%s, HF=%s, proxy=%s)",
|
||
ddg_ok, hf_ok, "yes" if proxy else "no")
|
||
else:
|
||
logger.debug("[network] 海外网络 (DDG=ok, HF=%s)", hf_ok)
|
||
|
||
return _network_cache
|
||
|
||
|
||
def get_httpx_kwargs(timeout: float = 10.0) -> dict:
|
||
"""
|
||
返回 httpx.Client / httpx.get / httpx.post 的通用 kwargs。
|
||
包含 proxy、timeout、follow_redirects。
|
||
"""
|
||
kwargs: dict = {
|
||
"timeout": timeout,
|
||
"follow_redirects": True,
|
||
}
|
||
proxy = get_proxy()
|
||
if proxy:
|
||
kwargs["proxy"] = proxy
|
||
return kwargs
|
||
|
||
|
||
def is_china_network() -> bool:
|
||
"""快速判断是否国内网络（使用缓存）。"""
|
||
return detect_network().get("china", False)
|