电商爬虫实战：拼多多商品价格监控系统搭建指南

python1headers = {
2    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
3    'Referer': 'https://pinduoduo.com/',
4    'Accept-Language': 'zh-CN,zh;q=0.9'
5}

python1from playwright.sync_api import sync_playwright
2
3def get_price_with_playwright(url):
4    with sync_playwright() as p:
5        browser = p.chromium.launch(headless=True)
6        page = browser.new_page()
7        page.goto(url, wait_until="networkidle")
8        
9        # 等待价格元素加载
10        price_element = page.locator('div.price-container >> nth=0')
11        price = price_element.inner_text()
12        
13        browser.close()
14        return price

python1def get_search_results(keyword):
2    url = f"https://mobile.yangkeduo.com/search_result.html?search_key={keyword}"
3    response = requests.get(url, headers=headers)
4    # 解析JSON数据提取商品ID
5    # 实际需分析接口返回结构
6    return goods_ids

python1def parse_goods_page(goods_id):
2    url = f"https://mobile.yangkeduo.com/goods.html?goods_id={goods_id}"
3    
4    # 使用Playwright获取动态内容
5    with sync_playwright() as p:
6        browser = p.chromium.launch(headless=True)
7        page = browser.new_page()
8        page.goto(url)
9        
10        # 提取关键数据
11        data = {
12            'title': page.title(),
13            'price': page.locator('span.price').inner_text(),
14            'stock': page.locator('div.stock').inner_text(),
15            'sales': page.locator('div.sales').inner_text()
16        }
17        browser.close()
18        return data

python1import pymysql
2
3conn = pymysql.connect(
4    host='localhost',
5    user='root',
6    password='password',
7    database='pinduoduo'
8)
9
10def save_to_mysql(data):
11    cursor = conn.cursor()
12    sql = """
13    INSERT INTO goods_price 
14    (goods_id, title, price, stock, sales, create_time)
15    VALUES (%s, %s, %s, %s, %s, NOW())
16    """
17    cursor.execute(sql, (data['goods_id'], data['title'], data['price'], data['stock'], data['sales']))
18    conn.commit()

python1from pymongo import MongoClient
2
3client = MongoClient('mongodb://localhost:27017/')
4db = client['pinduoduo']
5collection = db['goods_price']
6
7def save_to_mongo(data):
8    collection.insert_one(data)

python1from apscheduler.schedulers.blocking import BlockingScheduler
2
3def job_function():
4    goods_ids = get_search_results('iPhone13')
5    for goods_id in goods_ids[:5]:  # 只监控前5个商品
6        data = parse_goods_page(goods_id)
7        save_to_mongo(data)
8        print(f"已采集 {data['title']} 价格：{data['price']}")
9
10scheduler = BlockingScheduler()
11scheduler.add_job(job_function, 'interval', minutes=30)
12scheduler.start()

python1from pyecharts.charts import Line
2from pyecharts import options as opts
3
4def generate_price_chart(goods_id):
5    # 从数据库查询历史数据
6    prices = [...]  # 实际应从数据库获取
7    dates = [...]
8    
9    line = (
10        Line()
11        .add_xaxis(dates)
12        .add_yaxis("价格趋势", prices)
13        .set_global_opts(
14            title_opts=opts.TitleOpts(title=f"商品ID:{goods_id} 价格走势"),
15            tooltip_opts=opts.TooltipOpts(trigger="axis"),
16            yaxis_opts=opts.AxisOpts(name="价格（元）")
17        )
18    )
19    line.render("price_trend.html")

python1import random
2
3PROXY_POOL = [
4    "http://1.1.1.1:8080",
5    "http://2.2.2.2:8081",
6    # 更多代理...
7]
8
9def get_random_proxy():
10    return {"http": random.choice(PROXY_POOL)}

python1import requests
2
3def solve_captcha(image_bytes):
4    url = "http://api.dama2.com/solve"
5    params = {
6        'username': 'your_username',
7        'password': 'your_password',
8        'type': '1004'  # 滑块验证码类型
9    }
10    files = {'image': image_bytes}
11    response = requests.post(url, params=params, files=files)
12    return response.json().get('result')