AI 爬虫方案全解

import requests
import json
from bs4 import BeautifulSoup
from anthropic import Anthropic

# 步骤1：获取页面 HTML
response = requests.get(
    'https://example.com/product/123',
    headers={'User-Agent': 'Mozilla/5.0 ...'},
)
soup = BeautifulSoup(response.text, 'lxml')

# 步骤2：提取页面主要文本内容（去除导航/页脚等噪声）
# 关键：给 LLM 过多的 HTML 会让它分心和浪费 token
main_content = soup.get_text(separator='\n', strip=True)

# 如果内容太长，截取关键部分
if len(main_content) > 8000:
    main_content = main_content[:8000]

# 步骤3：调用 LLM 提取结构化数据
client = Anthropic()

message = client.messages.create(
    model='claude-sonnet-4-20250514',
    max_tokens=1024,
    messages=[{
        'role': 'user',
        'content': f"""从以下网页文本中提取商品信息，返回 JSON 格式。

要求：
- name: 商品名称
- price: 价格（数字）
- original_price: 原价（如有，否则 null）
- rating: 评分（数字，如 4.5）
- reviews_count: 评论数
- availability: 是否有货（true/false）
- specs: 规格参数字典

网页内容：
{main_content}

请只返回 JSON，不要其他文字。"""
    }],
)

# 步骤4：解析 LLM 返回的 JSON
result_text = message.content[0].text
# 尝试提取 JSON（LLM 可能返回带 ```json 标记的内容）
if '```json' in result_text:
    result_text = result_text.split('```json')[1].split('```')[0].strip()

product = json.loads(result_text)
print(json.dumps(product, ensure_ascii=False, indent=2))
# {"name": "iPhone 15 Pro Max", "price": 9999, "original_price": 10999, ...}

pip install scrapegraphai

# 需要 LLM API Key（支持 OpenAI / Anthropic / 本地模型）
export OPENAI_API_KEY=sk-...
# 或使用本地模型（免费，但需要 Ollama）
# pip install scrapegraphai[ollama]

import json
from scrapegraphai.graphs import SmartScraperGraph

# 定义数据 Schema（告诉 LLM 你要提取什么）
graph_config = {
    "llm": {
        "model": "openai/gpt-4o-mini",  # 或 "ollama/llama3"
        "api_key": "sk-...",             # 或不设，使用本地模型
    },
}

# 创建爬虫图
scraper = SmartScraperGraph(
    prompt="提取所有商品信息：名称、价格、评分、链接",
    source="https://example.com/products",
    config=graph_config,
)

# 执行爬取
result = scraper.run()
print(json.dumps(result, ensure_ascii=False, indent=2))
# {
#   "products": [
#     {"name": "Product A", "price": "$29.99", "rating": "4.5", "link": "..."},
#     {"name": "Product B", "price": "$49.99", "rating": "4.8", "link": "..."},
#   ]
# }

from scrapegraphai.graphs import SmartScraperGraph

# 爬取+搜索组合
graph_config = {
    "llm": {"model": "openai/gpt-4o-mini", "api_key": "sk-..."},
    "verbose": True,
}

scraper = SmartScraperGraph(
    prompt="从页面中提取所有博客文章的标题、摘要、发布日期和作者",
    source="https://blog.example.com",
    config=graph_config,
    schema_provided={  # 可选：提供 Schema 约束输出格式
        "type": "object",
        "properties": {
            "articles": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "title": {"type": "string"},
                        "summary": {"type": "string"},
                        "date": {"type": "string"},
                        "author": {"type": "string"},
                    }
                }
            }
        }
    }
)

result = scraper.run()

pip install crawl4ai

import asyncio
from crawl4ai import AsyncWebCrawler

async def crawl_example():
    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(
            url='https://example.com/article',
            # 自动去除导航、广告、页脚等噪声
            word_count_threshold=10,
            # 提取主要内容（类似浏览器阅读模式）
            css_selector='article',
            # 输出 Markdown 格式
            output_format='markdown',
        )

        print(result.markdown)  # 干净的 Markdown 文本
        print(result.cleaned_html)  # 清理后的 HTML

        # 提取结构化数据
        result = await crawler.arun(
            url='https://example.com/products',
            extraction_strategy='llm',  # LLM 辅助提取
            query='提取所有商品的名称和价格',
        )
        print(result.extracted_content)

asyncio.run(crawl_example())

