简介：一套基于Python2的电商价格监控项目，面向需要追踪商品降价动态的网购用户、电商运营者以及Python爬虫学习者。项目通过requests库抓取电商页面，结合SQLite保存商品与历史价格，并借助代理池规避反爬限制；当价格低于预设期望值时，自动调用SMTP发送邮件提醒，形成从商品采集、数据存储、价格检测到消息通知的完整闭环。资源压缩包共34个文件，其中18个py脚本构成项目主体，覆盖爬虫调度、页面解析、代理池管理、数据库操作、邮件发送、异常处理等模块；另有xml配置、md说明文档、txt数据、png示意图以及sqlite数据库文件，整体大小仅1.13MB，目录结构清晰，便于快速部署与二次开发。已有365人学习下载。阅读源码与文档，可掌握Python2环境下多线程爬虫、代理池维护、SQLite数据持久化及邮件告警的综合实现思路，还能了解定时任务调度与网页解析的落地方式，适合作为电商数据采集和自动化通知的参考范例，也可快速扩展用于监控其他类型网站。

1. 电商价格监控的隐藏难点：同样的代码，为什么别人监控没翻车？

做电商价格监控这件事，真正麻烦的从来不是把页面请求回来，而是拿到数据之后怎么存、怎么判断"降没降价"，以及请求发出去之后怎么应对封IP、验证码和接口改版。一套用Python2 + requests + sqlite3的电商价格监控资源，核心就是解决这三个问题：请求层用requests自带的重试机制配合自建的代理池做反爬对抗，数据层用sqlite3做增量存储，只记录价格变动而不是每次都塞一条新记录。这几天我把这套资源原样跑了一遍，发现它虽然是用老掉牙的Python2写的，但骨架思路放在今天依然管用。适合谁看？打算写第一版价格监控脚本的初学者，或者手里已经有一套采集程序但总是被反爬拦住、想在代码层面解决代理调度问题的从业者。

为什么强调"增量存储"而不是每次都写库？因为监控类任务的数据量不是最大的问题，数据噪声才是。价价格每天变动几十次，如果你每次都落一条记录，一个月后数据库里全是无效记录，查询还变慢。这套资源的做法是把每次抓到的价格、标题、促销信息跟库里的最新记录做比对，变化了才更新，没变化就跳过。思路不复杂，但手工用requests和sqlite3从零搭一次，能避开不少坑。

2. 抓取层设计：requests会话复用、超时与重试参数怎么定

2.1 为什么用requests.Session而不是每次新建请求

这套资源的抓取层用的是requests库，但仔细看就会发现，它没有每次请求都requests.get()，而是先建了一个requests.Session()。两者差别在实际运行中非常明显：Session会保存Cookie、被动维持TCP连接，监控多个商品页时能减少握手次数，抓取效率能提升一截。

# -*- coding: utf-8 -*- import requests import random # 创建一个全局会话，供所有产品页共用 session = requests.Session() session.headers.update({ "User-Agent": "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36", "Accept-Language": "zh-CN,zh;q=0.8", }) def build_proxies(proxy_dict): """把代理字典转成requests需要的proxies格式""" return { "http": "http://{}:{}".format(proxy_dict["ip"], proxy_dict["port"]), "https": "http://{}:{}".format(proxy_dict["ip"], proxy_dict["port"]), }

为什么这里的User-Agent保持恒定而不是每次随机换一个？这是这套资源引出的第一个值得商榷的细节。常见做法是给UA池子做轮换，但这个资源里偏爱"固定UA + 多IP"的组合——理由是电商反爬更多依赖IP维度而非UA维度，IP不够干净时换100个UA也白搭，UA换来换去反而容易触发行为一致性检测。这点不绝对正确，但设计取舍是清楚的。

2.2 超时与重试设置：让请求失败可控而非失控

价格监控跑在实际网络环境里，超时问题远比想象中高频。这套资源的做法是：连接超时给5秒，读取超时给10秒，配合requests内置的重试机制。注意，核心参数是total而不是每次失败立刻重试——重试次数给到3次以上就能明显提高抓取成功率，但也别给太多，否则代理池里的坏IP会在重试上浪费大量时间。

