【教程】用Python+Playwright打造AI智能网页爬虫，自动提取结构化数据 - 闲社

1. Playwright 加载目标网页并截图
2. 将页面HTML + 截图发送给LLM
3. LLM分析页面结构，返回数据提取策略
4. 按策略提取数据，输出结构化结果

复制代码

pip install playwright openai python-dotenv
playwright install chromium

复制代码

OPENAI_API_KEY=your_api_key_here
OPENAI_BASE_URL=https://api.openai.com/v1 # 或其他兼容接口

复制代码

import os
import json
import base64
from playwright.sync_api import sync_playwright
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
class AIScraper:
def __init__(self):
self.client = OpenAI(
api_key=os.getenv("OPENAI_API_KEY"),
base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
)
def capture_page(self, url):
"""用Playwright加载页面并截图"""
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
page = browser.new_page(viewport={"width": 1920, "height": 1080})
page.goto(url, wait_until="networkidle")
# 获取页面HTML
html = page.content()
# 截图并转为base64
screenshot = page.screenshot(type="jpeg", quality=80)
screenshot_b64 = base64.b64encode(screenshot).decode()
browser.close()
return html, screenshot_b64
def analyze_page(self, html, screenshot_b64, instruction):
"""调用LLM分析页面，返回提取策略"""
prompt = f"""
你是一个网页数据提取专家。请分析以下网页内容，帮我提取指定数据。
提取需求：{instruction}
页面HTML片段（前5000字符）：
{html[:5000]}
请返回JSON格式的提取策略，包含：
1. "data_type": 数据类型（list/table/detail）
2. "selectors": 具体的CSS选择器或提取逻辑
3. "fields": 需要提取的字段列表
4. "next_page": 是否有下一页，如何翻页（可选）
只返回JSON，不要其他解释。
"""
response = self.client.chat.completions.create(
model="gpt-4o-mini", # 或其他可用模型
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": prompt},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{screenshot_b64}"
}
}
]
}
],
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
def extract_data(self, url, instruction):
"""主流程：截图 -> AI分析 -> 提取数据"""
print(f"正在加载页面：{url}")
html, screenshot = self.capture_page(url)
print("正在分析页面结构...")
strategy = self.analyze_page(html, screenshot, instruction)
print(f"提取策略：{json.dumps(strategy, ensure_ascii=False, indent=2)}")
# 根据策略提取数据
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
page = browser.new_page()
page.goto(url, wait_until="networkidle")
data = self._execute_strategy(page, strategy)
browser.close()
return data
def _execute_strategy(self, page, strategy):
"""根据AI返回的策略执行数据提取"""
data_type = strategy.get("data_type", "list")
selectors = strategy.get("selectors", {})
fields = strategy.get("fields", [])
results = []
if data_type == "list":
items = page.query_selector_all(selectors.get("item", "body"))
for item in items[:20]: # 限制数量
row = {}
for field in fields:
name = field["name"]
selector = field.get("selector", "")
attr = field.get("attribute", "textContent")
el = item.query_selector(selector) if selector else item
if el:
if attr == "textContent":
row[name] = el.text_content().strip()
elif attr == "href":
row[name] = el.get_attribute("href")
else:
row[name] = el.get_attribute(attr)
else:
row[name] = None
results.append(row)
elif data_type == "table":
rows = page.query_selector_all(selectors.get("row", "tr"))
for row_el in rows[1:]: # 跳过表头
cells = row_el.query_selector_all("td")
row = {}
for i, field in enumerate(fields):
if i < len(cells):
row[field["name"]] = cells[i].text_content().strip()
results.append(row)
return results
# 使用示例
if __name__ == "__main__":
scraper = AIScraper()
# 示例：提取新闻列表
url = "https://news.ycombinator.com"
instruction = "提取首页所有新闻的标题、链接和评分"
data = scraper.extract_data(url, instruction)
print(json.dumps(data, ensure_ascii=False, indent=2))

复制代码

python ai_scraper.py

复制代码

def extract_with_pagination(self, start_url, instruction, max_pages=5):
"""支持自动翻页的数据采集"""
all_data = []
current_url = start_url
page_count = 0
while current_url and page_count < max_pages:
print(f"正在采集第 {page_count + 1} 页...")
html, screenshot = self.capture_page(current_url)
strategy = self.analyze_page(html, screenshot, instruction)
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
page = browser.new_page()
page.goto(current_url, wait_until="networkidle")
data = self._execute_strategy(page, strategy)
all_data.extend(data)
# 查找下一页链接
next_selector = strategy.get("next_page", {}).get("selector")
if next_selector:
next_el = page.query_selector(next_selector)
current_url = next_el.get_attribute("href") if next_el else None
if current_url and not current_url.startswith("http"):
from urllib.parse import urljoin
current_url = urljoin(start_url, current_url)
else:
current_url = None
browser.close()
page_count += 1
return all_data

复制代码