async def batch_crawl(urls: list[str]) -> list[dict]:
    """批量爬取多个 URL"""
    results = []
    async with AsyncWebCrawler() as crawler:
        for url in urls:
            result = await crawler.arun(
                url=url,
                word_count_threshold=10,
                output_format='markdown',
            )
            results.append({
                'url': url,
                'title': result.metadata.get('title', ''),
                'content': result.markdown,
            })
    return results

# pip install firecrawl-py
from firecrawl import FirecrawlApp

app = FirecrawlApp(api_key='fc-...')

# 爬取单个页面
result = app.scrape_url(
    'https://example.com/product/123',
    params={
        'formats': ['extract'],
        'extract': {
            'schema': {
                'type': 'object',
                'properties': {
                    'name': {'type': 'string'},
                    'price': {'type': 'number'},
                    'description': {'type': 'string'},
                }
            }
        }
    }
)
print(result['extract'])  # 结构化 JSON

# 爬取整个网站
result = app.crawl_url(
    'https://example.com',
    params={
        'limit': 100,
        'scrapeOptions': {
            'formats': ['markdown'],
        }
    }
)

pip install browser-use
playwright install chromium

from browser_use import Agent
from langchain_openai import ChatOpenAI

# 创建 Agent
agent = Agent(
    task="打开京东，搜索'iPhone 15 Pro'，获取前3个商品的价格和名称",
    llm=ChatOpenAI(model="gpt-4o"),
)

# 执行任务
result = await agent.run()
print(result)

from browser_use import Agent, Controller
from pydantic import BaseModel

class ProductInfo(BaseModel):
    name: str
    price: float
    url: str

controller = Controller(output_model=ProductInfo)

# 定义自定义操作
@controller.action("将商品信息保存到数据库")
def save_product(product: ProductInfo):
    # 这里可以对接 MongoDB / MySQL
    db.products.insert_one(product.model_dump())
    return f"已保存: {product.name}"

agent = Agent(
    task="爬取某电商网站的手机商品信息",
    llm=ChatOpenAI(model="gpt-4o"),
    controller=controller,
)

result = await agent.run()

import json
from anthropic import Anthropic
from playwright.sync_api import sync_playwright

class AICrawler:
    """AI Agent 爬虫 —— LLM 决策 + Playwright 执行"""

    def __init__(self):
        self.client = Anthropic()
        self.pw = sync_playwright().start()
        self.browser = self.pw.chromium.launch(headless=True)
        self.context = self.browser.new_context(
            user_agent='Mozilla/5.0 ...',
        )
        self.page = self.context.new_page()

    def crawl(self, url: str, goal: str) -> dict:
        """
        爬取目标 URL，根据 goal 提取数据。

        Args:
            url: 目标 URL
            goal: 提取目标描述（自然语言）
        """
        # 步骤1：访问页面
        self.page.goto(url, wait_until='networkidle')

        # 步骤2：获取页面内容
        html = self.page.content()

        # 步骤3：清理 HTML（只保留文本，减少 token 消耗）
        text_content = self.page.evaluate("""
            () => {
                // 去除 script、style 标签
                document.querySelectorAll('script, style, nav, footer, header').forEach(e => e.remove());
                return document.body.innerText;
            }
        """)

        # 步骤4：LLM 提取数据
        response = self.client.messages.create(
            model='claude-sonnet-4-20250514',
            max_tokens=2048,
            messages=[{
                'role': 'user',
                'content': f"""从以下网页内容中提取数据。

目标：{goal}

页面内容（截取前 6000 字符）：
{text_content[:6000]}

请返回 JSON 格式的数据，不要包含其他文字。"""
            }],
        )

        result_text = response.content[0].text
        if '```json' in result_text:
            result_text = result_text.split('```json')[1].split('```')[0].strip()

        return json.loads(result_text)

    def close(self):
        self.context.close()
        self.browser.close()
        self.pw.stop()


# 使用
crawler = AICrawler()
data = crawler.crawl(
    url='https://example.com/products',
    goal='提取所有商品的名称、价格、评分和库存状态',
)
print(json.dumps(data, ensure_ascii=False, indent=2))
crawler.close()

import base64
from anthropic import Anthropic
from playwright.sync_api import sync_playwright