from requests.adapters import HTTPAdapter # 把重试策略挂在session上，而不是手动写死循环 adapter = HTTPAdapter( max_retries=3, pool_connections=10, pool_maxsize=10, ) session.mount("http://", adapter) session.mount("https://", adapter) def fetch_product_page(session, url, proxies=None, timeout=(5, 10)): try: resp = session.get(url, proxies=proxies, timeout=timeout) if resp.status_code == 200: resp.encoding = "utf-8" return resp.text return None except requests.exceptions.ProxyError as e: # 代理失效和连接失败的处理逻辑泾渭分明 return "proxy_error" except requests.exceptions.ConnectTimeout: return "timeout"

返回"proxy_error"或"timeout"而不是直接抛异常，这个设计很朴素但实用——上层调度逻辑可以直接根据返回值决定是换代理还是放慢频率，不必在每一层都做异常捕获。参数pool_connections和pool_maxsize在这个场景下不用调大，单机跑几十个商品页够了，调大了反而占用空闲连接。常犯的错误是把timeout=(5, 10)错写成timeout=5，后者只对连接阶段生效，读取阶段会无限制等待，程序早该报错的时候却挂在那里一动不动。

3. 数据层设计：sqlite3增量存储、去重与价格变动判定

3.1 表结构应该怎么建：不要把价格历史压成一行

页面拿到手之后，这套资源里最有参考价值的模块就是数据层。它用sqlite3做存储，却不是一个随随便便的"放个数据库存数据"的写法。表结构上，它拆成了两张表：一张存产品信息，一张存变价记录。产品信息表只保留当前快照，变价记录表按时间追加。这样做的好处是查询"最近一次降价"、"这个月调过几次价"都很快，坏处是如果你把两张表概念混在一起，去重逻辑就很难写了。

CREATE TABLE IF NOT EXISTS product ( id INTEGER PRIMARY KEY AUTOINCREMENT, sku TEXT UNIQUE, title TEXT, url TEXT, latest_price REAL, lowest_price REAL, highest_price REAL, update_time TEXT ); CREATE TABLE IF NOT EXISTS price_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, sku TEXT NOT NULL, price REAL, crawl_time TEXT, source TEXT DEFAULT 'manual' );

唯一键加在sku上，是这套资源一个值得抄的细节。不要用id或title做唯一键——同一件商品换了个促销标题，title就变了，去重立刻失效。而sku是电商体系里相对稳定的商品标识，就算商品的标题和价格都变，sku通常不变。最早我看这套代码时还觉得多余，结果跑了两周数据后发现，真正能扛住标题漂移的就是这唯一约束。

3.2 增量更新逻辑：先查后写，比replace更省资源

写数据时最常见的错误就是一上来INSERT OR REPLACE。这套资源的做法是先用sku查一次当前价，比对后再决定更新还是跳过。这个操作看着多了一次查询，但实际上大幅减少了无效写入，数据库体积也控制得住。

import sqlite3 DB_PATH = "price_monitor.db" def update_price(sku, title, current_price, url): conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() try: # 查询库内最新价格 cursor.execute("SELECT latest_price FROM product WHERE sku = ?", (sku,)) row = cursor.fetchone() if row is None: # 从未监控过该商品，先插入 # price_history里存一条初始记录 cursor.execute("INSERT INTO product (sku, title, url, latest_price, lowest_price, highest_price, update_time) VALUES (?, ?, ?, ?, ?, ?, datetime('now'))", (sku, title, url, current_price, current_price, current_price)) cursor.execute("INSERT INTO price_history (sku, price, crawl_time) VALUES (?, ?, datetime('now'))", (sku, current_price)) elif abs(row[0] - current_price) > 0.01: # 原价与新价差一分钱以上才认定价格变动 # 避免因为浮点精度问题频繁更新 cursor.execute("UPDATE product SET latest_price = ?, update_time = datetime('now'), lowest_price = CASE WHEN ? < lowest_price THEN ? ELSE lowest_price END, highest_price = CASE WHEN ? > highest_price THEN ? ELSE highest_price END WHERE sku = ?", (current_price, current_price, current_price, current_price, current_price, sku)) cursor.execute("INSERT INTO price_history (sku, price, crawl_time) VALUES (?, ?, datetime('now'))", (sku, current_price)) return "updated" else: return "unchanged" conn.commit() return "inserted" if row is None else "updated" finally: conn.close()

