AI-Goofish-Monitor：基于 Playwright 与多模态模型的闲鱼商品监控工具

# 操作系统：Windows/Linux/macOS
# Python: 3.10+ (推荐 3.10 或更高版本)
# 内存：4GB+ RAM
# 存储：1GB+ 可用空间
# 网络：稳定互联网连接

# 生产环境推荐
# CPU: 4 核+
# 内存：8GB+
# 存储：SSD, 10GB+ 可用空间
# 网络：100Mbps+ 带宽

浏览器方面推荐使用 Chrome/Chromium 最新版本，Playwright 会自动管理浏览器驱动。

安装方式

源码安装：

# 克隆仓库
git clone https://github.com/dingyufei615/ai-goofish-monitor.git
cd ai-goofish-monitor

# 安装依赖
pip install -r requirements.txt

# 安装 Playwright 浏览器
playwright install chromium

Docker 安装（推荐）：

# 使用 docker-compose
docker-compose up --build -d

# 或者直接使用 Docker
docker build -t ai-goofish-monitor .
docker run -d -p 8000:8000 --name monitor ai-goofish-monitor

环境配置

创建环境变量文件 .env：

# AI 服务配置
OPENAI_API_KEY=your_api_key_here
OPENAI_BASE_URL=https://api.example.com/v1
OPENAI_MODEL_NAME=gemini-2.5-pro

# 代理配置（可选）
PROXY_URL=http://127.0.0.1:7890

# 通知配置
NTFY_TOPIC_URL=https://ntfy.sh/your_topic
BARK_URL=https://api.day.app/your_key
WX_BOT_URL="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=your_key"

# Web 界面认证
WEB_USERNAME=admin
WEB_PASSWORD=your_strong_password_here

# 运行配置
RUN_HEADLESS=true
AI_DEBUG_MODE=false
SERVER_PORT=8000

配置文件 config.json 示例：

{
  "tasks": [
    {
      "name": "索尼相机监控",
      "keywords": ["索尼 A7M4", "Sony 相机"],
      "price_range": [5000, 13000],
      "cron_expression": "0 */2 * * *",
      "ai_prompt": "寻找 95 新以上，快门数低于 5000 的索尼相机"
    }
  ],
  "notification": {
    "enabled": true,
    "services": ["ntfy", "wechat"],
    "min_confidence": 0.8
  }
}

登录配置

获取闲鱼登录状态有两种主要方式：

1. 使用登录脚本：需要图形界面支持

   python login.py
   

2. 通过 Web 界面更新（推荐）：启动服务后访问 http://localhost:8000，在系统设置中手动更新登录状态。

若需通过 Chrome 扩展提取状态，请确保已加载相关扩展程序并正确复制状态信息到 Web 界面。

使用指南

启动服务

# 启动 Web 服务
python web_server.py

# 或者使用 gunicorn（生产环境）
gunicorn -w 4 -b 0.0.0.0:8000 web_server:app

访问管理界面 http://localhost:8000，使用配置的 WEB_USERNAME 和 WEB_PASSWORD 登录。

任务管理

进入'任务管理'页面，点击'创建新任务'。系统会根据你的描述自动生成监控参数和 AI 分析标准。

你可以直接用自然语言描述需求，例如：

'我想买一台 95 新以上的索尼 A7M4 相机，预算 1 万 3 以内，快门数低于 5000'

任务配置选项结构如下：

{
  "name": "任务名称",
  "keywords": ["关键词 1", "关键词 2"],
  "price_range": [1000, 5000],
  "location": "上海",
  "cron_expression": "0 */1 * * *",
  "ai_prompt": "自定义 AI 分析指令",
  "enabled": true
}

AI 分析配置

支持自定义 Prompt 模板来调整分析逻辑：

# prompts/analysis_template.txt
你是一个专业的二手商品评估专家。请分析以下闲鱼商品信息：
商品标题：{{title}}
商品价格：{{price}}元
商品描述：{{description}}
卖家信息：{{seller_info}}
图片信息：{{image_descriptions}}

请根据以下标准进行评估：
1. 商品成色是否达到 95 新以上
2. 价格是否在合理范围内
3. 卖家信用是否良好
4. 是否存在潜在风险

请输出 JSON 格式的分析结果：
{
  "recommendation": "推荐/不推荐",
  "confidence": 0.95,
  "reasons": ["原因 1", "原因 2"],
  "risk_factors": ["风险因素 1", "风险因素 2"]
}

