以下是实现的闲鱼商品监控脚本，它通过定时抓取商品信息，并在检测到更新时通过钉钉机器人推送通知。

###核心功能与依赖| 功能模块 | 说明 | 关键依赖库 |
| :--- | :--- | :--- |
| 数据抓取 | 模拟请求获取闲鱼商品页面信息 | requests, BeautifulSoup4 |
| 定时任务 | 按设定频率执行监控任务 | schedule 或 time |
| 消息推送 | 将监控结果发送到钉钉群聊 | requests (调用钉钉Webhook) |
| 异常处理 | 保证脚本稳定运行 | try...except |

完整代码实现

import requests
import json
import time
import schedule
from bs4 import BeautifulSoup

class XianyuMonitor:
    """
    闲鱼商品监控与钉钉推送类
    """
    def __init__(self, target_url, dingding_webhook):
        """
        初始化监控器 :param target_url: 要监控的闲鱼商品URL :param dingding_webhook: 钉钉机器人的Webhook地址 """
        self.target_url = target_url self.dingding_webhook = dingding_webhook
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
        }
        # 用于记录上一次检查的商品信息，实现更新检测
        self.last_item_info = {}

    def fetch_item_info(self):
        """
        抓取闲鱼商品页面的关键信息 :return: 商品信息字典，若抓取失败返回None
        """
        try:
            response = requests.get(self.target_url, headers=self.headers, timeout=10)
            response.raise_for_status()  # 检查HTTP请求是否成功
            soup = BeautifulSoup(response.text, 'html.parser')

            # 示例解析逻辑：实际需根据闲鱼页面HTML结构调整选择器 
            # 1. 获取商品标题
            title_tag = soup.find('h1', class_='item-title')  # 示例选择器 title = title_tag.get_text(strip=True) if title_tag else '未找到标题'

            # 2. 获取商品价格 price_tag = soup.find('span', class_='price')  # 示例选择器
            price = price_tag.get_text(strip=True) if price_tag else '未找到价格'

            # 3. 获取卖家信息（示例）
            seller_tag = soup.find('a', class_='seller-name')
            seller = seller_tag.get_text(strip=True) if seller_tag else '未知卖家'

            # 4. 获取商品发布时间或状态（用于判断是否更新）
            # 注意：闲鱼页面结构可能变化，此处为示例
            time_tag = soup.find('span', class_='publish-time')
            publish_time = time_tag.get_text(strip=True) if time_tag else '未知时间'

            item_info = {
                'title': title,
                'price': price,
                'seller': seller,
                'publish_time': publish_time,
                'url': self.target_url
            }
            return item_info except requests.exceptions.RequestException as e:
            print(f"网络请求失败: {e}")
            return None
        except Exception as e:
            print(f"解析页面时出现错误: {e}")
            return None def compare_and_notify(self, current_info):
        """
        比较当前信息与上次信息，如果发生变化则发送钉钉通知 :param current_info: 当前抓取到的商品信息        """
        if not current_info:
            print("当前未获取到商品信息，跳过比较。")
            return

        # 首次运行，只记录信息不推送 if not self.last_item_info:
            print("首次运行，记录初始商品信息。")
            self.last_item_info = current_info
            return # 判断信息是否发生变化的逻辑（此处以标题和价格为例）
        is_changed = (current_info['title'] != self.last_item_info['title'] or
                      current_info['price'] != self.last_item_info['price'])

        if is_changed:
            print("检测到商品信息变化，准备发送通知...")
            self.send_dingding_notification(current_info, self.last_item_info)
            # 更新最后一次记录的信息 self.last_item_info = current_info
        else:
            print(f"商品信息未变化。上次检查时间: {time.strftime('%Y-%m-%d %H:%M:%S')}")

    def send_dingding_notification(self, current_info, old_info):
        """
        构造并发送钉钉Markdown格式的通知消息 
        :param current_info: 当前商品信息 :param old_info: 上一次的商品信息
        """
        # 构造Markdown消息内容
        message = {
            "msgtype": "markdown",
            "markdown": {
                "title": "闲鱼商品监控提醒",
                "text": f"""## ⚠️ 闲鱼监控商品信息已更新！
**商品标题**：
-旧：{old_info['title']}
- **新：{current_info['title']}**

**商品价格**：
-旧：{old_info['price']}
- **新：{current_info['price']}**

**卖家**：{current_info['seller']}
**发布时间**：{current_info['publish_time']}

**商品链接**：[点击查看]({current_info['url']})

*监控时间：{time.strftime('%Y-%m-%d %H:%M:%S')}*
"""
            },
            "at": {
                "isAtAll": False  # 是否@所有人，可根据需要修改 }
        }

        try:
            headers = {'Content-Type': 'application/json'}
            response = requests.post(self.dingding_webhook,
                                     data=json.dumps(message),
                                     headers=headers,
                                     timeout=5)
            if response.status_code == 200:
                print("钉钉通知发送成功！")
            else:
                print(f"钉钉通知发送失败，状态码：{response.status_code}, 响应：{response.text}")
        except Exception as e:
            print(f"发送钉钉通知时出现异常: {e}")

    def monitor_job(self):
        """定时执行的任务函数"""
        print(f"
开始执行监控任务 ({time.strftime('%Y-%m-%d %H:%M:%S')})...")
        current_info = self.fetch_item_info()
        self.compare_and_notify(current_info)

    def run(self, interval_minutes=0.1):
        """
        启动定时监控任务 :param interval_minutes: 监控间隔时间（分钟），默认0.1分钟（即6秒）
        """
        print(f"闲鱼商品监控已启动，监控URL: {self.target_url}")
        print(f"监控频率: 每 {interval_minutes} 分钟检查一次")
        # 使用schedule库定时执行
        schedule.every(interval_minutes).minutes.do(self.monitor_job)

        # 立即执行一次        self.monitor_job()

        while True:
            schedule.run_pending()
            time.sleep(1)  # 避免CPU空转

# ==================== 配置与使用示例 ====================
if __name__ == "__main__":
    # 1. 配置参数（用户需修改此处）
    # 要监控的闲鱼商品URL TARGET_ITEM_URL = "https://2.taobao.com/item.htm?id=你的商品ID"
    # 钉钉群机器人的Webhook地址（需在钉钉群添加自定义机器人获取）
    DINGDING_WEBHOOK_URL = "https://oapi.dingtalk.com/robot/send?access_token=你的token"

    # 2. 创建监控实例并运行 monitor = XianyuMonitor(target_url=TARGET_ITEM_URL,
                            dingding_webhook=DINGDING_WEBHOOK_URL)
    # 启动监控，每5分钟检查一次（可根据需要调整间隔）
    monitor.run(interval_minutes=5)

关键配置与使用步骤

1. 安装依赖库：

   pip install requests beautifulsoup4 schedule
   

2. 获取钉钉机器人Webhook：

   • 在钉钉群中，点击“群设置” -> “智能群助手” -> “添加机器人” -> “自定义”。
   • 设置机器人名称和安全设置（例如选择“加签”或“IP地址段”）。
   • 复制生成的 Webhook URL，填入代码中的 DINGDING_WEBHOOK_URL 变量。

3. 设置监控目标：

   • 将 TARGET_ITEM_URL 变量替换为你要监控的闲鱼商品页面的真实URL。

4. 运行脚本：

   python xianyu_monitor.py
   

   脚本将按设定间隔（默认为5分钟）自动检查商品信息，并在检测到变化时推送钉钉通知。

注意事项与优化建议

• HTML结构变化：闲鱼网页结构可能更新，导致 BeautifulSoup 选择器失效。若抓取失败，需手动检查页面元素并更新代码中的选择器（如 find('h1', class_='item-title')）。
• 反爬虫策略：频繁请求可能触发反爬机制。可考虑：
  • 增加随机延迟（time.sleep(random.uniform(1, 3))）。
  • 使用代理IP池。
  • 添加更真实的请求头（User-Agent, Referer 等）。
• 推送频率限制：钉钉机器人有消息频率限制（默认20条/分钟），请合理设置监控间隔，避免超限 。
• 错误处理与日志：生产环境建议将 print 替换为更完善的日志记录（如 logging 模块），并增加更细致的异常捕获和重试机制。
• 信息存储：对于需要长期监控或对比历史数据，可将 self.last_item_info 持久化到文件或数据库中。
• 多商品监控：可通过创建 XianyuMonitor 实例列表，或修改代码循环处理多个URL，实现对多个商品的监控 。

参考来源