def solve_image_captcha(page, captcha_selector: str) -> str:
    """使用 Claude Vision 识别图像验证码"""

    # 步骤1：截取验证码图片
    captcha_element = page.locator(captcha_selector)
    screenshot_bytes = captcha_element.screenshot()

    # 步骤2：编码为 base64
    img_b64 = base64.b64encode(screenshot_bytes).decode('utf-8')

    # 步骤3：调用多模态 LLM 识别
    client = Anthropic()
    response = client.messages.create(
        model='claude-sonnet-4-20250514',
        max_tokens=100,
        messages=[{
            'role': 'user',
            'content': [
                {
                    'type': 'image',
                    'source': {
                        'type': 'base64',
                        'media_type': 'image/png',
                        'data': img_b64,
                    }
                },
                {
                    'type': 'text',
                    'text': '这是一个验证码图片，请识别其中的字符，只返回字符本身，不要其他文字。'
                }
            ]
        }]
    )

    captcha_text = response.content[0].text.strip()

    # 步骤4：填入验证码
    page.fill('input[name="captcha"]', captcha_text)
    page.click('button[type="submit"]')

    return captcha_text

# 使用
with sync_playwright() as pw:
    browser = pw.chromium.launch(headless=False)
    page = browser.new_page()
    page.goto('https://example.com/login')
    code = solve_image_captcha(page, '#captcha-img')
    print(f"识别结果: {code}")

import random
import math
from playwright.sync_api import sync_playwright

def generate_human_like_drag(distance: int) -> list[dict]:
    """
    生成仿人类的鼠标拖拽轨迹。

    使用贝塞尔曲线 + 随机抖动，避免被行为检测识别为机器人。

    Args:
        distance: 需要拖拽的像素距离

    Returns:
        鼠标移动步骤列表 [{x, y, delay}, ...]
    """
    steps = []
    # 人类拖拽特征：起步慢 → 中间快 → 末尾减速
    total_steps = random.randint(30, 50)

    for i in range(total_steps):
        # 使用缓动函数模拟加速-减速
        progress = i / total_steps
        # ease-out cubic：快速起步，缓慢结束
        eased = 1 - (1 - progress) ** 3

        x = distance * eased
        # 添加纵向随机抖动（人类拖拽不是完全水平的）
        y = random.gauss(0, 1.5)
        # 每步间隔 8-15ms（人类反应速度）
        delay = random.uniform(8, 15)

        steps.append({'x': x, 'y': y, 'delay': delay})

    # 末尾微调（人类通常不会精确到位，需要修正）
    steps.append({'x': distance, 'y': 0, 'delay': random.uniform(50, 100)})

    return steps


def solve_slider_captcha(page, slider_selector: str, distance: int) -> None:
    """使用仿人类轨迹拖拽滑块验证码"""

    slider = page.locator(slider_selector)
    box = slider.bounding_box()

    if not box:
        raise ValueError("滑块元素未找到")

    # 移动到滑块中心
    start_x = box['x'] + box['width'] / 2
    start_y = box['y'] + box['height'] / 2

    page.mouse.move(start_x, start_y)
    page.mouse.down()

    # 按照仿人类轨迹拖拽
    trajectory = generate_human_like_drag(distance)
    for step in trajectory:
        page.mouse.move(
            start_x + step['x'],
            start_y + step['y'],
        )
        page.wait_for_timeout(int(step['delay']))

    page.mouse.up()

import base64
import json
from anthropic import Anthropic
from playwright.sync_api import sync_playwright

def solve_click_captcha(page, captcha_img_selector: str, prompt: str) -> list[dict]:
    """
    使用 LLM 视觉能力解决点选验证码。

    Args:
        page: Playwright 页面对象
        captcha_img_selector: 验证码图片选择器
        prompt: 点击提示（如"请依次点击：猫、狗、鸟"）

    Returns:
        点击坐标列表 [{x, y}, ...]
    """
    # 截取验证码图片
    element = page.locator(captcha_img_selector)
    img_bytes = element.screenshot()
    img_b64 = base64.b64encode(img_bytes).decode('utf-8')

    # 获取图片在页面中的位置
    box = element.bounding_box()

    # LLM 识别并返回坐标
    client = Anthropic()
    response = client.messages.create(
        model='claude-sonnet-4-20250514',
        max_tokens=1024,
        messages=[{
            'role': 'user',
            'content': [
                {
                    'type': 'image',
                    'source': {
                        'type': 'base64',
                        'media_type': 'image/png',
                        'data': img_b64,
                    }
                },
                {
                    'type': 'text',
                    'text': f"""这张图片的验证码提示是：{prompt}

请找出需要点击的目标，返回它们的坐标（相对于图片左上角的像素偏移）。

返回 JSON 格式：
{{"clicks": [{{"x": 100, "y": 80, "label": "猫"}}, ...]}}

只返回 JSON，不要其他文字。"""
                }
            ]
        }]
    )

    result_text = response.content[0].text.strip()
    if '```json' in result_text:
        result_text = result_text.split('```json')[1].split('```')[0].strip()

    data = json.loads(result_text)

    # 将相对坐标转为页面绝对坐标并点击
    clicks = []
    for click in data['clicks']:
        abs_x = box['x'] + click['x']
        abs_y = box['y'] + click['y']
        clicks.append({'x': abs_x, 'y': abs_y, 'label': click['label']})

        # 模拟人类点击（添加随机延迟）
        page.mouse.click(abs_x, abs_y, delay=random.randint(50, 150))
        page.wait_for_timeout(random.randint(300, 600))

    return clicks

