1. 项目概述为什么我们需要一个价格追踪器在电商购物、投资理财甚至日常消费中价格波动是常态。你是否曾为错过心仪商品的历史低价而懊恼是否在购买电子产品、机票或基金时因无法把握最佳时机而犹豫不决手动刷新比价网站不仅效率低下还容易遗漏关键节点。这正是“Python Price Tracker”价格追踪器项目诞生的初衷。它不是一个简单的脚本而是一个自动化、可定制、能持续运行的数据监控解决方案核心目标是通过编程手段解放你的时间和精力让你在价格博弈中占据信息优势。简单来说这个项目就是用Python写一个“智能小助手”它能按照你设定的规则自动抓取指定商品或资产的价格信息并在满足特定条件如价格低于阈值、出现大幅波动时通过邮件、短信或应用通知提醒你。这背后涉及的核心技术点包括网络数据抓取爬虫、数据解析、定时任务调度以及消息推送。无论你是想追踪显卡的降价情况、监控特定股票的价格还是关注某个航线的机票折扣这个工具都能派上用场。它适合有一定Python基础希望将编程技能应用于解决实际问题的开发者、数据分析爱好者以及精明的消费者。2. 核心思路与技术选型解析构建一个健壮的价格追踪器远不止写几行requests和BeautifulSoup那么简单。我们需要从架构层面思考其稳定性、可扩展性和易用性。整个系统可以拆解为四个核心模块数据获取模块、数据处理与存储模块、监控与逻辑判断模块以及通知告警模块。2.1 数据获取爬虫策略与反反爬应对数据是追踪器的血液。获取价格数据通常有两种途径通过目标网站的公开API或者直接解析网页Web Scraping。对于大多数电商网站公开API要么不存在要么有严格的调用限制和认证要求。因此网页抓取成为了更通用的选择。技术选型Requests BeautifulSoup 还是 Selenium这是一个关键抉择。Requests库高效、轻量适合抓取静态HTML页面。配合BeautifulSoup或lxml进行解析是绝大多数场景下的首选。然而许多现代网站尤其是使用了React、Vue等前端框架的站点的内容是动态加载的初始HTML中并不包含价格数据。这时就需要Selenium或Playwright这类浏览器自动化工具来模拟真实用户操作等待JavaScript执行完毕后再获取完整的页面内容。实操心得优先尝试Requests。先用浏览器开发者工具F12的Network面板查看页面加载过程。如果能在XHR或Fetch请求中找到直接返回价格数据的API接口通常是JSON格式那就用Requests模拟这个请求这是最稳定、最高效的方式。如果价格数据直接嵌在初始HTML中也用Requests。只有当数据确系动态渲染且找不到隐藏接口时才考虑使用Selenium因为它资源消耗大、速度慢。反反爬虫策略设置请求头Headers务必模拟真实浏览器的User-Agent并带上常见的Accept、Accept-Language等字段。控制请求频率在代码中引入随机延时如time.sleep(random.uniform(1, 3))避免短时间内高频请求导致IP被封。使用代理IP池对于需要大规模、高频追踪的场景可以考虑使用付费或自建的代理IP服务来轮换IP地址。处理Cookie和Session对于需要登录后才能查看价格的网站需要维护会话状态。2.2 数据处理与存储从临时变量到持久化记录抓取到的原始数据需要被清洗、提取出目标价格并保存下来以便进行历史对比和趋势分析。数据解析使用BeautifulSoup或lxml通过CSS选择器或XPath定位到包含价格的HTML元素。这里最大的挑战是网站页面结构可能会变动。因此选择器要尽可能精准且具有一定的容错性。提取出文本后需要用正则表达式或字符串方法清理掉货币符号、空格等非数字字符转换为浮点数。数据存储选型SQLite轻量级无需安装额外服务整个数据库就是一个文件。非常适合单机、个人使用的追踪器简单可靠。CSV/JSON文件最简单但频繁读写可能遇到并发问题且查询历史数据不如数据库方便。MySQL/PostgreSQL功能强大适合多用户、需要复杂查询的分布式应用。时序数据库如InfluxDB如果追踪标的物非常多且对时间序列数据的写入和查询性能有极高要求可以考虑。对于个人项目SQLite通常是完美选择。它足以记录每个商品每次抓取的时间戳、价格、以及其他你可能关心的信息如库存状态。2.3 监控逻辑与任务调度让程序自动运转起来追踪器的核心逻辑是“定期执行抓取任务并判断价格是否符合预设条件”。这涉及到定时任务调度。技术选型简单循环 time.sleep在脚本中使用while True循环每次执行完抓取和判断逻辑后休眠一段时间如1小时。这是最直白的实现但缺点是睡眠期间程序独占一个终端且不易管理。操作系统级任务调度在Linux上使用cron在Windows上使用“任务计划程序”。让操作系统定时调用你的Python脚本。这种方式将执行逻辑和调度逻辑解耦更清晰也便于日志管理。Python库调度使用schedule或APScheduler这类库在Python进程内实现更复杂的调度规则如每周一上午10点执行。它比系统级调度更灵活但需要保证Python进程常驻。监控逻辑设计 判断条件不应只是“当前价格低于设定值”。更聪明的策略包括历史价格对比当前价格是否低于过去7天/30天的平均价价格波动预警单日涨跌幅是否超过某个百分比组合条件仅在同时满足“价格低于X元”且“库存状态为‘有货’”时触发提醒。2.4 通知告警如何第一时间获知信息当监控逻辑被触发你需要被及时告知。通知方式应可靠且符合你的使用习惯。常见通知方式电子邮件SMTP最通用、最稳定的方式。使用Python的smtplib和email库可以轻松实现邮件发送。你需要一个发件邮箱建议使用QQ、163等提供的SMTP服务并开启授权码。桌面通知适用于脚本就在本地电脑运行的情况。可以使用plyer或win10toast仅Windows库发送系统桌面通知。手机推送集成推送服务使用如PushDeer、Server酱等第三方服务它们提供简单的API可以将消息推送到你的手机App。Telegram Bot创建一个Telegram机器人通过其API发送消息到你的私聊或频道。这种方式在技术圈非常流行稳定且功能强大。短信成本较高且依赖于第三方短信网关API一般不作为首选。注意事项无论选择哪种方式务必处理好通知发送失败的情况如网络错误、API限额。可以考虑加入重试机制并将发送失败的日志记录下来。3. 分步实现构建一个基础的电商价格追踪器下面我们以追踪一个亚马逊商品例如一本特定的书的价格为例手把手实现一个基础版本。我们将选择Requests BeautifulSoup SQLite 邮件通知 系统Cron调度这条技术栈因为它平衡了复杂度、稳定性和学习价值。3.1 环境准备与依赖安装首先确保你的Python环境是3.6以上版本。创建一个新的项目目录并初始化虚拟环境是一个好习惯。mkdir python_price_tracker cd python_price_tracker python -m venv venv # Windows: venv\Scripts\activate source venv/bin/activate # Linux/Mac安装必要的Python库pip install requests beautifulsoup4 lxmlrequests: 用于发送HTTP请求。beautifulsoup4和lxml: 用于解析HTML。lxml是一个解析器比Python内置的html.parser更快、更强大。我们暂时不安装数据库和邮件相关的库将在后续步骤中引入。3.2 编写核心抓取与解析函数创建一个名为tracker.py的文件。我们将首先实现最核心的功能给定一个商品URL抓取页面并提取价格。import requests from bs4 import BeautifulSoup import re import time import random def fetch_price(url): 从给定的URL抓取并解析价格。 返回一个字典包含价格、商品标题和抓取状态。 # 模拟浏览器的请求头 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: en-US,en;q0.9, } try: # 发送GET请求 response requests.get(url, headersheaders, timeout10) response.raise_for_status() # 如果状态码不是200抛出HTTPError异常 # 使用BeautifulSoup和lxml解析器解析HTML soup BeautifulSoup(response.content, lxml) # **关键步骤定位价格元素** # 你需要使用浏览器开发者工具找到价格所在的HTML元素和其选择器。 # 以下选择器是示例针对亚马逊页面可能有效但网站改版后会失效。 # 方法在商品页面上右键“检查”找到价格数字右键其父元素 - Copy - Copy selector price_selector #corePriceDisplay_desktop_feature_div div.a-section.a-spacing-none.aok-align-center span.a-price.aok-align-center.reinventPricePriceToPayMargin.priceToPay span:nth-child(2) span.a-price-whole price_element soup.select_one(price_selector) if price_element: # 提取文本并清理非数字字符如逗号、货币符号 price_text price_element.get_text(stripTrue) # 使用正则表达式提取数字包括小数点 price_match re.search(r[\d,]\.?\d*, price_text.replace(,, )) if price_match: price float(price_match.group()) else: price None else: price None print(f警告在 {url} 中未找到价格元素。选择器可能已过期。) # 尝试获取商品标题可选用于通知中显示 title_selector #productTitle title_element soup.select_one(title_selector) title title_element.get_text(stripTrue) if title_element else 未知商品 return { success: True, price: price, title: title, url: url, timestamp: time.time() } except requests.exceptions.RequestException as e: print(f请求失败: {e}) return {success: False, error: str(e), url: url} except Exception as e: print(f解析过程发生未知错误: {e}) return {success: False, error: str(e), url: url} # 测试函数 if __name__ __main__: # 替换成你想追踪的实际商品URL test_url https://www.amazon.com/dp/B08N5WRWNW # 示例URL result fetch_price(test_url) print(result) if result[success] and result[price]: print(f商品: {result[title][:50]}...) print(f价格: ${result[price]:.2f})重要提示上述代码中的price_selector和title_selector是示例来源于特定时间点的亚马逊页面结构。网站前端随时可能改版导致选择器失效。这是网页抓取项目最大的维护成本。你需要根据目标网站的实际HTML结构使用开发者工具重新定位并更新这些选择器。3.3 集成SQLite进行数据持久化我们需要一个地方来存储每次抓取到的价格记录。修改tracker.py增加数据库操作。首先安装SQLite驱动Python标准库已包含无需额外安装。我们创建一个数据库帮助类。import sqlite3 from datetime import datetime class PriceDB: def __init__(self, db_pathprices.db): self.conn sqlite3.connect(db_path) self.create_table() def create_table(self): 创建存储价格记录的表 cursor self.conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS price_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, product_url TEXT NOT NULL, product_title TEXT, price REAL, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP ) ) # 为商品URL和日期创建索引加速查询 cursor.execute(CREATE INDEX IF NOT EXISTS idx_url ON price_history (product_url)) cursor.execute(CREATE INDEX IF NOT EXISTS idx_timestamp ON price_history (timestamp)) self.conn.commit() def insert_price(self, url, title, price): 插入一条新的价格记录 cursor self.conn.cursor() cursor.execute( INSERT INTO price_history (product_url, product_title, price) VALUES (?, ?, ?) , (url, title, price)) self.conn.commit() return cursor.lastrowid def get_price_history(self, url, limit30): 获取某个商品最近的价格历史 cursor self.conn.cursor() cursor.execute( SELECT price, timestamp FROM price_history WHERE product_url ? ORDER BY timestamp DESC LIMIT ? , (url, limit)) return cursor.fetchall() def close(self): self.conn.close()然后修改主逻辑在抓取成功后插入数据。def track_product(url, db): 追踪单个商品抓取价格并存入数据库 result fetch_price(url) if result[success] and result[price] is not None: db.insert_price(url, result[title], result[price]) print(f[{datetime.now()}] 已记录 {result[title][:30]}... 价格: ${result[price]:.2f}) return result else: print(f[{datetime.now()}] 抓取失败: {result.get(error, 未知错误)}) return None if __name__ __main__: db PriceDB() test_url https://www.amazon.com/dp/B08N5WRWNW track_product(test_url, db) # 可以查询历史记录 history db.get_price_history(test_url, 5) for price, ts in history: print(f{ts}: ${price}) db.close()3.4 实现邮件通知功能当价格低于我们的目标价时发送邮件提醒。我们需要配置发件邮箱。这里以QQ邮箱的SMTP服务为例。首先你需要登录QQ邮箱在“设置”-“账户”中开启POP3/SMTP服务并获取一个授权码不是邮箱密码。修改tracker.py添加邮件发送功能import smtplib from email.mime.text import MIMEText from email.header import Header class EmailNotifier: def __init__(self, smtp_server, smtp_port, sender_email, sender_auth_code): self.smtp_server smtp_server self.smtp_port smtp_port self.sender_email sender_email self.sender_auth_code sender_auth_code def send(self, to_email, subject, body): 发送邮件 message MIMEText(body, plain, utf-8) message[From] Header(fPrice Tracker {self.sender_email}, utf-8) message[To] Header(to_email, utf-8) message[Subject] Header(subject, utf-8) try: # 使用SSL加密连接 with smtplib.SMTP_SSL(self.smtp_server, self.smtp_port) as server: server.login(self.sender_email, self.sender_auth_code) server.sendmail(self.sender_email, [to_email], message.as_string()) print(f[{datetime.now()}] 邮件已发送至 {to_email}) return True except Exception as e: print(f[{datetime.now()}] 邮件发送失败: {e}) return False # 配置你的邮箱信息务必保密不要将授权码提交到Git等公开仓库 SMTP_SERVER smtp.qq.com SMTP_PORT 465 SENDER_EMAIL your_qqqq.com # 你的QQ邮箱 SENDER_AUTH_CODE your_authorization_code # 你的授权码 TO_EMAIL your_notification_emailexample.com # 接收通知的邮箱 def check_and_notify(url, target_price, db, notifier): 检查价格并决定是否发送通知 result track_product(url, db) # 抓取并记录 if result and result[price] target_price: subject f价格提醒{result[title][:50]}... 已降至目标价以下 body f 商品{result[title]} 当前价格${result[price]:.2f} 目标价格${target_price:.2f} 商品链接{url} 抓取时间{datetime.fromtimestamp(result[timestamp]).strftime(%Y-%m-%d %H:%M:%S)} notifier.send(TO_EMAIL, subject, body) return True # 表示已发送通知 return False if __name__ __main__: db PriceDB() notifier EmailNotifier(SMTP_SERVER, SMTP_PORT, SENDER_EMAIL, SENDER_AUTH_CODE) # 配置要追踪的商品和目标价 products_to_track [ {url: https://www.amazon.com/dp/B08N5WRWNW, target_price: 30.0}, # 可以添加更多商品 ] for product in products_to_track: print(f正在检查 {product[url]}...) check_and_notify(product[url], product[target_price], db, notifier) # 礼貌性延时避免请求过快 time.sleep(random.uniform(2, 5)) db.close()3.5 使用系统Cron实现定时任务现在我们的核心脚本已经完成。我们需要让它定时比如每天上午10点和晚上8点各运行一次自动执行。在Linux/Mac上使用Cron:打开终端输入crontab -e编辑当前用户的cron任务。在文件末尾添加一行假设你的脚本位于/home/username/projects/price_tracker/tracker.py且虚拟环境位于同目录下的venv0 10,20 * * * cd /home/username/projects/price_tracker /home/username/projects/price_tracker/venv/bin/python tracker.py /home/username/projects/price_tracker/tracker.log 210 10,20 * * *表示在每天的第10小时和第20小时的0分即上午10点和晚上8点执行。cd ...切换到项目目录。venv/bin/python tracker.py使用虚拟环境中的Python解释器运行脚本。 tracker.log 21将脚本的标准输出和错误输出都重定向追加到tracker.log文件中方便查看运行日志和排错。在Windows上使用任务计划程序:搜索并打开“任务计划程序”。点击“创建基本任务”。按照向导设置任务名称、触发器例如“每天”、开始时间。在“操作”步骤选择“启动程序”。在“程序或脚本”中填写你的Python解释器完整路径例如C:\Users\YourName\projects\price_tracker\venv\Scripts\python.exe。在“添加参数”中填写你的脚本完整路径例如C:\Users\YourName\projects\price_tracker\tracker.py。在“起始于”中填写你的脚本所在目录例如C:\Users\YourName\projects\price_tracker。完成创建。4. 进阶优化与问题排查一个基础版本已经可以工作但要投入实际长期使用还需要考虑更多细节。4.1 增强健壮性错误处理与日志记录我们的脚本需要应对网络波动、网站改版、服务暂时不可用等各种异常。更全面的异常捕获在fetch_price函数中我们已经捕获了网络请求异常。还应该考虑BeautifulSoup解析异常、数据库连接和写入异常、邮件发送异常等。结构化日志使用Python内置的logging模块替代print语句。可以配置日志级别DEBUG, INFO, WARNING, ERROR并将日志同时输出到控制台和文件方便后期排查问题。import logging logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(tracker.log), logging.StreamHandler() ] ) logger logging.getLogger(__name__) # 使用 logger.info(消息) 代替 print设置重试机制对于网络请求失败可以加入简单的重试逻辑。import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def create_session_with_retries(retries3, backoff_factor0.5): session requests.Session() retry_strategy Retry( totalretries, backoff_factorbackoff_factor, # 重试间隔0.5s, 1s, 2s... status_forcelist[429, 500, 502, 503, 504], # 遇到这些状态码才重试 ) adapter HTTPAdapter(max_retriesretry_strategy) session.mount(http://, adapter) session.mount(https://, adapter) return session # 在fetch_price中使用 session create_session_with_retries(); response session.get(...)4.2 应对网站反爬策略升级如果发现抓取频繁失败返回403错误或验证码页面说明你的请求被识别为爬虫了。检查并更新请求头确保User-Agent是常见的浏览器字符串可以准备一个列表随机选择。添加Referer、Accept-Encoding等字段。使用代理IP这是应对IP封锁最有效的方法。可以使用一些免费的代理IP网站但稳定性差或购买可靠的代理服务。在requests中设置代理很简单proxies {http: http://10.10.1.10:3128, https: http://10.10.1.10:1080}然后传入requests.get。降低请求频率进一步增加请求间的随机延时并避免在固定时间点抓取。模拟浏览器行为终极方案如果上述方法都无效网站可能使用了复杂的JavaScript反爬。这时只能祭出Selenium或Playwright它们能完全模拟真人操作浏览器但代价是速度极慢、资源消耗大。仅作为最后的手段。4.3 扩展功能从单一到多源与可视化支持多个网站不同的网站页面结构不同。可以为每个网站如Amazon、BestBuy、Steam编写一个单独的解析函数或者创建一个配置系统将URL、价格选择器、标题选择器等作为配置项存入数据库或JSON文件。主程序根据URL的域名自动调用对应的解析配置。价格趋势可视化定期比如每周运行一个分析脚本从SQLite数据库中读取历史价格数据使用matplotlib或plotly生成价格走势图并通过邮件发送报告。构建Web界面使用Flask或Django搭建一个简单的Web后台让你可以通过网页添加/删除追踪商品、设置目标价、查看历史图表而无需修改代码和Cron配置。容器化部署使用Docker将整个应用Python环境、代码、Cron打包成一个镜像。这样可以在任何支持Docker的服务器或NAS上轻松部署和迁移环境隔离也更干净。4.4 常见问题排查速查表问题现象可能原因排查步骤与解决方案脚本运行无任何输出/报错1. Cron任务未正确执行。2. Python路径或虚拟环境路径错误。3. 脚本有语法错误在导入阶段就崩溃了。1. 检查Cron日志sudo grep CRON /var/log/syslog或查看任务计划程序历史记录。2. 在Cron命令中直接使用绝对路径。手动在终端运行完整的Cron命令看是否成功。3. 手动运行python -m py_compile tracker.py检查语法。抓取失败返回403 ForbiddenIP或请求头被识别为爬虫。1. 检查并随机化User-Agent。2. 添加Referer等头信息。3. 显著降低请求频率加入随机延时。4. 考虑使用代理IP。抓取成功但价格提取为None网页HTML结构已更改CSS选择器失效。1. 打开目标网页使用开发者工具重新检查价格元素的HTML结构。2. 更新代码中的price_selector。考虑使用更宽松但精准的父级选择器或结合多个特征定位。邮件发送失败1. SMTP服务器地址、端口错误。2. 邮箱账号、密码授权码错误。3. 发件邮箱未开启SMTP服务。4. 网络问题。1. 核对QQ/163等邮箱的官方SMTP设置文档。2. 确认使用的是授权码而非登录密码。3. 登录网页版邮箱在设置中确认POP3/SMTP服务已开启。4. 尝试使用telnet smtp.qq.com 465测试网络连通性。数据库写入错误1. 数据库文件被锁定多进程同时写。2. 磁盘空间不足。3. 数据库表结构不匹配。1. 确保同一时间只有一个脚本实例在运行。检查Cron任务是否有重叠。2. 检查磁盘空间。3. 如果修改了表结构可能需要删除旧的.db文件重新运行或编写数据库迁移脚本。这个项目就像一把瑞士军刀核心框架搭建好后你可以根据具体需求无限扩展其功能。从追踪一个商品开始逐步完善错误处理、增加数据源、美化通知、搭建面板整个过程本身就是一次绝佳的Python全栈实践。最关键的是它解决的是一个真实、高频的痛点这种从零到一创造工具并服务于自身需求的过程带来的成就感远超完成一个教程里的练习项目。 Python价格追踪器：从爬虫到自动化通知的完整实现
1. 项目概述为什么我们需要一个价格追踪器在电商购物、投资理财甚至日常消费中价格波动是常态。你是否曾为错过心仪商品的历史低价而懊恼是否在购买电子产品、机票或基金时因无法把握最佳时机而犹豫不决手动刷新比价网站不仅效率低下还容易遗漏关键节点。这正是“Python Price Tracker”价格追踪器项目诞生的初衷。它不是一个简单的脚本而是一个自动化、可定制、能持续运行的数据监控解决方案核心目标是通过编程手段解放你的时间和精力让你在价格博弈中占据信息优势。简单来说这个项目就是用Python写一个“智能小助手”它能按照你设定的规则自动抓取指定商品或资产的价格信息并在满足特定条件如价格低于阈值、出现大幅波动时通过邮件、短信或应用通知提醒你。这背后涉及的核心技术点包括网络数据抓取爬虫、数据解析、定时任务调度以及消息推送。无论你是想追踪显卡的降价情况、监控特定股票的价格还是关注某个航线的机票折扣这个工具都能派上用场。它适合有一定Python基础希望将编程技能应用于解决实际问题的开发者、数据分析爱好者以及精明的消费者。2. 核心思路与技术选型解析构建一个健壮的价格追踪器远不止写几行requests和BeautifulSoup那么简单。我们需要从架构层面思考其稳定性、可扩展性和易用性。整个系统可以拆解为四个核心模块数据获取模块、数据处理与存储模块、监控与逻辑判断模块以及通知告警模块。2.1 数据获取爬虫策略与反反爬应对数据是追踪器的血液。获取价格数据通常有两种途径通过目标网站的公开API或者直接解析网页Web Scraping。对于大多数电商网站公开API要么不存在要么有严格的调用限制和认证要求。因此网页抓取成为了更通用的选择。技术选型Requests BeautifulSoup 还是 Selenium这是一个关键抉择。Requests库高效、轻量适合抓取静态HTML页面。配合BeautifulSoup或lxml进行解析是绝大多数场景下的首选。然而许多现代网站尤其是使用了React、Vue等前端框架的站点的内容是动态加载的初始HTML中并不包含价格数据。这时就需要Selenium或Playwright这类浏览器自动化工具来模拟真实用户操作等待JavaScript执行完毕后再获取完整的页面内容。实操心得优先尝试Requests。先用浏览器开发者工具F12的Network面板查看页面加载过程。如果能在XHR或Fetch请求中找到直接返回价格数据的API接口通常是JSON格式那就用Requests模拟这个请求这是最稳定、最高效的方式。如果价格数据直接嵌在初始HTML中也用Requests。只有当数据确系动态渲染且找不到隐藏接口时才考虑使用Selenium因为它资源消耗大、速度慢。反反爬虫策略设置请求头Headers务必模拟真实浏览器的User-Agent并带上常见的Accept、Accept-Language等字段。控制请求频率在代码中引入随机延时如time.sleep(random.uniform(1, 3))避免短时间内高频请求导致IP被封。使用代理IP池对于需要大规模、高频追踪的场景可以考虑使用付费或自建的代理IP服务来轮换IP地址。处理Cookie和Session对于需要登录后才能查看价格的网站需要维护会话状态。2.2 数据处理与存储从临时变量到持久化记录抓取到的原始数据需要被清洗、提取出目标价格并保存下来以便进行历史对比和趋势分析。数据解析使用BeautifulSoup或lxml通过CSS选择器或XPath定位到包含价格的HTML元素。这里最大的挑战是网站页面结构可能会变动。因此选择器要尽可能精准且具有一定的容错性。提取出文本后需要用正则表达式或字符串方法清理掉货币符号、空格等非数字字符转换为浮点数。数据存储选型SQLite轻量级无需安装额外服务整个数据库就是一个文件。非常适合单机、个人使用的追踪器简单可靠。CSV/JSON文件最简单但频繁读写可能遇到并发问题且查询历史数据不如数据库方便。MySQL/PostgreSQL功能强大适合多用户、需要复杂查询的分布式应用。时序数据库如InfluxDB如果追踪标的物非常多且对时间序列数据的写入和查询性能有极高要求可以考虑。对于个人项目SQLite通常是完美选择。它足以记录每个商品每次抓取的时间戳、价格、以及其他你可能关心的信息如库存状态。2.3 监控逻辑与任务调度让程序自动运转起来追踪器的核心逻辑是“定期执行抓取任务并判断价格是否符合预设条件”。这涉及到定时任务调度。技术选型简单循环 time.sleep在脚本中使用while True循环每次执行完抓取和判断逻辑后休眠一段时间如1小时。这是最直白的实现但缺点是睡眠期间程序独占一个终端且不易管理。操作系统级任务调度在Linux上使用cron在Windows上使用“任务计划程序”。让操作系统定时调用你的Python脚本。这种方式将执行逻辑和调度逻辑解耦更清晰也便于日志管理。Python库调度使用schedule或APScheduler这类库在Python进程内实现更复杂的调度规则如每周一上午10点执行。它比系统级调度更灵活但需要保证Python进程常驻。监控逻辑设计 判断条件不应只是“当前价格低于设定值”。更聪明的策略包括历史价格对比当前价格是否低于过去7天/30天的平均价价格波动预警单日涨跌幅是否超过某个百分比组合条件仅在同时满足“价格低于X元”且“库存状态为‘有货’”时触发提醒。2.4 通知告警如何第一时间获知信息当监控逻辑被触发你需要被及时告知。通知方式应可靠且符合你的使用习惯。常见通知方式电子邮件SMTP最通用、最稳定的方式。使用Python的smtplib和email库可以轻松实现邮件发送。你需要一个发件邮箱建议使用QQ、163等提供的SMTP服务并开启授权码。桌面通知适用于脚本就在本地电脑运行的情况。可以使用plyer或win10toast仅Windows库发送系统桌面通知。手机推送集成推送服务使用如PushDeer、Server酱等第三方服务它们提供简单的API可以将消息推送到你的手机App。Telegram Bot创建一个Telegram机器人通过其API发送消息到你的私聊或频道。这种方式在技术圈非常流行稳定且功能强大。短信成本较高且依赖于第三方短信网关API一般不作为首选。注意事项无论选择哪种方式务必处理好通知发送失败的情况如网络错误、API限额。可以考虑加入重试机制并将发送失败的日志记录下来。3. 分步实现构建一个基础的电商价格追踪器下面我们以追踪一个亚马逊商品例如一本特定的书的价格为例手把手实现一个基础版本。我们将选择Requests BeautifulSoup SQLite 邮件通知 系统Cron调度这条技术栈因为它平衡了复杂度、稳定性和学习价值。3.1 环境准备与依赖安装首先确保你的Python环境是3.6以上版本。创建一个新的项目目录并初始化虚拟环境是一个好习惯。mkdir python_price_tracker cd python_price_tracker python -m venv venv # Windows: venv\Scripts\activate source venv/bin/activate # Linux/Mac安装必要的Python库pip install requests beautifulsoup4 lxmlrequests: 用于发送HTTP请求。beautifulsoup4和lxml: 用于解析HTML。lxml是一个解析器比Python内置的html.parser更快、更强大。我们暂时不安装数据库和邮件相关的库将在后续步骤中引入。3.2 编写核心抓取与解析函数创建一个名为tracker.py的文件。我们将首先实现最核心的功能给定一个商品URL抓取页面并提取价格。import requests from bs4 import BeautifulSoup import re import time import random def fetch_price(url): 从给定的URL抓取并解析价格。 返回一个字典包含价格、商品标题和抓取状态。 # 模拟浏览器的请求头 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: en-US,en;q0.9, } try: # 发送GET请求 response requests.get(url, headersheaders, timeout10) response.raise_for_status() # 如果状态码不是200抛出HTTPError异常 # 使用BeautifulSoup和lxml解析器解析HTML soup BeautifulSoup(response.content, lxml) # **关键步骤定位价格元素** # 你需要使用浏览器开发者工具找到价格所在的HTML元素和其选择器。 # 以下选择器是示例针对亚马逊页面可能有效但网站改版后会失效。 # 方法在商品页面上右键“检查”找到价格数字右键其父元素 - Copy - Copy selector price_selector #corePriceDisplay_desktop_feature_div div.a-section.a-spacing-none.aok-align-center span.a-price.aok-align-center.reinventPricePriceToPayMargin.priceToPay span:nth-child(2) span.a-price-whole price_element soup.select_one(price_selector) if price_element: # 提取文本并清理非数字字符如逗号、货币符号 price_text price_element.get_text(stripTrue) # 使用正则表达式提取数字包括小数点 price_match re.search(r[\d,]\.?\d*, price_text.replace(,, )) if price_match: price float(price_match.group()) else: price None else: price None print(f警告在 {url} 中未找到价格元素。选择器可能已过期。) # 尝试获取商品标题可选用于通知中显示 title_selector #productTitle title_element soup.select_one(title_selector) title title_element.get_text(stripTrue) if title_element else 未知商品 return { success: True, price: price, title: title, url: url, timestamp: time.time() } except requests.exceptions.RequestException as e: print(f请求失败: {e}) return {success: False, error: str(e), url: url} except Exception as e: print(f解析过程发生未知错误: {e}) return {success: False, error: str(e), url: url} # 测试函数 if __name__ __main__: # 替换成你想追踪的实际商品URL test_url https://www.amazon.com/dp/B08N5WRWNW # 示例URL result fetch_price(test_url) print(result) if result[success] and result[price]: print(f商品: {result[title][:50]}...) print(f价格: ${result[price]:.2f})重要提示上述代码中的price_selector和title_selector是示例来源于特定时间点的亚马逊页面结构。网站前端随时可能改版导致选择器失效。这是网页抓取项目最大的维护成本。你需要根据目标网站的实际HTML结构使用开发者工具重新定位并更新这些选择器。3.3 集成SQLite进行数据持久化我们需要一个地方来存储每次抓取到的价格记录。修改tracker.py增加数据库操作。首先安装SQLite驱动Python标准库已包含无需额外安装。我们创建一个数据库帮助类。import sqlite3 from datetime import datetime class PriceDB: def __init__(self, db_pathprices.db): self.conn sqlite3.connect(db_path) self.create_table() def create_table(self): 创建存储价格记录的表 cursor self.conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS price_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, product_url TEXT NOT NULL, product_title TEXT, price REAL, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP ) ) # 为商品URL和日期创建索引加速查询 cursor.execute(CREATE INDEX IF NOT EXISTS idx_url ON price_history (product_url)) cursor.execute(CREATE INDEX IF NOT EXISTS idx_timestamp ON price_history (timestamp)) self.conn.commit() def insert_price(self, url, title, price): 插入一条新的价格记录 cursor self.conn.cursor() cursor.execute( INSERT INTO price_history (product_url, product_title, price) VALUES (?, ?, ?) , (url, title, price)) self.conn.commit() return cursor.lastrowid def get_price_history(self, url, limit30): 获取某个商品最近的价格历史 cursor self.conn.cursor() cursor.execute( SELECT price, timestamp FROM price_history WHERE product_url ? ORDER BY timestamp DESC LIMIT ? , (url, limit)) return cursor.fetchall() def close(self): self.conn.close()然后修改主逻辑在抓取成功后插入数据。def track_product(url, db): 追踪单个商品抓取价格并存入数据库 result fetch_price(url) if result[success] and result[price] is not None: db.insert_price(url, result[title], result[price]) print(f[{datetime.now()}] 已记录 {result[title][:30]}... 价格: ${result[price]:.2f}) return result else: print(f[{datetime.now()}] 抓取失败: {result.get(error, 未知错误)}) return None if __name__ __main__: db PriceDB() test_url https://www.amazon.com/dp/B08N5WRWNW track_product(test_url, db) # 可以查询历史记录 history db.get_price_history(test_url, 5) for price, ts in history: print(f{ts}: ${price}) db.close()3.4 实现邮件通知功能当价格低于我们的目标价时发送邮件提醒。我们需要配置发件邮箱。这里以QQ邮箱的SMTP服务为例。首先你需要登录QQ邮箱在“设置”-“账户”中开启POP3/SMTP服务并获取一个授权码不是邮箱密码。修改tracker.py添加邮件发送功能import smtplib from email.mime.text import MIMEText from email.header import Header class EmailNotifier: def __init__(self, smtp_server, smtp_port, sender_email, sender_auth_code): self.smtp_server smtp_server self.smtp_port smtp_port self.sender_email sender_email self.sender_auth_code sender_auth_code def send(self, to_email, subject, body): 发送邮件 message MIMEText(body, plain, utf-8) message[From] Header(fPrice Tracker {self.sender_email}, utf-8) message[To] Header(to_email, utf-8) message[Subject] Header(subject, utf-8) try: # 使用SSL加密连接 with smtplib.SMTP_SSL(self.smtp_server, self.smtp_port) as server: server.login(self.sender_email, self.sender_auth_code) server.sendmail(self.sender_email, [to_email], message.as_string()) print(f[{datetime.now()}] 邮件已发送至 {to_email}) return True except Exception as e: print(f[{datetime.now()}] 邮件发送失败: {e}) return False # 配置你的邮箱信息务必保密不要将授权码提交到Git等公开仓库 SMTP_SERVER smtp.qq.com SMTP_PORT 465 SENDER_EMAIL your_qqqq.com # 你的QQ邮箱 SENDER_AUTH_CODE your_authorization_code # 你的授权码 TO_EMAIL your_notification_emailexample.com # 接收通知的邮箱 def check_and_notify(url, target_price, db, notifier): 检查价格并决定是否发送通知 result track_product(url, db) # 抓取并记录 if result and result[price] target_price: subject f价格提醒{result[title][:50]}... 已降至目标价以下 body f 商品{result[title]} 当前价格${result[price]:.2f} 目标价格${target_price:.2f} 商品链接{url} 抓取时间{datetime.fromtimestamp(result[timestamp]).strftime(%Y-%m-%d %H:%M:%S)} notifier.send(TO_EMAIL, subject, body) return True # 表示已发送通知 return False if __name__ __main__: db PriceDB() notifier EmailNotifier(SMTP_SERVER, SMTP_PORT, SENDER_EMAIL, SENDER_AUTH_CODE) # 配置要追踪的商品和目标价 products_to_track [ {url: https://www.amazon.com/dp/B08N5WRWNW, target_price: 30.0}, # 可以添加更多商品 ] for product in products_to_track: print(f正在检查 {product[url]}...) check_and_notify(product[url], product[target_price], db, notifier) # 礼貌性延时避免请求过快 time.sleep(random.uniform(2, 5)) db.close()3.5 使用系统Cron实现定时任务现在我们的核心脚本已经完成。我们需要让它定时比如每天上午10点和晚上8点各运行一次自动执行。在Linux/Mac上使用Cron:打开终端输入crontab -e编辑当前用户的cron任务。在文件末尾添加一行假设你的脚本位于/home/username/projects/price_tracker/tracker.py且虚拟环境位于同目录下的venv0 10,20 * * * cd /home/username/projects/price_tracker /home/username/projects/price_tracker/venv/bin/python tracker.py /home/username/projects/price_tracker/tracker.log 210 10,20 * * *表示在每天的第10小时和第20小时的0分即上午10点和晚上8点执行。cd ...切换到项目目录。venv/bin/python tracker.py使用虚拟环境中的Python解释器运行脚本。 tracker.log 21将脚本的标准输出和错误输出都重定向追加到tracker.log文件中方便查看运行日志和排错。在Windows上使用任务计划程序:搜索并打开“任务计划程序”。点击“创建基本任务”。按照向导设置任务名称、触发器例如“每天”、开始时间。在“操作”步骤选择“启动程序”。在“程序或脚本”中填写你的Python解释器完整路径例如C:\Users\YourName\projects\price_tracker\venv\Scripts\python.exe。在“添加参数”中填写你的脚本完整路径例如C:\Users\YourName\projects\price_tracker\tracker.py。在“起始于”中填写你的脚本所在目录例如C:\Users\YourName\projects\price_tracker。完成创建。4. 进阶优化与问题排查一个基础版本已经可以工作但要投入实际长期使用还需要考虑更多细节。4.1 增强健壮性错误处理与日志记录我们的脚本需要应对网络波动、网站改版、服务暂时不可用等各种异常。更全面的异常捕获在fetch_price函数中我们已经捕获了网络请求异常。还应该考虑BeautifulSoup解析异常、数据库连接和写入异常、邮件发送异常等。结构化日志使用Python内置的logging模块替代print语句。可以配置日志级别DEBUG, INFO, WARNING, ERROR并将日志同时输出到控制台和文件方便后期排查问题。import logging logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(tracker.log), logging.StreamHandler() ] ) logger logging.getLogger(__name__) # 使用 logger.info(消息) 代替 print设置重试机制对于网络请求失败可以加入简单的重试逻辑。import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def create_session_with_retries(retries3, backoff_factor0.5): session requests.Session() retry_strategy Retry( totalretries, backoff_factorbackoff_factor, # 重试间隔0.5s, 1s, 2s... status_forcelist[429, 500, 502, 503, 504], # 遇到这些状态码才重试 ) adapter HTTPAdapter(max_retriesretry_strategy) session.mount(http://, adapter) session.mount(https://, adapter) return session # 在fetch_price中使用 session create_session_with_retries(); response session.get(...)4.2 应对网站反爬策略升级如果发现抓取频繁失败返回403错误或验证码页面说明你的请求被识别为爬虫了。检查并更新请求头确保User-Agent是常见的浏览器字符串可以准备一个列表随机选择。添加Referer、Accept-Encoding等字段。使用代理IP这是应对IP封锁最有效的方法。可以使用一些免费的代理IP网站但稳定性差或购买可靠的代理服务。在requests中设置代理很简单proxies {http: http://10.10.1.10:3128, https: http://10.10.1.10:1080}然后传入requests.get。降低请求频率进一步增加请求间的随机延时并避免在固定时间点抓取。模拟浏览器行为终极方案如果上述方法都无效网站可能使用了复杂的JavaScript反爬。这时只能祭出Selenium或Playwright它们能完全模拟真人操作浏览器但代价是速度极慢、资源消耗大。仅作为最后的手段。4.3 扩展功能从单一到多源与可视化支持多个网站不同的网站页面结构不同。可以为每个网站如Amazon、BestBuy、Steam编写一个单独的解析函数或者创建一个配置系统将URL、价格选择器、标题选择器等作为配置项存入数据库或JSON文件。主程序根据URL的域名自动调用对应的解析配置。价格趋势可视化定期比如每周运行一个分析脚本从SQLite数据库中读取历史价格数据使用matplotlib或plotly生成价格走势图并通过邮件发送报告。构建Web界面使用Flask或Django搭建一个简单的Web后台让你可以通过网页添加/删除追踪商品、设置目标价、查看历史图表而无需修改代码和Cron配置。容器化部署使用Docker将整个应用Python环境、代码、Cron打包成一个镜像。这样可以在任何支持Docker的服务器或NAS上轻松部署和迁移环境隔离也更干净。4.4 常见问题排查速查表问题现象可能原因排查步骤与解决方案脚本运行无任何输出/报错1. Cron任务未正确执行。2. Python路径或虚拟环境路径错误。3. 脚本有语法错误在导入阶段就崩溃了。1. 检查Cron日志sudo grep CRON /var/log/syslog或查看任务计划程序历史记录。2. 在Cron命令中直接使用绝对路径。手动在终端运行完整的Cron命令看是否成功。3. 手动运行python -m py_compile tracker.py检查语法。抓取失败返回403 ForbiddenIP或请求头被识别为爬虫。1. 检查并随机化User-Agent。2. 添加Referer等头信息。3. 显著降低请求频率加入随机延时。4. 考虑使用代理IP。抓取成功但价格提取为None网页HTML结构已更改CSS选择器失效。1. 打开目标网页使用开发者工具重新检查价格元素的HTML结构。2. 更新代码中的price_selector。考虑使用更宽松但精准的父级选择器或结合多个特征定位。邮件发送失败1. SMTP服务器地址、端口错误。2. 邮箱账号、密码授权码错误。3. 发件邮箱未开启SMTP服务。4. 网络问题。1. 核对QQ/163等邮箱的官方SMTP设置文档。2. 确认使用的是授权码而非登录密码。3. 登录网页版邮箱在设置中确认POP3/SMTP服务已开启。4. 尝试使用telnet smtp.qq.com 465测试网络连通性。数据库写入错误1. 数据库文件被锁定多进程同时写。2. 磁盘空间不足。3. 数据库表结构不匹配。1. 确保同一时间只有一个脚本实例在运行。检查Cron任务是否有重叠。2. 检查磁盘空间。3. 如果修改了表结构可能需要删除旧的.db文件重新运行或编写数据库迁移脚本。这个项目就像一把瑞士军刀核心框架搭建好后你可以根据具体需求无限扩展其功能。从追踪一个商品开始逐步完善错误处理、增加数据源、美化通知、搭建面板整个过程本身就是一次绝佳的Python全栈实践。最关键的是它解决的是一个真实、高频的痛点这种从零到一创造工具并服务于自身需求的过程带来的成就感远超完成一个教程里的练习项目。