assistant/spike/embed/probe.py

75 lines
3.0 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""技术验证:browser-use 通过 CDP 连接 Electron 自带 Chromium 里的 WebContentsView 并跑一个小任务。
先启动:npx electron spike/embed/main.js
再运行:.venv/Scripts/python.exe spike/embed/probe.py
"""
import asyncio
import json
import os
import sys
import urllib.request
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))
from dotenv import load_dotenv # noqa: E402
load_dotenv(ROOT / ".env")
CDP = f"http://127.0.0.1:{os.environ.get('SPIKE_CDP_PORT', '9333')}"
def targets() -> list[dict]:
return json.loads(urllib.request.urlopen(CDP + "/json", timeout=5).read().decode())
async def main() -> int:
from browser_use import Agent, Browser, Tools
from backend.browser.llm_factory import create_llm
before = targets()
print("targets before:", [(t["type"], t["url"][:60], t.get("title", "")[:20], t["id"][-6:]) for t in before])
view = next((t for t in before if t["type"] == "page" and t["url"] == "about:blank"), None)
if not view:
print("no about:blank page target found")
return 1
# 内嵌模式不传 allowed_domains:browser-use 的安全看门狗会处理不在白名单里的标签页(含我们的界面页),
# 域名限制改由 Electron 在导航层面执行
# 白名单里带上界面页的主机名 ui(app://ui/...),学校域名照常;Electron 侧另做导航拦截
browser = Browser(cdp_url=CDP, keep_alive=True, allowed_domains=["ui", "school.miaopinai.com", "*.miaopinai.com"])
print("connecting...", flush=True)
await browser.start()
print("connected", flush=True)
print("focus after start:", (browser.agent_focus_target_id or "")[-6:], "| want:", view["id"][-6:])
await browser.get_or_create_cdp_session(view["id"], focus=True)
print("focus after steer :", (browser.agent_focus_target_id or "")[-6:])
tools = Tools(exclude_actions=["switch", "close", "search", "upload_file", "write_file", "replace_file", "read_file"])
agent = Agent(
task="打开 https://school.miaopinai.com:29800/xuegong/ ,读出页面标题,以及登录页「测试环境快捷登录」区域有哪些按钮,用 done 汇报。只读,不要点击任何按钮。",
llm=create_llm(),
browser=browser,
tools=tools,
use_vision=False,
max_actions_per_step=2,
step_timeout=90,
use_judge=False,
)
history = await agent.run(max_steps=6)
print("steps:", history.number_of_steps(), "| done:", history.is_done(), "| ok:", history.is_successful())
print("urls:", history.urls())
print("final:", str(history.final_result())[:400])
shots = [s for s in (history.screenshots() or []) if s]
print("screenshots captured:", len(shots))
after = targets()
print("targets after:", [(t["type"], t["url"][:60], t["id"][-6:]) for t in after])
print("page targets before/after:", sum(t["type"] == "page" for t in before), "/", sum(t["type"] == "page" for t in after))
return 0
if __name__ == "__main__":
raise SystemExit(asyncio.run(main()))