import random

from playwright.sync_api import sync_playwright

def create_stealth_browser(pw):
    """创建带反检测能力的浏览器实例"""

    browser = pw.chromium.launch(
        headless=True,
        args=[
            '--disable-blink-features=AutomationControlled',  # 隐藏自动化标志
            '--no-sandbox',
        ]
    )

    context = browser.new_context(
        user_agent='Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) '
                   'AppleWebKit/537.36 (KHTML, like Gecko) '
                   'Chrome/125.0.0.0 Safari/537.36',
        viewport={'width': 1920, 'height': 1080},
        locale='zh-CN',
        timezone_id='Asia/Shanghai',
    )

    # 注入反检测脚本
    context.add_init_script("""
        // 1. 隐藏 webdriver 标志
        Object.defineProperty(navigator, 'webdriver', { get: () => undefined });

        // 2. 伪装 Chrome 特征
        window.chrome = { runtime: {}, csi: function(){}, loadTimes: function(){} };

        // 3. 伪装 permissions API
        const originalQuery = window.navigator.permissions.query;
        window.navigator.permissions.query = (parameters) =>
            parameters.name === 'notifications'
                ? Promise.resolve({ state: Notification.permission })
                : originalQuery(parameters);

        // 4. 随机化 Canvas 指纹
        const originalToDataURL = HTMLCanvasElement.prototype.toDataURL;
        HTMLCanvasElement.prototype.toDataURL = function(type) {
            // 在像素数据中添加微小的随机噪声
            const context = this.getContext('2d');
            if (context) {
                const imageData = context.getImageData(0, 0, 1, 1);
                imageData.data[3] = imageData.data[3] ^ (Math.random() * 2 | 0); // 最低位随机
                context.putImageData(imageData, 0, 0);
            }
            return originalToDataURL.apply(this, arguments);
        };

        // 5. 伪装 WebGL 渲染器信息
        const getParameter = WebGLRenderingContext.prototype.getParameter;
        WebGLRenderingContext.prototype.getParameter = function(parameter) {
            if (parameter === 37445) return 'Intel Inc.';         // UNMASKED_VENDOR
            if (parameter === 37446) return 'Intel Iris OpenGL';  // UNMASKED_RENDERER
            return getParameter.call(this, parameter);
        };

        // 6. 随机化 AudioContext 指纹
        const audioCtx = window.AudioContext || window.webkitAudioContext;
        if (audioCtx) {
            const originalGetFloatFreqData = AnalyserNode.prototype.getFloatFrequencyData;
            AnalyserNode.prototype.getFloatFrequencyData = function(array) {
                originalGetFloatFreqData.call(this, array);
                // 添加微量随机噪声
                for (let i = 0; i < array.length; i++) {
                    array[i] += (Math.random() - 0.5) * 0.001;
                }
            };
        }
    """)

    return browser, context

# 使用
with sync_playwright() as pw:
    browser, context = create_stealth_browser(pw)
    page = context.new_page()
    page.goto('https://example.com')
    # ... 执行爬取操作

import random
import time

def human_type(page, selector: str, text: str) -> None:
    """模拟人类打字节奏：随机延迟 + 偶尔退格修正"""

    element = page.locator(selector)
    element.click()

    for i, char in enumerate(text):
        # 基础打字延迟：50-150ms（正常人打字速度）
        delay = random.gauss(100, 30)
        delay = max(40, min(250, delay))  # 钳制到合理范围

        # 模拟偶尔的"打错字 → 退格修正"
        if random.random() < 0.05 and len(text) > 3:  # 5% 概率打错
            # 打一个错误的字符
            wrong_chars = 'abcdefghijklmnopqrstuvwxyz'
            wrong_char = random.choice(wrong_chars)
            page.keyboard.type(wrong_char, delay=0)
            time.sleep(random.uniform(0.1, 0.3))  # 发现错误的延迟
            page.keyboard.press('Backspace')
            time.sleep(random.uniform(0.05, 0.15))

        page.keyboard.type(char, delay=0)
        time.sleep(delay / 1000)

        # 偶尔在标点符号后停顿更长
        if char in '，。！？；：':
            time.sleep(random.uniform(0.2, 0.5))

