博客 / 詳情

返回

爬蟲_20251211_Browser-Use_MCP_Selenium_爬蟲+LLM

爬蟲_20251211

Browser-Use

Browser-Use 下載安裝

Github 倉庫鏈接: https://github.com/browser-use/browser-use

檢查 Windows 中是否已經安裝 uv:

uv --version

升級 uv 版本:

uv self update

安裝方法:

  1. 用 pip 安裝 uv:
pip install uv
  1. 用官方腳本安裝:
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

通過檢查 uv 的安裝路徑來判斷自己的安裝方式。在 PowerShell 裏運行:

Get-Command uv

如果是用 pip 安裝的 uv（Python 包），路徑通常是:

C:\Users\yuanz\AppData\Local\Programs\Python\Python311\Scripts\uv.exe
C:\Users\yuanz\AppData\Roaming\Python\Python311\Scripts\uv.exe

如果是用官方腳本安裝的 uv（獨立可執行文件），路徑通常是：

C:\Users\yuanz\.local\bin\uv.exe
C:\Users\yuanz\AppData\Local\uv\bin\uv.exe

檢查是否是 pip 管理的版本:

pip list | findstr uv

解除 PowerShell .ps1 執行禁令:

# 在 PowerShell 以管理員身份運行的情況下執行，以解禁 PowerShell 的默認禁止執行 .ps1 腳本禁令
# 這條命令的意思是允許當前用户運行本地的 PowerShell 腳本（如 .venv\Scripts\activate.ps1），但仍阻止來自互聯網且未簽名的腳本。
Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser

創建並激活虛擬環境:

cd <項目路徑>
uv venv
.venv\Scripts\activate

命令行前多出 (.venv) 説明虛擬環境激活成功。

安裝 Browser-Use:

uv init
uv add browser-use
uv sync
uvx browser-use install

安裝 dotenv:

uv pip install python-dotenv

創建 .env 文件注意事項:

  • 不要加引號，不要加空格。

  • .env 文件名一定要以點開頭。

  • 確保它和你的 Python 腳本在同一文件夾下。

Browser-Use 使用示例:

# run_agent.py
# -*- coding: utf-8 -*-
"""
一個開箱即用的 browser-use 示例：
- 讀取 .env 裏的 GEMINI_API_KEY
- 默認用【本機瀏覽器】；需要可一鍵切換到雲端瀏覽器
- 支持為網頁流量設置“本地代理”，同時保證本地 CDP (127.0.0.1) 不走代理
- 兼容 browser-use 0.8.x（Browser() 只支持 use_cloud/headless/proxy）
"""

import os
import sys
import traceback
from dotenv import load_dotenv

# 1) 載入 .env（放在項目根目錄，內容：GEMINI_API_KEY=你的key）
load_dotenv()

# 2) —— 關鍵：保證本地調試端口不走代理（否則會 JSONDecodeError）
os.environ["NO_PROXY"]  = "localhost,127.0.0.1"
os.environ["no_proxy"]  = "localhost,127.0.0.1"
# 如果之前在系統/終端裏設置過以下代理變量，這裏強制清理（僅當前進程）
for k in ("HTTP_PROXY", "HTTPS_PROXY", "http_proxy", "https_proxy"):
    os.environ.pop(k, None)

# 3) 根據你的代理情況配置網頁訪問代理（如果需要）
#    例如你的本地 VPN 端口是 25378，則設為：
PROXY_SERVER = os.getenv("PROXY_SERVER", "http://127.0.0.1:25378")
USE_PROXY = os.getenv("USE_PROXY", "false").lower() in ("1", "true", "yes")

# 4) 是否使用雲端瀏覽器（本機不穩定時可切換為 True；需先執行 `browser-use auth` 完成登錄）
USE_CLOUD = os.getenv("USE_CLOUD", "false").lower() in ("1", "true", "yes")

# 5) 其餘參數
HEADLESS = os.getenv("HEADLESS", "false").lower() in ("1", "true", "yes")
MODEL_ID = os.getenv("GEMINI_MODEL", "gemini-2.5-flash")

# 6) 導入 browser-use（放到 NO_PROXY 設置之後）
from browser_use import Agent, ChatGoogle, Browser