逻辑不说多高深，但胜在每一步都有取舍。abs(row[0] - current_price) > 0.01这行判断，是用一分钱作为价格变动的最小单位，避免浮点误差带来的误判。库里的lowest_price和highest_price是直接通过CASE WHEN在SQL里维护的，不需要先从库里取出历史极值再在Python里比较，省了一次查询。这套代码在一个小监控任务里跑了两个月，库里总共只有几百条变价记录，查询和备份都轻松。

3.3 字段类型与编码处理：中文乱码的根源在哪

sqlite3在Python2下处理中文最经典的问题就是编码。页面请求回来的时候，用resp.encoding = "utf-8"保证了解码正确，但如果网页实际是GBK，这个固定设置就会把整个页面解成乱码。这套资源的做法是先从响应头里读Content-Type的charset，拿不到就尝试用chardet猜，实在猜不出来才兜底到utf-8。

def guess_encoding(resp): ctype = resp.headers.get("Content-Type", "") if "charset=" in ctype: return ctype.split("charset=")[-1].strip() import chardet detect_result = chardet.detect(resp.content[:500]) encoding = detect_result.get("encoding") return encoding if encoding else "utf-8"

在Python2环境里，从网页里提取到的中文字符串要存进sqlite3，最稳妥的做法是统一在写库前转成unicode类型。如果你在代码里看到一堆u"中文"字面量，不要觉得是怪癖，那是Python2环境下避免隐式编码转换的常规做法。跑这套代码时如果你发现库里存的是ä¸æ–‡这类乱码，基本可以断定是哪一步用了str而没有用unicode。

4. 代理池设计与调度：怎样做到"坏一个补一个，挂一批换一批"

4.1 代理池的基本框架：校验、标权、剔除三件事

电商监控领域代理池几乎是标配。这套资源里的代理池不是把所有代理放进列表里随机选就完事，而是维护了一个带状态标记的小数据库，每个代理有存活次数、连续失败次数、最后校验时间三个字段。代理失效在监控场景里不是偶发事件——免费代理跑一天能有四分之一失效，所以淘汰逻辑必须自动运行。

CREATE TABLE IF NOT EXISTS proxies ( id INTEGER PRIMARY KEY AUTOINCREMENT, ip TEXT NOT NULL, port INTEGER NOT NULL, protocol TEXT DEFAULT 'http', fail_count INTEGER DEFAULT 0, success_count INTEGER DEFAULT 0, last_check TEXT, status TEXT DEFAULT 'alive' );

连续失败三次就把status置成dead，下次调度直接跳过。阈值设成三次是权衡过的——设成一次会导致瞬时网络抖动就把代理误杀了，设成五次则会让坏代理多存活很久，拖慢抓取。这套资源的做法是：每次请求失败就fail_count + 1，成功一次就清零并success_count + 1。这样代理从alive变dead的路径是可追踪的，而不是一个黑匣子。

4.2 调度策略：只有"随机选"是不够的

一般常见做法是随机从代理池里挑一个。但这套资源有一个小优化：随机挑一个之后，在请求前先用短超时（比如3秒）去访问一个稳定网站做连通性预检，通了才真正发给目标网站，不通就直接换下一个。如果预检本身耗时过大，就从代理池里剔除它。