import time
import random
import logging
from dataclasses import dataclass, field

logger = logging.getLogger(__name__)

@dataclass
class AdaptiveScheduler:
    """
    自适应请求调度器。

    根据服务器响应时间动态调整请求间隔，
    遇到反爬时自动增加延迟，恢复正常后逐步缩短。
    """

    base_delay: float = 1.0          # 基础延迟（秒）
    current_delay: float = 1.0       # 当前延迟
    min_delay: float = 0.5           # 最小延迟
    max_delay: float = 30.0          # 最大延迟
    backoff_factor: float = 1.5      # 退避因子
    recovery_factor: float = 0.9     # 恢复因子
    consecutive_success: int = 0     # 连续成功次数
    consecutive_failure: int = 0     # 连续失败次数

    def on_success(self, response_time: float) -> None:
        """请求成功回调：逐步缩短延迟"""
        self.consecutive_success += 1
        self.consecutive_failure = 0

        # 连续 5 次成功，缩短延迟
        if self.consecutive_success >= 5:
            self.current_delay = max(
                self.min_delay,
                self.current_delay * self.recovery_factor
            )
            self.consecutive_success = 0
            logger.info(f"恢复正常，延迟缩短至 {self.current_delay:.1f}s")

    def on_failure(self, status_code: int) -> float:
        """请求失败回调：指数退避"""
        self.consecutive_failure += 1
        self.consecutive_success = 0

        if status_code == 429:
            # 429 Too Many Requests：大幅增加延迟
            self.current_delay = min(
                self.max_delay,
                self.current_delay * self.backoff_factor * 2
            )
        elif status_code in (403, 503):
            # 403 Forbidden / 503 Service Unavailable
            self.current_delay = min(
                self.max_delay,
                self.current_delay * self.backoff_factor
            )

        logger.warning(f"请求失败({status_code})，延迟增加至 {self.current_delay:.1f}s")
        return self.current_delay

    def wait(self) -> None:
        """等待当前延迟时间（添加随机抖动）"""
        actual_delay = self.current_delay + random.uniform(-0.1, 0.3)
        time.sleep(max(0, actual_delay))

import base64
import json
from anthropic import Anthropic
from playwright.sync_api import sync_playwright

def extract_table_from_screenshot(
    url: str,
    table_selector: str = 'table',
) -> list[dict]:
    """
    使用多模态 LLM 从页面截图中提取表格数据。

    适用于：
    - 表格渲染在 Canvas/SVG 中，无法从 HTML 提取
    - 表格结构复杂（合并单元格、嵌套表头）
    - 表格数据来自动态 JS 渲染
    """
    with sync_playwright() as pw:
        browser = pw.chromium.launch(headless=True)
        page = browser.new_page(viewport={'width': 1920, 'height': 1080})
        page.goto(url, wait_until='networkidle')

        # 截取表格区域
        table = page.locator(table_selector)
        screenshot_bytes = table.screenshot()
        browser.close()

    # 发送给多模态 LLM 识别
    img_b64 = base64.b64encode(screenshot_bytes).decode('utf-8')

    client = Anthropic()
    response = client.messages.create(
        model='claude-sonnet-4-20250514',
        max_tokens=4096,
        messages=[{
            'role': 'user',
            'content': [
                {
                    'type': 'image',
                    'source': {
                        'type': 'base64',
                        'media_type': 'image/png',
                        'data': img_b64,
                    }
                },
                {
                    'type': 'text',
                    'text': """请识别这张截图中的表格数据，转换为 JSON 格式。

要求：
1. 表头作为 JSON 的键名
2. 每行数据为一个对象
3. 数字类型保持数值（不要转为字符串）
4. 合并单元格的值填充到所有对应的行

返回格式：
{"headers": ["列1", "列2", ...], "rows": [{...}, ...]}

只返回 JSON，不要其他文字。"""
                }
            ]
        }]
    )

    result_text = response.content[0].text.strip()
    if '```json' in result_text:
        result_text = result_text.split('```json')[1].split('```')[0].strip()

    return json.loads(result_text)

# 使用
data = extract_table_from_screenshot('https://example.com/data-table')
print(json.dumps(data, ensure_ascii=False, indent=2))

import base64
import json
from anthropic import Anthropic