def make_browser():
    """
    兼容 browser-use 0.7.x：
    Browser() 支持的參數有限：use_cloud、headless、proxy、profile_name
    其中 proxy 傳遞為 playwright 兼容的 dict：
      {"server": "http://host:port", "username": "...", "password": "..."}
    """
    kwargs = {
        "use_cloud": USE_CLOUD,
        "headless": HEADLESS,
    }
    if (not USE_CLOUD) and USE_PROXY:
        kwargs["proxy"] = {"server": PROXY_SERVER}
        # 如需賬號密碼，改成：
        # kwargs["proxy"] = {"server": PROXY_SERVER, "username": "your_user", "password": "your_pass"}

    return Browser(**kwargs)


# 不再需要文件解析和 CSV 生成功能，數據直接輸出到 terminal
def main():
    # 小檢查：API Key
    key = os.getenv("GEMINI_API_KEY")
    if not key:
        print("❌ 未檢測到 GEMINI_API_KEY，請在項目根目錄創建 .env 並寫入：")
        print("GEMINI_API_KEY=你的key")
        sys.exit(1)

    print("=== Config ===")
    print(f"USE_CLOUD : {USE_CLOUD}")
    print(f"HEADLESS  : {HEADLESS}")
    print(f"USE_PROXY : {USE_PROXY}  ({PROXY_SERVER if USE_PROXY else 'no proxy'})")
    print(f"MODEL_ID  : {MODEL_ID}")
    print("NO_PROXY  :", os.environ.get("NO_PROXY"))

    browser = make_browser()

    agent = Agent(
        task="""
            【嚴格動作協議】
            - 你是瀏覽器自動化 Agent。你每一步“必須”返回一個且僅一個動作（action），動作必須是下列之一：
            navigate / click / input / send_keys / wait / scroll / find_text / extract / evaluate / read_file / replace_file / done
            - 除最終 `done` 外，不要輸出任何自然語言或總結；若需要記錄進度，使用 `replace_file`。
            - 如不確定，也必須給出一個動作（例如 wait）。絕不能只輸出思考（thinking）。

            【任務】
            1) 打開 https://www.cn-healthcare.com/ 。
            2) 點擊頂欄“搜索/放大鏡”圖標 <div class="ni_head_search_wrap">，先等待url變為 "https://www.cn-healthcare.com/search/"，再等待輸入框出現；
            3) 然後在頂部的 <input class="search-input"> 的輸入框中輸入：公立醫院，並按 Enter，等待 10 秒；若無明顯結果，再按一次 Enter 或再點搜索按鈕。
            4) 滾動到底部，等待 5 秒，只執行這一次操作。

            4) 從當前頁抽取文章卡片（僅 cn-healthcare.com 域）：
            - 每個卡片元素：<div class="search-item"> 
            - 標題：<div class="search-item"> 內的 <h5 class="tit"></h5> 下的 <a> 文本 
            - 網址：<h5 class="tit"></h5> 內的 <a> 標籤內的 href 參數內容 
            - 作者：<div class="footer"> 下的 <a> 標籤內的 span.author 的所有文本內容 
            - 發佈日期：<div class="footer"> 下的 <span class="date"></span> 內的所有文本內容（格式 yyyy/mm/dd）
            5) 過濾：
            - 僅保留 URL 含 /article /content /articlewm，且不是圖片/視頻/PDF 等後綴
            - 日期在 2023-05-01 至 2025-10-31（含端點）；相對時間如“剛剛/幾天前”跳過
            - URL 去重

            【抽取與落地（必須）】
            - 完成搜索與滾動後，執行一次 extract：
            - 每個卡片元素：<div class="search-item"> 
            - 標題：<div class="search-item"> 內的 <h5 class="tit"></h5> 下的 <a> 文本 
            - 網址：<h5 class="tit"></h5> 內的 <a> 標籤內的 href 參數內容 
            - 作者：<div class="footer"> 下的 <a> 標籤內的 span.author 的所有文本內容 
            - 發佈日期：<div class="footer"> 下的 <span class="date"></span> 內的所有文本內容（格式 yyyy/mm/dd）
            - 執行完 evaluate 後，Agent 立即結束（調用 done）。
            - 數據會直接顯示在 terminal 中，無需寫入文件。

            【禁止事項】
            - 除最終 `done` 外，任何步驟禁止輸出自然語言文本、代碼塊或截圖。

        """,
        llm=ChatGoogle(model=MODEL_ID),
        browser=browser,
    )


    # 運行 Agent，數據會直接輸出到 terminal
    try:
        print("🚀 開始爬取數據...")
        agent.run_sync()   # 關鍵：同步執行
        print("✅ 數據提取完成！")

    except Exception:
        print("❌ Agent 運行失敗，堆棧如下：")
        traceback.print_exc()
        print("\n常見修復：")
        print("1) 若開啓了系統全局代理，請關閉或確保排除 127.0.0.1；")
        print("2) 若本機仍不穩，設置環境變量 USE_CLOUD=true 後再跑（先 `browser-use auth`）；")
        print("3) 升級到新版：uv pip install -U 'browser-use[cli]'；")
        print("4) 以有頭模式調試：設 HEADLESS=false。")