import random def get_working_proxy(proxy_pool, check_url="http://www.baidu.com", timeout=3): # 最多尝试5个代理，找到一个能用的就返回 for _ in range(5): proxy = random.choice(proxy_pool) if proxy["fail_count"] >= 3: continue try: resp = requests.get(check_url, proxies=build_proxies(proxy), timeout=timeout) if resp.status_code == 200: proxy["success_count"] += 1 proxy["fail_count"] = 0 # 先取出再归还，避免同一个代理被并发拿两次 proxy_pool.remove(proxy) proxy_pool.append(proxy) return proxy except: proxy["fail_count"] += 1 return None

为什么先remove再append？因为单机多线程抓取时，两个线程同时拿到同一个代理就会互相干扰——一个请求失败导致两个线程都在标错。这套资源并没有上锁，而是通过"取出即删除、用完归还"的方式避免了重复获取。代价是代理池里的元素位置会频繁变动，但结合前面提到的连续失败阈值，这种无锁设计在小规模抓取场景里足够稳定。

4.3 代理池的自动补充：运行中热更新，而不是重启脚本

代理失效是常态，只靠剔除不补充，代理池会越跑越空。这套资源的补充方式是从备用URL里抓取新代理列表，入库前先做格式校验和连通性测试。注意，它补充代理的时机不是"池子空了才补"，而是每隔一段时间主动拉取一次，保证池子始终维持在一个可用数量。

def refresh_proxy_pool(proxy_api_url, min_alive=20): try: resp = requests.get(proxy_api_url, timeout=10) proxy_list = resp.json() except: return # 校验后再入库，绝不直接信任接口返回的字段 for item in proxy_list: ip = item.get("ip") port = item.get("port") if not ip or not port: continue if not (0 < int(port) < 65535): continue cursor.execute("INSERT OR IGNORE INTO proxies (ip, port, last_check, status) VALUES (?, ?, datetime('now'), 'alive')", (ip, port)) # 统计当前存活代理数量 cursor.execute("SELECT COUNT(*) FROM proxies WHERE status='alive'") alive_count = cursor.fetchone()[0] # 少于阈值时把状态为dead的代理清掉，避免下次遍历浪费 if alive_count < min_alive: cursor.execute("DELETE FROM proxies WHERE status='dead'")

这里要强调一个反直觉的细节：代理池资源往往把代理质量放在第一位，但真正影响监控稳定性的其实是"池子不能出现真空期"。低价代理源本身鱼龙混杂，入库时不做高并发校验，先靠请求失败计数来动态淘汰，反而是性价比更高的方案。许多项目初期在代理校验上花了大量功夫，结果免费代理几分钟就失效，预校验的成本完全收不回来。

5. 避坑实战：Python2迁移、反爬适配与代理池失效排查记录

5.1 现象：Python2的urllib报错，requests导入正常但在pip install时装不上

老资源最常见的问题是运行环境。

现象：在一台只有Python3的机器上直接运行脚本，提示ModuleNotFoundError: No module named requests，而pip install requests又会提示当前环境是Python3，或者装上之后import时报编码错误。

原因：这套资源是为Python2写的。Python2和Python3在字符串处理、print语法、异常写法上都不兼容，直接用Python3跑Python2代码基本等于碰运气。

解决：用python2虚拟环境跑，或者用2to3做一次代码转换。我的处理习惯是装一个Python2.7（比如从系统包管理器装），然后在虚拟环境里安装requests==2.19.1和pysqlite2这类尽量贴近当年的库版本。

# 用virtualenv隔离老环境 virtualenv -p /usr/bin/python2 env2 source env2/bin/activate pip install requests==2.19.1

注意，requests==2.19.1是Python2下可用的最终一代requests，再新的版本就放弃Python2了。另一个大坑是Python2的print——如果你在代码里看到print resp.status_code，在Python3里得改成print(resp.status_code)，这个低级报错会消耗你大量时间。现在回看这套资源，最省事的做法就是老老实实用Python2环境，而不是花大力气移植到Python3——毕竟代码量不大，但每个细节都踩在Python2的特性上，强行移植往往得不偿失。

5.2 现象：页面能打开，但提取到的价格字段要么是零要么是None