def extract_chart_data(image_path: str, chart_type: str = 'bar') -> dict:
    """
    使用多模态 LLM 从图表截图中估算数据。

    注意：LLM 提取图表数据是**估算**，精度有限。
    对于精确数据，应优先寻找图表的原始数据源（API 接口）。
    """
    with open(image_path, 'rb') as f:
        img_b64 = base64.b64encode(f.read()).decode('utf-8')

    client = Anthropic()
    response = client.messages.create(
        model='claude-sonnet-4-20250514',
        max_tokens=2048,
        messages=[{
            'role': 'user',
            'content': [
                {
                    'type': 'image',
                    'source': {
                        'type': 'base64',
                        'media_type': 'image/png',
                        'data': img_b64,
                    }
                },
                {
                    'type': 'text',
                    'text': f"""这是一个{chart_type}图（柱状图），请估算每个柱子对应的数据值。

返回 JSON 格式：
{{"title": "图表标题", "x_axis_label": "...", "y_axis_label": "...", "data": [{{"label": "...", "value": 123}}, ...]}}

只返回 JSON。"""
                }
            ]
        }]
    )

    result_text = response.content[0].text.strip()
    if '```json' in result_text:
        result_text = result_text.split('```json')[1].split('```')[0].strip()

    return json.loads(result_text)

import json
from anthropic import Anthropic

def extract_pdf_tables(pdf_path: str, api_key: str = None) -> list[dict]:
    """
    使用 PyMuPDF + LLM 联合方案解析 PDF 中的表格。

    流程：
    1. PyMuPDF 提取 PDF 页面文本
    2. LLM 理解上下文并结构化提取
    """
    import fitz  # PyMuPDF

    doc = fitz.open(pdf_path)
    results = []

    for page_num in range(len(doc)):
        page = doc[page_num]

        # 提取文本（保留格式信息）
        text = page.get_text("text")

        if not text.strip():
            continue

        # 如果文本过长，只发送前 8000 字符
        truncated = text[:8000]

        # 调用 LLM 结构化提取
        client = Anthropic() if api_key is None else Anthropic(api_key=api_key)
        response = client.messages.create(
            model='claude-sonnet-4-20250514',
            max_tokens=4096,
            messages=[{
                'role': 'user',
                'content': f"""以下是从 PDF 第 {page_num + 1} 页提取的文本内容。

请识别其中的表格数据，转换为结构化 JSON 格式。
如果没有表格，返回 {{"has_table": false}}。

PDF 内容：
{truncated}

返回格式：
{{"has_table": true, "tables": [{{"headers": [...], "rows": [...]}}]}}"""
            }]
        )

        result_text = response.content[0].text.strip()
        if '```json' in result_text:
            result_text = result_text.split('```json')[1].split('```')[0].strip()

        try:
            page_data = json.loads(result_text)
            page_data['page'] = page_num + 1
            results.append(page_data)
        except json.JSONDecodeError:
            logger.warning(f"第 {page_num + 1} 页 LLM 输出无法解析为 JSON")

    doc.close()
    return results

import logging
logger = logging.getLogger(__name__)

npm install stagehand
# 或 Python 版本
pip install stagehand

from stagehand import Stagehand, StagehandConfig

async def stagehand_example():
    config = StagehandConfig(
        env="LOCAL",  # 本地浏览器
        api_key="your-api-key",
        model_name="gpt-4o",
    )

    stagehand = Stagehand(config)
    await stagehand.init()

    page = stagehand.page
    await page.goto("https://example.com/login")

    # act：AI 执行操作（用自然语言描述）
    await page.act("在用户名输入框中输入 admin")
    await page.act("在密码输入框中输入 password123")
    await page.act("点击登录按钮")

    # extract：AI 提取结构化数据
    data = await page.extract({
        "prompt": "提取页面上所有商品信息",
        "schema": {
            "type": "object",
            "properties": {
                "products": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "name": {"type": "string"},
                            "price": {"type": "number"},
                        }
                    }
                }
            }
        }
    })

    # observe：AI 观察页面，返回可执行的操作列表
    actions = await page.observe("如何找到商品详情页？")

    await stagehand.close()
    return data

# Skyvern 通过 API 调用
import requests

response = requests.post(
    "http://localhost:8000/api/v1/tasks",
    json={
        "url": "https://example.com/form",
        "navigation_goal": "填写注册表单并提交",
        "data_extraction_goal": "提取注册成功后的确认信息",
        "extracted_information_schema": {
            "type": "object",
            "properties": {
                "confirmation_id": {"type": "string"},
                "status": {"type": "string"},
            }
        }
    }
)

task_id = response.json()["id"]
# 轮询任务状态...