if __name__ == "__main__":
    main()

命令行指令

查看 Windows 系統層面的 WinHTTP 代理設置:

PS C:\Users\yuanz\Desktop\scraper-wjw> netsh winhttp show proxy

當前的 WinHTTP 代理服務器設置:

    直接訪問(沒有代理服務器)。

查看網絡端口占用:

PS C:\Users\yuanz\Desktop\scraper-wjw> netstat -ano | findstr 25378
  TCP    127.0.0.1:25378        0.0.0.0:0              LISTENING       82344

  • netstat -ano：列出所有端口

  • a = 所有連接和監聽端口

  • n = 以數字格式顯示（不反查域名）

  • o = 顯示 PID（哪個程序佔用）

  • | findstr 25378：過濾包含 “25378” 的行

  • TCP: 協議類型

  • 127.0.0.1:23578: 本機監聽端口25378

  • 0.0.0.0:0: 無特定遠端連接（表示監聽狀態）

  • LISTENING: 端口正在被程序監聽

  • 82344: 佔用這個端口的進程 PID

tasklist 用於列出當前所有運行中的程序（任務管理器的命令行版本）:

PS C:\Users\yuanz\Desktop\scraper-wjw> tasklist | findstr 82344
SSTap.exe                    82344 Console                    2     25,892 K
  • PID = 82344 的進程是 SSTap.exe
  • 端口 25378 是 SSTap 提供的本地代理端口（HTTP/SOCKS 本地監聽）

大模型 API

AIIAI: https://api.aiiai.top/

GalaAPI: https://www.galaapi.com/

MCP (Model Context Protocol)

MCP 是 OpenAI 在 2024 年提出的一個 標準化協議，主要用來讓 AI 模型（如 ChatGPT、Gemini 等）和外部工具 / 數據源 / 插件 進行交互。

  • Cursor / Claude Desktop = MCP Client

  • 自己編寫的 Python FastMCP 程序 = MCP Server

用 Python 寫 MCP Server 時，想觸發 Cursor 調用 (Call) 至少需要完成兩件事：

  1. 寫一個 MCP Server (Python / FastMCP)
# server_fastmcp.py

from mcp.server.fastmcp import FastMCP
import httpx

app = FastMCP("BAS Security Platform")

@app.tool()
async def create_attack_task(...):
    ...
    return "任務創建成功"

啓動 server:

uv run server_fastmcp.py
  1. 讓 Cursor 識別這個 MCP Server

Cursor 支持 MCP，需要在 Cursor 裏創建一個 MCP 配置文件 .cursor/mcp.json:

{
  "servers": [
    {
      "name": "bas-mcp",
      "command": ["python", "server_fastmcp.py"],
      "type": "stdio"
    }
  ]
}

MCP Client 的職責，就是負責調用（call/invoke） MCP Server 暴露出來的工具（tools）、資源（resources）、提示模板（prompts）等能力。

換句話説：

  • MCP Server = 工具提供者（提供能力）

  • MCP Client = 調用者（消費能力）

