电商新品上架：从零搭建自动化监控与数据采集系统

一、系统架构与工具选型

本系统将实现对新品链接的自动监控、价格与销量数据采集、以及异常波动提醒。整套方案基于Python，使用轻量级工具，无需复杂部署，运行在一台普通电脑或云服务器上即可。

1.1 核心组件清单

• 数据采集：Requests + BeautifulSoup4 (用于静态页面) / Selenium (用于动态加载页面)
• 定时任务：APScheduler
• 数据存储：SQLite3 (轻量，单文件数据库)
• 通知提醒：Server酱 (微信通知) / 钉钉机器人

1.2 环境准备

确保你的电脑已安装Python 3.8或以上版本。打开终端（Windows为CMD或PowerShell），执行以下命令安装所有依赖包：

``` pip install requests beautifulsoup4 apscheduler selenium ```

对于需要抓取JavaScript渲染页面的情况，还需下载对应浏览器的WebDriver。以Chrome为例：

1. 查看你的Chrome浏览器版本（在浏览器地址栏输入 chrome://version/）。
2. 访问 https://chromedriver.chromium.org/ 下载对应版本的ChromeDriver。
3. 将下载的chromedriver.exe文件放在一个固定路径（例如 C:\WebDriver\），并将此路径添加到系统的环境变量PATH中。

二、数据库设计与初始化

我们使用SQLite存储监控的商品信息及采集到的历史数据。创建一个名为product_monitor.db的数据库文件。

2.1 创建数据库表

新建一个Python脚本init_database.py，写入以下代码：

``` import sqlite3 conn = sqlite3.connect('product_monitor.db') cursor = conn.cursor() 创建商品监控列表 cursor.execute(''' CREATE TABLE IF NOT EXISTS products ( id INTEGER PRIMARY KEY AUTOINCREMENT, product_name TEXT NOT NULL, platform TEXT NOT NULL, -- 如：淘宝、京东、拼多多 product_url TEXT NOT NULL UNIQUE, selector_price TEXT NOT NULL, -- 价格元素的CSS选择器 selector_sales TEXT, -- 销量元素的CSS选择器（可选） is_active INTEGER DEFAULT 1 -- 1表示监控中，0表示已停止 ) ''') 创建价格历史记录表 cursor.execute(''' CREATE TABLE IF NOT EXISTS price_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, product_id INTEGER, price REAL NOT NULL, sales INTEGER, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (product_id) REFERENCES products (id) ) ''') conn.commit() conn.close() print("数据库表创建成功！") ```

运行此脚本完成初始化。

三、核心采集器实现

我们将编写两个采集函数：一个用于静态页面（Requests+BS4），一个用于动态页面（Selenium）。

3.1 静态页面采集函数

新建文件crawler_static.py：

``` import requests from bs4 import BeautifulSoup import time import random def fetch_static(product_url, selector_price, selector_sales=None, headers=None): """ 采集静态页面商品信息 :param product_url: 商品链接 :param selector_price: 价格CSS选择器 :param selector_sales: 销量CSS选择器（可选） :param headers: 请求头，模拟浏览器 :return: (price, sales) 元组，采集失败返回(None, None) """ if headers is None: headers = { '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' } try: 添加随机延迟，避免请求过快 time.sleep(random.uniform(1, 3)) response = requests.get(product_url, headers=headers, timeout=10) response.raise_for_status() response.encoding = response.apparent_encoding soup = BeautifulSoup(response.text, 'html.parser') 提取价格 price_element = soup.select_one(selector_price) if not price_element: return None, None price_text = price_element.get_text().strip() 简单清理价格文本，提取数字 import re price_num = re.search(r'(\d+\.?\d)', price_text.replace(',', '')) price = float(price_num.group(1)) if price_num else None 提取销量（如果提供了选择器） sales = None if selector_sales: sales_element = soup.select_one(selector_sales) if sales_element: sales_text = sales_element.get_text().strip() sales_num = re.search(r'(\d+)', sales_text.replace('人付款', '').replace('销量', '')) sales = int(sales_num.group(1)) if sales_num else None return price, sales except Exception as e: print(f"采集失败 {product_url}: {e}") return None, None ```