from camoufox.sync_api import Camoufox

def camoufox_example():
    with Camoufox(headless=True) as browser:
        page = browser.new_page()
        page.goto("https://browserleaks.com/canvas")

        # Camoufox 自动处理：
        # - Canvas 指纹随机化
        # - WebGL 渲染器伪装
        # - AudioContext 指纹伪装
        # - Navigator 属性伪装
        # - 屏幕分辨率和色深随机化

        content = page.content()
        print("指纹检测结果:", content[:200])

# 对比

import json
import logging
from dataclasses import dataclass
from typing import Any

logger = logging.getLogger(__name__)

@dataclass
class HybridCrawler:
    """
    混合爬虫：自动在传统爬虫和 AI 爬虫之间选择。

    策略：
    1. 优先尝试传统选择器提取（成本为零）
    2. 选择器提取失败时，回退到 AI 辅助提取
    3. AI 提取结果与传统结果交叉验证
    """

    selectors: dict[str, str]   # CSS 选择器映射
    llm_client: Any = None      # LLM 客户端（延迟初始化）
    confidence_threshold: float = 0.8

    def extract(self, html: str, url: str) -> dict:
        """混合提取策略"""

        # 阶段1：尝试传统选择器提取（成本 ≈ 0）
        traditional_result = self._extract_with_selectors(html)
        if traditional_result and self._validate(traditional_result):
            logger.info(f"[传统提取成功] {url}")
            return {**traditional_result, '_method': 'selector', '_confidence': 1.0}

        # 阶段2：回退到 AI 辅助提取（有 API 成本）
        logger.info(f"[选择器失败，回退 AI] {url}")
        ai_result = self._extract_with_llm(html, url)
        if ai_result:
            return {**ai_result, '_method': 'llm', '_confidence': ai_result.get('_confidence', 0.85)}

        # 阶段3：两种方法都失败
        logger.warning(f"[提取失败] {url}")
        return {'_method': 'failed', '_confidence': 0.0}

    def _extract_with_selectors(self, html: str) -> dict | None:
        """使用 CSS 选择器提取"""
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(html, 'lxml')

        result = {}
        for field, selector in self.selectors.items():
            element = soup.select_one(selector)
            if element:
                result[field] = element.get_text(strip=True)
            else:
                return None  # 任一字段提取失败，回退到 AI

        return result if result else None

    def _extract_with_llm(self, html: str, url: str) -> dict | None:
        """使用 LLM 辅助提取"""
        if not self.llm_client:
            return None

        from bs4 import BeautifulSoup
        soup = BeautifulSoup(html, 'lxml')
        text = soup.get_text(separator='\n', strip=True)[:6000]

        # ... 调用 LLM API 提取 ...
        # 详见第1节"AI 辅助提取"
        return None  # 简化示例

    def _validate(self, result: dict) -> bool:
        """验证提取结果是否合理"""
        if not result:
            return False
        # 检查关键字段是否为空
        for key, value in result.items():
            if not value or (isinstance(value, str) and len(value.strip()) == 0):
                return False
        return True

# 本地模型部署方案
# 使用 Ollama 部署本地 LLM（零 API 成本）

# 1. 安装 Ollama
# curl -fsSL https://ollama.com/install.sh | sh

# 2. 拉取模型
# ollama pull llama3        # 8B 参数，~4.7GB
# ollama pull mistral       # 7B 参数，~4.1GB
# ollama pull qwen2:7b      # 中文优化，~4.4GB

# 3. Python 调用
import requests

def extract_with_local_llm(text: str, goal: str) -> dict:
    """使用 Ollama 本地模型提取数据（零 API 成本）"""

    response = requests.post(
        'http://localhost:11434/api/generate',
        json={
            'model': 'qwen2:7b',  # 中文场景推荐
            'prompt': f"""从以下网页内容中提取数据。

目标：{goal}

内容：
{text[:4000]}

请返回 JSON 格式。""",
            'stream': False,
            'options': {
                'temperature': 0,  # 确定性输出
                'num_predict': 1024,
            }
        },
        timeout=60,
    )

    result = response.json()
    return result['response']

import json
from dataclasses import dataclass