LLM（ChatGPT、Claude、Cursor 內置模型）則是運行在 MCP Client 裏的一部分，它會根據對話推理出“需要調用哪個 tool”，然後由 MCP Client 發起調用。

┌───────────────────────────────┐
│        MCP Client             │
│  (Cursor / Claude / VSCode)   │
│                               │
│  - 解析用户輸入                │
│  - 判斷是否要調用 tool        │
│  - 發送 MCP Request →         │
└──────────────┬────────────────┘
               │ (MCP Protocol)
               ▼
┌───────────────────────────────┐
│        MCP Server             │
│  (你寫的 server.py / Node.js) │
│                               │
│  - @mcp.tool() 暴露工具        │
│  - 執行業務邏輯               │
│  - 返回結果給 Client          │
└───────────────────────────────┘

一個完整的 BAS MCP 示例:

假設：

  • MCP Server 文件名為 bas_server.py

  • bas_server.py 中有一個 tool:

    @app.tool()
async def start_attack(type: str, target: str) -> str:
    return bas.start(type, target)

  • 還有更多 tool，比如 get_status(task_id)

那麼 Cursor 的 MCP 配置文件 (mcp.json) 應該是:

{
  "servers": [
    {
      "name": "bas-mcp",    // MCP Server 在 Cursor 左側面板中的名字，自定義即可
      "type": "command",     // 告訴 Cursor：這個 MCP server 是通過子進程（stdio）啓動的
      "command": "python",     // Cursor 會用這個命令啓動 server
      "args": ["bas_server.py"],    // 代表 MCP Server 執行：python bas_server.py
      "env": {
        "BAS_API_URL": "http://localhost:8080",
        "BAS_TOKEN": "your-secret-token"
      }
    }
  ]
}

當我在 Cursor 對話框中輸入：“請幫我對 192.168.0.5 發起 SQL 注入攻擊”，工具觸發流程是:

  1. Cursor（MCP Client）內置的 LLM 自動生成 MCP 調用：
call tool start_attack({"type":"sql_injection","target":"192.168.0.5"})
  1. Cursor 會發送 JSON-RPC:
{
  "method": "tools.call",
  "params": {
    "name": "start_attack",
    "arguments": {
      "type": "sql_injection",
      "target": "192.168.0.5"
    }
  }
}

MCP Server 執行工具函數:

@app.tool()
async def start_attack(type: str, target: str) -> str:
    return bas.start(type, target)

Selenium 爬蟲

固定開頭:

def build_driver(headless: bool = True) -> webdriver.Chrome:
    chrome_opts = Options()
    if headless:
        chrome_opts.add_argument("--headless=new")
    chrome_opts.add_argument("--disable-gpu")
    chrome_opts.add_argument("--no-sandbox")
    chrome_opts.add_argument("--window-size=1400,900")
    chrome_opts.add_argument(
        '--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
        'AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36'
    )
    chrome_opts.add_argument("--disable-blink-features=AutomationControlled")

    service = Service(ChromeDriverManager().install())
    driver = webdriver.Chrome(service=service, options=chrome_opts)
    driver.set_page_load_timeout(180)
    driver.implicitly_wait(3)
    return driver


def human_like_scroll(driver: webdriver.Chrome):
    """Simulate human scrolling behavior, to prevent being too "robotic"."""
    try:
        # Scroll a few times, each time to a different height, with random pauses in between
        scroll_steps = random.randint(2, 5)
        for _ in range(scroll_steps):
            # 0.3 ~ 1.0 times the page height randomly
            factor = random.uniform(0.3, 1.0)
            driver.execute_script(
                "window.scrollTo(0, document.body.scrollHeight * arguments[0]);",
                factor,
            )
            time.sleep(random.uniform(0.5, 1.5))
    except Exception as e:
        print("[WARN] human_like_scroll error: ", e)

搜索框鍵入搜索:

