1. 项目概述为什么需要一个“价格监控眼”作为一名经常在网上“剁手”的程序员我发现自己和身边的朋友们都有一个共同的痛点看中的商品价格总是在波动今天一个价明天一个价永远不知道什么时候才是真正的“好价”。手动去刷新页面不仅效率低下还容易错过最佳时机。尤其是在像京东这样的大型电商平台促销活动频繁价格变化更是瞬息万变。于是一个想法自然而然地诞生了能不能写一个程序让它像一只不知疲倦的眼睛24小时帮我盯着心仪商品的价格一旦降到我的心理价位就立刻通知我这就是“京东商品价格监控眼”项目的初衷。它不仅仅是一个简单的Python爬虫练习更是一个能解决实际生活问题的自动化工具。通过这个项目你将学会如何从零开始构建一个结构完整、运行稳定、具有一定健壮性的爬虫应用。我们会从最基础的网页请求开始一步步解析复杂的页面结构处理可能遇到的反爬机制并将获取到的数据存储下来最终实现定时监控和消息通知。整个过程我会手把手带你走一遍并提供完整的源码供你参考和修改。无论你是刚入门Python的新手想找一个有成就感的实战项目还是有一定基础想深化爬虫技能的朋友这个项目都能让你收获满满。2. 核心思路与技术选型如何构建一个稳健的监控系统在动手写代码之前我们先来梳理一下整个系统的核心思路。一个完整的监控系统绝不仅仅是“抓取网页-解析数据”这么简单我们需要考虑其可持续性、稳定性和易用性。2.1 系统工作流程设计整个监控眼的工作流程可以抽象为以下几个核心环节目标输入用户提供想要监控的商品链接URL。数据抓取程序模拟浏览器访问该链接获取网页的HTML源代码。数据解析从复杂的HTML代码中精准地提取出商品名称、当前价格等关键信息。数据存储将提取到的信息商品名、价格、抓取时间保存下来以便后续分析和对比。逻辑判断将当前价格与用户设定的目标价格或历史价格进行对比。消息通知如果当前价格满足条件如低于目标价则通过预设的渠道如邮件、微信发送提醒。定时任务让以上所有步骤能够自动地、周期性地执行。2.2 关键技术栈选型与理由为什么选择这些工具每一个选择背后都有其考量。请求库requestshttpx(备用)requests库是Python界进行HTTP请求的“事实标准”其API设计优雅文档丰富社区支持极好对于京东这样的常规页面请求完全够用。我们将主要使用它。但考虑到部分复杂的动态页面或未来可能遇到的反爬升级我也会简单介绍httpx这个支持HTTP/2和异步请求的后起之秀作为备选方案它对于处理一些需要更高效并发或更现代协议的场景有优势。解析库BeautifulSoup4lxml解析器从HTML中提取数据BeautifulSoup4(简称bs4) 是不二之选。它提供了非常直观的、类似DOM操作的API即使你不熟悉复杂的XPath也能通过标签名、属性、CSS选择器轻松定位元素。搭配lxml作为解析引擎速度比Python内置的html.parser快很多稳定性也更好。数据存储SQLite3(内置)对于个人使用或轻量级监控我们不需要启动一个庞大的MySQL或PostgreSQL数据库。Python标准库自带的SQLite3完美契合需求——它是一个轻量级的、服务器进程的数据库整个数据库就是一个文件无需安装和配置非常适合本项目。我们将用它来创建商品信息表和价格历史记录表。定时任务schedule库我们需要程序能每隔一段时间比如每30分钟自动运行一次。schedule库提供了非常人性化的API可以用接近自然语言的方式设置定时任务如schedule.every(30).minutes.do(job)比直接使用复杂的cron或apscheduler对于新手来说更友好。消息通知smtplib(邮件) 与requests(Server酱等微信通知)价格提醒需要触达用户。最通用和可靠的方式是邮件Python内置的smtplib库可以轻松实现。对于更即时的提醒我们可以集成像“Server酱”一个将服务器消息推送到微信的工具这样的第三方服务本质上也是通过requests库发送一个HTTP请求到其API。注意在开始任何爬虫项目前务必、务必、务必查看目标网站的robots.txt文件通常在网站根目录如https://www.jd.com/robots.txt。这个文件规定了网络爬虫哪些页面可以抓取哪些不可以。尊重robots.txt是爬虫开发者的基本素养也是对网站资源的尊重。我们的监控目标仅限于具体的商品详情页通常这类页面是对搜索引擎开放的但依然建议控制请求频率避免对服务器造成过大压力。3. 环境准备与核心库安装工欲善其事必先利其器。让我们先把开发环境搭建起来。3.1 Python环境确认与虚拟环境创建首先确保你的电脑上安装了Python建议版本3.7及以上。打开终端Windows是CMD或PowerShellMac/Linux是Terminal输入python --version或python3 --version查看版本。强烈建议为这个项目创建一个独立的虚拟环境这样可以避免不同项目间的库版本冲突。# 安装虚拟环境管理工具如果尚未安装 pip install virtualenv # 创建一个名为 jd_monitor 的虚拟环境 virtualenv jd_monitor # 激活虚拟环境 # 在 Windows 上 jd_monitor\Scripts\activate # 在 Mac/Linux 上 source jd_monitor/bin/activate激活后你的命令行提示符前面应该会出现(jd_monitor)字样表示你已经在这个虚拟环境中了。3.2 安装所需第三方库在激活的虚拟环境中一次性安装我们需要的所有库pip install requests beautifulsoup4 lxml schedulerequests: 用于发送HTTP请求。beautifulsoup4: 用于解析HTML。lxml: 作为bs4的解析引擎需要单独安装。schedule: 用于管理定时任务。安装完成后可以通过pip list命令查看已安装的包确认它们都在列表中。3.3 项目目录结构规划一个好的项目结构能让代码更清晰后期维护也更方便。我们先创建好目录和文件jd_price_monitor/ ├── config.py # 配置文件存放数据库路径、邮件设置、监控间隔等 ├── database.py # 数据库操作相关函数创建表、插入数据、查询数据 ├── fetcher.py # 网页抓取和解析的核心函数 ├── notifier.py # 通知发送功能邮件、微信等 ├── monitor.py # 主程序串联整个流程包含定时任务逻辑 ├── requirements.txt # 项目依赖库列表 └── products.json # 可选存放待监控的商品链接列表你可以使用以下命令快速创建mkdir jd_price_monitor cd jd_price_monitor touch config.py database.py fetcher.py notifier.py monitor.py requirements.txt4. 网页抓取与解析攻克京东商品页这是整个项目的核心也是最容易出问题的一步。京东的页面结构可能会变动反爬策略也可能升级因此我们的代码需要有一定的容错和适应性。4.1 分析页面结构与定位价格元素首先我们需要手动打开一个京东商品页例如一个手机页面使用浏览器的“开发者工具”按F12来观察。找到价格元素在页面上右键点击商品价格选择“检查”。你会发现京东的商品价格通常在一个类名为price的标签内但为了更精确我们需要找到其唯一的ID或特定的CSS选择器路径。经过分析当前京东商品页面的当前价格通常位于一个类似#priceSale或.price.J-p-{商品ID}的标签内。请注意这个选择器可能会随着京东前端的改版而变化这是爬虫需要维护的主要原因之一。找到商品标题同样方法商品标题通常在一个ID为itemName或类名包含sku-name的h1标签内。4.2 编写抓取与解析函数 (fetcher.py)我们将抓取和解析的逻辑封装在一个函数里。# fetcher.py import requests from bs4 import BeautifulSoup import re import time def fetch_product_info(url, headersNone): 抓取指定URL的商品信息标题、价格。 参数: url (str): 商品详情页的URL。 headers (dict): 可选的请求头用于模拟浏览器。 返回: dict: 包含商品标题和价格的字典如果失败则返回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, Accept-Language: zh-CN,zh;q0.9, } try: # 1. 发送HTTP GET请求 response requests.get(url, headersheaders, timeout10) response.raise_for_status() # 如果状态码不是200抛出HTTPError异常 response.encoding response.apparent_encoding or utf-8 # 自动检测编码 # 2. 使用BeautifulSoup解析HTML soup BeautifulSoup(response.text, lxml) # 3. 提取商品标题 - 需要根据实际页面调整选择器 # 示例选择器可能需要调整 title_elem soup.select_one(.sku-name) or soup.select_one(#itemName) if title_elem: product_title title_elem.get_text(stripTrue) else: # 如果找不到尝试其他选择器或记录错误 product_title 未知商品 print(f警告在 {url} 中未找到商品标题。) # 4. 提取商品价格 - 这是最关键的也是最容易失效的部分 price None # 方法A尝试通过标签属性查找 price_elem soup.select_one(.price) if price_elem: price_text price_elem.get_text(stripTrue) else: # 方法B尝试在脚本数据中查找京东价格有时通过JS动态加载 script_tags soup.find_all(script) for script in script_tags: if script.string and price in script.string.lower(): # 使用正则表达式从JS代码中匹配价格数字 match re.search(rprice\s*:\s*[\]?([\d.]), script.string) if match: price_text match.group(1) break else: price_text # 清理价格文本提取纯数字 if price_text: # 匹配数字和可能的小数点 match re.search(r[\d,.], price_text.replace(,, )) if match: price float(match.group()) if price is None: print(f错误无法从 {url} 解析出价格。) return None return { title: product_title, price: price, url: url } except requests.exceptions.RequestException as e: print(f网络请求出错 ({url}): {e}) return None except Exception as e: print(f解析过程出错 ({url}): {e}) return None # 简单的测试代码 if __name__ __main__: # 替换成一个真实的京东商品URL进行测试 test_url https://item.jd.com/100012043978.html # 示例链接请勿频繁请求 info fetch_product_info(test_url) if info: print(f商品: {info[title]}) print(f价格: {info[price]}) else: print(抓取失败)实操心得京东页面的价格信息可能存在于多个地方直接的HTML标签、script标签内的JSON数据、甚至是异步接口返回的。上面的代码提供了两种查找策略。在实际运行中如果第一种方法失效第二种方法从JS中正则匹配往往能作为备用方案。最可靠的方法是直接分析其数据接口但这涉及到更复杂的逆向工程对于入门项目我们先从HTML解析入手。记得在测试时务必使用真实的商品URL并控制请求频率比如在测试代码中加上time.sleep(3)避免请求过快。4.3 处理动态加载与反爬策略现代网站大量使用JavaScript动态加载内容直接请求HTML可能拿不到价格。此时有几种策略寻找数据接口打开开发者工具的“网络”(Network)选项卡刷新页面筛选XHR/Fetch请求寻找包含价格数据的API接口。直接请求这个接口会更简单、更稳定。但京东的接口通常带有复杂的签名参数如h5st逆向难度较大。使用无头浏览器如Selenium或Playwright。它们能模拟真实浏览器行为执行JS等页面完全加载后再获取HTML。这种方法能解决几乎所有动态加载问题但代价是资源消耗大、速度慢。对于监控这种需要长期、低频运行的任务优先推荐寻找接口其次再考虑无头浏览器。请求头与Cookie模拟更真实的浏览器行为添加完整的User-Agent、Accept、Referer等头部信息。有时还需要维护一个会话Session来携带必要的Cookie。在我们的fetcher.py中我们暂时使用最基本的请求头。如果后续发现抓取失败可以按需增强headers字典。5. 数据存储与历史记录使用SQLite抓取到的数据需要持久化存储这样才能进行价格对比和生成历史价格曲线。我们使用SQLite数据库。5.1 设计数据库表结构我们需要两张表products存储被监控商品的基本信息。price_history存储每次抓取的价格记录。5.2 编写数据库操作模块 (database.py)# database.py import sqlite3 from datetime import datetime import os DB_PATH price_monitor.db # 数据库文件路径可在config.py中配置 def get_db_connection(): 创建并返回一个数据库连接。 conn sqlite3.connect(DB_PATH) # 设置返回的行对象为字典类型方便操作 conn.row_factory sqlite3.Row return conn def init_database(): 初始化数据库创建所需的表如果不存在。 conn get_db_connection() cursor conn.cursor() # 创建商品表 cursor.execute( CREATE TABLE IF NOT EXISTS products ( id INTEGER PRIMARY KEY AUTOINCREMENT, url TEXT UNIQUE NOT NULL, title TEXT, target_price REAL, is_active INTEGER DEFAULT 1, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) # 创建价格历史表 cursor.execute( CREATE TABLE IF NOT EXISTS price_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, product_id INTEGER NOT NULL, price REAL NOT NULL, fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (product_id) REFERENCES products (id) ) ) # 为价格历史表创建索引加快按商品和时间的查询速度 cursor.execute(CREATE INDEX IF NOT EXISTS idx_product_time ON price_history (product_id, fetched_at)) conn.commit() conn.close() print(数据库初始化完成。) def add_or_update_product(url, titleNone, target_priceNone): 添加一个新商品或更新已存在商品的信息如标题。 参数: url: 商品URL作为唯一标识。 title: 商品标题。 target_price: 目标提醒价格。 返回: 商品的数据库ID。 conn get_db_connection() cursor conn.cursor() # 首先尝试插入如果冲突URL已存在则更新标题和目标价 cursor.execute( INSERT INTO products (url, title, target_price) VALUES (?, ?, ?) ON CONFLICT(url) DO UPDATE SET title excluded.title, target_price excluded.target_price , (url, title, target_price)) # 获取商品的ID cursor.execute(SELECT id FROM products WHERE url ?, (url,)) product cursor.fetchone() product_id product[id] if product else None conn.commit() conn.close() if product_id: print(f商品已添加/更新ID: {product_id}) else: print(操作失败未能获取商品ID。) return product_id def insert_price_record(product_id, price): 插入一条价格历史记录。 conn get_db_connection() cursor conn.cursor() cursor.execute(INSERT INTO price_history (product_id, price) VALUES (?, ?), (product_id, price)) conn.commit() conn.close() def get_product_by_url(url): 根据URL获取商品信息。 conn get_db_connection() cursor conn.cursor() cursor.execute(SELECT * FROM products WHERE url ?, (url,)) product cursor.fetchone() conn.close() return dict(product) if product else None def get_all_active_products(): 获取所有活跃的监控商品。 conn get_db_connection() cursor conn.cursor() cursor.execute(SELECT * FROM products WHERE is_active 1) products [dict(row) for row in cursor.fetchall()] conn.close() return products def get_price_history(product_id, limit100): 获取某个商品最近的价格历史。 conn get_db_connection() cursor conn.cursor() cursor.execute( SELECT price, fetched_at FROM price_history WHERE product_id ? ORDER BY fetched_at DESC LIMIT ? , (product_id, limit)) history [dict(row) for row in cursor.fetchall()] conn.close() return history # 初始化数据库如果第一次运行 if __name__ __main__: init_database() # 测试添加一个商品 test_url https://item.jd.com/100012043978.html pid add_or_update_product(test_url, 测试商品, 2999.0) if pid: insert_price_record(pid, 2899.0) print(f测试数据插入成功商品ID: {pid})这个模块提供了完整的数据库操作封装。init_database函数应该在主程序启动时调用一次。add_or_update_product函数实现了“有则更新无则插入”的逻辑非常实用。6. 消息通知让提醒触手可及监控到了目标价格必须能及时通知我们。这里实现两种最常用的方式邮件和微信通过Server酱。6.1 邮件通知配置与发送你需要有一个支持SMTP的邮箱如QQ邮箱、163邮箱、Gmail等并开启SMTP服务获取授权码不是登录密码。# notifier.py import smtplib from email.mime.text import MIMEText from email.header import Header import requests import json class Notifier: def __init__(self, config): 初始化通知器。 参数: config: 包含通知配置的字典或对象。 需要包含 mail_config 和 wechat_config。 self.mail_config config.get(mail, {}) self.wechat_config config.get(wechat, {}) def send_email(self, subject, content, to_addr): 发送邮件通知。 参数: subject: 邮件主题。 content: 邮件正文纯文本。 to_addr: 收件人邮箱。 返回: bool: 发送成功返回True否则False。 if not all([self.mail_config.get(smtp_server), self.mail_config.get(smtp_port), self.mail_config.get(from_addr), self.mail_config.get(password)]): print(邮件配置不完整跳过邮件通知。) return False try: # 构造邮件 msg MIMEText(content, plain, utf-8) msg[From] Header(f价格监控机器人 {self.mail_config[from_addr]}, utf-8) msg[To] Header(to_addr, utf-8) msg[Subject] Header(subject, utf-8) # 连接服务器并发送 server smtplib.SMTP_SSL(self.mail_config[smtp_server], self.mail_config[smtp_port]) server.login(self.mail_config[from_addr], self.mail_config[password]) server.sendmail(self.mail_config[from_addr], [to_addr], msg.as_string()) server.quit() print(f邮件已发送至 {to_addr}) return True except Exception as e: print(f发送邮件失败: {e}) return False def send_wechat(self, title, content): 通过Server酱发送微信通知。 参数: title: 消息标题。 content: 消息内容。 返回: bool: 发送成功返回True否则False。 send_key self.wechat_config.get(send_key) if not send_key: print(未配置Server酱SendKey跳过微信通知。) return False url fhttps://sctapi.ftqq.com/{send_key}.send data { title: title, desp: content # Server酱的desp字段支持Markdown } try: resp requests.post(url, datadata, timeout10) result resp.json() if result.get(code) 0: print(微信通知发送成功。) return True else: print(f微信通知发送失败: {result.get(message)}) return False except Exception as e: print(f发送微信通知请求失败: {e}) return False def notify_price_drop(self, product_info, current_price, target_priceNone): 发送降价通知。 参数: product_info: 商品信息字典包含 title, url 等。 current_price: 当前价格。 target_price: 目标价格可选。 subject f【价格监控】{product_info[title]} 降价啦 content f 商品{product_info[title]} 当前价格{current_price} {目标价格 str(target_price) if target_price else } 商品链接{product_info[url]} 抓取时间{datetime.now().strftime(%Y-%m-%d %H:%M:%S)} 快去看看是否入手好时机 # 同时发送邮件和微信如果配置了 email_sent self.send_email(subject, content, self.mail_config.get(to_addr, )) wechat_sent self.send_wechat(subject, content) return email_sent or wechat_sent # 只要有一个成功就算通知成功6.2 配置说明你需要创建一个config.py文件来存放敏感的配置信息# config.py # 数据库配置 DB_PATH price_monitor.db # 邮件配置 (以QQ邮箱为例) MAIL_CONFIG { smtp_server: smtp.qq.com, # QQ邮箱SMTP服务器 smtp_port: 465, # SSL端口 from_addr: your_emailqq.com, # 发件邮箱 password: your_authorization_code, # 授权码不是密码 to_addr: receiverexample.com # 收件邮箱 } # 微信通知配置 (Server酱 https://sct.ftqq.com/) WECHAT_CONFIG { send_key: 你的SendKey # 在Server酱官网获取 } # 监控配置 MONITOR_INTERVAL_MINUTES 30 # 监控间隔分钟 REQUEST_TIMEOUT 10 # 请求超时时间秒 USER_AGENT Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ... # 你的User-Agent重要提示config.py中包含密码和密钥千万不要将其上传到GitHub等公开代码仓库你应该将其添加到.gitignore文件中。一个更好的实践是使用环境变量来存储这些敏感信息。7. 主程序串联与定时任务现在我们将所有模块像拼图一样组合起来并让它们定时运行。7.1 编写核心监控逻辑 (monitor.py)# monitor.py import time import schedule from datetime import datetime from config import DB_PATH, MONITOR_INTERVAL_MINUTES, REQUEST_TIMEOUT, USER_AGENT from database import init_database, get_all_active_products, add_or_update_product, insert_price_record, get_product_by_url from fetcher import fetch_product_info from notifier import Notifier import json # 初始化通知器 from config import MAIL_CONFIG, WECHAT_CONFIG notifier Notifier({mail: MAIL_CONFIG, wechat: WECHAT_CONFIG}) def load_products_from_file(filepathproducts.json): 从JSON文件加载监控商品列表。 try: with open(filepath, r, encodingutf-8) as f: products json.load(f) return products except FileNotFoundError: print(f商品列表文件 {filepath} 未找到将监控数据库中的商品。) return [] except json.JSONDecodeError: print(f商品列表文件 {filepath} 格式错误。) return [] def check_single_product(product_url, target_priceNone): 检查单个商品的价格。 参数: product_url: 商品URL。 target_price: 预设的目标价格为None则只记录不提醒。 返回: bool: 如果触发提醒返回True否则False。 print(f[{datetime.now().strftime(%H:%M:%S)}] 正在检查: {product_url}) # 1. 抓取商品信息 product_info fetch_product_info(product_url, headers{User-Agent: USER_AGENT}) if not product_info: print(f 抓取失败跳过。) return False # 2. 更新或添加商品到数据库 db_product get_product_by_url(product_url) if db_product: product_id db_product[id] # 如果数据库中的标题为空或与抓取到的不同则更新 if not db_product[title] or db_product[title] ! product_info[title]: add_or_update_product(product_url, product_info[title], target_price) else: product_id add_or_update_product(product_url, product_info[title], target_price) if not product_id: print(f 无法获取或创建商品记录跳过。) return False current_price product_info[price] # 3. 插入价格历史记录 insert_price_record(product_id, current_price) print(f 商品: {product_info[title][:30]}...) print(f 当前价格: {current_price}) # 4. 逻辑判断是否触发通知 triggered False # 情况A提供了目标价且当前价 目标价 if target_price is not None and current_price target_price: print(f - 达到目标价格 {target_price}触发通知。) notifier.notify_price_drop(product_info, current_price, target_price) triggered True # 情况B未提供目标价但与上次价格相比有显著下降可选逻辑这里简单示例 # 你可以从数据库查询上一次价格进行比较 # elif ... : # ... return triggered def monitor_job(): 定时任务要执行的主函数。 print(f\n 开始执行监控任务 ({datetime.now().strftime(%Y-%m-%d %H:%M:%S)}) ) # 方式一从数据库读取活跃商品 active_products get_all_active_products() for prod in active_products: # 检查并等待一小会儿避免请求过快 check_single_product(prod[url], prod.get(target_price)) time.sleep(2) # 友好的延迟避免对服务器造成压力 # 方式二也可以从文件读取适合临时添加 # file_products load_products_from_file() # for item in file_products: # check_single_product(item[url], item.get(target_price)) # time.sleep(2) print(f 监控任务完成 \n) def main(): 主函数初始化并启动定时任务。 print(京东商品价格监控眼启动中...) # 1. 初始化数据库 init_database() # 2. 可以在这里添加初始商品示例 # initial_url https://item.jd.com/100012043978.html # add_or_update_product(initial_url, 示例商品, 2500.0) # 3. 立即运行一次监控 monitor_job() # 4. 设置定时任务 schedule.every(MONITOR_INTERVAL_MINUTES).minutes.do(monitor_job) print(f定时监控已启动每 {MONITOR_INTERVAL_MINUTES} 分钟运行一次。) print(程序将持续运行按 CtrlC 终止。) # 5. 循环执行定时任务 try: while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次是否有任务需要执行 except KeyboardInterrupt: print(\n监控程序已手动停止。) if __name__ __main__: main()7.2 如何添加监控商品你有两种方式添加想要监控的商品通过数据库直接操作推荐持久化你可以写一个简单的脚本或者直接使用SQLite浏览器工具向products表插入记录。url是必填项title可以不填程序会自动抓取target_price是你的目标价is_active设为1。通过JSON文件临时、灵活创建一个products.json文件格式如下[ { url: https://item.jd.com/100012043978.html, target_price: 2800.0 }, { url: https://item.jd.com/100123456789.html, target_price: 1500.0 } ]然后在monitor_job函数中取消注释从文件加载的代码部分。8. 部署运行与进阶优化8.1 如何运行这个监控眼确保所有文件 (config.py,database.py,fetcher.py,notifier.py,monitor.py) 在同一目录下。根据你的邮箱和Server酱配置正确修改config.py。在终端中进入项目目录并确保虚拟环境已激活。首次运行直接执行python monitor.py程序会先初始化数据库立即运行一次监控任务然后进入定时循环。8.2 让程序在后台长期运行如果你有一台一直开机的电脑比如家里的NAS或旧笔记本可以直接在上面运行。对于云服务器有以下几种方式Screen/Tmux在Linux服务器上使用screen或tmux创建一个持久会话在其中运行程序即使断开SSH连接程序也会继续。screen -S price_monitor python monitor.py # 按 CtrlA, 再按 D 分离会话 # 重新连接screen -r price_monitor系统服务 (Systemd)在Linux上可以创建一个systemd服务文件让系统来管理进程的启动、停止和重启以及日志收集。这是最专业的方式。Docker容器将整个项目Docker化可以更方便地在任何环境部署和迁移。8.3 常见问题与排查技巧实录在实际运行中你几乎一定会遇到下面这些问题问题抓取失败返回403或请求被重定向到登录页。原因请求头不够“像”浏览器或IP请求频率过高被暂时限制。排查在fetcher.py的fetch_product_info函数中打印出response.status_code和response.text的前500个字符看看返回了什么。如果是验证页面说明触发了反爬。解决增强请求头复制你浏览器开发者工具中“网络”选项卡里某个请求的完整Headers特别是User-Agent,Accept,Accept-Language,Referer(可以设为商品列表页或首页)以及Cookie谨慎使用涉及隐私。降低频率增加time.sleep()的间隔时间比如从2秒增加到5秒甚至更长。在config.py中设置MONITOR_INTERVAL_MINUTES为更长的时间如60分钟。使用代理IP如果单个IP被限制可以考虑使用代理IP池但这会大大增加项目复杂度。问题无法解析出价格price总是None。原因京东页面结构发生变化之前写的CSS选择器或正则表达式失效了。排查将抓取到的HTML保存到本地文件仔细分析新的页面结构。with open(debug_page.html, w, encodingutf-8) as f: f.write(response.text)解决更新fetcher.py中的解析逻辑。重点检查价格是否在script标签的JSON数据里尝试搜索p-price、j-sku-price等关键词。是否变成了异步加载在“网络”选项卡中寻找包含价格的XHR请求。考虑引入Selenium作为兜底方案。问题邮件发送失败报错ssl.SSLError或认证失败。原因邮箱SMTP配置错误或授权码不对或网络问题。排查检查config.py中的smtp_server,smtp_port,from_addr,password是否正确。QQ邮箱的端口是465或587密码是授权码。可以先用简单的脚本单独测试邮件发送功能。解决确保邮箱已开启SMTP服务。对于Gmail等海外邮箱可能需要允许“不够安全的应用”访问不推荐或使用App专用密码。问题程序运行一段时间后自动停止或报错。原因网络波动、数据库锁、未处理的异常导致进程崩溃。排查查看程序打印的日志。增加更详细的try...except捕获异常并将错误信息记录到文件而不是仅仅打印。解决增加异常捕获的广度在monitor_job和check_single_product函数外层包裹try...except确保一个商品的失败不会影响整个任务。使用日志模块用Python内置的logging模块替代print可以输出到文件并设置不同等级INFO, ERROR。考虑进程守护使用systemd或supervisor等工具当进程意外退出时自动重启。8.4 进阶优化方向当你成功运行起基础版本后可以考虑以下优化让监控眼更强大多平台支持将抓取解析部分抽象成基类然后为京东、淘宝、拼多多等不同平台编写子类实现一个统一的多平台监控系统。价格策略多样化不止是“低于目标价”可以设置“历史最低价”、“30天内最低价”、“降价幅度超过X%”等多种提醒策略。数据可视化使用matplotlib或pyecharts等库读取price_history表的数据生成商品价格历史走势图通过邮件附件或网页形式展示。Web管理界面使用Flask或Django搭建一个简单的Web页面方便地添加、删除、修改监控商品查看监控状态和历史图表。容器化部署编写Dockerfile将整个项目打包成Docker镜像实现一键部署。这个“京东商品价格监控眼”项目从构思到实现涵盖了爬虫项目的核心环节环境搭建、页面分析、数据抓取、数据解析、数据存储、业务逻辑、消息通知和任务调度。它就像一把瑞士军刀虽然小巧但功能齐全。希望你在动手实现的过程中不仅能学会技术更能体会到用代码自动化解决生活问题的乐趣。编程最大的成就感莫过于此。
1. 项目概述为什么需要一个“价格监控眼”作为一名经常在网上“剁手”的程序员我发现自己和身边的朋友们都有一个共同的痛点看中的商品价格总是在波动今天一个价明天一个价永远不知道什么时候才是真正的“好价”。手动去刷新页面不仅效率低下还容易错过最佳时机。尤其是在像京东这样的大型电商平台促销活动频繁价格变化更是瞬息万变。于是一个想法自然而然地诞生了能不能写一个程序让它像一只不知疲倦的眼睛24小时帮我盯着心仪商品的价格一旦降到我的心理价位就立刻通知我这就是“京东商品价格监控眼”项目的初衷。它不仅仅是一个简单的Python爬虫练习更是一个能解决实际生活问题的自动化工具。通过这个项目你将学会如何从零开始构建一个结构完整、运行稳定、具有一定健壮性的爬虫应用。我们会从最基础的网页请求开始一步步解析复杂的页面结构处理可能遇到的反爬机制并将获取到的数据存储下来最终实现定时监控和消息通知。整个过程我会手把手带你走一遍并提供完整的源码供你参考和修改。无论你是刚入门Python的新手想找一个有成就感的实战项目还是有一定基础想深化爬虫技能的朋友这个项目都能让你收获满满。2. 核心思路与技术选型如何构建一个稳健的监控系统在动手写代码之前我们先来梳理一下整个系统的核心思路。一个完整的监控系统绝不仅仅是“抓取网页-解析数据”这么简单我们需要考虑其可持续性、稳定性和易用性。2.1 系统工作流程设计整个监控眼的工作流程可以抽象为以下几个核心环节目标输入用户提供想要监控的商品链接URL。数据抓取程序模拟浏览器访问该链接获取网页的HTML源代码。数据解析从复杂的HTML代码中精准地提取出商品名称、当前价格等关键信息。数据存储将提取到的信息商品名、价格、抓取时间保存下来以便后续分析和对比。逻辑判断将当前价格与用户设定的目标价格或历史价格进行对比。消息通知如果当前价格满足条件如低于目标价则通过预设的渠道如邮件、微信发送提醒。定时任务让以上所有步骤能够自动地、周期性地执行。2.2 关键技术栈选型与理由为什么选择这些工具每一个选择背后都有其考量。请求库requestshttpx(备用)requests库是Python界进行HTTP请求的“事实标准”其API设计优雅文档丰富社区支持极好对于京东这样的常规页面请求完全够用。我们将主要使用它。但考虑到部分复杂的动态页面或未来可能遇到的反爬升级我也会简单介绍httpx这个支持HTTP/2和异步请求的后起之秀作为备选方案它对于处理一些需要更高效并发或更现代协议的场景有优势。解析库BeautifulSoup4lxml解析器从HTML中提取数据BeautifulSoup4(简称bs4) 是不二之选。它提供了非常直观的、类似DOM操作的API即使你不熟悉复杂的XPath也能通过标签名、属性、CSS选择器轻松定位元素。搭配lxml作为解析引擎速度比Python内置的html.parser快很多稳定性也更好。数据存储SQLite3(内置)对于个人使用或轻量级监控我们不需要启动一个庞大的MySQL或PostgreSQL数据库。Python标准库自带的SQLite3完美契合需求——它是一个轻量级的、服务器进程的数据库整个数据库就是一个文件无需安装和配置非常适合本项目。我们将用它来创建商品信息表和价格历史记录表。定时任务schedule库我们需要程序能每隔一段时间比如每30分钟自动运行一次。schedule库提供了非常人性化的API可以用接近自然语言的方式设置定时任务如schedule.every(30).minutes.do(job)比直接使用复杂的cron或apscheduler对于新手来说更友好。消息通知smtplib(邮件) 与requests(Server酱等微信通知)价格提醒需要触达用户。最通用和可靠的方式是邮件Python内置的smtplib库可以轻松实现。对于更即时的提醒我们可以集成像“Server酱”一个将服务器消息推送到微信的工具这样的第三方服务本质上也是通过requests库发送一个HTTP请求到其API。注意在开始任何爬虫项目前务必、务必、务必查看目标网站的robots.txt文件通常在网站根目录如https://www.jd.com/robots.txt。这个文件规定了网络爬虫哪些页面可以抓取哪些不可以。尊重robots.txt是爬虫开发者的基本素养也是对网站资源的尊重。我们的监控目标仅限于具体的商品详情页通常这类页面是对搜索引擎开放的但依然建议控制请求频率避免对服务器造成过大压力。3. 环境准备与核心库安装工欲善其事必先利其器。让我们先把开发环境搭建起来。3.1 Python环境确认与虚拟环境创建首先确保你的电脑上安装了Python建议版本3.7及以上。打开终端Windows是CMD或PowerShellMac/Linux是Terminal输入python --version或python3 --version查看版本。强烈建议为这个项目创建一个独立的虚拟环境这样可以避免不同项目间的库版本冲突。# 安装虚拟环境管理工具如果尚未安装 pip install virtualenv # 创建一个名为 jd_monitor 的虚拟环境 virtualenv jd_monitor # 激活虚拟环境 # 在 Windows 上 jd_monitor\Scripts\activate # 在 Mac/Linux 上 source jd_monitor/bin/activate激活后你的命令行提示符前面应该会出现(jd_monitor)字样表示你已经在这个虚拟环境中了。3.2 安装所需第三方库在激活的虚拟环境中一次性安装我们需要的所有库pip install requests beautifulsoup4 lxml schedulerequests: 用于发送HTTP请求。beautifulsoup4: 用于解析HTML。lxml: 作为bs4的解析引擎需要单独安装。schedule: 用于管理定时任务。安装完成后可以通过pip list命令查看已安装的包确认它们都在列表中。3.3 项目目录结构规划一个好的项目结构能让代码更清晰后期维护也更方便。我们先创建好目录和文件jd_price_monitor/ ├── config.py # 配置文件存放数据库路径、邮件设置、监控间隔等 ├── database.py # 数据库操作相关函数创建表、插入数据、查询数据 ├── fetcher.py # 网页抓取和解析的核心函数 ├── notifier.py # 通知发送功能邮件、微信等 ├── monitor.py # 主程序串联整个流程包含定时任务逻辑 ├── requirements.txt # 项目依赖库列表 └── products.json # 可选存放待监控的商品链接列表你可以使用以下命令快速创建mkdir jd_price_monitor cd jd_price_monitor touch config.py database.py fetcher.py notifier.py monitor.py requirements.txt4. 网页抓取与解析攻克京东商品页这是整个项目的核心也是最容易出问题的一步。京东的页面结构可能会变动反爬策略也可能升级因此我们的代码需要有一定的容错和适应性。4.1 分析页面结构与定位价格元素首先我们需要手动打开一个京东商品页例如一个手机页面使用浏览器的“开发者工具”按F12来观察。找到价格元素在页面上右键点击商品价格选择“检查”。你会发现京东的商品价格通常在一个类名为price的标签内但为了更精确我们需要找到其唯一的ID或特定的CSS选择器路径。经过分析当前京东商品页面的当前价格通常位于一个类似#priceSale或.price.J-p-{商品ID}的标签内。请注意这个选择器可能会随着京东前端的改版而变化这是爬虫需要维护的主要原因之一。找到商品标题同样方法商品标题通常在一个ID为itemName或类名包含sku-name的h1标签内。4.2 编写抓取与解析函数 (fetcher.py)我们将抓取和解析的逻辑封装在一个函数里。# fetcher.py import requests from bs4 import BeautifulSoup import re import time def fetch_product_info(url, headersNone): 抓取指定URL的商品信息标题、价格。 参数: url (str): 商品详情页的URL。 headers (dict): 可选的请求头用于模拟浏览器。 返回: dict: 包含商品标题和价格的字典如果失败则返回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, Accept-Language: zh-CN,zh;q0.9, } try: # 1. 发送HTTP GET请求 response requests.get(url, headersheaders, timeout10) response.raise_for_status() # 如果状态码不是200抛出HTTPError异常 response.encoding response.apparent_encoding or utf-8 # 自动检测编码 # 2. 使用BeautifulSoup解析HTML soup BeautifulSoup(response.text, lxml) # 3. 提取商品标题 - 需要根据实际页面调整选择器 # 示例选择器可能需要调整 title_elem soup.select_one(.sku-name) or soup.select_one(#itemName) if title_elem: product_title title_elem.get_text(stripTrue) else: # 如果找不到尝试其他选择器或记录错误 product_title 未知商品 print(f警告在 {url} 中未找到商品标题。) # 4. 提取商品价格 - 这是最关键的也是最容易失效的部分 price None # 方法A尝试通过标签属性查找 price_elem soup.select_one(.price) if price_elem: price_text price_elem.get_text(stripTrue) else: # 方法B尝试在脚本数据中查找京东价格有时通过JS动态加载 script_tags soup.find_all(script) for script in script_tags: if script.string and price in script.string.lower(): # 使用正则表达式从JS代码中匹配价格数字 match re.search(rprice\s*:\s*[\]?([\d.]), script.string) if match: price_text match.group(1) break else: price_text # 清理价格文本提取纯数字 if price_text: # 匹配数字和可能的小数点 match re.search(r[\d,.], price_text.replace(,, )) if match: price float(match.group()) if price is None: print(f错误无法从 {url} 解析出价格。) return None return { title: product_title, price: price, url: url } except requests.exceptions.RequestException as e: print(f网络请求出错 ({url}): {e}) return None except Exception as e: print(f解析过程出错 ({url}): {e}) return None # 简单的测试代码 if __name__ __main__: # 替换成一个真实的京东商品URL进行测试 test_url https://item.jd.com/100012043978.html # 示例链接请勿频繁请求 info fetch_product_info(test_url) if info: print(f商品: {info[title]}) print(f价格: {info[price]}) else: print(抓取失败)实操心得京东页面的价格信息可能存在于多个地方直接的HTML标签、script标签内的JSON数据、甚至是异步接口返回的。上面的代码提供了两种查找策略。在实际运行中如果第一种方法失效第二种方法从JS中正则匹配往往能作为备用方案。最可靠的方法是直接分析其数据接口但这涉及到更复杂的逆向工程对于入门项目我们先从HTML解析入手。记得在测试时务必使用真实的商品URL并控制请求频率比如在测试代码中加上time.sleep(3)避免请求过快。4.3 处理动态加载与反爬策略现代网站大量使用JavaScript动态加载内容直接请求HTML可能拿不到价格。此时有几种策略寻找数据接口打开开发者工具的“网络”(Network)选项卡刷新页面筛选XHR/Fetch请求寻找包含价格数据的API接口。直接请求这个接口会更简单、更稳定。但京东的接口通常带有复杂的签名参数如h5st逆向难度较大。使用无头浏览器如Selenium或Playwright。它们能模拟真实浏览器行为执行JS等页面完全加载后再获取HTML。这种方法能解决几乎所有动态加载问题但代价是资源消耗大、速度慢。对于监控这种需要长期、低频运行的任务优先推荐寻找接口其次再考虑无头浏览器。请求头与Cookie模拟更真实的浏览器行为添加完整的User-Agent、Accept、Referer等头部信息。有时还需要维护一个会话Session来携带必要的Cookie。在我们的fetcher.py中我们暂时使用最基本的请求头。如果后续发现抓取失败可以按需增强headers字典。5. 数据存储与历史记录使用SQLite抓取到的数据需要持久化存储这样才能进行价格对比和生成历史价格曲线。我们使用SQLite数据库。5.1 设计数据库表结构我们需要两张表products存储被监控商品的基本信息。price_history存储每次抓取的价格记录。5.2 编写数据库操作模块 (database.py)# database.py import sqlite3 from datetime import datetime import os DB_PATH price_monitor.db # 数据库文件路径可在config.py中配置 def get_db_connection(): 创建并返回一个数据库连接。 conn sqlite3.connect(DB_PATH) # 设置返回的行对象为字典类型方便操作 conn.row_factory sqlite3.Row return conn def init_database(): 初始化数据库创建所需的表如果不存在。 conn get_db_connection() cursor conn.cursor() # 创建商品表 cursor.execute( CREATE TABLE IF NOT EXISTS products ( id INTEGER PRIMARY KEY AUTOINCREMENT, url TEXT UNIQUE NOT NULL, title TEXT, target_price REAL, is_active INTEGER DEFAULT 1, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) # 创建价格历史表 cursor.execute( CREATE TABLE IF NOT EXISTS price_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, product_id INTEGER NOT NULL, price REAL NOT NULL, fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (product_id) REFERENCES products (id) ) ) # 为价格历史表创建索引加快按商品和时间的查询速度 cursor.execute(CREATE INDEX IF NOT EXISTS idx_product_time ON price_history (product_id, fetched_at)) conn.commit() conn.close() print(数据库初始化完成。) def add_or_update_product(url, titleNone, target_priceNone): 添加一个新商品或更新已存在商品的信息如标题。 参数: url: 商品URL作为唯一标识。 title: 商品标题。 target_price: 目标提醒价格。 返回: 商品的数据库ID。 conn get_db_connection() cursor conn.cursor() # 首先尝试插入如果冲突URL已存在则更新标题和目标价 cursor.execute( INSERT INTO products (url, title, target_price) VALUES (?, ?, ?) ON CONFLICT(url) DO UPDATE SET title excluded.title, target_price excluded.target_price , (url, title, target_price)) # 获取商品的ID cursor.execute(SELECT id FROM products WHERE url ?, (url,)) product cursor.fetchone() product_id product[id] if product else None conn.commit() conn.close() if product_id: print(f商品已添加/更新ID: {product_id}) else: print(操作失败未能获取商品ID。) return product_id def insert_price_record(product_id, price): 插入一条价格历史记录。 conn get_db_connection() cursor conn.cursor() cursor.execute(INSERT INTO price_history (product_id, price) VALUES (?, ?), (product_id, price)) conn.commit() conn.close() def get_product_by_url(url): 根据URL获取商品信息。 conn get_db_connection() cursor conn.cursor() cursor.execute(SELECT * FROM products WHERE url ?, (url,)) product cursor.fetchone() conn.close() return dict(product) if product else None def get_all_active_products(): 获取所有活跃的监控商品。 conn get_db_connection() cursor conn.cursor() cursor.execute(SELECT * FROM products WHERE is_active 1) products [dict(row) for row in cursor.fetchall()] conn.close() return products def get_price_history(product_id, limit100): 获取某个商品最近的价格历史。 conn get_db_connection() cursor conn.cursor() cursor.execute( SELECT price, fetched_at FROM price_history WHERE product_id ? ORDER BY fetched_at DESC LIMIT ? , (product_id, limit)) history [dict(row) for row in cursor.fetchall()] conn.close() return history # 初始化数据库如果第一次运行 if __name__ __main__: init_database() # 测试添加一个商品 test_url https://item.jd.com/100012043978.html pid add_or_update_product(test_url, 测试商品, 2999.0) if pid: insert_price_record(pid, 2899.0) print(f测试数据插入成功商品ID: {pid})这个模块提供了完整的数据库操作封装。init_database函数应该在主程序启动时调用一次。add_or_update_product函数实现了“有则更新无则插入”的逻辑非常实用。6. 消息通知让提醒触手可及监控到了目标价格必须能及时通知我们。这里实现两种最常用的方式邮件和微信通过Server酱。6.1 邮件通知配置与发送你需要有一个支持SMTP的邮箱如QQ邮箱、163邮箱、Gmail等并开启SMTP服务获取授权码不是登录密码。# notifier.py import smtplib from email.mime.text import MIMEText from email.header import Header import requests import json class Notifier: def __init__(self, config): 初始化通知器。 参数: config: 包含通知配置的字典或对象。 需要包含 mail_config 和 wechat_config。 self.mail_config config.get(mail, {}) self.wechat_config config.get(wechat, {}) def send_email(self, subject, content, to_addr): 发送邮件通知。 参数: subject: 邮件主题。 content: 邮件正文纯文本。 to_addr: 收件人邮箱。 返回: bool: 发送成功返回True否则False。 if not all([self.mail_config.get(smtp_server), self.mail_config.get(smtp_port), self.mail_config.get(from_addr), self.mail_config.get(password)]): print(邮件配置不完整跳过邮件通知。) return False try: # 构造邮件 msg MIMEText(content, plain, utf-8) msg[From] Header(f价格监控机器人 {self.mail_config[from_addr]}, utf-8) msg[To] Header(to_addr, utf-8) msg[Subject] Header(subject, utf-8) # 连接服务器并发送 server smtplib.SMTP_SSL(self.mail_config[smtp_server], self.mail_config[smtp_port]) server.login(self.mail_config[from_addr], self.mail_config[password]) server.sendmail(self.mail_config[from_addr], [to_addr], msg.as_string()) server.quit() print(f邮件已发送至 {to_addr}) return True except Exception as e: print(f发送邮件失败: {e}) return False def send_wechat(self, title, content): 通过Server酱发送微信通知。 参数: title: 消息标题。 content: 消息内容。 返回: bool: 发送成功返回True否则False。 send_key self.wechat_config.get(send_key) if not send_key: print(未配置Server酱SendKey跳过微信通知。) return False url fhttps://sctapi.ftqq.com/{send_key}.send data { title: title, desp: content # Server酱的desp字段支持Markdown } try: resp requests.post(url, datadata, timeout10) result resp.json() if result.get(code) 0: print(微信通知发送成功。) return True else: print(f微信通知发送失败: {result.get(message)}) return False except Exception as e: print(f发送微信通知请求失败: {e}) return False def notify_price_drop(self, product_info, current_price, target_priceNone): 发送降价通知。 参数: product_info: 商品信息字典包含 title, url 等。 current_price: 当前价格。 target_price: 目标价格可选。 subject f【价格监控】{product_info[title]} 降价啦 content f 商品{product_info[title]} 当前价格{current_price} {目标价格 str(target_price) if target_price else } 商品链接{product_info[url]} 抓取时间{datetime.now().strftime(%Y-%m-%d %H:%M:%S)} 快去看看是否入手好时机 # 同时发送邮件和微信如果配置了 email_sent self.send_email(subject, content, self.mail_config.get(to_addr, )) wechat_sent self.send_wechat(subject, content) return email_sent or wechat_sent # 只要有一个成功就算通知成功6.2 配置说明你需要创建一个config.py文件来存放敏感的配置信息# config.py # 数据库配置 DB_PATH price_monitor.db # 邮件配置 (以QQ邮箱为例) MAIL_CONFIG { smtp_server: smtp.qq.com, # QQ邮箱SMTP服务器 smtp_port: 465, # SSL端口 from_addr: your_emailqq.com, # 发件邮箱 password: your_authorization_code, # 授权码不是密码 to_addr: receiverexample.com # 收件邮箱 } # 微信通知配置 (Server酱 https://sct.ftqq.com/) WECHAT_CONFIG { send_key: 你的SendKey # 在Server酱官网获取 } # 监控配置 MONITOR_INTERVAL_MINUTES 30 # 监控间隔分钟 REQUEST_TIMEOUT 10 # 请求超时时间秒 USER_AGENT Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ... # 你的User-Agent重要提示config.py中包含密码和密钥千万不要将其上传到GitHub等公开代码仓库你应该将其添加到.gitignore文件中。一个更好的实践是使用环境变量来存储这些敏感信息。7. 主程序串联与定时任务现在我们将所有模块像拼图一样组合起来并让它们定时运行。7.1 编写核心监控逻辑 (monitor.py)# monitor.py import time import schedule from datetime import datetime from config import DB_PATH, MONITOR_INTERVAL_MINUTES, REQUEST_TIMEOUT, USER_AGENT from database import init_database, get_all_active_products, add_or_update_product, insert_price_record, get_product_by_url from fetcher import fetch_product_info from notifier import Notifier import json # 初始化通知器 from config import MAIL_CONFIG, WECHAT_CONFIG notifier Notifier({mail: MAIL_CONFIG, wechat: WECHAT_CONFIG}) def load_products_from_file(filepathproducts.json): 从JSON文件加载监控商品列表。 try: with open(filepath, r, encodingutf-8) as f: products json.load(f) return products except FileNotFoundError: print(f商品列表文件 {filepath} 未找到将监控数据库中的商品。) return [] except json.JSONDecodeError: print(f商品列表文件 {filepath} 格式错误。) return [] def check_single_product(product_url, target_priceNone): 检查单个商品的价格。 参数: product_url: 商品URL。 target_price: 预设的目标价格为None则只记录不提醒。 返回: bool: 如果触发提醒返回True否则False。 print(f[{datetime.now().strftime(%H:%M:%S)}] 正在检查: {product_url}) # 1. 抓取商品信息 product_info fetch_product_info(product_url, headers{User-Agent: USER_AGENT}) if not product_info: print(f 抓取失败跳过。) return False # 2. 更新或添加商品到数据库 db_product get_product_by_url(product_url) if db_product: product_id db_product[id] # 如果数据库中的标题为空或与抓取到的不同则更新 if not db_product[title] or db_product[title] ! product_info[title]: add_or_update_product(product_url, product_info[title], target_price) else: product_id add_or_update_product(product_url, product_info[title], target_price) if not product_id: print(f 无法获取或创建商品记录跳过。) return False current_price product_info[price] # 3. 插入价格历史记录 insert_price_record(product_id, current_price) print(f 商品: {product_info[title][:30]}...) print(f 当前价格: {current_price}) # 4. 逻辑判断是否触发通知 triggered False # 情况A提供了目标价且当前价 目标价 if target_price is not None and current_price target_price: print(f - 达到目标价格 {target_price}触发通知。) notifier.notify_price_drop(product_info, current_price, target_price) triggered True # 情况B未提供目标价但与上次价格相比有显著下降可选逻辑这里简单示例 # 你可以从数据库查询上一次价格进行比较 # elif ... : # ... return triggered def monitor_job(): 定时任务要执行的主函数。 print(f\n 开始执行监控任务 ({datetime.now().strftime(%Y-%m-%d %H:%M:%S)}) ) # 方式一从数据库读取活跃商品 active_products get_all_active_products() for prod in active_products: # 检查并等待一小会儿避免请求过快 check_single_product(prod[url], prod.get(target_price)) time.sleep(2) # 友好的延迟避免对服务器造成压力 # 方式二也可以从文件读取适合临时添加 # file_products load_products_from_file() # for item in file_products: # check_single_product(item[url], item.get(target_price)) # time.sleep(2) print(f 监控任务完成 \n) def main(): 主函数初始化并启动定时任务。 print(京东商品价格监控眼启动中...) # 1. 初始化数据库 init_database() # 2. 可以在这里添加初始商品示例 # initial_url https://item.jd.com/100012043978.html # add_or_update_product(initial_url, 示例商品, 2500.0) # 3. 立即运行一次监控 monitor_job() # 4. 设置定时任务 schedule.every(MONITOR_INTERVAL_MINUTES).minutes.do(monitor_job) print(f定时监控已启动每 {MONITOR_INTERVAL_MINUTES} 分钟运行一次。) print(程序将持续运行按 CtrlC 终止。) # 5. 循环执行定时任务 try: while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次是否有任务需要执行 except KeyboardInterrupt: print(\n监控程序已手动停止。) if __name__ __main__: main()7.2 如何添加监控商品你有两种方式添加想要监控的商品通过数据库直接操作推荐持久化你可以写一个简单的脚本或者直接使用SQLite浏览器工具向products表插入记录。url是必填项title可以不填程序会自动抓取target_price是你的目标价is_active设为1。通过JSON文件临时、灵活创建一个products.json文件格式如下[ { url: https://item.jd.com/100012043978.html, target_price: 2800.0 }, { url: https://item.jd.com/100123456789.html, target_price: 1500.0 } ]然后在monitor_job函数中取消注释从文件加载的代码部分。8. 部署运行与进阶优化8.1 如何运行这个监控眼确保所有文件 (config.py,database.py,fetcher.py,notifier.py,monitor.py) 在同一目录下。根据你的邮箱和Server酱配置正确修改config.py。在终端中进入项目目录并确保虚拟环境已激活。首次运行直接执行python monitor.py程序会先初始化数据库立即运行一次监控任务然后进入定时循环。8.2 让程序在后台长期运行如果你有一台一直开机的电脑比如家里的NAS或旧笔记本可以直接在上面运行。对于云服务器有以下几种方式Screen/Tmux在Linux服务器上使用screen或tmux创建一个持久会话在其中运行程序即使断开SSH连接程序也会继续。screen -S price_monitor python monitor.py # 按 CtrlA, 再按 D 分离会话 # 重新连接screen -r price_monitor系统服务 (Systemd)在Linux上可以创建一个systemd服务文件让系统来管理进程的启动、停止和重启以及日志收集。这是最专业的方式。Docker容器将整个项目Docker化可以更方便地在任何环境部署和迁移。8.3 常见问题与排查技巧实录在实际运行中你几乎一定会遇到下面这些问题问题抓取失败返回403或请求被重定向到登录页。原因请求头不够“像”浏览器或IP请求频率过高被暂时限制。排查在fetcher.py的fetch_product_info函数中打印出response.status_code和response.text的前500个字符看看返回了什么。如果是验证页面说明触发了反爬。解决增强请求头复制你浏览器开发者工具中“网络”选项卡里某个请求的完整Headers特别是User-Agent,Accept,Accept-Language,Referer(可以设为商品列表页或首页)以及Cookie谨慎使用涉及隐私。降低频率增加time.sleep()的间隔时间比如从2秒增加到5秒甚至更长。在config.py中设置MONITOR_INTERVAL_MINUTES为更长的时间如60分钟。使用代理IP如果单个IP被限制可以考虑使用代理IP池但这会大大增加项目复杂度。问题无法解析出价格price总是None。原因京东页面结构发生变化之前写的CSS选择器或正则表达式失效了。排查将抓取到的HTML保存到本地文件仔细分析新的页面结构。with open(debug_page.html, w, encodingutf-8) as f: f.write(response.text)解决更新fetcher.py中的解析逻辑。重点检查价格是否在script标签的JSON数据里尝试搜索p-price、j-sku-price等关键词。是否变成了异步加载在“网络”选项卡中寻找包含价格的XHR请求。考虑引入Selenium作为兜底方案。问题邮件发送失败报错ssl.SSLError或认证失败。原因邮箱SMTP配置错误或授权码不对或网络问题。排查检查config.py中的smtp_server,smtp_port,from_addr,password是否正确。QQ邮箱的端口是465或587密码是授权码。可以先用简单的脚本单独测试邮件发送功能。解决确保邮箱已开启SMTP服务。对于Gmail等海外邮箱可能需要允许“不够安全的应用”访问不推荐或使用App专用密码。问题程序运行一段时间后自动停止或报错。原因网络波动、数据库锁、未处理的异常导致进程崩溃。排查查看程序打印的日志。增加更详细的try...except捕获异常并将错误信息记录到文件而不是仅仅打印。解决增加异常捕获的广度在monitor_job和check_single_product函数外层包裹try...except确保一个商品的失败不会影响整个任务。使用日志模块用Python内置的logging模块替代print可以输出到文件并设置不同等级INFO, ERROR。考虑进程守护使用systemd或supervisor等工具当进程意外退出时自动重启。8.4 进阶优化方向当你成功运行起基础版本后可以考虑以下优化让监控眼更强大多平台支持将抓取解析部分抽象成基类然后为京东、淘宝、拼多多等不同平台编写子类实现一个统一的多平台监控系统。价格策略多样化不止是“低于目标价”可以设置“历史最低价”、“30天内最低价”、“降价幅度超过X%”等多种提醒策略。数据可视化使用matplotlib或pyecharts等库读取price_history表的数据生成商品价格历史走势图通过邮件附件或网页形式展示。Web管理界面使用Flask或Django搭建一个简单的Web页面方便地添加、删除、修改监控商品查看监控状态和历史图表。容器化部署编写Dockerfile将整个项目打包成Docker镜像实现一键部署。这个“京东商品价格监控眼”项目从构思到实现涵盖了爬虫项目的核心环节环境搭建、页面分析、数据抓取、数据解析、数据存储、业务逻辑、消息通知和任务调度。它就像一把瑞士军刀虽然小巧但功能齐全。希望你在动手实现的过程中不仅能学会技术更能体会到用代码自动化解决生活问题的乐趣。编程最大的成就感莫过于此。