@dataclass
class ExtractionValidator:
    """AI 爬虫提取结果校验器"""

    schema: dict       # JSON Schema 定义
    required_fields: list[str]

    def validate(self, result: dict) -> tuple[bool, float, list[str]]:
        """
        校验提取结果。

        Returns:
            (是否通过, 置信度 0-1, 问题列表)
        """
        issues = []

        # 1. 检查必需字段
        for field in self.required_fields:
            if field not in result or result[field] is None:
                issues.append(f"缺少必需字段: {field}")

        # 2. 检查类型一致性
        for field, expected_type in self.schema.items():
            if field in result and result[field] is not None:
                actual_type = type(result[field]).__name__
                if actual_type != expected_type:
                    issues.append(f"字段 {field} 类型错误: 期望 {expected_type}, 实际 {actual_type}")

        # 3. 检查值合理性（基于业务规则）
        if 'price' in result and isinstance(result['price'], (int, float)):
            if result['price'] < 0 or result['price'] > 1_000_000:
                issues.append(f"价格不合理: {result['price']}")

        # 4. 计算置信度
        total_checks = len(self.required_fields) + len(self.schema)
        passed_checks = total_checks - len(issues)
        confidence = passed_checks / total_checks if total_checks > 0 else 0.0

        passed = len(issues) == 0 and confidence >= 0.8

        return passed, confidence, issues

    def cross_validate(
        self,
        text_result: dict,
        vision_result: dict,
    ) -> tuple[dict, float]:
        """
        多通道交叉验证：文本通道 vs 视觉通道。

        两个通道提取的结果互相验证，一致的字段置信度更高。
        """
        merged = {}
        total_confidence = 0.0
        field_count = 0

        all_keys = set(text_result.keys()) | set(vision_result.keys())

        for key in all_keys:
            text_val = text_result.get(key)
            vision_val = vision_result.get(key)

            if text_val == vision_val and text_val is not None:
                # 两通道一致：高置信度
                merged[key] = text_val
                total_confidence += 1.0
            elif text_val is not None and vision_val is None:
                # 仅文本通道有值
                merged[key] = text_val
                total_confidence += 0.7
            elif vision_val is not None and text_val is None:
                # 仅视觉通道有值
                merged[key] = vision_val
                total_confidence += 0.7
            else:
                # 两通道不一致：取文本通道（通常更精确）
                merged[key] = text_val
                total_confidence += 0.5

            field_count += 1

        avg_confidence = total_confidence / field_count if field_count else 0
        return merged, avg_confidence

import time
from collections import defaultdict
from dataclasses import dataclass, field

@dataclass
class CrawlMonitor:
    """AI 爬虫监控器"""

    start_time: float = field(default_factory=time.time)
    total_tasks: int = 0
    success_tasks: int = 0
    total_tokens: int = 0
    llm_calls: int = 0
    confidence_scores: list[float] = field(default_factory=list)
    method_counts: dict[str, int] = field(default_factory=lambda: defaultdict(int))

    def record_task(
        self,
        method: str,       # 'selector' / 'llm' / 'agent'
        success: bool,
        confidence: float,
        tokens_used: int = 0,
    ) -> None:
        """记录一次任务结果"""
        self.total_tasks += 1
        if success:
            self.success_tasks += 1
        self.method_counts[method] += 1
        self.confidence_scores.append(confidence)
        self.total_tokens += tokens_used
        if tokens_used > 0:
            self.llm_calls += 1

        # 检查告警条件
        if self.total_tasks >= 10:
            success_rate = self.success_tasks / self.total_tasks
            if success_rate < 0.8:
                self._alert(f"提取成功率过低: {success_rate:.1%}")

            avg_confidence = sum(self.confidence_scores) / len(self.confidence_scores)
            if avg_confidence < 0.7:
                self._alert(f"平均置信度过低: {avg_confidence:.2f}")

            ai_ratio = (self.method_counts.get('llm', 0) + self.method_counts.get('agent', 0)) / self.total_tasks
            if ai_ratio > 0.5:
                self._alert(f"AI 调用占比过高: {ai_ratio:.1%}，选择器可能大面积失效")

    def report(self) -> dict:
        """生成监控报告"""
        elapsed = time.time() - self.start_time
        return {
            'elapsed_minutes': round(elapsed / 60, 1),
            'total_tasks': self.total_tasks,
            'success_rate': f"{self.success_tasks / self.total_tasks:.1%}" if self.total_tasks else 'N/A',
            'total_tokens': self.total_tokens,
            'llm_calls': self.llm_calls,
            'avg_confidence': f"{sum(self.confidence_scores) / len(self.confidence_scores):.2f}" if self.confidence_scores else 'N/A',
            'method_distribution': dict(self.method_counts),
        }

    def _alert(self, message: str) -> None:
        """发送告警（实际项目中对接钉钉/企业微信/Slack）"""
        import logging
        logging.getLogger(__name__).warning(f"[AI爬虫告警] {message}")