监控流程

通知配置

支持多平台通知设置，包括 ntfy、企业微信、Bark 等：

notification_config = {
    "ntfy": { "enabled": True, "topic": "your_ntfy_topic", "priority": "high" },
    "wechat": { "enabled": True, "webhook": "企业微信群机器人 Webhook", "mention_users": ["@all"] },
    "bark": { "enabled": True, "device_key": "your_bark_key", "sound": "alarm" },
    "webhook": { "enabled": False, "url": "https://example.com/webhook", "method": "POST" }
}

应用场景实例

案例 1：二手相机监控

场景：摄影爱好者寻找特定型号的二手相机。

解决方案：

class CameraMonitor:
    def __init__(self):
        self.config = {
            "name": "索尼相机专业监控",
            "keywords": ["索尼 A7M4", "Sony A7M4", "A7M4"],
            "price_range": [8000, 13000],
            "conditions": {
                "min_condition": "95 新",
                "max_shutter_count": 5000,
                "require_original_box": True,
                "exclude_professional_sellers": True
            },
            "schedule": "0 */3 * * *",
            "notification_channels": ["wechat", "ntfy"]
        }

    def create_monitor_task(self):
        # 寻找 95 新以上的索尼 A7M4 相机，要求快门数低于 5000，有原装包装盒
        # 个人卖家，非专业商家，价格在 8000-13000 元之间，上海地区优先
        response = requests.post(
            "http://localhost:8000/api/tasks/create",
            json={"description": "natural_language", "config": self.config},
            auth=("admin", "password123")
        )
        return response.json()

    def handle_notification(self, item):
        message = f"""📸 发现符合条件的相机！
型号：{item['title']}
价格：{item['price']}元
成色：{item['condition']}
快门数：{item['shutter_count']}
链接：{item['url']}
AI 推荐度：{item['confidence']}%"""
        self.send_notification(message)

# 使用示例
monitor = CameraMonitor()
task_id = monitor.create_monitor_task()
print(f"监控任务已创建：{task_id}")

案例 2：限量版球鞋监控

场景：球鞋收藏家寻找限量版球鞋。

解决方案：

class SneakerMonitor:
    def __init__(self):
        self.sneaker_models = ["Nike Dunk SB", "Air Jordan 1", "Yeezy 350", "限量版球鞋", "联名款球鞋"]

    def setup_monitoring(self):
        tasks = []
        for model in self.sneaker_models:
            task_config = {
                "name": f"{model}监控",
                "keywords": [model, "限量", "联名"],
                "price_range": [500, 3000],
                "filters": { "exclude_fake": True, "min_authenticity_score": 0.9 },
                "ai_prompt": "分析球鞋真伪和收藏价值，检查图片是否实拍图，评估价格是否合理",
                "schedule": "*/30 * * * *"
            }
            tasks.append(task_config)
        return tasks

    def is_worth_buying(self, item):
        criteria = [
            item['authenticity_score'] > 0.95,
            item['price'] <= item['market_value'] * 0.8,
            item['seller_rating'] >= 4.8,
            item['condition'] in ['全新', '99 新']
        ]
        return all(criteria)

# 使用示例
sneaker_bot = SneakerMonitor()
tasks = sneaker_bot.setup_monitoring()
print(f"创建了 {len(tasks)} 个球鞋监控任务")

案例 3：电子产品降价监控

场景：等待特定电子产品降价时购买。

解决方案：

class PriceDropMonitor:
    def __init__(self, target_product, target_price):
        self.target_product = target_product
        self.target_price = target_price
        self.price_history = {}

    def monitor_price_changes(self):
        config = {
            "name": f"{self.target_product}降价监控",
            "keywords": [self.target_product],
            "price_range": [0, self.target_price * 1.2],
            "ai_prompt": "重点关注价格变化趋势和降价时机，识别真实降价，判断是否达到目标价位",
            "schedule": "0 */1 * * *"
        }
        return config

    def analyze_price_trend(self, items):
        for item in items:
            product_id = item['id']
            current_price = item['price']
            if product_id not in self.price_history:
                self.price_history[product_id] = []
            self.price_history[product_id].append({'price': current_price, 'timestamp': datetime.now()})
            
            if current_price <= self.target_price:
                self.send_price_alert(item)

    def send_price_alert(self, item):
        message = f"""🎉 价格达标提醒！
商品：{item['title']}
当前价格：{item['price']}元
目标价格：{self.target_price}元
链接：{item['url']}⏰ 尽快购买，可能很快售罄！"""
        self.send_notification(message)

