实时监控 1688 商品价格变化的爬虫系统实现

原创

小白学大数据

发布于 2025-12-15 16:47:55

5900

在电商运营、市场调研以及个人网购决策中，商品价格的实时监控具有重要的价值。1688 作为国内头部的批发电商平台，其商品价格的波动直接反映了供应链、市场需求的变化。本文将详细介绍如何搭建一套实时监控 1688 商品价格变化的爬虫系统，从技术选型、核心逻辑实现到数据存储与告警机制，完整呈现系统的构建过程。一、技术选型与系统架构（一）技术栈选择爬虫系统的核心需求是页面解析、定时任务执行、数据存储和价格变化告警，结合 1688 的页面特性，我们选择以下技术栈：编程语言：Python。Python 拥有丰富的爬虫库（如 Requests、BeautifulSoup、Selenium）和数据处理库，开发效率高。请求库：Requests（处理常规接口请求）+ Selenium（处理动态渲染页面）。1688 部分商品页面采用 JavaScript 动态渲染，单纯的 Requests 无法获取完整数据，需结合 Selenium 模拟浏览器行为。解析库：BeautifulSoup4。用于解析 HTML 页面，提取商品名称、价格、规格等核心信息。定时任务：APScheduler。实现定时爬取商品价格的功能，支持灵活的时间配置（如每隔 1 小时爬取一次）。数据存储：SQLite。轻量级关系型数据库，无需额外部署，适合小型监控系统；若需扩容，可无缝迁移至 MySQL。告警机制：SMTP 协议。当价格发生变化时，通过邮件发送告警信息。（二）系统架构系统分为四个核心模块：爬虫模块：负责向 1688 发送请求，获取商品页面并解析关键数据。定时任务模块：触发爬虫模块按指定频率执行爬取操作。数据存储模块：存储商品的历史价格数据和基础信息。告警模块：对比当前价格与历史价格，若发生变化则触发告警。二、环境搭建（一）数据库表设计我们创建两张表：goods_info（存储商品基础信息）和price_record（存储商品价格记录）。python运行

import sqlite3

# 初始化数据库
def init_db():
    conn = sqlite3.connect('1688_price_monitor.db')
    cursor = conn.cursor()
    # 创建商品信息表
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS goods_info (
        goods_id TEXT PRIMARY KEY,
        goods_name TEXT,
        goods_url TEXT UNIQUE,
        create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
    ''')
    # 创建价格记录表
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS price_record (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        goods_id TEXT,
        price FLOAT,
        crawl_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
        FOREIGN KEY (goods_id) REFERENCES goods_info (goods_id)
    )
    ''')
    conn.commit()
    conn.close()

if __name__ == '__main__':
    init_db()

（二）爬虫模块实现1688 商品链接格式通常为https://detail.1688.com/offer/[商品ID].html，我们需要解析页面中的商品名称、价格信息。对于动态渲染的页面，使用 Selenium 加载页面后再解析。python运行

import requests
from bs4 import BeautifulSoup
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
import time

# 配置Chrome无头模式，避免弹出浏览器窗口
def get_chrome_driver():
    chrome_options = Options()
    chrome_options.add_argument('--headless')
    chrome_options.add_argument('--no-sandbox')
    chrome_options.add_argument('--disable-dev-shm-usage')
    driver = webdriver.Chrome(options=chrome_options)
    return driver

# 解析商品信息
def parse_goods_info(goods_url):
    try:
        # 初始化驱动
        driver = get_chrome_driver()
        driver.get(goods_url)
        # 等待页面加载完成
        time.sleep(3)
        page_source = driver.page_source
        soup = BeautifulSoup(page_source, 'html.parser')
        
        # 提取商品ID（从URL中截取）
        goods_id = goods_url.split('/')[-1].replace('.html', '')
        
        # 提取商品名称（不同页面可能有不同的class，需根据实际情况调整）
        goods_name = soup.find('h1', class_='title').get_text().strip() if soup.find('h1', class_='title') else '未知名称'
        
        # 提取商品价格（1688价格可能存在多个规格，此处提取最低价格）
        price_tag = soup.find('span', class_='price')
        price = float(price_tag.get_text().strip()) if price_tag else 0.0
        
        driver.quit()
        return {
            'goods_id': goods_id,
            'goods_name': goods_name,
            'goods_url': goods_url,
            'price': price
        }
    except Exception as e:
        print(f'解析商品信息失败：{e}')
        return None

（三）数据存储与价格对比将爬取的商品信息存入数据库，并对比历史价格，判断是否发生变化。python运行

import sqlite3
from datetime import datetime

# 插入商品基础信息
def insert_goods_info(goods_data):
    conn = sqlite3.connect('1688_price_monitor.db')
    cursor = conn.cursor()
    try:
        cursor.execute('''
        INSERT OR IGNORE INTO goods_info (goods_id, goods_name, goods_url)
        VALUES (?, ?, ?)
        ''', (goods_data['goods_id'], goods_data['goods_name'], goods_data['goods_url']))
        conn.commit()
    except Exception as e:
        print(f'插入商品信息失败：{e}')
    finally:
        conn.close()

