From b0dfc6214fc49b454243fbcc19caf8bcf2dae3d6 Mon Sep 17 00:00:00 2001 From: hjq <770690987@qq.com> Date: Tue, 23 Jun 2026 11:17:17 +0800 Subject: [PATCH] =?UTF-8?q?BOM=E5=8F=91=E6=96=99=E5=AF=B9=E6=AF=94?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- browser_login/login.py | 71 +++++++++++++++++++++++++++++++++++------- 1 file changed, 59 insertions(+), 12 deletions(-) diff --git a/browser_login/login.py b/browser_login/login.py index cb58cdc..7ef154f 100644 --- a/browser_login/login.py +++ b/browser_login/login.py @@ -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}")