3.2 动态页面采集函数

新建文件crawler_dynamic.py：

``` from selenium import webdriver from selenium.webdriver.chrome.options import Options from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC import time def fetch_dynamic(product_url, selector_price, selector_sales=None): """ 采集动态加载页面商品信息 :param product_url: 商品链接 :param selector_price: 价格CSS选择器 :param selector_sales: 销量CSS选择器（可选） :return: (price, sales) 元组 """ 配置Chrome无头模式，不显示浏览器窗口 chrome_options = Options() chrome_options.add_argument('--headless') chrome_options.add_argument('--disable-gpu') chrome_options.add_argument('--no-sandbox') driver = None try: driver = webdriver.Chrome(options=chrome_options) driver.get(product_url) 显式等待关键元素加载，最多等10秒 wait = WebDriverWait(driver, 10) 等待价格元素出现 price_element = wait.until( EC.presence_of_element_located((By.CSS_SELECTOR, selector_price)) ) price_text = price_element.text.strip() import re price_num = re.search(r'(\d+\.?\d)', price_text.replace(',', '')) price = float(price_num.group(1)) if price_num else None 提取销量 sales = None if selector_sales: try: sales_element = driver.find_element(By.CSS_SELECTOR, selector_sales) sales_text = sales_element.text.strip() sales_num = re.search(r'(\d+)', sales_text) sales = int(sales_num.group(1)) if sales_num else None except: sales = None driver.quit() return price, sales except Exception as e: if driver: driver.quit() print(f"动态采集失败 {product_url}: {e}") return None, None ```

四、数据入库与任务调度

创建主调度脚本main_scheduler.py，它将定时执行采集任务并保存数据。

4.1 添加待监控商品

在运行调度前，需要将新品链接添加到监控列表。创建一个add_product.py脚本：

``` import sqlite3 def add_product_to_monitor(db_path, product_name, platform, product_url, selector_price, selector_sales=''): conn = sqlite3.connect(db_path) cursor = conn.cursor() try: cursor.execute(''' INSERT INTO products (product_name, platform, product_url, selector_price, selector_sales) VALUES (?, ?, ?, ?, ?) ''', (product_name, platform, product_url, selector_price, selector_sales)) conn.commit() print(f"成功添加商品：{product_name}") except sqlite3.IntegrityError: print("该商品链接已存在于监控列表中。") finally: conn.close() 示例：添加一个淘宝商品 if __name__ == '__main__': 你需要手动获取商品页面的CSS选择器，可以使用浏览器的开发者工具（F12）检查元素获取 add_product_to_monitor( db_path='product_monitor.db', product_name='示例商品名称', platform='淘宝', product_url='https://item.taobao.com/item.htm?id=xxxx', selector_price='.tb-rmb-num', 价格选择器示例，实际需根据页面调整 selector_sales='.tb-sell-counter' 销量选择器示例 ) ```

4.2 主调度与监控逻辑

以下是main_scheduler.py的完整内容：

