我们计划编写一个闲鱼监控脚本，主要功能是监控特定关键词的商品，并在发现新商品时通过钉钉机器人推送通知。

```python
import requests
import time
import json
import hashlib
import os
from datetime import datetime# ===== 配置区域 =====
KEYWORDS = ["显卡", "游戏本"]  # 监控关键词列表
MONITOR_INTERVAL = 1  # 监控间隔(秒)
DINGTALK_WEBHOOK = "https://oapi.dingtalk.com/robot/send?access_token=your_token_here"  # 钉钉机器人Webhook
DATA_FILE = "seen_items.json"  # 已出现商品存储文件# 请求头(模拟浏览器)
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","Referer": "https://s.2.taobao.com/"
}# ===== 功能函数 =====
def get_xianyu_items(keyword):"""获取闲鱼关键词搜索结果"""try:url = f"https://s.2.taobao.com/list/list.htm?q={keyword}&search_type=item"response = requests.get(url, headers=HEADERS, timeout=10)response.raise_for_status()# 解析商品数据（实际需要更复杂的解析）# 这里简化处理，真实环境需分析页面结构items = []# 示例解析逻辑（需根据实际页面调整）：# soup = BeautifulSoup(response.text, 'html.parser')# for item in soup.select('.item-card'):#     title = item.select_one('.title').text.strip()#     price = item.select_one('.price').text.strip()#     link = item.select_one('a')['href']#     items.append({"title": title, "price": price, "link": link})# 模拟数据print(f"正在模拟获取关键词 [{keyword}] 的商品...")return [{"title": f"{keyword}商品示例1", "price": "999", "link": "https://item.2.taobao.com/item1"},{"title": f"{keyword}商品示例2", "price": "1500", "link": "https://item.2.taobao.com/item2"}]except Exception as e:print(f"获取商品失败: {e}")return []def generate_item_id(item):"""生成商品唯一ID(MD5哈希)"""data = f"{item['title']}{item['link']}"return hashlib.md5(data.encode()).hexdigest()def load_seen_items():"""加载已出现商品记录"""if not os.path.exists(DATA_FILE):return set()try:with open(DATA_FILE, 'r') as f:return set(json.load(f))except:return set()def save_seen_items(seen_set):"""保存已出现商品记录"""with open(DATA_FILE, 'w') as f:json.dump(list(seen_set), f)def send_dingtalk_message(item, keyword):"""发送钉钉消息通知"""message = {"msgtype": "markdown","markdown": {"title": "闲鱼新商品通知","text": f"### 🔔 闲鱼监控通知\n\n"f"**关键词**：{keyword}\n\n"f"**商品标题**：{item['title']}\n\n"f"**价格**：{item['price']}元\n\n"f"**链接**：\n{item['link']}\n\n"f"**发现时间**：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"}}try:response = requests.post(DINGTALK_WEBHOOK,headers={"Content-Type": "application/json"},data=json.dumps(message))print("钉钉通知发送状态:", response.status_code)except Exception as e:print(f"钉钉消息发送失败: {e}")# ===== 主监控循环 =====
def main_monitor():print("=== 闲鱼商品监控程序启动 ===")seen_items = load_seen_items()while True:print(f"\n{datetime.now()} 开始新一轮监控...")for keyword in KEYWORDS:items = get_xianyu_items(keyword)print(f"关键词 [{keyword}] 获取到 {len(items)} 个商品")for item in items:item_id = generate_item_id(item)if item_id not in seen_items:print(f"发现新商品: {item['title']}")send_dingtalk_message(item, keyword)seen_items.add(item_id)save_seen_items(seen_items)time.sleep(1)  # 避免消息发送过快print(f"本轮监控完成，等待 {MONITOR_INTERVAL} 秒...")time.sleep(MONITOR_INTERVAL)if __name__ == "__main__":main_monitor()
```---### 功能描述#### 1. 监控功能
- **关键词监控**：可同时监控多个关键词（在`KEYWORDS`中配置）
- **定时扫描**：可设置监控间隔时间（`MONITOR_INTERVAL`）
- **新品识别**：通过MD5哈希生成商品唯一ID，避免重复通知
- **数据持久化**：使用JSON文件存储已发现商品记录#### 2. 钉钉推送功能
- **Markdown格式**：清晰展示商品信息
- **包含内容**：- 监控关键词- 商品标题- 商品价格- 商品链接- 发现时间
- **错误处理**：网络异常时捕获错误并打印日志#### 3. 技术实现特点
1. **反爬虫处理**：- 使用真实浏览器User-Agent- 设置Referer头- 请求间隔时间控制2. **稳定性设计**：- 异常捕获机制- 数据持久化存储- 网络请求超时设置3. **唯一性校验**：- 通过标题+链接生成MD5哈希ID- 避免同一商品重复通知---### 使用说明1. **配置钉钉机器人**：- 在钉钉群创建自定义机器人- 复制Webhook地址替换`DINGTALK_WEBHOOK`2. **自定义监控参数**：- 修改`KEYWORDS`为需要监控的商品关键词- 调整`MONITOR_INTERVAL`设置监控频率3. **运行环境**：```bashpip install requestspython xianyu_monitor.py```4. **实际使用建议**：- 添加代理支持（在requests中添加proxies参数）- 使用BeautifulSoup等库完善HTML解析- 部署到云服务器实现24小时监控- 添加日志记录功能>

以上是一个闲鱼商品监控脚本的Python实现，包含钉钉推送功能及详细功能描述

**注意**：闲鱼页面结构可能变化，需要定期更新解析逻辑。当前示例包含模拟数据获取逻辑，实际使用时需要根据闲鱼页面结构实现真实解析