def test_search(url: str, keyword: str):
    driver = build_driver(headless=False)   # 方便你看到效果
    driver.get(url)
    time.sleep(1)

    # 找到搜索框
    search_input = driver.find_element(By.CSS_SELECTOR, ".search-container .sh-inpt input")
    search_input.clear()
    search_input.send_keys(keyword)
    time.sleep(0.3)

    # 點擊“搜索”按鈕（Selenium 4 推薦方式）
    search_button = driver.find_element(By.CSS_SELECTOR, ".search-container .sh-btn")
    search_button.click()

    # 等待頁面加載
    time.sleep(3)

    # 模擬滾動
    human_like_scroll(driver)

    print("頁面標題:", driver.title)
    print("當前URL:", driver.current_url)

    # 你可在這裏加“爬取結果”的代碼
    # html = driver.page_source
    # print(html[:300])

    time.sleep(2)
    driver.quit()


if __name__ == "__main__":
    test_search(
        url="http://search.people.cn/",     # ← 要測試的網站
        keyword="數字化轉型"            # ← 測試關鍵詞
    )

爬蟲 + LLM

在本地調用大模型 API，不能對網址鏈接進行訪問。因為大模型本身是語言模型，只能處理文本輸入輸出，無法直接發起HTTP請求，無法執行瀏覽器操作。

所以一般的方法是，先用爬蟲工具，如 BeautifulSoup4、requests、Selenium 等將文本類型的數據爬取下來，然後將文本數據導入大模型進行語義分析和目標關鍵數據的提取。

配置環境變量:

from dotenv import load_dotenv
load_dotenv()

讀取環境變量:

base_url = os.getenv("AI_BASE_URL", "")     # e.g. https://api.aiiai.top/v1
api_key = os.getenv("AI_API_KEY", "")
model = os.getenv("AI_MODEL_TYPE", "")      # e.g. gemini-2.5-pro

用 OpenAI 兼容 SDK 調用:

client = OpenAI(
    base_url=base_url,
    api_key=api_key,
)

completion = client.chat.completions.create(
    model=model,
    messages=messages,
)
content = completion.choices[0].message.content

  • 使用 openai 的 OpenAI 客户端，但是把 base_url 改成自己的網關 (https://api.aiiai.top/v1)，從而兼容各種自託管 / 代理服務。

  • 調用方式是標準的 Chat Completions 接口 (聊天接口)：傳入 model 和 messages。具體詳見 OpenAI API 文檔: https://platform.openai.com/docs/api-reference/chat/create。

OpenAI 文檔明確説明請求 JSON 必須包含：

{
  "model": "gpt-5.2",
  "messages": [
    {"role": "system", "content": "..." },
    {"role": "user", "content": "..." }
  ]
}

實例:

def call_local_llm(messages: List[Dict[str, str]]) -> Any:
    """
    Call the local / OpenAI compatible model:
    - Read AI_BASE_URL / AI_API_KEY / AI_MODEL_TYPE from environment variables
    - Default base_url = https://api.aiiai.top/v1
    - Default model = gemini-2.5-pro
    - Return the Python object (dict / list / None) after JSON parsing
    """
    init_logger()

    base_url = os.getenv("AI_BASE_URL", "")
    api_key = os.getenv("AI_API_KEY", "")
    model = os.getenv("AI_MODEL_TYPE", "")

    if not base_url:
        base_url = "https://api.aiiai.top/v1"
    if not model:
        model = "gemini-2.5-pro"

    if not api_key:
        logger.error("API key is empty")
        raise ValueError("API key is empty")

    client = OpenAI(
        base_url=base_url,
        api_key=api_key,
    )

    logger.info(f"Calling LLM, model={model}")
    completion = client.chat.completions.create(
        model=model,
        messages=messages,
    )

    content = completion.choices[0].message.content
    if content is None:
        logger.error("LLM returned empty content")
        raise RuntimeError("LLM returned empty content")

    content = content.strip()
    logger.info(f"LLM raw output: {content}")

    # Try JSON parsing, compatible with ```json ... ``` wrapped cases
    if isinstance(content, str):
        try:
            return json.loads(content)
        except json.JSONDecodeError:
            cleaned = content.strip().strip("`")
            lower = cleaned.lower()
            if lower.startswith("json\n") or lower.startswith("json\r\n"):
                cleaned = "\n".join(cleaned.splitlines()[1:])
            return json.loads(cleaned)
    else:
        # It should not reach here, but keep compatible
        return content

SDK = Software Development Kit（軟件開發工具包）。它是一套官方提供的工具，用來方便開發者調用某個服務。

對於 OpenAI，SDK 封裝了 HTTP 請求，不需要自己寫複雜的 POST body、headers，SDK 會自動處理錯誤、重試、流式輸出等。

用 SDK 的寫法：

completion = client.chat.completions.create(
    model=model,
    messages=messages,
)

如果不用 SDK，就必須手寫 HTTP:

import requests

requests.post(
    "https://api.openai.com/v1/chat/completions",
    headers={"Authorization": f"Bearer {API_KEY}"},
    json={
        "model": "...",
        "messages": [...],
    }
)

把要分析的文本傳入大模型:

  1. SYSTEM_PROMPT: 從外部文件加載規則
def load_system_prompt():
    base_dir = os.path.dirname(os.path.abspath(__file__))
    ht_path = os.path.join(base_dir, "ht_jg.txt")
    ...
SYSTEM_PROMPT = load_system_prompt()
  1. messages: 一條一條地喂合同記錄
def build_messages_for_single_record(
    region: str,
    organization_name: Optional[str],
    contract_record: Dict[str, Any],
) -> List[Dict[str, str]]:
    """
    Construct the messages for a single contract record:
    - system: system instruction (SYSTEM_PROMPT), which should clearly state: this time only process this one contract
    - user:   contains region / organization_name / this current contract_record / demo_str
    """
    user_payload = {
        "region": region,
        "organization_name": organization_name or "",
        "contract_record": contract_record,
        "demo_str": demo_str,
    }

    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {
            "role": "user",
            "content": (
                "下面是本次任務的具體輸入參數 JSON（僅包含一條合同記錄），"
                "你只能基於這條記錄判斷是否與目標地區/機構相關，並抽取結構化信息：\n\n"
                + json.dumps(user_payload, ensure_ascii=False, indent=2)
            ),
        },
    ]
    return messages

