119 lines
3.9 KiB
Python
119 lines
3.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
比特浏览器 API 封装。
|
|
提供：列出窗口、打开窗口、关闭窗口，返回 CDP 地址供 DrissionPage 连接。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
import requests
|
|
|
|
logger = logging.getLogger("worker.bit_browser")
|
|
|
|
|
|
class BitBrowserAPI:
|
|
"""比特浏览器本地 API 客户端。"""
|
|
|
|
def __init__(self, base_url: str = "http://127.0.0.1:54345") -> None:
|
|
self.base_url = base_url.rstrip("/")
|
|
|
|
def _post(self, path: str, data: dict, timeout: int = 15) -> dict:
|
|
url = f"{self.base_url}{path}"
|
|
r = requests.post(url, json=data, timeout=timeout)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
# ─── 列出浏览器窗口 ───
|
|
|
|
def list_browsers(
|
|
self,
|
|
name: Optional[str] = None,
|
|
remark: Optional[str] = None,
|
|
page: int = 0,
|
|
page_size: int = 100,
|
|
) -> List[Dict[str, Any]]:
|
|
"""获取比特浏览器窗口列表。可按 name / remark 模糊筛选。"""
|
|
data: Dict[str, Any] = {"page": page, "pageSize": page_size}
|
|
if name is not None:
|
|
data["name"] = name
|
|
if remark is not None:
|
|
data["remark"] = remark
|
|
|
|
res = self._post("/browser/list", data)
|
|
if not res.get("success"):
|
|
raise RuntimeError(f"list 失败: {res.get('msg', res)}")
|
|
|
|
list_data = res.get("data")
|
|
if not list_data:
|
|
return []
|
|
if isinstance(list_data, list):
|
|
return list_data
|
|
if isinstance(list_data, dict) and "list" in list_data:
|
|
return list_data["list"]
|
|
return []
|
|
|
|
# ─── 打开浏览器窗口 ───
|
|
|
|
def open_browser(
|
|
self,
|
|
browser_id: Optional[str] = None,
|
|
name: Optional[str] = None,
|
|
remark: Optional[str] = None,
|
|
) -> Tuple[str, int, str]:
|
|
"""
|
|
打开指定浏览器窗口。
|
|
返回 (cdp_addr, port, browser_id)。
|
|
"""
|
|
if browser_id is None:
|
|
items = self.list_browsers(name=name, remark=remark, page_size=10)
|
|
if not items:
|
|
raise RuntimeError("没有匹配的浏览器窗口")
|
|
browser_id = items[0].get("id")
|
|
if not browser_id:
|
|
raise RuntimeError("列表项中没有 id")
|
|
|
|
res = self._post("/browser/open", {"id": browser_id})
|
|
if not res.get("success"):
|
|
raise RuntimeError(f"open 失败: {res.get('msg', res)}")
|
|
|
|
data = res.get("data") or {}
|
|
http_addr = data.get("http")
|
|
if not http_addr:
|
|
raise RuntimeError(f"返回中无 http 地址: {data}")
|
|
|
|
if ":" in str(http_addr):
|
|
host, port_str = str(http_addr).strip().rsplit(":", 1)
|
|
port = int(port_str)
|
|
cdp_addr = f"{host}:{port}"
|
|
else:
|
|
port = int(http_addr)
|
|
cdp_addr = f"127.0.0.1:{port}"
|
|
|
|
logger.info("已打开浏览器 %s, CDP: %s", browser_id, cdp_addr)
|
|
return cdp_addr, port, browser_id
|
|
|
|
# ─── 关闭浏览器窗口 ───
|
|
|
|
def close_browser(self, browser_id: str) -> bool:
|
|
"""关闭指定浏览器窗口。"""
|
|
try:
|
|
res = self._post("/browser/close", {"id": browser_id})
|
|
return res.get("success", False)
|
|
except Exception as e:
|
|
logger.warning("关闭浏览器 %s 失败: %s", browser_id, e)
|
|
return False
|
|
|
|
# ─── 便捷方法：获取供 DrissionPage 连接的信息 ───
|
|
|
|
def get_browser_for_drission(
|
|
self,
|
|
browser_id: Optional[str] = None,
|
|
name: Optional[str] = None,
|
|
remark: Optional[str] = None,
|
|
) -> Tuple[str, int]:
|
|
"""按需打开浏览器，返回 (addr, port) 供 DrissionPage 使用。"""
|
|
cdp_addr, port, _ = self.open_browser(browser_id=browser_id, name=name, remark=remark)
|
|
return cdp_addr, port
|