简介面向电商价格监控需求的 Python 2 项目依托 requests、sqlite 与代理池构建从商品页抓取、价格解析到邮件提醒的完整监控链路适合需要自动追踪降价动态、或希望学习爬虫与自动化监控的开发者。资源共 34 个文件含 18 个 Python 脚本、SQLite 数据库、JSON 配置、Markdown 文档与图片等压缩包约 1.13MBpy 文件覆盖代理池管理、网页解析、邮件发送、定时调度等核心模块db 文件保存监控商品与历史价格md/txt 提供说明与配置参考。已有 363 人学习下载。整套代码可作为直接运行的价格监控工具也能帮助读者掌握代理 IP 随机轮换、多页面数据提取、异常处理与数据持久化等实战思路适合在旧版 Python 环境下进行二次开发和功能扩展。 这两年电商价格监控的需求一直挺旺我自己就接过好几个类似的项目。客户需求大同小异盯着竞品价格变了就通知最好还能存个历史趋势。技术选型上有人用Node、有人用Go但我手里这套老项目是用Python 2 requests sqlite 代理池搭的运行了一两年日均采集几万条商品数据还算稳定。这项目标题叫“Python-Pricemonitorpy2电商价格监控”核心就干三件事用requests抓取商品页面用sqlite做本地存储用代理池解决封IP的问题。它会碰到一个所有爬虫都绕不开的坎——429 Too Many Requests。这个HTTP状态码几乎是每个爬虫开发者的“成人礼”遇不到说明你的量还不够大遇到了反而是好事说明你的爬虫真的在生产环境里跑了。写这篇东西是想把整套方案掰开揉碎讲清楚。适合谁看呢第一类想自己搞个价格监控小工具、又不想用现成SaaS的人。第二类正在学Python爬虫想看看真实项目里requests、sqlite、代理池是怎么配合的。第三类接了类似外包活需要一个能直接改改就上线的底子。这篇不会从头教Python语法但代码部分我尽量写得能直接跑你拿到手改改选择器和价格提取规则就能用。1. 项目定位与整体架构设计1.1 需求拆解价格监控到底在监控什么表面上看价格监控就是“定时抓个页面把价格存下来”。但真把需求摊开里面藏着不少细节。我接手这个项目时的原始需求只有一句话“盯着几个电商平台上我们竞品的价格变了要提醒我们。”但这句话背后至少要拆出这么几个问题抓哪些商品商品ID怎么维护新增商品和下线商品怎么处理多久抓一次实时盯盘还是每天两次频次直接决定被封IP的概率。价格变了怎么通知邮件、短信、还是钉钉/企业微信机器人历史价格存多久客户要的是周报趋势图还是只看当前价格这个项目最终定下来的方案是商品列表维护在sqlite里抓取频率默认每30分钟一次价格变化通过邮件通知历史数据永久保留。存储用sqlite而不是MySQL原因是单机部署、数据量不大几百万条以内、不想额外维护一个数据库服务。这个选型在当时看是非常务实的。1.2 技术选型为什么是Python2 requests sqlite 代理池先说Python2。这项目是几年前起的那时候Python2还在维护期第三方库的兼容性也最稳。现在做新项目肯定上Python3但如果你手里有老项目要维护或者看到某些生产环境还在跑Python2脚本别急着嗤之以鼻——能稳定跑一年的代码就是好代码。requests库是Python里最顺手的HTTP客户端对比标准库urllib2注意Python2里是urllib2不是urllib3requests的session管理、headers定制、超时控制都人性化得多。咱们这代写爬虫的基本都是从requests入门它让HTTP请求这件事变得像“发个快递”一样简单填好地址URL、贴好单子Headers、选好快递方式GET/POST寄出去等着就行了。sqlite用于本地存储选它的逻辑很简单文件即数据库零配置。对于价格监控这种单机应用sqlite的性能绰绰有余。Python2标准库里自带sqlite3模块不用装任何额外的东西。代理池是这套架构里最核心也最容易被新手忽略的部分。电商平台的反爬策略不是吃素的同一IP高频访问轻则给你个验证码重则直接封IP段。代理池就是你的“IP弹药库”一批被封了下一批顶上。1.3 整体架构一条数据从网页到数据库的旅程整个系统的数据流大概是这样商品配置表里维护着要监控的商品ID和URL模板。调度器一个简单的while循环加sleep按固定间隔触发抓取任务。抓取模块通过代理池获取可用代理带着代理IP去请求目标页面。拿到HTML后用正则或简单的字符串匹配提取价格、标题、库存状态。最后把提取结果写入sqlite的价格历史表同时和上一次的价格做对比如果变化幅度超过阈值触发通知逻辑。这个链路不复杂但每个环节都有坑。我后面会按模块把代码和设计思路一块儿说清楚。2. 核心模块解析与关键技术点2.1 requests在Python2中的正确打开方式Python2的requests库和Python3版在用法上没太大区别但有几个细节容易踩坑。第一是响应内容的编码处理。Python2里requests的r.text默认会通过headers里的charset猜测编码但很多电商页面不严格按规范来猜错的概率不低。最稳妥的做法是从r.contentbytes里手动解编码或者直接用r.encoding utf-8强制指定。第二是连接复用。requests的Session对象会复用底层TCP连接对同一域名的大量请求能显著减少握手开销同时也能更好地模拟浏览器行为。Session在Python2里的用法和Python3一致但你在老代码里经常能看到裸露的requests.get()那其实每次都在创建新连接效率低不少。第三是超时设置。requests.get()如果不加timeout参数默认是无限等待。在生产环境里一个慢接口能拖住整个抓取线程。我的实践是统一设置timeout(3.05, 10)分别表示连接超时和读取超时。这个值的含义是3.05秒内没建立连接就放弃10秒内没读到数据也放弃。2.2 代理池的落地实现不能用的时候才是关键代理池这个东西网上有很多花里胡哨的实现什么Redis队列、什么SSDB存储、什么定时校验我这个项目一个都没用。原因很简单就一台服务器没必要引那么多中间件。代理池的核心需求就两个一个池子用来存代理一套机制来保证池子里的代理尽量可用。我的实现是用Python的list做存储配合一个简单的校验线程。代理来源是几个免费代理网站的公开代理每5分钟抓一轮测速后写入池子。校验方式是对目标电商域名发一个带代理的请求能在一个可接受的时间比如8秒内返回且状态码为200就认为是可用代理。这里有个很重要的点代理的有效性是针对目标站点的。一个代理对你测试的A站可用不代表它对B站也可用。有些代理出口IP本身就在电商平台的黑名单里怎么测都是废的。所以校验必须直接拿目标站点测不能图省事随便找个通用的检测URL。代理池的另一个设计细节是“超额请求”的处理。如果一个代理在连续N次请求中失败次数过多就把它暂时移出池子待会儿再回来验证。这能避免同一个坏代理反复被选中白白浪费时间。这个策略被后面要讲的429错误直接验证了价值。2.3 sqlite在Python2中的存储方案sqlite3模块在Python2.5之后就是标准库了用法很稳定。价格监控的数据模型就两张表goods表商品ID、商品名称、平台、商品URL、状态启用/停用price_history表自增ID、商品ID、价格、采集时间、原始标题快照price_history表是典型的时序数据会持续膨胀。我的策略是每季度归档一次把超过一年的数据导出成CSV后从主表删除。查询价格趋势时可视化工具直接读sqlite文件省去导出数据的麻烦。写入方面用executemany批量插入比逐条execute快一个数量级。还有一个小技巧给price_history表建复合索引索引字段是(goods_id, created_at)这样按商品查历史趋势的SQL会快很多。不加索引的话数据量上了百万一条查询能卡好几秒。对了sqlite对并发写入的支持很弱Python2的线程里如果多线程同时写同一个库文件容易报database is locked。我的方案是全局只用一个写连接所有写入操作串行化。这样做的代价是峰值吞吐量受限但对价格监控这种低频写入的场景完全够用。3. 实操过程与核心代码实现3.1 环境准备一台Linux服务器Python2环境部署环境是CentOS 7自带Python2.7.5。用virtualenv隔离项目依赖避免污染系统Python。需要装的包就三个requests、BeautifulSoup4解析页面用、还有标准库的sqlite3和smtplib发邮件用。用pip安装时要注意Python2的pip版本必须小于21.0否则装不了带二进制扩展的包。安装命令是pip install requests2.22.0 beautifulsoup44.6.0为什么锁版本因为更高版本的requests已经放弃Python2支持了4.6.0的BeautifulSoup是最后一个支持Python2的版本。这种老项目依赖锁定版本是保平安的唯一方式。3.2 数据模型与初始化建表先初始化数据库。这个脚本可以直接在命令行跑每个部分独立可测CREATE TABLE IF NOT EXISTS goods ( id INTEGER PRIMARY KEY AUTOINCREMENT, sku TEXT NOT NULL UNIQUE, name TEXT NOT NULL DEFAULT , platform TEXT NOT NULL DEFAULT , goods_url TEXT NOT NULL, enabled INTEGER NOT NULL DEFAULT 1, created_at TIMESTAMP DEFAULT (datetime(now, localtime)) ); CREATE TABLE IF NOT EXISTS price_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, goods_id INTEGER NOT NULL, price REAL NOT NULL, title TEXT, created_at TIMESTAMP DEFAULT (datetime(now, localtime)) ); CREATE INDEX IF NOT EXISTS idx_price_history_goods_time ON price_history(goods_id, created_at);sku字段是商品唯一标识用于和电商平台页面对应。goods表本身的增删就是运营后台的活这里先不展开。3.3 requests抓取封装Session复用代理注入重试机制上核心代码。这是一个抓取模块的封装把代理注入、超时、重试都揉在一起# -*- coding: utf-8 -*- import requests import random import logging import time logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) class Fetcher(object): def __init__(self, proxy_pool): self.proxy_pool proxy_pool self.session requests.Session() self.session.headers.update({ User-Agent: self._fake_ua(), Accept-Language: zh-CN,zh;q0.8, Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, }) def _fake_ua(self): ua_list [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/83.0.4103.116 Safari/537.36, Mozilla/5.0 (iPhone; CPU iPhone OS 12_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/12.0 Mobile/15A372 Safari/604.1 ] return random.choice(ua_list) def fetch(self, url, retry_times3): for attempt in range(retry_times): proxy self.proxy_pool.get_one() proxies {http: http://%s % proxy, https: http://%s % proxy} try: resp self.session.get(url, proxiesproxies, timeout(3.05, 10)) if resp.status_code 200: resp.encoding utf-8 return resp.text elif resp.status_code 429: logger.warning(Proxy %s got 429, remove it and retry., proxy) self.proxy_pool.remove(proxy) time.sleep(0.5) continue else: logger.warning(Unexpected status code: %s, proxy: %s, resp.status_code, proxy) continue except requests.RequestException as e: logger.warning(Request failed via %s: %s, proxy, e) self.proxy_pool.remove(proxy) time.sleep(0.3) return None几个设计细节拆解一下。UA轮换。UAUser-Agent是浏览器在HTTP请求头里标注“我是谁”的字段。很多网站反爬的第一步就是过滤没有UA的请求。我不会只用一个固定UA而是准备一个池子随机换虽然不能完全规避风控但能少碰一些傻黑名单。重试策略。fetch方法里的retry_times3意味着每个页面最多尝试三次。每次重试都会换一个代理因为失败原因很可能是当前代理已经被目标站标记了。三次都失败就返回None由上层逻辑决定是记日志还是跳过。针对性处理429。这个状态码在这个项目里简直比200还常见。429意味着请求频率过高触发了目标站的限流策略。代码里的处理方式是遇到429判定当前代理已经“不干净”直接从池子里移除让上层换下一个代理。这一步对长期稳定运行至关重要。3.4 价格提取正则表达式才是王道提取价格我选择用正则而不是BeautifulSoup。原因很实际电商页面的价格往往在动态渲染的节点里BeautifulSoup按class取节点经常扑空但价格数字本身在HTML源代码里反而有比较稳定的特征。举个例子某平台的页面源码里价格长这样span classprice-now i¥/i199.00 /span我的提取正则就是import re PRICE_PATTERNS [ rprice\s*:\s*?(\d(?:\.\d{1,2})?)?,, rclass(?:price|price-now|current-price)[^]*\s*(?:[^])*\s*(?:¥|)?\s*(\d(?:\.\d{1,2})?), r(\d\.\d{2}), ] def extract_price(html): for pattern in PRICE_PATTERNS: m re.search(pattern, html) if m: try: return float(m.group(1)) except ValueError: continue return None写正则提取价格要注意几个边界情况有的页面价格带千分位分隔符1,999.00有的价格是区间199.00-299.00有的商品无货时价格显示为--。这些都得在正则里或者后续解析时处理掉。如果正则提取出来的是一个区间我会取较低值并在日志里标记。你可以直接用这个正则跑一遍自己盯着的平台看看命中率。如果命中率低就按你自己看到的页面源码特征调整。这一步是整个项目的核心页面上没那么多花哨的库比得上一个能匹配到实际数据模式的正则。3.5 数据入库与变化检测价格拿到手后先写price_history表再和最后一条历史记录比较。如果价格变动超过设置的阈值触发通知。import sqlite3 import smtplib from email.mime.text import MIMEText class PriceMonitor(object): def __init__(self, db_path, smtp_conf): self.conn sqlite3.connect(db_path) self.smtp_conf smtp_conf def save_price(self, goods_id, price, title): cur self.conn.cursor() cur.execute( INSERT INTO price_history (goods_id, price, title) VALUES (?, ?, ?), (goods_id, price, title) ) self.conn.commit() last_price cur.execute( SELECT price FROM price_history WHERE goods_id? ORDER BY id DESC LIMIT 1, (goods_id,) ).fetchone() return last_price[0] if last_price else None def check_and_notify(self, goods_id, goods_name, new_price, last_price, threshold0.03): if last_price is None: return change abs(new_price - last_price) / float(last_price) if change threshold: subject [价变提醒] %s 从 %.2f 变为 %.2f % (goods_name, last_price, new_price) body 商品 %sID%d价格变化 %.2f%% % (goods_name, goods_id, change * 100) self._send_email(subject, body)save_price里注意一个细节我先插入新记录再去查最后一条价格。这样查到的就是刚插入的这条用它拿到“当前最新价”。而真正用于对比的“上一次价格”应该在插入之前查。这个逻辑在完整代码里要调整一下顺序上面这段示例仅为演示表结构实际用的时候建议把查询放到insert之前避免把刚写入的新纪录当成对比基准。3.6 调度器最简单的while循环就是最稳的调度就一个while循环。但要注意几个细节用time.sleep控制间隔每次循环检查一下代理池是不是空的空了就触发补货线程捕获所有异常避免整个进程因单个商品解析失败而退出。def run(self): while True: start time.time() goods_list self.get_enabled_goods() for goods in goods_list: try: html self.fetcher.fetch(goods[goods_url]) if html is None: continue price extract_price(html) if price is None: logger.warning(No price found for %s, goods[sku]) continue last_price self.get_last_price(goods[id]) self.save_price(goods[id], price, goods[name]) self.check_and_notify(goods[id], goods[name], price, last_price) except Exception as e: logger.exception(Error processing goods %s: %s, goods[sku], e) continue elapsed time.time() - start sleep_time max(0, self.interval - elapsed) logger.info(Round done in %.2f seconds, sleep for %.2f seconds, elapsed, sleep_time) time.sleep(sleep_time)这里把整个抓取round的耗时算出来用interval减去elapsed保证每轮周期稳定在设定的间隔左右而不是“每轮开始后固定睡N秒”导致实际频率偏快。执行久了你会发现这个简单的调度器才是最省心的部件。4. 常见问题与排查技巧实录4.1 429 Too Many Requests限流的正确应对姿势这个项目上线后我遇到最多的就是429。它的含义是请求太频繁目标服务器不想理你了。区分一下两种典型的“请求过多”一种是单IP维度IP频率触发阈值此时需要换代理IP另一种是账号/设备维度Cookies可能被标记了此时需要清Cookie、换身份。排查429的第一步确认是不是同一个IP高频请求导致。看日志里失败请求用的代理如果一段时间内同一个代理IP反复触发429说明它已经进了目标站的黑名单移出池子就行。如果所有代理都触发429说明是整个出口网段被盯上了或者你请求的频率真的太高需要拉大间隔。第二步观察429响应里是否带Retry-After头。有些平台会告诉你“多少秒后再来”。带上这个头做sleep比无脑重试更友好也更不容易把平台惹毛。第三步也是很多人忽略的429不代表必须硬刚换IP。有时候降低单个商品页面的请求频率就能解决。比如30分钟一轮改成45分钟一轮牺牲一点实时性换取稳定运行长远来看更划算。4.2 代理池枯竭免费代理说挂就挂免费代理的稳定性用过的人都懂。今天测出来还能用的10个代理明天可能就剩2个。代理池枯竭的典型症状是日志里大量删除代理的记录然后fetch返回None整个采集轮次颗粒无收。解决办法有两个方向一个是从源头扩充代理来源多爬几个免费代理站做去重和合并另一个是降低对代理池的依赖频率比如设置一个“本地直连模式”当代理池可用数量低于阈值时同一轮内改用直连不经过代理试探性采集几个页面。注意直连模式仅适合低频抓取高频还是会被拦。我后期的做法更省心给代理池增加了一个“持久化”功能把验证通过的代理写进sqlite表里重启进程后自动加载。免费代理虽说不稳定但这些IP从“存活”到“失效”的时间窗口往往足够跑完下一轮采集。这个状态下至少代理池不会因为进程重启而归零。4.3 数据库锁定与并发写冲突sqlite的database is locked错误我在项目早期遇到得挺多。原因是Python2的sqlite3模块默认会为每个连接创建独立的事务不同线程同时写就锁住了。我的解决方案是所有写操作统一走一个连接这个连接在程序启动时创建整个生命周期不关闭。并且所有写操作都在同一个线程里执行。如果你有多线程读的需求读可以用只读连接connect时带上file:...?modero写还是走那唯一一个。这套方案在数据量到百万级之前都没问题。如果真有一天数据量大到sqlite扛不住再考虑平滑迁移到PostgreSQL也不迟但至少目前这个项目跑了这么久sqlite完全没成为瓶颈。4.4 字符集与乱码Python2的招牌暗坑Python2处理字符串编码说多了都是泪。响应内容如果是UTF-8requests会正确处理但也可能返回GBK编码的页面。此时resp.text乱码而resp.content是一堆bytes。我的经验是不要依赖requests自动解码一律resp.encoding utf-8如果发现乱码再把它改成gbk实测目标平台页面编码后固定下来。另外在Python2里从页面正则提取出的字符串要decode成unicode写进sqlite或者拼邮件正文时保持一致不然就是经典的UnicodeEncodeError。如果你的环境已经切到Python3这块就不用操心了。但如果你在维护老项目字符串编码问题建议在所有入口处以utf-8为准统一下来能省掉大量调试时间。4.5 504与网关超时不是你的错也别硬刚有时候请求返回504 Gateway Timeout这往往是目标站反代或者CDN层面的问题代理本身可能还是好的。遇到504正确做法是标记该代理“本次失败但不清出池子”下次还可以继续用。如果连续多次都504再考虑淘汰。这里的判断逻辑可以在代理池的fail计数里做连续失败5次才移除而不是一次失败就remove。这个阈值可以调我调试下来发现3到5次比较合理。太低会误杀偶然失败的代理太高会拖慢整体速度。5. 几个容易被忽略但很重要的细节5.1 日志与可观测性没人愿意盯着黑框框爬虫程序如果不打日志出了问题就是灾难。我的日志规范是抓取开始和结束各有一次INFO日志失败和异常用WARNING或ERROR价格变化单独打一条醒目日志方便事后统计触发次数。日志格式统一包含时间戳、级别、模块名、消息。生产环境建议再配一个日志周期归档用Python的logging.handlers.TimedRotatingFileHandler按天切割日志文件保留最近30天。这样日志文件不会无限膨胀排错时也能快速定位某天的记录。5.2 请求频率设计追求稳定不追求激进电商价格监控每分钟都在变的价格意义不大。一小时内的价格波动对普通消费者或者经销商来说参考价值极低。所以我的建议是每小时2到4轮就足够也就是15到30分钟间隔。太低频抓不到瞬时价格高频只会徒增封号风险。如果你监控的商品量大还可以把商品列表分组每组错峰抓取。比如100个商品分成4组每组25个每组间隔7分钟启动这样请求在时间轴上摊开不容易形成瞬时脉冲。5.3 邮件通知的发送频率控制价格变化通知如果不做频率控制可能一天收到几十封邮件最后被当成垃圾邮件忽略了。我的经验是单商品每天最多提醒一次。具体做法是在goods表里加last_notify_date字段当天已经提醒过的商品即使价格再变也不再发。这个“一天一次”的约束在客户那边反馈非常好。他们不是要实时盯盘而是想知道“今天哪几个商品值得关注”。把噪音降到最低通知的价值才会被放大。6. 一套更干净的改进方向如果你不介意用Python3重写一遍几个改动会让这个项目舒服很多用schedule库替代手写while循环调度用SQLAlchemy替代裸sqlite3ORM写起来省事用Celery定时任务替代常驻进程天然支持分布式用Redis做代理池和管理队列支持多台机器共享但我要提醒一句改动越大调试成本越高。如果目标只是跑通价格监控这套老方案完全够用。想升级也要在“线上稳定运行”的前提下逐步替换模块千万别一次性重写。最后再分享一个小经验。这类跑了好几年的脚本最大的风险不是技术过时而是没人维护。建议给每个模块写清楚注释把代理池的验证逻辑、价格提取正则怎么改、通知频率在哪调全部写进README里。哪怕半年后再接手也不至于对着代码发呆。我这套方案在你自己的场景里跑起来可能还会遇到一些问题欢迎带着具体的报错来交流。爬虫这条路踩坑是常态但把坑填平了后面就顺了。本文还有配套的精品资源点击获取 Python电商价格监控实战：requests+代理池+sqlite全解析
简介面向电商价格监控需求的 Python 2 项目依托 requests、sqlite 与代理池构建从商品页抓取、价格解析到邮件提醒的完整监控链路适合需要自动追踪降价动态、或希望学习爬虫与自动化监控的开发者。资源共 34 个文件含 18 个 Python 脚本、SQLite 数据库、JSON 配置、Markdown 文档与图片等压缩包约 1.13MBpy 文件覆盖代理池管理、网页解析、邮件发送、定时调度等核心模块db 文件保存监控商品与历史价格md/txt 提供说明与配置参考。已有 363 人学习下载。整套代码可作为直接运行的价格监控工具也能帮助读者掌握代理 IP 随机轮换、多页面数据提取、异常处理与数据持久化等实战思路适合在旧版 Python 环境下进行二次开发和功能扩展。 这两年电商价格监控的需求一直挺旺我自己就接过好几个类似的项目。客户需求大同小异盯着竞品价格变了就通知最好还能存个历史趋势。技术选型上有人用Node、有人用Go但我手里这套老项目是用Python 2 requests sqlite 代理池搭的运行了一两年日均采集几万条商品数据还算稳定。这项目标题叫“Python-Pricemonitorpy2电商价格监控”核心就干三件事用requests抓取商品页面用sqlite做本地存储用代理池解决封IP的问题。它会碰到一个所有爬虫都绕不开的坎——429 Too Many Requests。这个HTTP状态码几乎是每个爬虫开发者的“成人礼”遇不到说明你的量还不够大遇到了反而是好事说明你的爬虫真的在生产环境里跑了。写这篇东西是想把整套方案掰开揉碎讲清楚。适合谁看呢第一类想自己搞个价格监控小工具、又不想用现成SaaS的人。第二类正在学Python爬虫想看看真实项目里requests、sqlite、代理池是怎么配合的。第三类接了类似外包活需要一个能直接改改就上线的底子。这篇不会从头教Python语法但代码部分我尽量写得能直接跑你拿到手改改选择器和价格提取规则就能用。1. 项目定位与整体架构设计1.1 需求拆解价格监控到底在监控什么表面上看价格监控就是“定时抓个页面把价格存下来”。但真把需求摊开里面藏着不少细节。我接手这个项目时的原始需求只有一句话“盯着几个电商平台上我们竞品的价格变了要提醒我们。”但这句话背后至少要拆出这么几个问题抓哪些商品商品ID怎么维护新增商品和下线商品怎么处理多久抓一次实时盯盘还是每天两次频次直接决定被封IP的概率。价格变了怎么通知邮件、短信、还是钉钉/企业微信机器人历史价格存多久客户要的是周报趋势图还是只看当前价格这个项目最终定下来的方案是商品列表维护在sqlite里抓取频率默认每30分钟一次价格变化通过邮件通知历史数据永久保留。存储用sqlite而不是MySQL原因是单机部署、数据量不大几百万条以内、不想额外维护一个数据库服务。这个选型在当时看是非常务实的。1.2 技术选型为什么是Python2 requests sqlite 代理池先说Python2。这项目是几年前起的那时候Python2还在维护期第三方库的兼容性也最稳。现在做新项目肯定上Python3但如果你手里有老项目要维护或者看到某些生产环境还在跑Python2脚本别急着嗤之以鼻——能稳定跑一年的代码就是好代码。requests库是Python里最顺手的HTTP客户端对比标准库urllib2注意Python2里是urllib2不是urllib3requests的session管理、headers定制、超时控制都人性化得多。咱们这代写爬虫的基本都是从requests入门它让HTTP请求这件事变得像“发个快递”一样简单填好地址URL、贴好单子Headers、选好快递方式GET/POST寄出去等着就行了。sqlite用于本地存储选它的逻辑很简单文件即数据库零配置。对于价格监控这种单机应用sqlite的性能绰绰有余。Python2标准库里自带sqlite3模块不用装任何额外的东西。代理池是这套架构里最核心也最容易被新手忽略的部分。电商平台的反爬策略不是吃素的同一IP高频访问轻则给你个验证码重则直接封IP段。代理池就是你的“IP弹药库”一批被封了下一批顶上。1.3 整体架构一条数据从网页到数据库的旅程整个系统的数据流大概是这样商品配置表里维护着要监控的商品ID和URL模板。调度器一个简单的while循环加sleep按固定间隔触发抓取任务。抓取模块通过代理池获取可用代理带着代理IP去请求目标页面。拿到HTML后用正则或简单的字符串匹配提取价格、标题、库存状态。最后把提取结果写入sqlite的价格历史表同时和上一次的价格做对比如果变化幅度超过阈值触发通知逻辑。这个链路不复杂但每个环节都有坑。我后面会按模块把代码和设计思路一块儿说清楚。2. 核心模块解析与关键技术点2.1 requests在Python2中的正确打开方式Python2的requests库和Python3版在用法上没太大区别但有几个细节容易踩坑。第一是响应内容的编码处理。Python2里requests的r.text默认会通过headers里的charset猜测编码但很多电商页面不严格按规范来猜错的概率不低。最稳妥的做法是从r.contentbytes里手动解编码或者直接用r.encoding utf-8强制指定。第二是连接复用。requests的Session对象会复用底层TCP连接对同一域名的大量请求能显著减少握手开销同时也能更好地模拟浏览器行为。Session在Python2里的用法和Python3一致但你在老代码里经常能看到裸露的requests.get()那其实每次都在创建新连接效率低不少。第三是超时设置。requests.get()如果不加timeout参数默认是无限等待。在生产环境里一个慢接口能拖住整个抓取线程。我的实践是统一设置timeout(3.05, 10)分别表示连接超时和读取超时。这个值的含义是3.05秒内没建立连接就放弃10秒内没读到数据也放弃。2.2 代理池的落地实现不能用的时候才是关键代理池这个东西网上有很多花里胡哨的实现什么Redis队列、什么SSDB存储、什么定时校验我这个项目一个都没用。原因很简单就一台服务器没必要引那么多中间件。代理池的核心需求就两个一个池子用来存代理一套机制来保证池子里的代理尽量可用。我的实现是用Python的list做存储配合一个简单的校验线程。代理来源是几个免费代理网站的公开代理每5分钟抓一轮测速后写入池子。校验方式是对目标电商域名发一个带代理的请求能在一个可接受的时间比如8秒内返回且状态码为200就认为是可用代理。这里有个很重要的点代理的有效性是针对目标站点的。一个代理对你测试的A站可用不代表它对B站也可用。有些代理出口IP本身就在电商平台的黑名单里怎么测都是废的。所以校验必须直接拿目标站点测不能图省事随便找个通用的检测URL。代理池的另一个设计细节是“超额请求”的处理。如果一个代理在连续N次请求中失败次数过多就把它暂时移出池子待会儿再回来验证。这能避免同一个坏代理反复被选中白白浪费时间。这个策略被后面要讲的429错误直接验证了价值。2.3 sqlite在Python2中的存储方案sqlite3模块在Python2.5之后就是标准库了用法很稳定。价格监控的数据模型就两张表goods表商品ID、商品名称、平台、商品URL、状态启用/停用price_history表自增ID、商品ID、价格、采集时间、原始标题快照price_history表是典型的时序数据会持续膨胀。我的策略是每季度归档一次把超过一年的数据导出成CSV后从主表删除。查询价格趋势时可视化工具直接读sqlite文件省去导出数据的麻烦。写入方面用executemany批量插入比逐条execute快一个数量级。还有一个小技巧给price_history表建复合索引索引字段是(goods_id, created_at)这样按商品查历史趋势的SQL会快很多。不加索引的话数据量上了百万一条查询能卡好几秒。对了sqlite对并发写入的支持很弱Python2的线程里如果多线程同时写同一个库文件容易报database is locked。我的方案是全局只用一个写连接所有写入操作串行化。这样做的代价是峰值吞吐量受限但对价格监控这种低频写入的场景完全够用。3. 实操过程与核心代码实现3.1 环境准备一台Linux服务器Python2环境部署环境是CentOS 7自带Python2.7.5。用virtualenv隔离项目依赖避免污染系统Python。需要装的包就三个requests、BeautifulSoup4解析页面用、还有标准库的sqlite3和smtplib发邮件用。用pip安装时要注意Python2的pip版本必须小于21.0否则装不了带二进制扩展的包。安装命令是pip install requests2.22.0 beautifulsoup44.6.0为什么锁版本因为更高版本的requests已经放弃Python2支持了4.6.0的BeautifulSoup是最后一个支持Python2的版本。这种老项目依赖锁定版本是保平安的唯一方式。3.2 数据模型与初始化建表先初始化数据库。这个脚本可以直接在命令行跑每个部分独立可测CREATE TABLE IF NOT EXISTS goods ( id INTEGER PRIMARY KEY AUTOINCREMENT, sku TEXT NOT NULL UNIQUE, name TEXT NOT NULL DEFAULT , platform TEXT NOT NULL DEFAULT , goods_url TEXT NOT NULL, enabled INTEGER NOT NULL DEFAULT 1, created_at TIMESTAMP DEFAULT (datetime(now, localtime)) ); CREATE TABLE IF NOT EXISTS price_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, goods_id INTEGER NOT NULL, price REAL NOT NULL, title TEXT, created_at TIMESTAMP DEFAULT (datetime(now, localtime)) ); CREATE INDEX IF NOT EXISTS idx_price_history_goods_time ON price_history(goods_id, created_at);sku字段是商品唯一标识用于和电商平台页面对应。goods表本身的增删就是运营后台的活这里先不展开。3.3 requests抓取封装Session复用代理注入重试机制上核心代码。这是一个抓取模块的封装把代理注入、超时、重试都揉在一起# -*- coding: utf-8 -*- import requests import random import logging import time logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) class Fetcher(object): def __init__(self, proxy_pool): self.proxy_pool proxy_pool self.session requests.Session() self.session.headers.update({ User-Agent: self._fake_ua(), Accept-Language: zh-CN,zh;q0.8, Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, }) def _fake_ua(self): ua_list [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/83.0.4103.116 Safari/537.36, Mozilla/5.0 (iPhone; CPU iPhone OS 12_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/12.0 Mobile/15A372 Safari/604.1 ] return random.choice(ua_list) def fetch(self, url, retry_times3): for attempt in range(retry_times): proxy self.proxy_pool.get_one() proxies {http: http://%s % proxy, https: http://%s % proxy} try: resp self.session.get(url, proxiesproxies, timeout(3.05, 10)) if resp.status_code 200: resp.encoding utf-8 return resp.text elif resp.status_code 429: logger.warning(Proxy %s got 429, remove it and retry., proxy) self.proxy_pool.remove(proxy) time.sleep(0.5) continue else: logger.warning(Unexpected status code: %s, proxy: %s, resp.status_code, proxy) continue except requests.RequestException as e: logger.warning(Request failed via %s: %s, proxy, e) self.proxy_pool.remove(proxy) time.sleep(0.3) return None几个设计细节拆解一下。UA轮换。UAUser-Agent是浏览器在HTTP请求头里标注“我是谁”的字段。很多网站反爬的第一步就是过滤没有UA的请求。我不会只用一个固定UA而是准备一个池子随机换虽然不能完全规避风控但能少碰一些傻黑名单。重试策略。fetch方法里的retry_times3意味着每个页面最多尝试三次。每次重试都会换一个代理因为失败原因很可能是当前代理已经被目标站标记了。三次都失败就返回None由上层逻辑决定是记日志还是跳过。针对性处理429。这个状态码在这个项目里简直比200还常见。429意味着请求频率过高触发了目标站的限流策略。代码里的处理方式是遇到429判定当前代理已经“不干净”直接从池子里移除让上层换下一个代理。这一步对长期稳定运行至关重要。3.4 价格提取正则表达式才是王道提取价格我选择用正则而不是BeautifulSoup。原因很实际电商页面的价格往往在动态渲染的节点里BeautifulSoup按class取节点经常扑空但价格数字本身在HTML源代码里反而有比较稳定的特征。举个例子某平台的页面源码里价格长这样span classprice-now i¥/i199.00 /span我的提取正则就是import re PRICE_PATTERNS [ rprice\s*:\s*?(\d(?:\.\d{1,2})?)?,, rclass(?:price|price-now|current-price)[^]*\s*(?:[^])*\s*(?:¥|)?\s*(\d(?:\.\d{1,2})?), r(\d\.\d{2}), ] def extract_price(html): for pattern in PRICE_PATTERNS: m re.search(pattern, html) if m: try: return float(m.group(1)) except ValueError: continue return None写正则提取价格要注意几个边界情况有的页面价格带千分位分隔符1,999.00有的价格是区间199.00-299.00有的商品无货时价格显示为--。这些都得在正则里或者后续解析时处理掉。如果正则提取出来的是一个区间我会取较低值并在日志里标记。你可以直接用这个正则跑一遍自己盯着的平台看看命中率。如果命中率低就按你自己看到的页面源码特征调整。这一步是整个项目的核心页面上没那么多花哨的库比得上一个能匹配到实际数据模式的正则。3.5 数据入库与变化检测价格拿到手后先写price_history表再和最后一条历史记录比较。如果价格变动超过设置的阈值触发通知。import sqlite3 import smtplib from email.mime.text import MIMEText class PriceMonitor(object): def __init__(self, db_path, smtp_conf): self.conn sqlite3.connect(db_path) self.smtp_conf smtp_conf def save_price(self, goods_id, price, title): cur self.conn.cursor() cur.execute( INSERT INTO price_history (goods_id, price, title) VALUES (?, ?, ?), (goods_id, price, title) ) self.conn.commit() last_price cur.execute( SELECT price FROM price_history WHERE goods_id? ORDER BY id DESC LIMIT 1, (goods_id,) ).fetchone() return last_price[0] if last_price else None def check_and_notify(self, goods_id, goods_name, new_price, last_price, threshold0.03): if last_price is None: return change abs(new_price - last_price) / float(last_price) if change threshold: subject [价变提醒] %s 从 %.2f 变为 %.2f % (goods_name, last_price, new_price) body 商品 %sID%d价格变化 %.2f%% % (goods_name, goods_id, change * 100) self._send_email(subject, body)save_price里注意一个细节我先插入新记录再去查最后一条价格。这样查到的就是刚插入的这条用它拿到“当前最新价”。而真正用于对比的“上一次价格”应该在插入之前查。这个逻辑在完整代码里要调整一下顺序上面这段示例仅为演示表结构实际用的时候建议把查询放到insert之前避免把刚写入的新纪录当成对比基准。3.6 调度器最简单的while循环就是最稳的调度就一个while循环。但要注意几个细节用time.sleep控制间隔每次循环检查一下代理池是不是空的空了就触发补货线程捕获所有异常避免整个进程因单个商品解析失败而退出。def run(self): while True: start time.time() goods_list self.get_enabled_goods() for goods in goods_list: try: html self.fetcher.fetch(goods[goods_url]) if html is None: continue price extract_price(html) if price is None: logger.warning(No price found for %s, goods[sku]) continue last_price self.get_last_price(goods[id]) self.save_price(goods[id], price, goods[name]) self.check_and_notify(goods[id], goods[name], price, last_price) except Exception as e: logger.exception(Error processing goods %s: %s, goods[sku], e) continue elapsed time.time() - start sleep_time max(0, self.interval - elapsed) logger.info(Round done in %.2f seconds, sleep for %.2f seconds, elapsed, sleep_time) time.sleep(sleep_time)这里把整个抓取round的耗时算出来用interval减去elapsed保证每轮周期稳定在设定的间隔左右而不是“每轮开始后固定睡N秒”导致实际频率偏快。执行久了你会发现这个简单的调度器才是最省心的部件。4. 常见问题与排查技巧实录4.1 429 Too Many Requests限流的正确应对姿势这个项目上线后我遇到最多的就是429。它的含义是请求太频繁目标服务器不想理你了。区分一下两种典型的“请求过多”一种是单IP维度IP频率触发阈值此时需要换代理IP另一种是账号/设备维度Cookies可能被标记了此时需要清Cookie、换身份。排查429的第一步确认是不是同一个IP高频请求导致。看日志里失败请求用的代理如果一段时间内同一个代理IP反复触发429说明它已经进了目标站的黑名单移出池子就行。如果所有代理都触发429说明是整个出口网段被盯上了或者你请求的频率真的太高需要拉大间隔。第二步观察429响应里是否带Retry-After头。有些平台会告诉你“多少秒后再来”。带上这个头做sleep比无脑重试更友好也更不容易把平台惹毛。第三步也是很多人忽略的429不代表必须硬刚换IP。有时候降低单个商品页面的请求频率就能解决。比如30分钟一轮改成45分钟一轮牺牲一点实时性换取稳定运行长远来看更划算。4.2 代理池枯竭免费代理说挂就挂免费代理的稳定性用过的人都懂。今天测出来还能用的10个代理明天可能就剩2个。代理池枯竭的典型症状是日志里大量删除代理的记录然后fetch返回None整个采集轮次颗粒无收。解决办法有两个方向一个是从源头扩充代理来源多爬几个免费代理站做去重和合并另一个是降低对代理池的依赖频率比如设置一个“本地直连模式”当代理池可用数量低于阈值时同一轮内改用直连不经过代理试探性采集几个页面。注意直连模式仅适合低频抓取高频还是会被拦。我后期的做法更省心给代理池增加了一个“持久化”功能把验证通过的代理写进sqlite表里重启进程后自动加载。免费代理虽说不稳定但这些IP从“存活”到“失效”的时间窗口往往足够跑完下一轮采集。这个状态下至少代理池不会因为进程重启而归零。4.3 数据库锁定与并发写冲突sqlite的database is locked错误我在项目早期遇到得挺多。原因是Python2的sqlite3模块默认会为每个连接创建独立的事务不同线程同时写就锁住了。我的解决方案是所有写操作统一走一个连接这个连接在程序启动时创建整个生命周期不关闭。并且所有写操作都在同一个线程里执行。如果你有多线程读的需求读可以用只读连接connect时带上file:...?modero写还是走那唯一一个。这套方案在数据量到百万级之前都没问题。如果真有一天数据量大到sqlite扛不住再考虑平滑迁移到PostgreSQL也不迟但至少目前这个项目跑了这么久sqlite完全没成为瓶颈。4.4 字符集与乱码Python2的招牌暗坑Python2处理字符串编码说多了都是泪。响应内容如果是UTF-8requests会正确处理但也可能返回GBK编码的页面。此时resp.text乱码而resp.content是一堆bytes。我的经验是不要依赖requests自动解码一律resp.encoding utf-8如果发现乱码再把它改成gbk实测目标平台页面编码后固定下来。另外在Python2里从页面正则提取出的字符串要decode成unicode写进sqlite或者拼邮件正文时保持一致不然就是经典的UnicodeEncodeError。如果你的环境已经切到Python3这块就不用操心了。但如果你在维护老项目字符串编码问题建议在所有入口处以utf-8为准统一下来能省掉大量调试时间。4.5 504与网关超时不是你的错也别硬刚有时候请求返回504 Gateway Timeout这往往是目标站反代或者CDN层面的问题代理本身可能还是好的。遇到504正确做法是标记该代理“本次失败但不清出池子”下次还可以继续用。如果连续多次都504再考虑淘汰。这里的判断逻辑可以在代理池的fail计数里做连续失败5次才移除而不是一次失败就remove。这个阈值可以调我调试下来发现3到5次比较合理。太低会误杀偶然失败的代理太高会拖慢整体速度。5. 几个容易被忽略但很重要的细节5.1 日志与可观测性没人愿意盯着黑框框爬虫程序如果不打日志出了问题就是灾难。我的日志规范是抓取开始和结束各有一次INFO日志失败和异常用WARNING或ERROR价格变化单独打一条醒目日志方便事后统计触发次数。日志格式统一包含时间戳、级别、模块名、消息。生产环境建议再配一个日志周期归档用Python的logging.handlers.TimedRotatingFileHandler按天切割日志文件保留最近30天。这样日志文件不会无限膨胀排错时也能快速定位某天的记录。5.2 请求频率设计追求稳定不追求激进电商价格监控每分钟都在变的价格意义不大。一小时内的价格波动对普通消费者或者经销商来说参考价值极低。所以我的建议是每小时2到4轮就足够也就是15到30分钟间隔。太低频抓不到瞬时价格高频只会徒增封号风险。如果你监控的商品量大还可以把商品列表分组每组错峰抓取。比如100个商品分成4组每组25个每组间隔7分钟启动这样请求在时间轴上摊开不容易形成瞬时脉冲。5.3 邮件通知的发送频率控制价格变化通知如果不做频率控制可能一天收到几十封邮件最后被当成垃圾邮件忽略了。我的经验是单商品每天最多提醒一次。具体做法是在goods表里加last_notify_date字段当天已经提醒过的商品即使价格再变也不再发。这个“一天一次”的约束在客户那边反馈非常好。他们不是要实时盯盘而是想知道“今天哪几个商品值得关注”。把噪音降到最低通知的价值才会被放大。6. 一套更干净的改进方向如果你不介意用Python3重写一遍几个改动会让这个项目舒服很多用schedule库替代手写while循环调度用SQLAlchemy替代裸sqlite3ORM写起来省事用Celery定时任务替代常驻进程天然支持分布式用Redis做代理池和管理队列支持多台机器共享但我要提醒一句改动越大调试成本越高。如果目标只是跑通价格监控这套老方案完全够用。想升级也要在“线上稳定运行”的前提下逐步替换模块千万别一次性重写。最后再分享一个小经验。这类跑了好几年的脚本最大的风险不是技术过时而是没人维护。建议给每个模块写清楚注释把代理池的验证逻辑、价格提取正则怎么改、通知频率在哪调全部写进README里。哪怕半年后再接手也不至于对着代码发呆。我这套方案在你自己的场景里跑起来可能还会遇到一些问题欢迎带着具体的报错来交流。爬虫这条路踩坑是常态但把坑填平了后面就顺了。本文还有配套的精品资源点击获取