现象：代理、请求头都配置好了，但解析出来的价格在数据库里显示为0，或者偶尔能取到偶尔取不到。

原因：电商页面经常在价格位置做动态渲染，价格不在HTML源码里，而是通过异步接口加载。还有一种隐蔽情况：页面里存在多个价格标签，一个显示原价一个显示促销价，解析代码取到了原价字段，误当成当前价。

解决：先抓取页面时把完整HTML整体存下来做检查，再看页面结构里价格字段是否在同一个DOM层级下。这套资源的处理方式采取了一个比较实用的兜底逻辑：如果解析出来的价格低于某个阈值（比如1元），就判定为解析失败，重新抓取一次而不是直接入库。

def parse_price(html, sku): # 解析价格，省略具体正则细节 price = extract_price_field(html) # 价格低于合理阈值，多半是页面结构变了或动态加载失败 if price is None or price < 1.0: return None return price

如果多次解析失败，就把这个sku标记为parse_failed，跳过后面的抓取，而不是死磕。这个动作看似简单，实际价值非常大——它让你的监控程序不会因为页面改版而疯狂写入错误数据。页面反爬升级导致解析失败时，如果在代码里硬等，整个任务会被拖垮；而标记之后，你能从数据库里一眼看出哪个商品出了问题。

5.3 现象：代理池里有代理，但请求全部超时，错误日志满屏ProxyError

现象：proxies表里活着的代理有几十个，但连续几分钟的抓取任务全部超时，几乎没有一条成功。

原因：代理的alive状态是基于上一次预检结果定的，但预检通过不代表目标网站也通——不少代理出口IP被目标电商联合封禁，访问百度和访问电商是完全不同的两条路。也就是说，一个代理能打开百度，不代表能打开目标电商网站。

解决：把预检的目标从http改成你实际要监控的电商域名，让代理校验贴近真实场景。同时在调度逻辑里增加一个"连续失败即降级"的机制——同一个代理在目标网站连续失败3次，无论预检是否通过，都先标dead。

def request_with_dynamic_proxy(session, url, proxy_pool): for _ in range(3): proxy = get_working_proxy(proxy_pool, check_url="http://www.baidu.com") if proxy is None: # 池里没有能用代理，等待并重试 time.sleep(30) continue result = fetch_product_page(session, url, proxies=build_proxies(proxy)) if result == "proxy_error": # 说明代理本身有网络问题，直接在库里标记 mark_proxy_dead(proxy) continue if result == "timeout": # 超时可能是代理慢，也可能是目标网站慢，给两次机会 proxy["fail_count"] += 1 if proxy["fail_count"] >= 2: mark_proxy_dead(proxy) continue return result return None

这套做法把"代理管理"和"请求逻辑"揉在了一个循环里，初看不够解耦，但胜在直观——每个失败都在最早的节点被识别和处置。真正在生产环境跑这个循环时，请注意time.sleep(30)不能省。代理池全线超时的情况下仍保持高频请求，结果是目标网站在反爬系统里把你的出口IP段标得更狠，相当于自己把自己关进小黑屋。

5.4 现象：sqlite3报database is locked，程序频繁卡死

现象：监控程序跑了几分钟后，sqlite3操作开始报database is locked，有时候一次卡住几十秒。

原因：sqlite3在Python2的默认配置下并发写支持很差，多个线程同时写同一个库文件时，后到的写操作会等待甚至直接失败。价格监控里有多个抓取线程在写数据，冲突概率很高。

解决：给连接设置一个超时，同时把PRAGMA journal_mode=WAL打开。WAL模式显著改善了并发表现，并且几乎不需要改代码。

def get_connection(): conn = sqlite3.connect(DB_PATH, timeout=15) conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA synchronous=NORMAL") return conn

注意，设了timeout=15并不代表所有锁冲突都能自行解决，但它给了足够的缓冲窗口。另一个小坑是，Python2模式下如果连接不关闭，文件句柄会泄露，跑几天之后系统会报"Too many open files"，所以前面代码里finally: conn.close()的做法请务必保留。直到现在，每当我看到有人用sqlite3跑多线程任务却不加WAL模式，我都想提醒一句：数据库锁的问题十有八九出在模式配置上，而不是出在并发量上。

