BOM发料对比

This commit is contained in:
hjq
2026-06-23 11:17:17 +08:00
parent 9be2b1373f
commit b0dfc6214f

View File

@@ -6,6 +6,7 @@ import os
import sys
import time
import shutil
import json
import datetime
import subprocess
import urllib.request
@@ -98,6 +99,45 @@ def cleanup_debug_port(address: str) -> None:
)
def probe_devtools_endpoints(address: str, log_output: bool = True) -> dict:
"""探测 DevTools HTTP 端点,并返回解析后的版本信息与标签页信息。"""
result = {"version": None, "list": None}
if not address:
return result
endpoint_map = {"version": "json/version", "list": "json/list"}
for key, endpoint in endpoint_map.items():
url = f"http://{address}/{endpoint}"
try:
with urllib.request.urlopen(url, timeout=2) as resp:
body = resp.read().decode("utf-8", errors="replace")
if log_output:
log("WARN", f"[DEBUG] DevTools 探测 {url} -> HTTP {resp.status}")
log("WARN", f"[DEBUG] DevTools 响应体 {endpoint}: {body[:1000]}")
result[key] = json.loads(body)
except Exception as probe_err:
if log_output:
log("WARN", f"[DEBUG] DevTools 探测失败 {url}: {probe_err}")
return result
def get_first_page_ws_address(devtools_payload: dict) -> str:
"""从 /json/list 响应中提取第一个可用 page/webview 的 ws 地址。"""
tabs = devtools_payload.get("list") or []
if not isinstance(tabs, list):
return ""
for tab in tabs:
if (
isinstance(tab, dict)
and tab.get("type") in ("page", "webview")
and not str(tab.get("url", "")).startswith("devtools://")
and tab.get("webSocketDebuggerUrl")
):
return tab["webSocketDebuggerUrl"]
return ""
# ── 日志 ──────────────────────────────────────────────────────────────────────
def log(level: str, msg: str):
icons = {"INFO": " ", "OK": "", "WARN": "⚠️ ", "ERR": ""}
@@ -179,7 +219,24 @@ def get_page(headless: bool = False, port: int = 9222) -> ChromiumPage:
return page
except Exception as e:
actual_address = opt.address or f"127.0.0.1:{port}"
log("WARN", f"[DEBUG] ChromiumPage 初始化失败: {e},尝试清理地址 {actual_address} 后重试...")
log("WARN", f"[DEBUG] ChromiumPage 初始化失败: {e}")
devtools_payload = probe_devtools_endpoints(actual_address, log_output=True) if opt.address else {"version": None, "list": None}
fallback_page_ws = get_first_page_ws_address(devtools_payload)
if is_linux_env() and fallback_page_ws and "Handshake status 404 Not Found" in str(e):
log("WARN", f"[DEBUG] Browser WS 握手失败,尝试降级连接 Page WS: {fallback_page_ws}")
try:
fallback_co = ChromiumOptions()
fallback_co.set_address(fallback_page_ws)
page = ChromiumPage(fallback_co)
log("OK", "[DEBUG] 已通过 Page WS 降级连接成功。")
return page
except Exception as ws_only_e:
log("ERR", f"[DEBUG] Page WS 降级连接失败: {ws_only_e}")
e = ws_only_e
log("WARN", f"[DEBUG] 尝试清理地址 {actual_address} 后重试...")
try:
cleanup_debug_port(actual_address)
time.sleep(1)
@@ -191,17 +248,7 @@ def get_page(headless: bool = False, port: int = 9222) -> ChromiumPage:
e = retry_e
# #region debug-point B:devtools-http-probe
if opt.address:
for endpoint in ("json/version", "json/list"):
url = f"http://{opt.address}/{endpoint}"
try:
with urllib.request.urlopen(url, timeout=2) as resp:
body = resp.read().decode("utf-8", errors="replace")
log("WARN", f"[DEBUG] DevTools 探测 {url} -> HTTP {resp.status}")
log("WARN", f"[DEBUG] DevTools 响应体 {endpoint}: {body[:1000]}")
except Exception as probe_err:
log("WARN", f"[DEBUG] DevTools 探测失败 {url}: {probe_err}")
else:
if not opt.address:
log("WARN", "[DEBUG] DevTools 探测跳过address 为空")
# #endregion
log("ERR", f"浏览器初始化失败: {e}")