逐條處理：每次只給模型一條 contract_record（title/url/date/content），杜絕“混合同”錯位。

混合同錯位（cross-record hallucination 或 cross-record mixup）指模型在處理一批合同記錄時，把 A 合同的 URL、B 合同的內容、C 合同的業務場景混在一起輸出。

表現為：

  • 輸出的 URL ≠ 它根據內容抽取出的業務場景

  • 輸出內容包含另一個合同的片段

  • 模型生成了不存在的合同（虛構 URL）

  • 第 N 條合同的輸出明顯引用了第 N + 1 條文本的信息

你之前遇到的這段就是典型“混合同”：模型把不存在的 URL 輸出成真實合同，並且內容來自完全不同文章

出現這樣的原因是：

  1. 大模型的 “上下文融合機制”

    LLM 的本質是把輸入的所有內容當成一個統一的語境進行概率預測。

    這意味着：如果你給模型一次性輸入了 200 條合同，模型不會理解“這是 200 條獨立樣本”，它會認為這是“一個巨大的語料庫”，並在其中“尋找它認為合理的關聯”。

    換句話説：LLM 不天然支持“一條一條分開處理”這個概念。批量輸入 = 讓模型混淆邊界 = 產生錯位。

  2. 模型只保證“語言一致性”，不保證“索引對應關係”

  3. 模型會自動“對齊模式”，而不是逐條分析

    在批量合同輸入中，你提供了大量文本，但模型看到的是：

     contracts_records: [
 { A },
 { B },
 { C },
 ...
 ]

    LLM 的行為通常是模式聚合（pattern aggregation），提取公共特徵 → 生成統一風格的輸出，不會保持每個 item 的邊界。

    這叫模式坍縮（pattern collapse），這是 LLM 的自然傾向，而不是 bug。

  4. 模型把你的任務理解成“總結一個列表”，而不是“輸出 N 個獨立結果”

  5. 模型可能會“幻想 URL”

    你給大模型看了很多 URL 模式，它就會學會 URL 的構造方式。當大模型需要輸出一個 URL 卻沒有事實依據時，它會“自己造一個”合理-looking 的 URL，這種行為叫結構性幻覺（structural hallucination）。

後端
user avatar
0 位用戶收藏了這個故事！

發佈 評論

Some HTML is okay.