5.5 现象：定时任务跑了一周后，监控延迟越来越高

现象：脚本用crontab设定每30分钟跑一次，跑了几天后发现每次执行时间明显变长，甚至出现上一次还没跑完下一次就启动的情况，两批任务互相打架。

原因：数据库涨大后，全表扫描变慢，重试机制又在高延迟代理上花了太多时间。更隐蔽的原因是，product表和price_history表没有建索引，每次WHERE sku = ?都做全表扫描，商品数量达到几百个时，延迟开始明显。

解决：在sku字段上加索引，并给price_history.crawl_time加索引，因为历史记录查询和清理都会按时间范围扫。

CREATE INDEX IF NOT EXISTS idx_product_sku ON product(sku); CREATE INDEX IF NOT EXISTS idx_history_sku ON price_history(sku); CREATE INDEX IF NOT EXISTS idx_history_time ON price_history(crawl_time);

另一个优化是，判断"要不要再跑一次"可以用时间戳控制，而不是裸crontab。脚本启动时先查库内最近一次update_time，小于阈值才执行抓取，否则直接退出。这套资源里没有现成实现，但我在复现时加了这一层——跑了一周后效果明显，任务重叠的问题彻底消失了。定时任务不是越频繁越好，关键是让"任务启动间隔"大于"单次任务耗时"，否则堆积是肉眼可见的。

6. 运行监控与异常自愈：把定期巡检做成脚本里的强制动作

监控脚本跑起来只是第一步，跑起来之后怎么知道它还在正常工作、出了问题怎么自动恢复，才是真正区分"能用"和"好用"的分水岭。

我自己习惯在监控脚本里加一个"心跳模块"，每次抓取循环结束后，把本轮抓取的成功率、失败率、代理池存活数量、数据库最新记录时间写入一个单独的health.log。不要小看这个动作，很多监控程序跑着跑着就静默死亡，重启时还找不到原因——有日志就能往前翻。这个健康信息的落库方式很简单：

def write_heartbeat(success_num, fail_num, alive_proxy_num): conn = sqlite3.connect(DB_PATH, timeout=10) conn.execute("INSERT INTO heartbeat (success_num, fail_num, alive_proxy_num, record_time) VALUES (?, ?, ?, datetime('now'))", (success_num, fail_num, alive_proxy_num)) conn.commit() conn.close()

接着，在每轮抓取结束后做一个基础诊断：连续两轮抓取成功率低于50%，就往日志里打一个WARN，并把代理池里所有存活的代理重新预检一次。这个动作看起来像是在代码里平白多写了逻辑，但实际运行中绝大多数"监控失效"都不是被封IP，而是代理池悄悄枯萎——预检一次就能救回来集体失效的代理。

验证这套资源有没有跑对，最直接的检查手段是看price_history表里时间段的分布：

sqlite3 price_monitor.db "SELECT count(*), substr(crawl_time,1,13) FROM price_history GROUP BY substr(crawl_time,1,13) ORDER BY 2 DESC LIMIT 24;"

如果某个小时段内记录数突降为0，那大概率是那个时段代理池全挂了。如果记录数奇高，说明价格变动频繁或者你的解析逻辑误把相同价格当成新记录写入了。这两类异常单看数据就能定位，不需要翻页面。

最后说一条个人习惯。从那以后，每次把监控脚本部署到新机器上，我都会强制走一遍"放慢刷新、开WAL、加索引、写心跳日志"这四步，宁可前期多花十几分钟，也不赌它自己稳定。跑了大半年，真正让脚本翻车的不是反爬玄学，而是数据库锁、编码错乱、代理池静默消失这些看起来不显眼的底层问题。希望这套资源的落地步骤和这里踩过的坑，能帮你少走一段弯路。

本文还有配套的精品资源，点击获取