This commit is contained in:
ddrwode
2026-02-27 17:36:31 +08:00
parent 19855500ab
commit db0fa81b2e
2 changed files with 58 additions and 21 deletions

View File

@@ -38,17 +38,38 @@ class TgeBrowserClient:
} }
def _post(self, path: str, json: Optional[dict] = None) -> dict: def _post(self, path: str, json: Optional[dict] = None) -> dict:
payload = json or {}
resp = requests.post( resp = requests.post(
f"{self.base_url}{path}", f"{self.base_url}{path}",
headers=self._headers, headers=self._headers,
json=json or {}, json=payload,
timeout=30, timeout=30,
) )
resp.raise_for_status() if not resp.ok:
detail = (resp.text or "").strip()
try:
err = resp.json()
if isinstance(err, dict):
message = err.get("message") or err.get("msg") or ""
code = err.get("code")
if message:
detail = f"{message}; code={code}; body={err}"
elif detail:
detail = f"{detail}; body={err}"
else:
detail = str(err)
except Exception:
pass
if not detail:
detail = "无响应体"
raise RuntimeError(
f"TgeBrowser API HTTP {resp.status_code} POST {path} 失败: {detail}; payload={payload}"
)
data = resp.json() data = resp.json()
if not data.get("success", False): if not data.get("success", False):
msg = data.get("message", "未知错误") msg = data.get("message", "未知错误")
raise RuntimeError(f"TgeBrowser API 失败: {msg}") raise RuntimeError(f"TgeBrowser API 失败: {msg}; body={data}; payload={payload}")
return data return data
def _get(self, path: str) -> dict: def _get(self, path: str) -> dict:
@@ -57,11 +78,13 @@ class TgeBrowserClient:
headers=self._headers, headers=self._headers,
timeout=30, timeout=30,
) )
resp.raise_for_status() if not resp.ok:
detail = (resp.text or "").strip() or "无响应体"
raise RuntimeError(f"TgeBrowser API HTTP {resp.status_code} GET {path} 失败: {detail}")
data = resp.json() data = resp.json()
if not data.get("success", False): if not data.get("success", False):
msg = data.get("message", "未知错误") msg = data.get("message", "未知错误")
raise RuntimeError(f"TgeBrowser API 失败: {msg}") raise RuntimeError(f"TgeBrowser API 失败: {msg}; body={data}")
return data return data
def status(self) -> dict: def status(self) -> dict:

View File

@@ -123,6 +123,13 @@ class FlowService:
def _choose_mobile_ua() -> str: def _choose_mobile_ua() -> str:
return random.choice(MOBILE_UA_POOL) return random.choice(MOBILE_UA_POOL)
@staticmethod
def _infer_fingerprint_os(mobile_ua: str) -> str:
ua = (mobile_ua or "").lower()
if "iphone" in ua or "ipad" in ua or "cpu iphone os" in ua:
return "iOS"
return "Android"
@staticmethod @staticmethod
def _connect_page_from_start_data(start_data: dict[str, Any]) -> ChromiumPage: def _connect_page_from_start_data(start_data: dict[str, Any]) -> ChromiumPage:
port = start_data.get("port") port = start_data.get("port")
@@ -143,30 +150,37 @@ class FlowService:
proxy_port: int, proxy_port: int,
mobile_ua: str, mobile_ua: str,
) -> dict[str, Any]: ) -> dict[str, Any]:
fingerprint_os = FlowService._infer_fingerprint_os(mobile_ua)
proxy_candidates = [ proxy_candidates = [
{"protocol": "http", "host": proxy_host, "port": proxy_port}, {"protocol": "http", "host": proxy_host, "port": proxy_port},
{"protocol": "http", "host": proxy_host, "port": str(proxy_port)}, {"protocol": "http", "host": proxy_host, "port": str(proxy_port)},
{"protocol": "http", "server": f"{proxy_host}:{proxy_port}"}, ]
fingerprint_candidates = [
{"os": fingerprint_os, "userAgent": mobile_ua},
{"os": fingerprint_os},
{"os": "Android"},
] ]
last_error: Optional[Exception] = None last_error: Optional[Exception] = None
errors: list[str] = []
for proxy_conf in proxy_candidates: for proxy_conf in proxy_candidates:
try: for fingerprint_conf in fingerprint_candidates:
return client.create_browser( try:
browser_name=browser_name, return client.create_browser(
start_page_url=start_page_url, browser_name=browser_name,
proxy=proxy_conf, start_page_url=start_page_url,
fingerprint={ proxy=proxy_conf,
"os": "Android", fingerprint=fingerprint_conf,
"platformVersion": 12, )
"userAgent": mobile_ua, except Exception as exc:
}, last_error = exc
) errors.append(
except Exception as exc: f"proxy={proxy_conf}, fingerprint={fingerprint_conf}, error={exc}"
last_error = exc )
continue continue
raise RuntimeError(f"创建浏览器失败(代理配置尝试均失败): {last_error}") detail = "; ".join(errors[-4:]) if errors else str(last_error)
raise RuntimeError(f"创建浏览器失败(代理/指纹配置尝试均失败): {detail}")
@staticmethod @staticmethod
def _apply_mobile_ua(page: ChromiumPage, ua: str) -> None: def _apply_mobile_ua(page: ChromiumPage, ua: str) -> None: