静态规则解析与动态行为分析结合的混合抽取框架

原创

jackcode

发布于 2025-11-05 12:00:00

590

——一次关于网页“行为语言”的深度调查

一、当规则不再可靠

在早期的网络世界，数据采集就像一个懂语法的阅读者。它根据固定规则（XPath、CSS Selector）解析网页，就能拿到想要的数据。可现在的网页已经变得更聪明——它们不再直接把内容写在HTML里，而是通过JavaScript渲染、懒加载、滚动触发等方式“临场发挥”。

于是问题来了：以前那些一眼就能看到的数据，现在被藏在脚本、接口和用户行为后面。静态规则变得越来越无力。

要想重新“看懂”网页，我们得学会两种语言：

一是结构语言——HTML的层次与标签规则；

二是行为语言——浏览器执行、脚本调用和接口触发的过程。

把这两者结合在一起，才算是真正意义上的“混合抽取框架”。这套方法既能快速匹配结构规律，又能模拟用户行为捕获真实数据，就像一个懂得“读心术”的侦探。

二、三个“隐藏层”的真相

在实际项目中，网页数据往往被藏在不同层级的“迷雾”之下。我把它们分成三类：

第一层：结构隐蔽。

内容确实在网页里，但被埋在复杂的标签、iframe或异步加载片段中。你能看到，但XPath找不到。

第二层：逻辑隐蔽。

某些字段看似明明白白，其实是由JavaScript动态拼出来的，比如价格被加密成一串看不懂的数字。

第三层：传输隐蔽。

真正的数据藏在XHR或fetch请求中，只有模拟真实操作（点击、滚动、延时）才能触发它出现。

为了应对这些情况，混合抽取框架通常分成两大模块：

静态层：用 requests + lxml 抓取能直接看到的内容。
动态层：用 Playwright 模拟浏览器行为，还原网页运行时的状态。

两者协同工作，就像两个调查员——一个分析现场痕迹，另一个重演案发过程。

三、代码实战：静态+动态的混合采集

下面是一段可以运行的混合抽取示例代码，用来采集新闻网站的标题、作者和发布时间。

在这个例子中，我分别使用 requests（静态层）和 Playwright（动态层），并接入了爬虫代理服务来提高访问稳定性。

import asyncio
import requests
from lxml import etree
from playwright.async_api import async_playwright

# ========= 代理配置（亿牛云示例） =========
proxy_host = "proxy.16yun.cn"    # 代理域名
proxy_port = "3100"                 # 代理端口
proxy_user = "16YUN"         # 代理用户名
proxy_pass = "16IP"         # 代理密码

proxy_meta = f"http://{proxy_user}:{proxy_pass}@{proxy_host}:{proxy_port}"

headers = {
    "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
                  "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0 Safari/537.36"
}

# ========= 静态层：结构化规则提取 =========
def static_extract(url):
    proxies = {"http": proxy_meta, "https": proxy_meta}
    response = requests.get(url, headers=headers, proxies=proxies, timeout=10)
    tree = etree.HTML(response.text)

    titles = tree.xpath('//h2/a/text()')
    links = tree.xpath('//h2/a/@href')

    data = [{"title": t.strip(), "link": l} for t, l in zip(titles, links)]
    print(f"[静态层] 抽取到 {len(data)} 条线索")
    return data

# ========= 动态层：模拟网页行为 =========
async def dynamic_extract(urls):
    results = []
    proxy_server = f"http://{proxy_host}:{proxy_port}"

    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        context = await browser.new_context(
            proxy={"server": proxy_server, "username": proxy_user, "password": proxy_pass}
        )

        for url in urls:
            page = await context.new_page()
            await page.goto(url)
            await page.wait_for_timeout(3000)  # 等待页面渲染
            try:
                title = await page.text_content("//h1")
                author = await page.text_content("//span[@class='author']")
                date = await page.text_content("//time")
                results.append({
                    "title": title.strip() if title else None,
                    "author": author.strip() if author else None,
                    "date": date.strip() if date else None,
                    "url": url
                })
            except Exception as e:
                print(f"[动态层] 解析失败: {url}, 原因: {e}")

        await browser.close()
    print(f"[动态层] 捕获 {len(results)} 条完整数据")
    return results

# ========= 数据融合 =========
def merge_results(static_data, dynamic_data):
    merged = []
    for s in static_data:
        d = next((x for x in dynamic_data if x["url"] == s["link"]), None)
        merged.append({
            "title": s["title"],
            "url": s["link"],
            "author": d["author"] if d else None,
            "date": d["date"] if d else None
        })
    return merged

# ========= 主流程 =========
async def main():
    url = "https://www.yicai.com/news/"
    static_data = static_extract(url)
    dynamic_data = await dynamic_extract([item["link"] for item in static_data[:5]])
    merged = merge_results(static_data, dynamic_data)

    for item in merged:
        print(item)

if __name__ == "__main__":
    asyncio.run(main())

这段代码其实就是一个最小可行版本的混合框架：

requests 负责快速采样网页结构，Playwright 则补齐动态内容。

在真实项目中，你还可以加入数据缓存、队列、日志、异常重试等模块，逐步扩展为生产级框架。