企业官网建设流程全解析

简介：这是一套面向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 高隐蔽性数据采集器的实现

采集器是直接与闲鱼“交锋”的前线，它的稳定性决定了整个系统的生死。

基础实现（使用Playwright）：

from playwright.sync_api import sync_playwright import time import random def fetch_xianyu_page_by_playwright(keyword, max_pages=3): """ 使用Playwright模拟浏览器搜索闲鱼关键词并翻页抓取 """ all_items = [] with sync_playwright() as p: # 1. 启动浏览器，推荐使用Chromium，可配合`--no-sandbox`等参数 browser = p.chromium.launch(headless=True) # 生产环境建议用headless context = browser.new_context( viewport={'width': 1920, 'height': 1080}, user_agent='Mozilla/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

实操要点与避坑指南：

1. User-Agent与浏览器指纹：不要使用默认的Playwright或Selenium UA。务必设置成主流浏览器的真实UA字符串。更高级的反爬会检测WebDriver特征，Playwright的chromium.launch可以通过args参数传入--disable-blink-features=AutomationControlled等来隐藏自动化痕迹。
2. 随机化等待时间：固定的sleep(5)是机器行为的明显标志。所有等待时间（页面加载后、点击前、翻页后）都应使用random.uniform(a, b)在一个区间内随机取值，模拟人类阅读和操作的不确定性。
3. 处理登录与验证码：频繁访问或触发某些规则后，可能会遇到登录墙或滑块验证码。对于个人低频使用，可以考虑手动登录一次，然后将浏览器上下文（Cookies）持久化保存，后续采集时复用。但这需要妥善保管Cookie文件。绝对不要尝试自动破解验证码，这违反平台规则。遇到验证码时应暂停任务，记录日志并发出告警，等待人工处理。
4. 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 class="item">`里 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(strip=True) if title_elem else 'N/A' # 2. 解析价格 - 通常包含货币符号，需要提取数字 price_elem = elem.select_one('div.price strong') price_text = price_elem.get_text(strip=True) 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(strip=True) if seller_elem else '匿名' # ... 解析其他字段 items.append(item) except Exception as e: # 记录解析单个商品失败的错误，但不影响其他商品 print(f"解析商品元素时出错: {e}， 跳过该商品。") continue # 跳过当前出错商品，继续下一个 return items

实操要点与避坑指南：

1. 选择器管理：将所有的CSS选择器或XPath集中存储在一个配置文件（如selectors.yaml或config.py）中。当闲鱼页面改版时，你只需要更新这个配置文件，而无需修改核心解析代码。
2. 多层解析与降级策略：不要指望一个选择器永远有效。可以为关键字段（如价格）设计2-3个备选选择器。当主选择器解析失败时，尝试使用备选选择器，这能大大提高系统的抗变化能力。
3. 异常捕获与数据清洗：每个字段的解析都应放在try-except块内。一个商品解析失败不应导致整个页面解析中断。对解析出的文本数据（如价格）要进行清洗，去除多余的空格、货币符号，并转换为正确的数据类型（整数、浮点数）。
4. 定期巡检与测试：编写一个简单的测试脚本，每天定时运行一次，用几个固定的关键词测试解析函数是否能成功提取数据。一旦测试失败，立即触发告警，提醒你更新选择器。

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_minutes=10, filters=None): """添加一个关键词监控任务""" 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(minutes=interval_minutes) job = self.scheduler.add_job( func=job_wrapper, trigger=trigger, id=task_id, name=f"监控-{keyword}", replace_existing=True # 如果存在则替换 ) 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_pages=2) # 示例抓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_id='switch_card_monitor', keyword='Switch 游戏卡', interval_minutes=15, filters={'max_price': 200} ) # 保持主线程运行 try: while True: time.sleep(1) except (KeyboardInterrupt, SystemExit): scheduler.shutdown()

实操要点与避坑指南：

1. 任务持久化：APScheduler默认将任务存储在内存中，一旦程序重启，所有任务都会丢失。在生产环境中，必须配置作业存储（Job Store），例如使用SQLAlchemyJobStore将任务存储到数据库中，确保系统重启后任务能恢复。
2. 并发控制与资源限制：如果有几十上百个监控任务，同时触发可能会耗尽网络或系统资源。需要设置调度器的最大并发实例数（executor），或者对任务进行分组，错峰执行。
3. 优雅停止：一定要捕获像KeyboardInterrupt（Ctrl+C）这样的中断信号，并在处理程序中调用scheduler.shutdown()，让正在执行的任务完成后再退出，避免数据不一致。
4. 状态记录与监控：除了任务本身，系统应该记录每个任务的执行日志、成功/失败状态、最后一次运行时间等。这有助于后期排查问题和评估系统健康度。

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_id="123456789", keyword="iphone13" fields: price=3888.00 time: 2023-10-27T14:30:00Z

4.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:password@localhost/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', inplace=True) 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'], marker='o', linestyle='-', linewidth=1, markersize=3) plt.title(f'商品价格走势图 (ID: {item_id})') plt.xlabel('时间') plt.ylabel('价格 (元)') plt.grid(True, which='both', linestyle='--', linewidth=0.5, alpha=0.7) plt.xticks(rotation=45) plt.tight_layout() # 保存图片或显示 plt.savefig(f'price_trend_{item_id}.png', dpi=150) # plt.show() print(f"走势图已保存为 'price_trend_{item_id}.png'") # 还可以分析市场整体情况，比如某个关键词下的平均价格分布、卖家集中度等。 def analyze_market(keyword, days=7): """分析过去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 """ # ... 执行查询并分析

实操要点与避坑指南：

1. 数据去重：在插入items表时，必须使用闲鱼商品的唯一ID（通常可以从商品URL中解析）作为唯一约束。避免同一商品被重复插入，导致数据冗余。
2. 增量更新：每次抓取时，应先根据商品ID查询数据库中是否存在。如果存在，则比较价格是否有变化，有变化则插入一条新的price_history记录，并更新items表的updated_at字段。如果不存在，则插入新的商品记录和第一条价格历史。
3. 连接管理：数据库连接是宝贵资源。务必使用连接池（如SQLAlchemy的create_engine默认提供），并在每次操作后正确关闭会话或连接，防止连接泄漏。
4. 异步操作：数据库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, secret=None): """ 初始化钉钉机器人 :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'), digestmod=hashlib.sha256 ).digest() sign = base64.b64encode(hmac_code).decode('utf-8') return timestamp, sign def send_markdown(self, title, text, at_mobiles=None, is_at_all=False): """ 发送Markdown格式消息 """ headers = {'Content-Type': 'application/json'} timestamp, sign = self._generate_sign() url = self.webhook_url if timestamp and sign: url += f'&timestamp={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, headers=headers, data=json.dumps(message), timeout=10) 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_url='https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN', secret='YOUR_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( title=f"闲鱼监控：{task_name}", text=markdown_text, # at_mobiles=['13800138000'] # 需要@特定人时填写 ) return success

5.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_email@qq.com', 'password': 'your_authorization_code', # 注意是授权码，不是邮箱密码 'sender': 'your_email@qq.com', 'use_tls': True } # 构建HTML内容 html_body = """ <h2>闲鱼监控通知</h2> <p>您监控的关键词 <strong>“Switch游戏卡”</strong> 发现了新的商品或价格变动。</p> <table border="1" cellpadding="5"> <tr><th>商品标题</th><th>价格</th><th>链接</th></tr> <tr><td>塞尔达传说 旷野之息</td><td style="color:red;">¥185</td><td><a href="https://...">查看</a></td></tr> </table> """ # send_email_alert(smtp_conf, ['recipient@example.com'], '闲鱼监控告警', html_body)

实操要点与避坑指南：

1. 通知去重与聚合：如果监控频率很高，短时间内可能触发多次相同或类似的告警（比如一个商品被多个任务监控到）。需要在通知前进行去重，或者将一段时间内的变化聚合到一条消息中发送，避免“轰炸”用户。
2. 多通道降级：重要的监控任务，可以配置“主备”通知通道。例如，优先发送钉钉，如果连续失败N次，则自动切换到邮件或短信，确保告警不丢失。
3. 消息模板化：将消息内容（如标题、正文格式）设计成模板，通过变量填充。这样便于统一风格，也方便后续修改。可以使用Jinja2等模板引擎。
4. 速率限制与礼貌性：遵守各通知渠道的发送频率限制。例如，钉钉机器人有频率限制（默认20条/分钟）。在代码中需要做简单的限流，避免触发平台限制。

6. 部署、运维与常见问题排查

一个开发完成的系统，需要稳定地跑起来才能发挥作用。

6.1 系统部署方案

对于个人或小团队，推荐以下两种部署方式：

• 方案A：本地服务器/常开电脑 + 后台运行

  • 优点：成本低，完全可控。
  • 做法：
    1. 将代码上传到一台长期开机的电脑或树莓派上。
    2. 使用systemd（Linux）或nssm（Windows）将主程序注册为系统服务，实现开机自启和进程守护。
    3. 使用cron或systemd timer定期执行数据备份和日志清理脚本。
  • 命令示例（Linux systemd服务文件/etc/systemd/system/xianyu-monitor.service）：

    [Unit] Description=Xianyu智能监控机器人 After=network.target [Service] Type=simple User=your_username WorkingDirectory=/path/to/your/code ExecStart=/usr/bin/python3 /path/to/your/code/main.py Restart=on-failure RestartSec=10s [Install] WantedBy=multi-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. 日志是生命线：一定要给系统加上详细且结构化的日志。不仅要记录信息（INFO），更要记录警告（WARN）和错误（ERROR）。使用logging模块，将日志按级别输出到文件和控制台，并设置日志轮转（RotatingFileHandler），方便事后排查。当系统出问题时，第一个查看的就是日志文件。
2. 设置“熔断”机制：如果连续多次抓取失败（比如连续10次解析不到数据），很可能是页面结构大改版或IP被彻底封禁。此时系统应自动暂停所有相关任务，并发送最高优先级的告警（如短信、电话）给管理员，而不是继续无意义地尝试，浪费资源。
3. 数据备份要定期：监控数据是你的核心资产。定期（如每天）将数据库导出备份到另一台机器或云存储。最简单的就是用pg_dump（PostgreSQL）命令配合cron任务。
4. 从简单开始，逐步迭代：不要一开始就追求大而全。先实现核心的“监控-通知”闭环，让它稳定跑起来。然后再逐步加入数据分析、Web管理界面、更复杂的过滤规则等功能。一个能稳定运行的基础系统，远比一个充满Bug的复杂系统有价值。

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