尧图网站建设 
简介这是一套面向Python开发者与电商数据分析师的闲鱼智能监控与分析工具解决人工盯梢效率低、筛选逻辑僵化、信息过载等痛点适用于二手商品比价、热门商品抢购、竞品动态追踪等实际场景。资源包共39个文件含11个核心Python脚本如web_server.py、scraper.py、ai_handler.py、2个Docker配置文件docker-compose.yaml、Dockerfile、4个图文素材PNG/JPG、3个HTML/CSS/JS前端页面及多个提示词模板prompts目录和示例配置config.json.example、.env.example整体12.31MB结构清晰模块职责分明。已有190人学习下载可直接运行Web管理界面获得自然语言创建任务、多模态AI图文分析、实时流式推送ntfy/企微/Bark、Cron定时调度及反爬增强等完整能力附带教程.docx与免责声明开箱即用。1. 项目概述为什么我们需要一个闲鱼智能监控机器人如果你在闲鱼上做过生意或者经常在上面淘货肯定有过这样的经历看中一个商品犹豫了一下或者想等卖家降价结果一转眼就被别人拍走了。又或者你是一个卖家想知道自己发布的商品在同类中的价格竞争力或者想监控某个关键词下新上架的商品动态。手动去刷、去比价不仅效率低下而且非常容易错过关键信息。这个“闲鱼智能监控分析系统”就是为了解决这些痛点而生的。简单来说它就是一个7x24小时不间断工作的“数字助理”。它的核心任务就是代替你按照你设定的规则比如关键词、价格区间、卖家信用等自动、持续地监控闲鱼平台上的商品信息变化并将关键数据如价格变动、上新提醒、商品下架实时推送给你甚至能进行初步的数据分析帮你做出更明智的决策。无论是个人买家想“捡漏”还是专业卖家或工作室需要进行市场调研和竞争分析这个系统都能大幅提升效率把人力从重复、枯燥的监控工作中解放出来。2. 系统核心架构与设计思路拆解一个完整的闲鱼监控系统远不止写个脚本爬取网页那么简单。它需要稳定、隐蔽、智能且易于维护。我们不能把它做成一个简单粗暴、频繁请求的爬虫那样极易被平台风控识别并封禁。因此系统的设计必须考虑合规性、健壮性和可扩展性。2.1 分层架构设计我设计的系统通常采用清晰的分层架构这有助于各模块解耦方便后续维护和功能扩展。数据采集层这是系统的“眼睛”和“手”。负责模拟浏览器行为访问闲鱼页面抓取HTML数据。这里的关键在于模拟真人操作。我们不能用简单的requests库直接访问因为闲鱼有完善的反爬机制。更稳妥的做法是使用Selenium或Playwright这类浏览器自动化工具配合真实的浏览器内核如Chrome并注入合理的鼠标移动、滚动、等待时间等行为脚本。对于大规模监控可以考虑使用puppeteer-extra及其stealth插件来进一步增强隐蔽性。数据处理与解析层这是系统的“大脑”。采集到的原始HTML是杂乱无章的这一层负责从中提取结构化信息。我们会使用BeautifulSoup或lxml来解析HTML定位商品标题、价格、图片、卖家信息、发布时间等关键字段。这里有一个难点闲鱼的页面结构可能会不定期调整导致解析规则失效。因此解析规则需要设计得具有一定的容错性和可配置性最好能通过配置文件来管理选择器CSS Selector或XPath。任务调度与监控核心层这是系统的“心脏”。它管理着所有监控任务Task。每个任务定义了监控目标如关键词“iPhone 13 256G”、过滤条件价格低于4000元卖家信用“极好”、执行频率如每5分钟一次和通知渠道。我们需要一个可靠的任务调度器比如APScheduler或Celery来定时、并发地触发数据采集任务。这一层还要负责任务的去重、优先级管理以及异常状态记录。数据存储与分析层这是系统的“记忆”和“分析中心”。抓取到的数据不能看一眼就丢需要持久化存储以便进行历史对比和趋势分析。我通常会选用时序数据库InfluxDB来存储价格、库存等随时间变化的数据方便绘制价格走势图。同时用PostgreSQL或MySQL来存储商品、卖家的详细属性信息。分析模块可以基于历史数据计算平均价格、价格波动率、热门卖家等指标。消息通知与告警层这是系统的“嘴巴”。当监控到符合条件的事件如目标商品降价、心仪关键词下有新商品上架时系统需要及时通知用户。通知方式可以多样化集成钉钉、企业微信、飞书的机器人进行群消息推送通过SMTP协议发送邮件甚至调用短信接口。告警规则需要可配置比如“当价格低于设定阈值时立即告警”或“同一卖家上新超过3件时提示”。2.2 关键技术选型考量为什么用Playwright/Selenium而不是纯请求闲鱼前端大量使用JavaScript渲染商品列表和详情数据很可能通过Ajax动态加载。纯HTTP请求如requests获取到的HTML是不完整的无法直接解析出商品数据。浏览器自动化工具能完整执行JS获取最终渲染的页面虽然速度稍慢但数据获取最可靠、最接近真人浏览。为什么需要任务调度器手动运行脚本不现实。我们需要系统能自动、周期性地执行监控任务。APScheduler是一个轻量级但功能强大的Python库支持定时、间隔、Cron式的任务触发非常适合本项目。如果未来监控任务量极大需要考虑分布式那么Celery是更专业的选择。数据存储的考量商品属性标题、描述、卖家ID这类关系型数据用PostgreSQL存储很合适。而价格变化是典型的时间序列数据每秒都可能产生新点。InfluxDB针对时间序列的写入、查询和聚合做了大量优化存储效率和查询速度远超传统关系型数据库做价格趋势图非常方便。注意任何针对第三方网站的自动化操作都必须严格遵守该网站的robots.txt协议并控制请求频率避免对目标服务器造成过大压力。本系统设计初衷是用于个人或小范围的合规市场调研严禁用于恶意爬取、数据盗用或任何干扰网站正常运营的行为。3. 核心模块实现细节与实操要点接下来我们深入到几个核心模块看看具体怎么实现以及里面有哪些容易踩坑的地方。3.1 高隐蔽性数据采集器的实现采集器是直接与闲鱼“交锋”的前线它的稳定性决定了整个系统的生死。基础实现使用Playwrightfrom playwright.sync_api import sync_playwright import time import random def fetch_xianyu_page_by_playwright(keyword, max_pages3): 使用Playwright模拟浏览器搜索闲鱼关键词并翻页抓取 all_items [] with sync_playwright() as p: # 1. 启动浏览器推荐使用Chromium可配合--no-sandbox等参数 browser p.chromium.launch(headlessTrue) # 生产环境建议用headless context browser.new_context( viewport{width: 1920, height: 1080}, user_agentMozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36... # 设置真实UA ) page context.new_page() try: # 2. 访问闲鱼首页或搜索页增加随机延迟模拟真人 page.goto(https://s.2.taobao.com/list/list.htm) time.sleep(random.uniform(2, 5)) # 首次访问等待 # 3. 模拟输入关键词并搜索需根据实际页面元素调整选择器 page.fill(input#search-input, keyword) # 选择器示例需实际查看 page.click(button.search-btn) page.wait_for_load_state(networkidle) # 等待网络空闲 time.sleep(random.uniform(3, 6)) current_page 1 while current_page max_pages: print(f正在抓取第 {current_page} 页...) # 4. 获取当前页面HTML内容 content page.content() # 这里调用解析函数下一节详述 items parse_page_items(content) all_items.extend(items) # 5. 模拟翻页 - 寻找“下一页”按钮并点击 next_button page.locator(a.next-page) # 选择器示例 if next_button.is_visible(): # 翻页前随机移动鼠标到按钮附近再点击 next_button.hover() time.sleep(random.uniform(1, 3)) next_button.click() page.wait_for_load_state(networkidle) time.sleep(random.uniform(4, 8)) # 翻页后等待更长时间 current_page 1 else: print(已到最后一页或未找到下一页按钮。) break except Exception as e: print(f抓取过程中发生错误: {e}) finally: browser.close() return all_items实操要点与避坑指南User-Agent与浏览器指纹不要使用默认的Playwright或Selenium UA。务必设置成主流浏览器的真实UA字符串。更高级的反爬会检测WebDriver特征Playwright的chromium.launch可以通过args参数传入--disable-blink-featuresAutomationControlled等来隐藏自动化痕迹。随机化等待时间固定的sleep(5)是机器行为的明显标志。所有等待时间页面加载后、点击前、翻页后都应使用random.uniform(a, b)在一个区间内随机取值模拟人类阅读和操作的不确定性。处理登录与验证码频繁访问或触发某些规则后可能会遇到登录墙或滑块验证码。对于个人低频使用可以考虑手动登录一次然后将浏览器上下文Cookies持久化保存后续采集时复用。但这需要妥善保管Cookie文件。绝对不要尝试自动破解验证码这违反平台规则。遇到验证码时应暂停任务记录日志并发出告警等待人工处理。IP代理池对于极高频率的监控单一IP容易被封。需要搭建或购买可靠的IP代理池并在请求时随机切换。但代理质量参差不齐会增加系统复杂度和不稳定因素个人或小规模使用初期可以不考虑。3.2 健壮的数据解析策略解析层必须足够健壮以应对网页结构的微小变动。from bs4 import BeautifulSoup import re def parse_page_items(html_content): 从闲鱼列表页HTML中解析出商品信息 soup BeautifulSoup(html_content, html.parser) items [] # 闲鱼商品列表项的选择器这个需要定期检查和更新 # 示例每个商品卡片可能在某个div classitem里 item_elements soup.select(div.card-item) # 请根据实际页面调整 for elem in item_elements: try: item {} # 1. 解析标题 - 增加容错找不到则赋默认值 title_elem elem.select_one(div.title a) item[title] title_elem.get_text(stripTrue) if title_elem else N/A # 2. 解析价格 - 通常包含货币符号需要提取数字 price_elem elem.select_one(div.price strong) price_text price_elem.get_text(stripTrue) if price_elem else 0 # 使用正则表达式提取数字包括小数 price_match re.search(r[\d\.], price_text) item[price] float(price_match.group()) if price_match else 0.0 # 3. 解析商品链接相对路径转绝对路径 link_elem elem.select_one(a[href*item.taobao.com]) if link_elem and link_elem.get(href): href link_elem[href] if href.startswith(//): item[url] https: href elif href.startswith(/): item[url] https://s.2.taobao.com href else: item[url] href else: item[url] # 4. 解析卖家信息、地点、发布时间等选择器需具体分析 seller_elem elem.select_one(div.seller-nickname) item[seller] seller_elem.get_text(stripTrue) if seller_elem else 匿名 # ... 解析其他字段 items.append(item) except Exception as e: # 记录解析单个商品失败的错误但不影响其他商品 print(f解析商品元素时出错: {e} 跳过该商品。) continue # 跳过当前出错商品继续下一个 return items实操要点与避坑指南选择器管理将所有的CSS选择器或XPath集中存储在一个配置文件如selectors.yaml或config.py中。当闲鱼页面改版时你只需要更新这个配置文件而无需修改核心解析代码。多层解析与降级策略不要指望一个选择器永远有效。可以为关键字段如价格设计2-3个备选选择器。当主选择器解析失败时尝试使用备选选择器这能大大提高系统的抗变化能力。异常捕获与数据清洗每个字段的解析都应放在try-except块内。一个商品解析失败不应导致整个页面解析中断。对解析出的文本数据如价格要进行清洗去除多余的空格、货币符号并转换为正确的数据类型整数、浮点数。定期巡检与测试编写一个简单的测试脚本每天定时运行一次用几个固定的关键词测试解析函数是否能成功提取数据。一旦测试失败立即触发告警提醒你更新选择器。3.3 任务调度与状态管理这是系统的指挥中心确保每个监控任务都能按时、正确地执行。from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.interval import IntervalTrigger import threading import time from datetime import datetime class MonitoringScheduler: def __init__(self): self.scheduler BackgroundScheduler() self.task_registry {} # 存储任务ID与任务信息的映射 self.lock threading.Lock() # 防止并发修改冲突 def add_keyword_task(self, task_id, keyword, interval_minutes10, filtersNone): 添加一个关键词监控任务 if filters is None: filters {} with self.lock: if task_id in self.task_registry: print(f任务 {task_id} 已存在先移除旧任务。) self.remove_task(task_id) # 包装任务函数传递参数 def job_wrapper(): self._execute_keyword_monitoring(task_id, keyword, filters) # 创建触发器每 interval_minutes 分钟执行一次 trigger IntervalTrigger(minutesinterval_minutes) job self.scheduler.add_job( funcjob_wrapper, triggertrigger, idtask_id, namef监控-{keyword}, replace_existingTrue # 如果存在则替换 ) self.task_registry[task_id] { keyword: keyword, interval: interval_minutes, filters: filters, job: job, last_run: None, last_result: None } print(f已添加任务: {task_id} ({keyword}), 间隔 {interval_minutes} 分钟。) def _execute_keyword_monitoring(self, task_id, keyword, filters): 实际执行监控任务的函数 print(f[{datetime.now()}] 开始执行任务 {task_id}: {keyword}) try: # 1. 调用采集器获取数据 items fetch_xianyu_page_by_playwright(keyword, max_pages2) # 示例抓2页 # 2. 应用过滤条件 (如价格范围、卖家信用等) filtered_items self._apply_filters(items, filters) # 3. 与上一次结果对比发现新商品或价格变动 new_items, price_changed_items self._detect_changes(task_id, filtered_items) # 4. 如果发现变化触发通知 if new_items or price_changed_items: self._send_notification(task_id, new_items, price_changed_items) # 5. 更新任务状态 with self.lock: if task_id in self.task_registry: self.task_registry[task_id][last_run] datetime.now() self.task_registry[task_id][last_result] { total_fetched: len(items), filtered: len(filtered_items), new: len(new_items), price_changed: len(price_changed_items) } print(f[{datetime.now()}] 任务 {task_id} 执行完毕。) except Exception as e: print(f[{datetime.now()}] 任务 {task_id} 执行失败: {e}) # 这里可以添加失败告警 def _apply_filters(self, items, filters): # 实现根据价格、地区、卖家信用等条件过滤商品 filtered items if max_price in filters: filtered [i for i in filtered if i.get(price, float(inf)) filters[max_price]] if min_price in filters: filtered [i for i in filtered if i.get(price, 0) filters[min_price]] # ... 其他过滤条件 return filtered def _detect_changes(self, task_id, current_items): # 需要与之前存储的数据进行对比这里简化处理 # 实际应查询数据库对比商品ID、价格等 new_items [] price_changed_items [] # ... 实现对比逻辑 return new_items, price_changed_items def _send_notification(self, task_id, new_items, price_changed_items): # 实现发送钉钉/微信/邮件通知的逻辑 message f监控任务 {task_id} 发现更新\n if new_items: message f新上架商品: {len(new_items)} 个\n if price_changed_items: message f价格变动商品: {len(price_changed_items)} 个\n print(f发送通知: {message}) # 调用具体的通知发送函数 def start(self): 启动调度器 self.scheduler.start() print(监控调度器已启动。) def shutdown(self): 关闭调度器 self.scheduler.shutdown() print(监控调度器已关闭。) # 使用示例 if __name__ __main__: scheduler MonitoringScheduler() scheduler.start() # 添加一个监控“Switch游戏卡”的任务价格低于200元每15分钟检查一次 scheduler.add_keyword_task( task_idswitch_card_monitor, keywordSwitch 游戏卡, interval_minutes15, filters{max_price: 200} ) # 保持主线程运行 try: while True: time.sleep(1) except (KeyboardInterrupt, SystemExit): scheduler.shutdown()实操要点与避坑指南任务持久化APScheduler默认将任务存储在内存中一旦程序重启所有任务都会丢失。在生产环境中必须配置作业存储Job Store例如使用SQLAlchemyJobStore将任务存储到数据库中确保系统重启后任务能恢复。并发控制与资源限制如果有几十上百个监控任务同时触发可能会耗尽网络或系统资源。需要设置调度器的最大并发实例数executor或者对任务进行分组错峰执行。优雅停止一定要捕获像KeyboardInterruptCtrlC这样的中断信号并在处理程序中调用scheduler.shutdown()让正在执行的任务完成后再退出避免数据不一致。状态记录与监控除了任务本身系统应该记录每个任务的执行日志、成功/失败状态、最后一次运行时间等。这有助于后期排查问题和评估系统健康度。4. 数据存储、分析与可视化实战数据只有被存储和分析才能产生长期价值。4.1 数据库设计我们需要至少两张核心表商品表 (items):存储商品的基本信息这些信息相对静态。CREATE TABLE items ( id BIGSERIAL PRIMARY KEY, item_id VARCHAR(255) UNIQUE NOT NULL, -- 闲鱼商品唯一ID (从URL中提取) title TEXT, url TEXT, seller_id VARCHAR(100), seller_name VARCHAR(255), location VARCHAR(100), created_at TIMESTAMP, -- 商品首次上架时间 first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP, -- 系统首次发现时间 updated_at TIMESTAMP -- 信息更新时间 ); CREATE INDEX idx_item_id ON items(item_id); CREATE INDEX idx_seller_id ON items(seller_id);价格历史表 (price_history):存储商品价格随时间的变化这是时序数据。CREATE TABLE price_history ( id BIGSERIAL PRIMARY KEY, item_id VARCHAR(255) NOT NULL, price DECIMAL(10, 2) NOT NULL, -- 价格 timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL, -- 记录时间点 FOREIGN KEY (item_id) REFERENCES items(item_id) ON DELETE CASCADE ); CREATE INDEX idx_price_history_item_time ON price_history(item_id, timestamp DESC);对于price_history如果数据量极大且查询频繁强烈建议使用InfluxDB。其插入和按时间范围查询的效率极高。其数据模型类似于measurement: item_prices tags: item_id123456789, keywordiphone13 fields: price3888.00 time: 2023-10-27T14:30:00Z4.2 数据分析示例有了数据我们可以进行一些简单的分析import pandas as pd import matplotlib.pyplot as plt from sqlalchemy import create_engine def analyze_price_trend(item_id): 分析某个商品的价格走势 # 连接数据库 engine create_engine(postgresql://user:passwordlocalhost/mydb) # 查询该商品的历史价格 query f SELECT timestamp, price FROM price_history WHERE item_id {item_id} ORDER BY timestamp ASC df pd.read_sql_query(query, engine) if df.empty: print(未找到该商品的价格历史数据。) return # 基础分析 df[timestamp] pd.to_datetime(df[timestamp]) df.set_index(timestamp, inplaceTrue) print(f价格分析报告 - 商品ID: {item_id}) print(f数据时间范围: {df.index.min()} 至 {df.index.max()}) print(f价格波动次数: {len(df)}) print(f最高价: {df[price].max():.2f}) print(f最低价: {df[price].min():.2f}) print(f平均价: {df[price].mean():.2f}) print(f当前价: {df[price].iloc[-1]:.2f}) # 简单可视化 plt.figure(figsize(12, 6)) plt.plot(df.index, df[price], markero, linestyle-, linewidth1, markersize3) plt.title(f商品价格走势图 (ID: {item_id})) plt.xlabel(时间) plt.ylabel(价格 (元)) plt.grid(True, whichboth, linestyle--, linewidth0.5, alpha0.7) plt.xticks(rotation45) plt.tight_layout() # 保存图片或显示 plt.savefig(fprice_trend_{item_id}.png, dpi150) # plt.show() print(f走势图已保存为 price_trend_{item_id}.png) # 还可以分析市场整体情况比如某个关键词下的平均价格分布、卖家集中度等。 def analyze_market(keyword, days7): 分析过去N天内某关键词的市场概况 query f SELECT DATE(timestamp) as date, COUNT(DISTINCT item_id) as new_items, AVG(price) as avg_price, MIN(price) as min_price, MAX(price) as max_price FROM price_history ph JOIN items i ON ph.item_id i.item_id WHERE i.title ILIKE %{keyword}% AND ph.timestamp NOW() - INTERVAL {days} days GROUP BY DATE(timestamp) ORDER BY date ASC # ... 执行查询并分析实操要点与避坑指南数据去重在插入items表时必须使用闲鱼商品的唯一ID通常可以从商品URL中解析作为唯一约束。避免同一商品被重复插入导致数据冗余。增量更新每次抓取时应先根据商品ID查询数据库中是否存在。如果存在则比较价格是否有变化有变化则插入一条新的price_history记录并更新items表的updated_at字段。如果不存在则插入新的商品记录和第一条价格历史。连接管理数据库连接是宝贵资源。务必使用连接池如SQLAlchemy的create_engine默认提供并在每次操作后正确关闭会话或连接防止连接泄漏。异步操作数据库IO尤其是插入大量价格历史记录是瓶颈。可以考虑使用异步数据库驱动如asyncpgfor PostgreSQL和异步框架如asyncio将IO操作与计算密集型任务分离提升系统吞吐量。5. 通知告警与系统集成监控到变化后如何及时、有效地通知用户是系统价值的最终体现。5.1 钉钉机器人通知实现钉钉群机器人是常用的通知方式配置简单到达率高。import requests import json import hashlib import base64 import hmac import time class DingTalkRobot: def __init__(self, webhook_url, secretNone): 初始化钉钉机器人 :param webhook_url: 完整的Webhook地址 :param secret: 安全设置中的加签密钥可选 self.webhook_url webhook_url self.secret secret def _generate_sign(self): 生成加签如果启用了安全设置 if not self.secret: return None, None timestamp str(round(time.time() * 1000)) string_to_sign f{timestamp}\n{self.secret} hmac_code hmac.new( self.secret.encode(utf-8), string_to_sign.encode(utf-8), digestmodhashlib.sha256 ).digest() sign base64.b64encode(hmac_code).decode(utf-8) return timestamp, sign def send_markdown(self, title, text, at_mobilesNone, is_at_allFalse): 发送Markdown格式消息 headers {Content-Type: application/json} timestamp, sign self._generate_sign() url self.webhook_url if timestamp and sign: url ftimestamp{timestamp}sign{sign} message { msgtype: markdown, markdown: { title: title, text: text }, at: { atMobiles: at_mobiles if at_mobiles else [], isAtAll: is_at_all } } try: response requests.post(url, headersheaders, datajson.dumps(message), timeout10) result response.json() if result.get(errcode) 0: print(钉钉消息发送成功) return True else: print(f钉钉消息发送失败: {result.get(errmsg)}) return False except Exception as e: print(f发送钉钉消息时发生异常: {e}) return False # 在监控到变化时调用 def send_dingtalk_alert(task_name, new_items, price_changed_items): robot DingTalkRobot( webhook_urlhttps://oapi.dingtalk.com/robot/send?access_tokenYOUR_TOKEN, secretYOUR_SECRET # 如果设置了加签 ) markdown_text f### 闲鱼监控告警 - {task_name}\n\n if new_items: markdown_text ** 新上架商品:**\n for item in new_items[:5]: # 最多显示5条避免消息过长 markdown_text f- [{item[title]}]({item[url]}) - **¥{item[price]}**\n if len(new_items) 5: markdown_text f... 等 {len(new_items)} 件商品\n if price_changed_items: markdown_text \n** 价格变动商品:**\n for item in price_changed_items[:5]: # 假设item中包含新旧价格信息 markdown_text f- [{item[title]}]({item[url]}) - ¥{item[old_price]} → **¥{item[new_price]}**\n if len(price_changed_items) 5: markdown_text f... 等 {len(price_changed_items)} 件商品\n markdown_text f\n---\n*触发时间: {time.strftime(%Y-%m-%d %H:%M:%S)}* success robot.send_markdown( titlef闲鱼监控{task_name}, textmarkdown_text, # at_mobiles[13800138000] # 需要特定人时填写 ) return success5.2 邮件通知实现对于不常用即时通讯工具的用户邮件通知是个可靠的备选。import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from email.header import Header def send_email_alert(smtp_config, to_addrs, subject, html_content): 通过SMTP发送HTML邮件 :param smtp_config: 字典包含 host, port, user, password, use_tls :param to_addrs: 收件人列表 :param subject: 邮件主题 :param html_content: HTML格式的邮件内容 msg MIMEMultipart(alternative) msg[From] smtp_config.get(sender, smtp_config[user]) msg[To] , .join(to_addrs) msg[Subject] Header(subject, utf-8) # 添加HTML内容 html_part MIMEText(html_content, html, utf-8) msg.attach(html_part) try: if smtp_config.get(use_tls, True): server smtplib.SMTP_SSL(smtp_config[host], smtp_config.get(port, 465)) else: server smtplib.SMTP(smtp_config[host], smtp_config.get(port, 25)) server.starttls() # 升级为TLS加密连接 server.login(smtp_config[user], smtp_config[password]) server.sendmail(msg[From], to_addrs, msg.as_string()) server.quit() print(邮件发送成功) return True except Exception as e: print(f邮件发送失败: {e}) return False # 配置示例 smtp_conf { host: smtp.qq.com, port: 465, user: your_emailqq.com, password: your_authorization_code, # 注意是授权码不是邮箱密码 sender: your_emailqq.com, use_tls: True } # 构建HTML内容 html_body h2闲鱼监控通知/h2 p您监控的关键词 strong“Switch游戏卡”/strong 发现了新的商品或价格变动。/p table border1 cellpadding5 trth商品标题/thth价格/thth链接/th/tr trtd塞尔达传说 旷野之息/tdtd stylecolor:red;¥185/tdtda hrefhttps://...查看/a/td/tr /table # send_email_alert(smtp_conf, [recipientexample.com], 闲鱼监控告警, html_body)实操要点与避坑指南通知去重与聚合如果监控频率很高短时间内可能触发多次相同或类似的告警比如一个商品被多个任务监控到。需要在通知前进行去重或者将一段时间内的变化聚合到一条消息中发送避免“轰炸”用户。多通道降级重要的监控任务可以配置“主备”通知通道。例如优先发送钉钉如果连续失败N次则自动切换到邮件或短信确保告警不丢失。消息模板化将消息内容如标题、正文格式设计成模板通过变量填充。这样便于统一风格也方便后续修改。可以使用Jinja2等模板引擎。速率限制与礼貌性遵守各通知渠道的发送频率限制。例如钉钉机器人有频率限制默认20条/分钟。在代码中需要做简单的限流避免触发平台限制。6. 部署、运维与常见问题排查一个开发完成的系统需要稳定地跑起来才能发挥作用。6.1 系统部署方案对于个人或小团队推荐以下两种部署方式方案A本地服务器/常开电脑 后台运行优点成本低完全可控。做法将代码上传到一台长期开机的电脑或树莓派上。使用systemdLinux或nssmWindows将主程序注册为系统服务实现开机自启和进程守护。使用cron或systemd timer定期执行数据备份和日志清理脚本。命令示例Linux systemd服务文件/etc/systemd/system/xianyu-monitor.service[Unit] DescriptionXianyu智能监控机器人 Afternetwork.target [Service] Typesimple Useryour_username WorkingDirectory/path/to/your/code ExecStart/usr/bin/python3 /path/to/your/code/main.py Restarton-failure RestartSec10s [Install] WantedBymulti-user.target然后使用sudo systemctl enable xianyu-monitor和sudo systemctl start xianyu-monitor来启用和启动服务。方案B云服务器部署优点网络稳定不受本地断电断网影响易于远程管理。做法购买一台入门级云服务器如1核2G安装Python环境、数据库。部署方式与本地类似但更推荐使用Docker容器化部署将应用、数据库、调度器打包成镜像用docker-compose管理迁移和升级更方便。简易Dockerfile示例FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple COPY . . # 安装Playwright浏览器 RUN playwright install chromium --with-deps CMD [python, main.py]6.2 常见问题与排查实录在系统运行过程中你肯定会遇到各种问题。以下是我踩过的一些坑和解决方法问题现象可能原因排查步骤与解决方案突然抓不到任何数据1. 闲鱼页面结构改版。2. IP被暂时限制。3. 触发了验证码。1.检查解析器手动访问目标页面用浏览器开发者工具检查商品列表的HTML结构是否变化更新选择器。2.检查网络尝试用同一IP的浏览器手动访问看是否正常。如果被限暂停任务几小时或更换IP如有代理。3.检查日志查看采集器是否返回了验证码页面或登录页面的HTML。数据库连接失败或变慢1. 连接数过多未释放。2. 数据库表未建索引查询慢。3. 磁盘空间不足。1.检查连接池确保每次数据库操作后正确关闭session/connection。使用连接池并设置最大连接数。2.分析慢查询对price_history的(item_id, timestamp)和items的item_id建立复合索引。3.监控磁盘定期清理过期数据如3个月前的价格历史或进行分表。调度任务不执行或重复执行1. 系统时间不同步。2. APScheduler配置了错误的时区。3. 多进程/多实例导致任务重复。1.同步时间使用ntpdate或systemd-timesyncd同步服务器时间。2.设置时区在创建scheduler时指定timezoneAsia/Shanghai。3.使用独占锁对于只能单实例运行的任务在任务函数开始前尝试获取一个文件锁或数据库锁确保同一时间只有一个进程执行。通知消息发送失败1. 钉钉/微信机器人密钥错误或过期。2. 网络问题导致API请求超时。3. 邮件SMTP配置错误。1.验证Token/Secret重新在钉钉/企业微信群添加机器人获取新的Webhook URL和密钥。2.增加重试机制对网络请求添加重试逻辑如tenacity库并设置超时时间。3.测试SMTP先用命令行telnet smtp.xxx.com 465测试连通性再检查邮箱是否开启SMTP服务并使用了正确的授权码。程序内存占用越来越高1. 存在内存泄漏如未关闭浏览器实例、全局列表无限增长。2. 抓取数据量过大一次性加载到内存。1.使用with语句确保Playwright/Selenium的browser和context对象在使用后被正确关闭。2.流式处理数据边抓取边解析边存储不要将所有商品数据都积累在一个大列表里。使用生成器yield逐条返回数据。我个人在实际运维中的几点深刻体会日志是生命线一定要给系统加上详细且结构化的日志。不仅要记录信息INFO更要记录警告WARN和错误ERROR。使用logging模块将日志按级别输出到文件和控制台并设置日志轮转RotatingFileHandler方便事后排查。当系统出问题时第一个查看的就是日志文件。设置“熔断”机制如果连续多次抓取失败比如连续10次解析不到数据很可能是页面结构大改版或IP被彻底封禁。此时系统应自动暂停所有相关任务并发送最高优先级的告警如短信、电话给管理员而不是继续无意义地尝试浪费资源。数据备份要定期监控数据是你的核心资产。定期如每天将数据库导出备份到另一台机器或云存储。最简单的就是用pg_dumpPostgreSQL命令配合cron任务。从简单开始逐步迭代不要一开始就追求大而全。先实现核心的“监控-通知”闭环让它稳定跑起来。然后再逐步加入数据分析、Web管理界面、更复杂的过滤规则等功能。一个能稳定运行的基础系统远比一个充满Bug的复杂系统有价值。本文还有配套的精品资源点击获取 Python自动化监控系统构建：从数据采集到智能告警的完整实践
简介这是一套面向Python开发者与电商数据分析师的闲鱼智能监控与分析工具解决人工盯梢效率低、筛选逻辑僵化、信息过载等痛点适用于二手商品比价、热门商品抢购、竞品动态追踪等实际场景。资源包共39个文件含11个核心Python脚本如web_server.py、scraper.py、ai_handler.py、2个Docker配置文件docker-compose.yaml、Dockerfile、4个图文素材PNG/JPG、3个HTML/CSS/JS前端页面及多个提示词模板prompts目录和示例配置config.json.example、.env.example整体12.31MB结构清晰模块职责分明。已有190人学习下载可直接运行Web管理界面获得自然语言创建任务、多模态AI图文分析、实时流式推送ntfy/企微/Bark、Cron定时调度及反爬增强等完整能力附带教程.docx与免责声明开箱即用。1. 项目概述为什么我们需要一个闲鱼智能监控机器人如果你在闲鱼上做过生意或者经常在上面淘货肯定有过这样的经历看中一个商品犹豫了一下或者想等卖家降价结果一转眼就被别人拍走了。又或者你是一个卖家想知道自己发布的商品在同类中的价格竞争力或者想监控某个关键词下新上架的商品动态。手动去刷、去比价不仅效率低下而且非常容易错过关键信息。这个“闲鱼智能监控分析系统”就是为了解决这些痛点而生的。简单来说它就是一个7x24小时不间断工作的“数字助理”。它的核心任务就是代替你按照你设定的规则比如关键词、价格区间、卖家信用等自动、持续地监控闲鱼平台上的商品信息变化并将关键数据如价格变动、上新提醒、商品下架实时推送给你甚至能进行初步的数据分析帮你做出更明智的决策。无论是个人买家想“捡漏”还是专业卖家或工作室需要进行市场调研和竞争分析这个系统都能大幅提升效率把人力从重复、枯燥的监控工作中解放出来。2. 系统核心架构与设计思路拆解一个完整的闲鱼监控系统远不止写个脚本爬取网页那么简单。它需要稳定、隐蔽、智能且易于维护。我们不能把它做成一个简单粗暴、频繁请求的爬虫那样极易被平台风控识别并封禁。因此系统的设计必须考虑合规性、健壮性和可扩展性。2.1 分层架构设计我设计的系统通常采用清晰的分层架构这有助于各模块解耦方便后续维护和功能扩展。数据采集层这是系统的“眼睛”和“手”。负责模拟浏览器行为访问闲鱼页面抓取HTML数据。这里的关键在于模拟真人操作。我们不能用简单的requests库直接访问因为闲鱼有完善的反爬机制。更稳妥的做法是使用Selenium或Playwright这类浏览器自动化工具配合真实的浏览器内核如Chrome并注入合理的鼠标移动、滚动、等待时间等行为脚本。对于大规模监控可以考虑使用puppeteer-extra及其stealth插件来进一步增强隐蔽性。数据处理与解析层这是系统的“大脑”。采集到的原始HTML是杂乱无章的这一层负责从中提取结构化信息。我们会使用BeautifulSoup或lxml来解析HTML定位商品标题、价格、图片、卖家信息、发布时间等关键字段。这里有一个难点闲鱼的页面结构可能会不定期调整导致解析规则失效。因此解析规则需要设计得具有一定的容错性和可配置性最好能通过配置文件来管理选择器CSS Selector或XPath。任务调度与监控核心层这是系统的“心脏”。它管理着所有监控任务Task。每个任务定义了监控目标如关键词“iPhone 13 256G”、过滤条件价格低于4000元卖家信用“极好”、执行频率如每5分钟一次和通知渠道。我们需要一个可靠的任务调度器比如APScheduler或Celery来定时、并发地触发数据采集任务。这一层还要负责任务的去重、优先级管理以及异常状态记录。数据存储与分析层这是系统的“记忆”和“分析中心”。抓取到的数据不能看一眼就丢需要持久化存储以便进行历史对比和趋势分析。我通常会选用时序数据库InfluxDB来存储价格、库存等随时间变化的数据方便绘制价格走势图。同时用PostgreSQL或MySQL来存储商品、卖家的详细属性信息。分析模块可以基于历史数据计算平均价格、价格波动率、热门卖家等指标。消息通知与告警层这是系统的“嘴巴”。当监控到符合条件的事件如目标商品降价、心仪关键词下有新商品上架时系统需要及时通知用户。通知方式可以多样化集成钉钉、企业微信、飞书的机器人进行群消息推送通过SMTP协议发送邮件甚至调用短信接口。告警规则需要可配置比如“当价格低于设定阈值时立即告警”或“同一卖家上新超过3件时提示”。2.2 关键技术选型考量为什么用Playwright/Selenium而不是纯请求闲鱼前端大量使用JavaScript渲染商品列表和详情数据很可能通过Ajax动态加载。纯HTTP请求如requests获取到的HTML是不完整的无法直接解析出商品数据。浏览器自动化工具能完整执行JS获取最终渲染的页面虽然速度稍慢但数据获取最可靠、最接近真人浏览。为什么需要任务调度器手动运行脚本不现实。我们需要系统能自动、周期性地执行监控任务。APScheduler是一个轻量级但功能强大的Python库支持定时、间隔、Cron式的任务触发非常适合本项目。如果未来监控任务量极大需要考虑分布式那么Celery是更专业的选择。数据存储的考量商品属性标题、描述、卖家ID这类关系型数据用PostgreSQL存储很合适。而价格变化是典型的时间序列数据每秒都可能产生新点。InfluxDB针对时间序列的写入、查询和聚合做了大量优化存储效率和查询速度远超传统关系型数据库做价格趋势图非常方便。注意任何针对第三方网站的自动化操作都必须严格遵守该网站的robots.txt协议并控制请求频率避免对目标服务器造成过大压力。本系统设计初衷是用于个人或小范围的合规市场调研严禁用于恶意爬取、数据盗用或任何干扰网站正常运营的行为。3. 核心模块实现细节与实操要点接下来我们深入到几个核心模块看看具体怎么实现以及里面有哪些容易踩坑的地方。3.1 高隐蔽性数据采集器的实现采集器是直接与闲鱼“交锋”的前线它的稳定性决定了整个系统的生死。基础实现使用Playwrightfrom playwright.sync_api import sync_playwright import time import random def fetch_xianyu_page_by_playwright(keyword, max_pages3): 使用Playwright模拟浏览器搜索闲鱼关键词并翻页抓取 all_items [] with sync_playwright() as p: # 1. 启动浏览器推荐使用Chromium可配合--no-sandbox等参数 browser p.chromium.launch(headlessTrue) # 生产环境建议用headless context browser.new_context( viewport{width: 1920, height: 1080}, user_agentMozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36... # 设置真实UA ) page context.new_page() try: # 2. 访问闲鱼首页或搜索页增加随机延迟模拟真人 page.goto(https://s.2.taobao.com/list/list.htm) time.sleep(random.uniform(2, 5)) # 首次访问等待 # 3. 模拟输入关键词并搜索需根据实际页面元素调整选择器 page.fill(input#search-input, keyword) # 选择器示例需实际查看 page.click(button.search-btn) page.wait_for_load_state(networkidle) # 等待网络空闲 time.sleep(random.uniform(3, 6)) current_page 1 while current_page max_pages: print(f正在抓取第 {current_page} 页...) # 4. 获取当前页面HTML内容 content page.content() # 这里调用解析函数下一节详述 items parse_page_items(content) all_items.extend(items) # 5. 模拟翻页 - 寻找“下一页”按钮并点击 next_button page.locator(a.next-page) # 选择器示例 if next_button.is_visible(): # 翻页前随机移动鼠标到按钮附近再点击 next_button.hover() time.sleep(random.uniform(1, 3)) next_button.click() page.wait_for_load_state(networkidle) time.sleep(random.uniform(4, 8)) # 翻页后等待更长时间 current_page 1 else: print(已到最后一页或未找到下一页按钮。) break except Exception as e: print(f抓取过程中发生错误: {e}) finally: browser.close() return all_items实操要点与避坑指南User-Agent与浏览器指纹不要使用默认的Playwright或Selenium UA。务必设置成主流浏览器的真实UA字符串。更高级的反爬会检测WebDriver特征Playwright的chromium.launch可以通过args参数传入--disable-blink-featuresAutomationControlled等来隐藏自动化痕迹。随机化等待时间固定的sleep(5)是机器行为的明显标志。所有等待时间页面加载后、点击前、翻页后都应使用random.uniform(a, b)在一个区间内随机取值模拟人类阅读和操作的不确定性。处理登录与验证码频繁访问或触发某些规则后可能会遇到登录墙或滑块验证码。对于个人低频使用可以考虑手动登录一次然后将浏览器上下文Cookies持久化保存后续采集时复用。但这需要妥善保管Cookie文件。绝对不要尝试自动破解验证码这违反平台规则。遇到验证码时应暂停任务记录日志并发出告警等待人工处理。IP代理池对于极高频率的监控单一IP容易被封。需要搭建或购买可靠的IP代理池并在请求时随机切换。但代理质量参差不齐会增加系统复杂度和不稳定因素个人或小规模使用初期可以不考虑。3.2 健壮的数据解析策略解析层必须足够健壮以应对网页结构的微小变动。from bs4 import BeautifulSoup import re def parse_page_items(html_content): 从闲鱼列表页HTML中解析出商品信息 soup BeautifulSoup(html_content, html.parser) items [] # 闲鱼商品列表项的选择器这个需要定期检查和更新 # 示例每个商品卡片可能在某个div classitem里 item_elements soup.select(div.card-item) # 请根据实际页面调整 for elem in item_elements: try: item {} # 1. 解析标题 - 增加容错找不到则赋默认值 title_elem elem.select_one(div.title a) item[title] title_elem.get_text(stripTrue) if title_elem else N/A # 2. 解析价格 - 通常包含货币符号需要提取数字 price_elem elem.select_one(div.price strong) price_text price_elem.get_text(stripTrue) if price_elem else 0 # 使用正则表达式提取数字包括小数 price_match re.search(r[\d\.], price_text) item[price] float(price_match.group()) if price_match else 0.0 # 3. 解析商品链接相对路径转绝对路径 link_elem elem.select_one(a[href*item.taobao.com]) if link_elem and link_elem.get(href): href link_elem[href] if href.startswith(//): item[url] https: href elif href.startswith(/): item[url] https://s.2.taobao.com href else: item[url] href else: item[url] # 4. 解析卖家信息、地点、发布时间等选择器需具体分析 seller_elem elem.select_one(div.seller-nickname) item[seller] seller_elem.get_text(stripTrue) if seller_elem else 匿名 # ... 解析其他字段 items.append(item) except Exception as e: # 记录解析单个商品失败的错误但不影响其他商品 print(f解析商品元素时出错: {e} 跳过该商品。) continue # 跳过当前出错商品继续下一个 return items实操要点与避坑指南选择器管理将所有的CSS选择器或XPath集中存储在一个配置文件如selectors.yaml或config.py中。当闲鱼页面改版时你只需要更新这个配置文件而无需修改核心解析代码。多层解析与降级策略不要指望一个选择器永远有效。可以为关键字段如价格设计2-3个备选选择器。当主选择器解析失败时尝试使用备选选择器这能大大提高系统的抗变化能力。异常捕获与数据清洗每个字段的解析都应放在try-except块内。一个商品解析失败不应导致整个页面解析中断。对解析出的文本数据如价格要进行清洗去除多余的空格、货币符号并转换为正确的数据类型整数、浮点数。定期巡检与测试编写一个简单的测试脚本每天定时运行一次用几个固定的关键词测试解析函数是否能成功提取数据。一旦测试失败立即触发告警提醒你更新选择器。3.3 任务调度与状态管理这是系统的指挥中心确保每个监控任务都能按时、正确地执行。from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.interval import IntervalTrigger import threading import time from datetime import datetime class MonitoringScheduler: def __init__(self): self.scheduler BackgroundScheduler() self.task_registry {} # 存储任务ID与任务信息的映射 self.lock threading.Lock() # 防止并发修改冲突 def add_keyword_task(self, task_id, keyword, interval_minutes10, filtersNone): 添加一个关键词监控任务 if filters is None: filters {} with self.lock: if task_id in self.task_registry: print(f任务 {task_id} 已存在先移除旧任务。) self.remove_task(task_id) # 包装任务函数传递参数 def job_wrapper(): self._execute_keyword_monitoring(task_id, keyword, filters) # 创建触发器每 interval_minutes 分钟执行一次 trigger IntervalTrigger(minutesinterval_minutes) job self.scheduler.add_job( funcjob_wrapper, triggertrigger, idtask_id, namef监控-{keyword}, replace_existingTrue # 如果存在则替换 ) self.task_registry[task_id] { keyword: keyword, interval: interval_minutes, filters: filters, job: job, last_run: None, last_result: None } print(f已添加任务: {task_id} ({keyword}), 间隔 {interval_minutes} 分钟。) def _execute_keyword_monitoring(self, task_id, keyword, filters): 实际执行监控任务的函数 print(f[{datetime.now()}] 开始执行任务 {task_id}: {keyword}) try: # 1. 调用采集器获取数据 items fetch_xianyu_page_by_playwright(keyword, max_pages2) # 示例抓2页 # 2. 应用过滤条件 (如价格范围、卖家信用等) filtered_items self._apply_filters(items, filters) # 3. 与上一次结果对比发现新商品或价格变动 new_items, price_changed_items self._detect_changes(task_id, filtered_items) # 4. 如果发现变化触发通知 if new_items or price_changed_items: self._send_notification(task_id, new_items, price_changed_items) # 5. 更新任务状态 with self.lock: if task_id in self.task_registry: self.task_registry[task_id][last_run] datetime.now() self.task_registry[task_id][last_result] { total_fetched: len(items), filtered: len(filtered_items), new: len(new_items), price_changed: len(price_changed_items) } print(f[{datetime.now()}] 任务 {task_id} 执行完毕。) except Exception as e: print(f[{datetime.now()}] 任务 {task_id} 执行失败: {e}) # 这里可以添加失败告警 def _apply_filters(self, items, filters): # 实现根据价格、地区、卖家信用等条件过滤商品 filtered items if max_price in filters: filtered [i for i in filtered if i.get(price, float(inf)) filters[max_price]] if min_price in filters: filtered [i for i in filtered if i.get(price, 0) filters[min_price]] # ... 其他过滤条件 return filtered def _detect_changes(self, task_id, current_items): # 需要与之前存储的数据进行对比这里简化处理 # 实际应查询数据库对比商品ID、价格等 new_items [] price_changed_items [] # ... 实现对比逻辑 return new_items, price_changed_items def _send_notification(self, task_id, new_items, price_changed_items): # 实现发送钉钉/微信/邮件通知的逻辑 message f监控任务 {task_id} 发现更新\n if new_items: message f新上架商品: {len(new_items)} 个\n if price_changed_items: message f价格变动商品: {len(price_changed_items)} 个\n print(f发送通知: {message}) # 调用具体的通知发送函数 def start(self): 启动调度器 self.scheduler.start() print(监控调度器已启动。) def shutdown(self): 关闭调度器 self.scheduler.shutdown() print(监控调度器已关闭。) # 使用示例 if __name__ __main__: scheduler MonitoringScheduler() scheduler.start() # 添加一个监控“Switch游戏卡”的任务价格低于200元每15分钟检查一次 scheduler.add_keyword_task( task_idswitch_card_monitor, keywordSwitch 游戏卡, interval_minutes15, filters{max_price: 200} ) # 保持主线程运行 try: while True: time.sleep(1) except (KeyboardInterrupt, SystemExit): scheduler.shutdown()实操要点与避坑指南任务持久化APScheduler默认将任务存储在内存中一旦程序重启所有任务都会丢失。在生产环境中必须配置作业存储Job Store例如使用SQLAlchemyJobStore将任务存储到数据库中确保系统重启后任务能恢复。并发控制与资源限制如果有几十上百个监控任务同时触发可能会耗尽网络或系统资源。需要设置调度器的最大并发实例数executor或者对任务进行分组错峰执行。优雅停止一定要捕获像KeyboardInterruptCtrlC这样的中断信号并在处理程序中调用scheduler.shutdown()让正在执行的任务完成后再退出避免数据不一致。状态记录与监控除了任务本身系统应该记录每个任务的执行日志、成功/失败状态、最后一次运行时间等。这有助于后期排查问题和评估系统健康度。4. 数据存储、分析与可视化实战数据只有被存储和分析才能产生长期价值。4.1 数据库设计我们需要至少两张核心表商品表 (items):存储商品的基本信息这些信息相对静态。CREATE TABLE items ( id BIGSERIAL PRIMARY KEY, item_id VARCHAR(255) UNIQUE NOT NULL, -- 闲鱼商品唯一ID (从URL中提取) title TEXT, url TEXT, seller_id VARCHAR(100), seller_name VARCHAR(255), location VARCHAR(100), created_at TIMESTAMP, -- 商品首次上架时间 first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP, -- 系统首次发现时间 updated_at TIMESTAMP -- 信息更新时间 ); CREATE INDEX idx_item_id ON items(item_id); CREATE INDEX idx_seller_id ON items(seller_id);价格历史表 (price_history):存储商品价格随时间的变化这是时序数据。CREATE TABLE price_history ( id BIGSERIAL PRIMARY KEY, item_id VARCHAR(255) NOT NULL, price DECIMAL(10, 2) NOT NULL, -- 价格 timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL, -- 记录时间点 FOREIGN KEY (item_id) REFERENCES items(item_id) ON DELETE CASCADE ); CREATE INDEX idx_price_history_item_time ON price_history(item_id, timestamp DESC);对于price_history如果数据量极大且查询频繁强烈建议使用InfluxDB。其插入和按时间范围查询的效率极高。其数据模型类似于measurement: item_prices tags: item_id123456789, keywordiphone13 fields: price3888.00 time: 2023-10-27T14:30:00Z4.2 数据分析示例有了数据我们可以进行一些简单的分析import pandas as pd import matplotlib.pyplot as plt from sqlalchemy import create_engine def analyze_price_trend(item_id): 分析某个商品的价格走势 # 连接数据库 engine create_engine(postgresql://user:passwordlocalhost/mydb) # 查询该商品的历史价格 query f SELECT timestamp, price FROM price_history WHERE item_id {item_id} ORDER BY timestamp ASC df pd.read_sql_query(query, engine) if df.empty: print(未找到该商品的价格历史数据。) return # 基础分析 df[timestamp] pd.to_datetime(df[timestamp]) df.set_index(timestamp, inplaceTrue) print(f价格分析报告 - 商品ID: {item_id}) print(f数据时间范围: {df.index.min()} 至 {df.index.max()}) print(f价格波动次数: {len(df)}) print(f最高价: {df[price].max():.2f}) print(f最低价: {df[price].min():.2f}) print(f平均价: {df[price].mean():.2f}) print(f当前价: {df[price].iloc[-1]:.2f}) # 简单可视化 plt.figure(figsize(12, 6)) plt.plot(df.index, df[price], markero, linestyle-, linewidth1, markersize3) plt.title(f商品价格走势图 (ID: {item_id})) plt.xlabel(时间) plt.ylabel(价格 (元)) plt.grid(True, whichboth, linestyle--, linewidth0.5, alpha0.7) plt.xticks(rotation45) plt.tight_layout() # 保存图片或显示 plt.savefig(fprice_trend_{item_id}.png, dpi150) # plt.show() print(f走势图已保存为 price_trend_{item_id}.png) # 还可以分析市场整体情况比如某个关键词下的平均价格分布、卖家集中度等。 def analyze_market(keyword, days7): 分析过去N天内某关键词的市场概况 query f SELECT DATE(timestamp) as date, COUNT(DISTINCT item_id) as new_items, AVG(price) as avg_price, MIN(price) as min_price, MAX(price) as max_price FROM price_history ph JOIN items i ON ph.item_id i.item_id WHERE i.title ILIKE %{keyword}% AND ph.timestamp NOW() - INTERVAL {days} days GROUP BY DATE(timestamp) ORDER BY date ASC # ... 执行查询并分析实操要点与避坑指南数据去重在插入items表时必须使用闲鱼商品的唯一ID通常可以从商品URL中解析作为唯一约束。避免同一商品被重复插入导致数据冗余。增量更新每次抓取时应先根据商品ID查询数据库中是否存在。如果存在则比较价格是否有变化有变化则插入一条新的price_history记录并更新items表的updated_at字段。如果不存在则插入新的商品记录和第一条价格历史。连接管理数据库连接是宝贵资源。务必使用连接池如SQLAlchemy的create_engine默认提供并在每次操作后正确关闭会话或连接防止连接泄漏。异步操作数据库IO尤其是插入大量价格历史记录是瓶颈。可以考虑使用异步数据库驱动如asyncpgfor PostgreSQL和异步框架如asyncio将IO操作与计算密集型任务分离提升系统吞吐量。5. 通知告警与系统集成监控到变化后如何及时、有效地通知用户是系统价值的最终体现。5.1 钉钉机器人通知实现钉钉群机器人是常用的通知方式配置简单到达率高。import requests import json import hashlib import base64 import hmac import time class DingTalkRobot: def __init__(self, webhook_url, secretNone): 初始化钉钉机器人 :param webhook_url: 完整的Webhook地址 :param secret: 安全设置中的加签密钥可选 self.webhook_url webhook_url self.secret secret def _generate_sign(self): 生成加签如果启用了安全设置 if not self.secret: return None, None timestamp str(round(time.time() * 1000)) string_to_sign f{timestamp}\n{self.secret} hmac_code hmac.new( self.secret.encode(utf-8), string_to_sign.encode(utf-8), digestmodhashlib.sha256 ).digest() sign base64.b64encode(hmac_code).decode(utf-8) return timestamp, sign def send_markdown(self, title, text, at_mobilesNone, is_at_allFalse): 发送Markdown格式消息 headers {Content-Type: application/json} timestamp, sign self._generate_sign() url self.webhook_url if timestamp and sign: url ftimestamp{timestamp}sign{sign} message { msgtype: markdown, markdown: { title: title, text: text }, at: { atMobiles: at_mobiles if at_mobiles else [], isAtAll: is_at_all } } try: response requests.post(url, headersheaders, datajson.dumps(message), timeout10) result response.json() if result.get(errcode) 0: print(钉钉消息发送成功) return True else: print(f钉钉消息发送失败: {result.get(errmsg)}) return False except Exception as e: print(f发送钉钉消息时发生异常: {e}) return False # 在监控到变化时调用 def send_dingtalk_alert(task_name, new_items, price_changed_items): robot DingTalkRobot( webhook_urlhttps://oapi.dingtalk.com/robot/send?access_tokenYOUR_TOKEN, secretYOUR_SECRET # 如果设置了加签 ) markdown_text f### 闲鱼监控告警 - {task_name}\n\n if new_items: markdown_text ** 新上架商品:**\n for item in new_items[:5]: # 最多显示5条避免消息过长 markdown_text f- [{item[title]}]({item[url]}) - **¥{item[price]}**\n if len(new_items) 5: markdown_text f... 等 {len(new_items)} 件商品\n if price_changed_items: markdown_text \n** 价格变动商品:**\n for item in price_changed_items[:5]: # 假设item中包含新旧价格信息 markdown_text f- [{item[title]}]({item[url]}) - ¥{item[old_price]} → **¥{item[new_price]}**\n if len(price_changed_items) 5: markdown_text f... 等 {len(price_changed_items)} 件商品\n markdown_text f\n---\n*触发时间: {time.strftime(%Y-%m-%d %H:%M:%S)}* success robot.send_markdown( titlef闲鱼监控{task_name}, textmarkdown_text, # at_mobiles[13800138000] # 需要特定人时填写 ) return success5.2 邮件通知实现对于不常用即时通讯工具的用户邮件通知是个可靠的备选。import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from email.header import Header def send_email_alert(smtp_config, to_addrs, subject, html_content): 通过SMTP发送HTML邮件 :param smtp_config: 字典包含 host, port, user, password, use_tls :param to_addrs: 收件人列表 :param subject: 邮件主题 :param html_content: HTML格式的邮件内容 msg MIMEMultipart(alternative) msg[From] smtp_config.get(sender, smtp_config[user]) msg[To] , .join(to_addrs) msg[Subject] Header(subject, utf-8) # 添加HTML内容 html_part MIMEText(html_content, html, utf-8) msg.attach(html_part) try: if smtp_config.get(use_tls, True): server smtplib.SMTP_SSL(smtp_config[host], smtp_config.get(port, 465)) else: server smtplib.SMTP(smtp_config[host], smtp_config.get(port, 25)) server.starttls() # 升级为TLS加密连接 server.login(smtp_config[user], smtp_config[password]) server.sendmail(msg[From], to_addrs, msg.as_string()) server.quit() print(邮件发送成功) return True except Exception as e: print(f邮件发送失败: {e}) return False # 配置示例 smtp_conf { host: smtp.qq.com, port: 465, user: your_emailqq.com, password: your_authorization_code, # 注意是授权码不是邮箱密码 sender: your_emailqq.com, use_tls: True } # 构建HTML内容 html_body h2闲鱼监控通知/h2 p您监控的关键词 strong“Switch游戏卡”/strong 发现了新的商品或价格变动。/p table border1 cellpadding5 trth商品标题/thth价格/thth链接/th/tr trtd塞尔达传说 旷野之息/tdtd stylecolor:red;¥185/tdtda hrefhttps://...查看/a/td/tr /table # send_email_alert(smtp_conf, [recipientexample.com], 闲鱼监控告警, html_body)实操要点与避坑指南通知去重与聚合如果监控频率很高短时间内可能触发多次相同或类似的告警比如一个商品被多个任务监控到。需要在通知前进行去重或者将一段时间内的变化聚合到一条消息中发送避免“轰炸”用户。多通道降级重要的监控任务可以配置“主备”通知通道。例如优先发送钉钉如果连续失败N次则自动切换到邮件或短信确保告警不丢失。消息模板化将消息内容如标题、正文格式设计成模板通过变量填充。这样便于统一风格也方便后续修改。可以使用Jinja2等模板引擎。速率限制与礼貌性遵守各通知渠道的发送频率限制。例如钉钉机器人有频率限制默认20条/分钟。在代码中需要做简单的限流避免触发平台限制。6. 部署、运维与常见问题排查一个开发完成的系统需要稳定地跑起来才能发挥作用。6.1 系统部署方案对于个人或小团队推荐以下两种部署方式方案A本地服务器/常开电脑 后台运行优点成本低完全可控。做法将代码上传到一台长期开机的电脑或树莓派上。使用systemdLinux或nssmWindows将主程序注册为系统服务实现开机自启和进程守护。使用cron或systemd timer定期执行数据备份和日志清理脚本。命令示例Linux systemd服务文件/etc/systemd/system/xianyu-monitor.service[Unit] DescriptionXianyu智能监控机器人 Afternetwork.target [Service] Typesimple Useryour_username WorkingDirectory/path/to/your/code ExecStart/usr/bin/python3 /path/to/your/code/main.py Restarton-failure RestartSec10s [Install] WantedBymulti-user.target然后使用sudo systemctl enable xianyu-monitor和sudo systemctl start xianyu-monitor来启用和启动服务。方案B云服务器部署优点网络稳定不受本地断电断网影响易于远程管理。做法购买一台入门级云服务器如1核2G安装Python环境、数据库。部署方式与本地类似但更推荐使用Docker容器化部署将应用、数据库、调度器打包成镜像用docker-compose管理迁移和升级更方便。简易Dockerfile示例FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple COPY . . # 安装Playwright浏览器 RUN playwright install chromium --with-deps CMD [python, main.py]6.2 常见问题与排查实录在系统运行过程中你肯定会遇到各种问题。以下是我踩过的一些坑和解决方法问题现象可能原因排查步骤与解决方案突然抓不到任何数据1. 闲鱼页面结构改版。2. IP被暂时限制。3. 触发了验证码。1.检查解析器手动访问目标页面用浏览器开发者工具检查商品列表的HTML结构是否变化更新选择器。2.检查网络尝试用同一IP的浏览器手动访问看是否正常。如果被限暂停任务几小时或更换IP如有代理。3.检查日志查看采集器是否返回了验证码页面或登录页面的HTML。数据库连接失败或变慢1. 连接数过多未释放。2. 数据库表未建索引查询慢。3. 磁盘空间不足。1.检查连接池确保每次数据库操作后正确关闭session/connection。使用连接池并设置最大连接数。2.分析慢查询对price_history的(item_id, timestamp)和items的item_id建立复合索引。3.监控磁盘定期清理过期数据如3个月前的价格历史或进行分表。调度任务不执行或重复执行1. 系统时间不同步。2. APScheduler配置了错误的时区。3. 多进程/多实例导致任务重复。1.同步时间使用ntpdate或systemd-timesyncd同步服务器时间。2.设置时区在创建scheduler时指定timezoneAsia/Shanghai。3.使用独占锁对于只能单实例运行的任务在任务函数开始前尝试获取一个文件锁或数据库锁确保同一时间只有一个进程执行。通知消息发送失败1. 钉钉/微信机器人密钥错误或过期。2. 网络问题导致API请求超时。3. 邮件SMTP配置错误。1.验证Token/Secret重新在钉钉/企业微信群添加机器人获取新的Webhook URL和密钥。2.增加重试机制对网络请求添加重试逻辑如tenacity库并设置超时时间。3.测试SMTP先用命令行telnet smtp.xxx.com 465测试连通性再检查邮箱是否开启SMTP服务并使用了正确的授权码。程序内存占用越来越高1. 存在内存泄漏如未关闭浏览器实例、全局列表无限增长。2. 抓取数据量过大一次性加载到内存。1.使用with语句确保Playwright/Selenium的browser和context对象在使用后被正确关闭。2.流式处理数据边抓取边解析边存储不要将所有商品数据都积累在一个大列表里。使用生成器yield逐条返回数据。我个人在实际运维中的几点深刻体会日志是生命线一定要给系统加上详细且结构化的日志。不仅要记录信息INFO更要记录警告WARN和错误ERROR。使用logging模块将日志按级别输出到文件和控制台并设置日志轮转RotatingFileHandler方便事后排查。当系统出问题时第一个查看的就是日志文件。设置“熔断”机制如果连续多次抓取失败比如连续10次解析不到数据很可能是页面结构大改版或IP被彻底封禁。此时系统应自动暂停所有相关任务并发送最高优先级的告警如短信、电话给管理员而不是继续无意义地尝试浪费资源。数据备份要定期监控数据是你的核心资产。定期如每天将数据库导出备份到另一台机器或云存储。最简单的就是用pg_dumpPostgreSQL命令配合cron任务。从简单开始逐步迭代不要一开始就追求大而全。先实现核心的“监控-通知”闭环让它稳定跑起来。然后再逐步加入数据分析、Web管理界面、更复杂的过滤规则等功能。一个能稳定运行的基础系统远比一个充满Bug的复杂系统有价值。本文还有配套的精品资源点击获取