diff --git a/tgebrowser_client.py b/tgebrowser_client.py index 746f844..bb6a3d1 100644 --- a/tgebrowser_client.py +++ b/tgebrowser_client.py @@ -38,17 +38,38 @@ class TgeBrowserClient: } def _post(self, path: str, json: Optional[dict] = None) -> dict: + payload = json or {} resp = requests.post( f"{self.base_url}{path}", headers=self._headers, - json=json or {}, + json=payload, 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() if not data.get("success", False): msg = data.get("message", "未知错误") - raise RuntimeError(f"TgeBrowser API 失败: {msg}") + raise RuntimeError(f"TgeBrowser API 失败: {msg}; body={data}; payload={payload}") return data def _get(self, path: str) -> dict: @@ -57,11 +78,13 @@ class TgeBrowserClient: headers=self._headers, 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() if not data.get("success", False): msg = data.get("message", "未知错误") - raise RuntimeError(f"TgeBrowser API 失败: {msg}") + raise RuntimeError(f"TgeBrowser API 失败: {msg}; body={data}") return data def status(self) -> dict: diff --git a/tyyp_app/services.py b/tyyp_app/services.py index 2661c6e..57b515d 100644 --- a/tyyp_app/services.py +++ b/tyyp_app/services.py @@ -123,6 +123,13 @@ class FlowService: def _choose_mobile_ua() -> str: 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 def _connect_page_from_start_data(start_data: dict[str, Any]) -> ChromiumPage: port = start_data.get("port") @@ -143,30 +150,37 @@ class FlowService: proxy_port: int, mobile_ua: str, ) -> dict[str, Any]: + fingerprint_os = FlowService._infer_fingerprint_os(mobile_ua) proxy_candidates = [ {"protocol": "http", "host": proxy_host, "port": 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 + errors: list[str] = [] for proxy_conf in proxy_candidates: - try: - return client.create_browser( - browser_name=browser_name, - start_page_url=start_page_url, - proxy=proxy_conf, - fingerprint={ - "os": "Android", - "platformVersion": 12, - "userAgent": mobile_ua, - }, - ) - except Exception as exc: - last_error = exc - continue + for fingerprint_conf in fingerprint_candidates: + try: + return client.create_browser( + browser_name=browser_name, + start_page_url=start_page_url, + proxy=proxy_conf, + fingerprint=fingerprint_conf, + ) + except Exception as exc: + last_error = exc + errors.append( + f"proxy={proxy_conf}, fingerprint={fingerprint_conf}, error={exc}" + ) + continue - raise RuntimeError(f"创建浏览器失败(代理配置尝试均失败): {last_error}") + detail = "; ".join(errors[-4:]) if errors else str(last_error) + raise RuntimeError(f"创建浏览器失败(代理/指纹配置尝试均失败): {detail}") @staticmethod def _apply_mobile_ua(page: ChromiumPage, ua: str) -> None: