多模态AI爬虫：文本+图像智能抓取实战

原创

jackcode

发布于 2025-06-12 03:57:36

13500

代码可运行

文章被收录于专栏：爬虫资料爬虫资料

运行总次数：0

代码可运行

一个AI模型训练任务引发的爬虫危机

上个月，公司AI组向我们数据组提出一个“看似简单”的需求：训练一个能识别商品种类的多模态模型，数据来源不限，但要求包含图像 + 商品文本 + 价格 + 折扣信息。初看似乎没问题，直到他们指明：

“先抓一批亚马逊上关键词为‘laptop’的商品吧，图片和描述要配套的，价格、库存、折扣最好也能标注。”

我心里一沉——亚马逊，图文混合，反爬一流，这可不是普通爬虫能搞定的活。

你以为的请求头伪装，根本不够

第一轮尝试，我们写了个基础版脚本，配了user-agent，伪装了headers，访问频率也控制得不错。

然而爬了几分钟后，页面直接跳转验证码，甚至被302重定向回首页。我们意识到：

IP频率检测非常敏感
navigator.webdriver 检测等自动化识别也在阻挡
图片链接多为懒加载或动态构造

于是我们准备全面升级：使用代理IP + 用户模拟持久化 + 多模态解析的组合拳。

多模态AI爬虫的架构搭建与实战代码

我们拆解了需求，决定使用 requests + lxml + Pillow，结合爬虫代理IP服务，实现一个具备图像和文本提取能力的智能爬虫。

核心技术点：

代理IP绕过限制（爬虫代理接入）
模拟用户防反爬识别
图片下载与保存
结构化存储：根据关键词分目录保存JSON和图片

代码实现如下（关键词以 `laptop` 为例）：

import requests
from lxml import html
import os
import json
from urllib.parse import quote
from PIL import Image
from io import BytesIO

# 设置关键词
keyword = "laptop"
search_url = f"https://www.amazon.com/s?k={quote(keyword)}"

# 设置爬虫代理（参考亿牛云爬虫加强版示例）
proxies = {
    "http": "http://用户名:密码@代理域名:端口",
    "https": "http://用户名:密码@代理域名:端口"
}

# 自定义请求头
headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
    "Accept-Language": "en-US,en;q=0.9",
    "Cookie": "你的cookie字符串"  # 可通过浏览器调试获取登录后的cookie
}

# 创建数据保存目录
os.makedirs(f"data/{keyword}/images", exist_ok=True)

# 发送请求获取页面
response = requests.get(search_url, headers=headers, proxies=proxies, timeout=15)
tree = html.fromstring(response.content)

# 解析商品列表（取前10个）
products = tree.xpath('//div[contains(@class,"s-result-item") and @data-asin]')
results = []

for idx, product in enumerate(products[:10]):
    try:
        asin = product.get("data-asin")
        title = product.xpath('.//h2//span/text()')
        price_whole = product.xpath('.//span[@class="a-price-whole"]/text()')
        price_frac = product.xpath('.//span[@class="a-price-fraction"]/text()')
        img_url = product.xpath('.//img[@class="s-image"]/@src')

        # 构造完整价格
        price = f"{price_whole[0]}.{price_frac[0]}" if price_whole and price_frac else "N/A"

        # 构造商品描述
        title = title[0].strip() if title else "N/A"

        # 图片下载与保存
        if img_url:
            img_response = requests.get(img_url[0], headers=headers, proxies=proxies)
            img = Image.open(BytesIO(img_response.content))
            img_path = f"data/{keyword}/images/{asin}.jpg"
            img.save(img_path)
        else:
            img_path = "N/A"

        # 构造数据结构
        item = {
            "asin": asin,
            "title": title,
            "price": price,
            "image_path": img_path,
            "keyword": keyword
        }

        results.append(item)

    except Exception as e:
        print(f"商品索引 {idx} 解析失败：{e}")
        continue

# 保存为JSON
with open(f"data/{keyword}/products.json", "w", encoding="utf-8") as f:
    json.dump(results, f, indent=2, ensure_ascii=False)

print(f"关键词 [{keyword}] 商品信息采集完成，共采集到 {len(results)} 条。")