# 使用示例
monitor = PriceDropMonitor("iPad Pro 2022", 4000)
config = monitor.monitor_price_changes()
print("降价监控已设置")

高级功能与定制

自定义 AI 分析

可以编写更复杂的分析模板，针对特定品类进行深度评估：

你是一个专业的{{item_category}}评估专家。请分析以下闲鱼商品：
商品信息：
- 标题：{{title}}
- 价格：{{price}}元
- 描述：{{description}}
- 卖家：{{seller_name}} (信用：{{seller_rating}})
- 图片：{{image_count}}张

评估要求：{{specific_requirements}}

请输出 JSON 格式的分析结果：
{
  "assessment": { "quality_rating": 0.95, "overall_recommendation": "推荐" },
  "details": { "strengths": ["优点 1"], "concerns": ["需要注意的点"] },
  "pricing_advice": { "fair_price_range": [1000, 1200] }
}

智能通知路由

根据置信度自动选择通知渠道：

class SmartNotifier:
    def __init__(self):
        self.notification_rules = {
            "urgent": { "channels": ["wechat", "bark", "sms"], "priority": "high" },
            "normal": { "channels": ["ntfy", "wechat"], "priority": "normal" },
            "low_priority": { "channels": ["ntfy"], "priority": "low" }
        }

    def route_notification(self, item, confidence):
        if confidence > 0.95:
            return self.notification_rules["urgent"]
        elif confidence > 0.8:
            return self.notification_rules["normal"]
        else:
            return self.notification_rules["low_priority"]

数据导出与分析

支持将监控数据导出为 CSV 或生成分析报告：

class DataExporter:
    def __init__(self, db_path="monitoring_data.db"):
        self.conn = sqlite3.connect(db_path)
        self.setup_database()

    def export_to_csv(self, output_file="monitoring_report.csv"):
        query = "SELECT i.title, i.price, i.seller_info, ... FROM items i ..."
        df = pd.read_sql_query(query, self.conn)
        df.to_csv(output_file, index=False, encoding='utf-8-sig')
        return output_file

故障排除与优化

常见问题解决

登录问题：

# 检查登录状态
curl -X GET http://localhost:8000/api/system/status
# 重新登录
python login.py --force

AI 分析失败：

def check_ai_service():
    import requests
    test_payload = { "model": os.getenv('OPENAI_MODEL_NAME'), "messages": [{"role": "user", "content": "test"}] }
    try:
        response = requests.post(
            f"{os.getenv('OPENAI_BASE_URL')}/chat/completions",
            headers={"Authorization": f"Bearer {os.getenv('OPENAI_API_KEY')}"},
            json=test_payload, timeout=10
        )
        return response.status_code == 200
    except Exception as e:
        print(f"AI 服务检查失败：{e}")
        return False

性能优化：

PERFORMANCE_OPTIMIZATION = {
    "browser": { "headless": True, "timeout": 30000, "concurrent_tabs": 3 },
    "ai": { "batch_size": 5, "timeout": 30000 },
    "network": { "proxy_rotation": True, "request_delay": [1000, 3000] }
}

日志与监控

配置日志记录以便追踪运行状态：

import logging
from logging.handlers import RotatingFileHandler

def setup_logging():
    logger = logging.getLogger("goofish_monitor")
    logger.setLevel(logging.INFO)
    file_handler = RotatingFileHandler("monitor.log", maxBytes=10 * 1024 * 1024, backupCount=5)
    console_handler = logging.StreamHandler()
    logger.addHandler(file_handler)
    logger.addHandler(console_handler)
    return logger

技术支持

如遇问题，请参考官方文档或提交 Issue。

• 项目文档：https://github.com/dingyufei615/ai-goofish-monitor
• 问题反馈：https://github.com/dingyufei615/ai-goofish-monitor/issues

免责声明

⚠️ 重要提示：

• 请遵守闲鱼用户协议和 robots.txt 规则
• 不要进行过于频繁的请求，避免对服务器造成负担
• 本项目仅供学习和技术研究使用
• 请勿用于非法用途或商业滥用

本项目采用 MIT 许可证发布，按'现状'提供，不提供任何形式的担保。