cfc/tools/nutrition-crawler/crawler.py

#!/usr/bin/env python3
"""营养产品爬虫（试点）— 抓取京东/Amazon 商品数据，输出 JSON 中间文件。

用法:
    python crawler.py --platform jd --query "益生菌"
    python crawler.py --platform amazon --query "probiotic"
    python crawler.py --seed              # 按 seed_products.json 抓取
"""
import argparse
import json
import os
import random
import time
from datetime import datetime

import requests
from bs4 import BeautifulSoup

OUTPUT_DIR = os.path.join(os.path.dirname(__file__), "output")
HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
                  "(KHTML, like Gecko) Chrome/120.0 Safari/537.36",
    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
}


def ensure_output_dir():
    os.makedirs(OUTPUT_DIR, exist_ok=True)


def fetch(url, headers=None, timeout=15):
    """带重试和随机延迟的请求"""
    h = {**HEADERS, **(headers or {})}
    for attempt in range(3):
        try:
            resp = requests.get(url, headers=h, timeout=timeout)
            resp.raise_for_status()
            time.sleep(random.uniform(1.0, 2.5))
            return resp
        except requests.RequestException as e:
            print(f"[{attempt+1}/3] 请求失败 {url}: {e}")
            time.sleep(2 * (attempt + 1))
    return None


def crawl_jd(query):
    """京东搜索抓取（返回商品名称/价格/链接）"""
    results = []
    url = f"https://search.jd.com/Search?keyword={query}&enc=utf-8"
    resp = fetch(url)
    if not resp:
        return results
    soup = BeautifulSoup(resp.text, "html.parser")
    items = soup.select(".gl-item")
    for item in items[:10]:
        try:
            name_el = item.select_one(".p-name em")
            price_el = item.select_one(".p-price strong i")
            link_el = item.select_one(".p-img a")
            name = name_el.get_text(strip=True) if name_el else None
            price_str = price_el.get_text(strip=True) if price_el else None
            link = "https:" + link_el["href"] if link_el and link_el.get("href") else None
            results.append({
                "platform": "京东",
                "query": query,
                "name": name,
                "price": float(price_str) if price_str else None,
                "url": link,
                "crawled_at": datetime.now().isoformat(),
            })
        except Exception as e:
            print(f"解析商品异常: {e}")
    return results


def crawl_amazon(query):
    """Amazon 搜索（中文/英文）抓取"""
    results = []
    url = f"https://www.amazon.com/s?k={query}"
    resp = fetch(url, headers={"User-Agent": HEADERS["User-Agent"]})
    if not resp:
        return results
    soup = BeautifulSoup(resp.text, "html.parser")
    items = soup.select('[data-component-type="s-search-result"]')
    for item in items[:10]:
        try:
            name_el = item.select_one("h2 span")
            price_whole = item.select_one(".a-price-whole")
            link_el = item.select_one("h2 a")
            name = name_el.get_text(strip=True) if name_el else None
            price_str = price_whole.get_text(strip=True) if price_whole else None
            link = "https://www.amazon.com" + link_el["href"] if link_el and link_el.get("href") else None
            results.append({
                "platform": "Amazon",
                "query": query,
                "name": name,
                "price": float(price_str.replace(",", "")) if price_str else None,
                "url": link,
                "crawled_at": datetime.now().isoformat(),
            })
        except Exception as e:
            print(f"解析商品异常: {e}")
    return results


def save(platform, data):
    ensure_output_dir()
    path = os.path.join(OUTPUT_DIR, f"{platform}.json")
    with open(path, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)
    print(f"已保存 {len(data)} 条到 {path}")


def main():
    parser = argparse.ArgumentParser(description="营养产品爬虫")
    parser.add_argument("--platform", choices=["jd", "amazon"], help="平台")
    parser.add_argument("--query", help="搜索关键词")
    parser.add_argument("--seed", action="store_true", help="按种子清单抓取")
    args = parser.parse_args()

    if args.seed:
        seed_path = os.path.join(os.path.dirname(__file__), "seed_products.json")
        with open(seed_path, encoding="utf-8") as f:
            seeds = json.load(f)
        all_jd, all_amazon = [], []
        for seed in seeds:
            print(f"抓取: {seed['name']}")
            all_jd.extend(crawl_jd(seed["keywords"]))
            all_amazon.extend(crawl_amazon(seed["keywords"]))
            time.sleep(random.uniform(2, 4))
        save("jd", all_jd)
        save("amazon", all_amazon)
    elif args.platform and args.query:
        if args.platform == "jd":
            save("jd", crawl_jd(args.query))
        elif args.platform == "amazon":
            save("amazon", crawl_amazon(args.query))
    else:
        parser.print_help()


if __name__ == "__main__":
    main()