AI智能体Skill实战：网页内容抓取从入门到避坑指南

1次阅读

共计 3485 个字符，预计需要花费 9 分钟才能阅读完成。

传统爬虫在应对现代网页时常常力不从心。很多网站采用动态加载技术，比如社交媒体的内容瀑布流、电商网站的懒加载商品列表。如果用 Requests 库直接请求，只能拿到一个空壳 HTML，真正需要的数据可能藏在 JavaScript 动态生成的代码里。

更麻烦的是反爬机制。许多网站会检测请求频率、校验 Headers 完整性，甚至通过鼠标轨迹识别机器人行为。我曾遇到过仅仅因为 User-Agent 不规范，就被封 IP 的尴尬情况。

刚开始学爬虫时，我也用 Requests+BeautifulSoup 这套经典组合。但很快就发现三个致命问题：

无法执行 JavaScript，动态内容直接丢失
缺乏浏览器环境，指纹容易被识别
同步请求导致效率低下

后来转向 Selenium，虽然能解决渲染问题，但存在明显短板：

浏览器启动速度慢（平均 3 - 5 秒）
资源占用高（每个实例消耗 300MB+ 内存）
API 设计不够现代

直到发现 Playwright 这个微软开源的宝藏工具，才算找到终极解决方案。它同时具备：

跨浏览器支持（Chromium/WebKit/Firefox）
自动等待机制（不用再写 sleep）
内置请求拦截和 mock 功能
异步 API 原生支持

安装 Playwright 时要注意两个步骤：

# 先安装 Python 包
pip install playwright

# 再下载浏览器内核
playwright install

建议使用虚拟环境，避免与其他项目产生依赖冲突。我习惯用 poetry 管理依赖，能自动处理版本兼容问题。

先看一个最小可用示例，抓取豆瓣电影 Top250：

import asyncio
from playwright.async_api import async_playwright

async def scrape_douban():
    async with async_playwright() as p:
        # 建议使用 Chromium，内存占用比 Firefox 低 30%
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()

        # 关键步骤：设置真实 User-Agent 和 Viewport
        await page.set_extra_http_headers({'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36...'
        })
        await page.set_viewport_size({"width": 1920, "height": 1080})

        # 智能等待策略：优先使用 networkidle
        await page.goto('https://movie.douban.com/top250', wait_until="networkidle")

        # 使用 CSS 选择器定位元素
        titles = await page.locator('span.title:nth-child(1)').all_text_contents()

        # 数据清洗小技巧：用列表推导式 +strip 快速处理
        clean_titles = [t.strip() for t in titles if t.strip()]
        print(clean_titles)

        await browser.close()

# 运行异步函数
asyncio.run(scrape_douban())

这个示例虽然简单，但已经包含几个关键技术点：

headless=True 表示无头模式，适合服务器环境
networkidle 会等待所有网络请求完成
locator API 比传统的 page.evaluate() 更易用

现代网页大量使用 Ajax 和 WebSocket，比如微博的滚动加载。处理这类场景需要事件监听：

async def scrape_infinite_scroll():
    async with async_playwright() as p:
        browser = await p.chromium.launch()
        page = await browser.new_page()

        await page.goto('https://weibo.com')

        # 核心技巧：监听 DOM 变化
        async def handle_new_items():
            items = await page.locator('.card-wrap').all()
            print(f"当前加载项数: {len(items)}")

        # 两种监听方式任选其一
        # 方案 A：定时检测
        while True:
            await handle_new_items()
            await page.evaluate('window.scrollBy(0, 500)')
            await asyncio.sleep(2)  # 适当控制频率

        # 方案 B：MutationObserver（更精确但复杂）# 代码略...

分享三个实战验证有效的策略：

请求头伪装
携带完整 Headers 链（Referer/Cookie/Accept）
随机生成 User-Agent 库：fake-useragent

行为模拟

# 人类输入模式
await page.type('#search', '关键词', delay=100)  # 100ms 间隔

# 随机鼠标移动
await page.mouse.move(random.randint(0, 1000),
    random.randint(0, 800)
)

代理轮换

browser = await p.chromium.launch(
    proxy={
        "server": "http://proxy.example.com:8080",
        "username": "user",
        "password": "pass"
    }
)

成熟的爬虫项目应该包含这些模块：

class WebScraper:
    def __init__(self):
        self.proxy_pool = []  # 代理池
        self.fail_count = 0   # 失败计数器

    async def scrape(self, url):
        try:
            async with async_playwright() as p:
                browser = await self._init_browser(p)
                page = await browser.new_page()

                # 核心爬取逻辑
                await self._apply_stealth(page)
                response = await page.goto(url)

                if response.status != 200:
                    raise Exception(f"Bad status: {response.status}")

                data = await self._parse(page)
                await self._save(data)

        except Exception as e:
            self._handle_error(e)
        finally:
            await browser.close()

    async def _init_browser(self, playwright):
        # 浏览器配置细节...
        pass

    async def _apply_stealth(self, page):
        # 反反爬措施...
        pass

内存泄漏：
定期重启浏览器实例（每 100 个请求）
用 await page.close() 显式关闭页面
验证码处理：
商业方案：打码平台（如超级鹰）
技术方案：Tesseract OCR（准确率约 60%）

数据存储优化：

# 使用生成器减少内存占用
async def stream_data():
    async for item in paginator:
        yield process_item(item)

# 分批写入文件
with open('output.json', 'a') as f:
    async for data in stream_data():
        f.write(json.dumps(data) + '\n')

在 4 核 8G 的服务器上测试：