``` import sqlite3 import time from apscheduler.schedulers.blocking import BlockingScheduler from crawler_static import fetch_static from crawler_dynamic import fetch_dynamic import logging 配置日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') def monitor_job(): """定时执行的任务：读取所有活跃商品，采集数据并入库""" conn = sqlite3.connect('product_monitor.db') cursor = conn.cursor() 获取所有监控中的商品 cursor.execute("SELECT id, product_name, product_url, selector_price, selector_sales, platform FROM products WHERE is_active = 1") products = cursor.fetchall() for prod in products: pid, name, url, sel_price, sel_sales, platform = prod 根据平台选择采集方式（示例逻辑，实际需根据网站特性调整） if platform in ['淘宝', '京东']: 这些网站通常动态加载较多，使用Selenium price, sales = fetch_dynamic(url, sel_price, sel_sales) else: 其他静态页面使用Requests price, sales = fetch_static(url, sel_price, sel_sales) if price is not None: 插入历史记录 cursor.execute(''' INSERT INTO price_history (product_id, price, sales) VALUES (?, ?, ?) ''', (pid, price, sales)) conn.commit() logging.info(f"采集成功 {name}: 价格={price}, 销量={sales}") 这里可以添加价格波动检查逻辑 check_price_alert(pid, price, name) else: logging.warning(f"采集失败 {name}") conn.close() def check_price_alert(product_id, current_price, product_name): """检查价格是否发生显著变化（示例：降价超过10%）""" conn = sqlite3.connect('product_monitor.db') cursor = conn.cursor() 获取上一次记录的价格 cursor.execute(''' SELECT price FROM price_history WHERE product_id = ? AND price IS NOT NULL ORDER BY timestamp DESC LIMIT 1,1 ''', (product_id,)) last_record = cursor.fetchone() if last_record: last_price = last_record[0] change_ratio = (current_price - last_price) / last_price 如果降价幅度超过10% if change_ratio <= -0.1: alert_msg = f"【价格提醒】{product_name} 价格从 {last_price} 降至 {current_price}，降幅 {abs(change_ratio100):.1f}%" send_notification(alert_msg) logging.info(alert_msg) conn.close() def send_notification(message): """发送通知（以Server酱为例）""" 你需要从Server酱官网（sct.ftqq.com）获取自己的SCKEY sckey = 'YOUR_SERVERCHAN_SCKEY_HERE' if sckey.startswith('YOUR_'): logging.warning("请配置Server酱的SCKEY以启用微信通知") return import requests url = f'https://sctapi.ftqq.com/{sckey}.send' data = { 'title': '电商新品监控提醒', 'desp': message } try: requests.post(url, data=data, timeout=5) except Exception as e: logging.error(f"发送通知失败: {e}") if __name__ == '__main__': scheduler = BlockingScheduler() 每30分钟执行一次监控任务 scheduler.add_job(monitor_job, 'interval', minutes=30, id='monitor_job') logging.info("电商新品监控系统已启动，每30分钟运行一次...") try: scheduler.start() except (KeyboardInterrupt, SystemExit): logging.info("监控系统已停止") ```

五、系统运行与数据查看

5.1 启动系统

在终端中，直接运行主调度脚本：

``` python main_scheduler.py ```

系统将开始每30分钟自动采集一次数据。你可以按Ctrl+C停止程序。

5.2 查询采集数据

创建一个简单的查询脚本check_data.py来验证结果：

``` import sqlite3 import pandas as pd conn = sqlite3.connect('product_monitor.db') 使用Pandas直接读取数据到DataFrame，方便查看 df_products = pd.read_sql_query("SELECT FROM products", conn) df_history = pd.read_sql_query("SELECT FROM price_history ORDER BY timestamp DESC LIMIT 20", conn) print("=== 监控商品列表 ===") print(df_products[['product_name', 'platform', 'product_url']].to_string()) print("\n=== 最近20条价格记录 ===") print(df_history.to_string()) conn.close() ```

5.3 关键CSS选择器获取方法

1. 打开电商新品页面，按F12打开开发者工具。
2. 点击左上角的箭头图标（或按Ctrl+Shift+C），然后将鼠标移动到商品价格上并点击。
3. 在右侧“元素”面板中，右键点击高亮的那行HTML代码，选择“复制” -> “复制selector”。
4. 将复制的内容粘贴到add_product.py的selector_price参数中。

对于销量等元素，重复上述步骤即可。

六、后续优化方向

• 多平台适配：为不同电商平台编写专门的解析模块，提高采集稳定性。
• 数据可视化：使用Matplotlib或Pyecharts生成价格走势图。
• 部署到服务器：使用systemd（Linux）或nssm（Windows）将Python脚本设置为后台服务，实现24小时无人值守运行。
• 异常重试机制：在网络请求失败时自动重试，并记录失败日志。

至此，一个完整的电商新品自动化监控与数据采集系统已搭建完成。所有代码均可直接复制使用，你只需替换商品链接和对应的CSS选择器即可开始监控自己的新品。