# 插入价格记录并对比历史价格
def insert_price_record(goods_data):
    conn = sqlite3.connect('1688_price_monitor.db')
    cursor = conn.cursor()
    try:
        # 获取最新的历史价格
        cursor.execute('''
        SELECT price FROM price_record WHERE goods_id = ? ORDER BY crawl_time DESC LIMIT 1
        ''', (goods_data['goods_id'],))
        history_price = cursor.fetchone()
        
        # 插入新价格记录
        cursor.execute('''
        INSERT INTO price_record (goods_id, price)
        VALUES (?, ?)
        ''', (goods_data['goods_id'], goods_data['price']))
        conn.commit()
        
        # 判断价格是否变化
        if history_price is None:
            # 首次爬取，无历史价格
            return (False, 0.0, goods_data['price'])
        else:
            history_price = history_price[0]
            if abs(goods_data['price'] - history_price) > 0.01:  # 价格变化阈值为0.01元
                return (True, history_price, goods_data['price'])
            else:
                return (False, history_price, goods_data['price'])
    except Exception as e:
        print(f'插入价格记录失败：{e}')
        return (False, 0.0, 0.0)
    finally:
        conn.close()

（四）告警模块实现当价格发生变化时，通过邮件发送告警信息。python运行

import sqlite3
from datetime import datetime

# 插入商品基础信息
def insert_goods_info(goods_data):
    conn = sqlite3.connect('1688_price_monitor.db')
    cursor = conn.cursor()
    try:
        cursor.execute('''
        INSERT OR IGNORE INTO goods_info (goods_id, goods_name, goods_url)
        VALUES (?, ?, ?)
        ''', (goods_data['goods_id'], goods_data['goods_name'], goods_data['goods_url']))
        conn.commit()
    except Exception as e:
        print(f'插入商品信息失败：{e}')
    finally:
        conn.close()

# 插入价格记录并对比历史价格
def insert_price_record(goods_data):
    conn = sqlite3.connect('1688_price_monitor.db')
    cursor = conn.cursor()
    try:
        # 获取最新的历史价格
        cursor.execute('''
        SELECT price FROM price_record WHERE goods_id = ? ORDER BY crawl_time DESC LIMIT 1
        ''', (goods_data['goods_id'],))
        history_price = cursor.fetchone()
        
        # 插入新价格记录
        cursor.execute('''
        INSERT INTO price_record (goods_id, price)
        VALUES (?, ?)
        ''', (goods_data['goods_id'], goods_data['price']))
        conn.commit()
        
        # 判断价格是否变化
        if history_price is None:
            # 首次爬取，无历史价格
            return (False, 0.0, goods_data['price'])
        else:
            history_price = history_price[0]
            if abs(goods_data['price'] - history_price) > 0.01:  # 价格变化阈值为0.01元
                return (True, history_price, goods_data['price'])
            else:
                return (False, history_price, goods_data['price'])
    except Exception as e:
        print(f'插入价格记录失败：{e}')
        return (False, 0.0, 0.0)
    finally:
        conn.close()

（五）定时任务模块使用 APScheduler 实现定时爬取，例如每隔 1 小时爬取一次指定商品。python运行

Python


from apscheduler.schedulers.blocking import BlockingScheduler
# 核心监控任务
def monitor_task(goods_url):
 # 1. 爬取商品信息
 goods_data = parse_goods_info(goods_url)
 if not goods_data:
 return
 # 2. 插入商品基础信息
 insert_goods_info(goods_data)
 # 3. 插入价格记录并对比
 price_changed, history_price, current_price = insert_price_record(goods_data)
 # 4. 价格变化则发送告警
 if price_changed:
 send_price_alert(goods_data, history_price, current_price)
 else:
 print(f'商品{goods_data["goods_name"]}价格未变化，当前价格：{current_price}元')
if __name__ == '__main__':
 # 初始化数据库
 init_db()
 # 待监控的商品链接
 target_goods_url = 'https://detail.1688.com/offer/123456789.html' # 替换为实际商品链接
 # 创建调度器
 scheduler = BlockingScheduler()
 # 添加定时任务，每隔1小时执行一次（cron表达式：0 */1 * * *）
 scheduler.add_job(monitor_task, 'cron', hour='*/1', args=[target_goods_url])
 print('1688价格监控系统已启动，每隔1小时执行一次爬取任务...')
 try:
 scheduler.start()
 except KeyboardInterrupt:
 print('系统已停止')

四、系统优化与注意事项（一）反爬机制应对1688 具有反爬机制，直接频繁爬取可能会被封 IP，因此需要做以下优化：添加请求头：在 Requests 或 Selenium 中添加 User-Agent、Referer 等请求头，模拟浏览器请求。设置随机延迟：爬取任务之间设置随机的时间间隔，避免固定频率请求。使用代理 IP：对于大规模监控，可使用代理 IP 池轮换 IP，降低被封风险。推荐使用亿牛云代理限制爬取频率：根据 1688 的 robots 协议，合理设置爬取频率，避免对服务器造成压力。（二）页面解析适配1688 的页面结构可能会更新，导致解析失败。因此，在代码中需要增加异常处理，并定期检查页面结构的变化，及时调整解析规则。（三）数据扩容当监控的商品数量增多时，SQLite 的性能可能不足，可将数据库迁移至 MySQL，并添加索引优化查询速度。同时，可引入 Redis 缓存常用的商品信息，减少数据库访问压力。五、总结本文搭建的 1688 商品价格监控系统，通过 Python 实现了爬虫、数据存储、定时任务和告警的全流程功能。该系统可满足个人或小型团队的价格监控需求，通过简单的扩展（如增加多商品监控、可视化数据展示），还能适配更复杂的场景。在实际使用中，需注意遵守网站的 robots 协议和相关法律法规，避免恶意爬取行为。同时，针对反爬机制和页面结构变化的问题，需要持续优化代码，确保系统的稳定性和可用性。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python-3.4