ReaderMX.
Python技术迷

Python实现闲鱼商品监控，ai goofish monitor

Ppython约 7 分钟

那天半夜我在沙发上刷闲鱼，你们懂的，本来说就随便看看，结果看着看着，一张 3060 的卡，成色新得发亮，价格还贼香，我手一抖准备点进去，刷新一下——没了。 就那种“本商品已下架或被神仙买走”的感觉，我整个人都不好了。

我当时心里就一个念头：这不行，这太被动了，得搞个程序给我盯着，不然总被别人截胡，老脸往哪放对吧。

后来我就整了个小玩意，名字想了半天，最后叫：ai goofish monitor。听着特高级，其实就是“Python + 一点点AI + 土方法防重复”的闲鱼商品监控脚本，说白了，就是帮你 7×24 小时刷闲鱼那种打工人机器人。

我慢慢跟你聊，你别着急，代码一点点来。

一开始我也没想那么复杂，就先把需求念叨一遍：

  • 监控某几个关键词，比如“3060 显卡”“switch 红蓝机”
  • 限定一个大概价格区间，太贵直接忽略
  • 新上架的，立刻提醒我（比如发到企业微信 / 钉钉机器人那种）
  • 最好用个小“AI”过滤一下垃圾信息，比如那种“图文不符”“钓鱼文案”少推点

于是我随手就先写了个配置，先把东西写死，后面再慢慢改成读配置文件的，那会儿已经困得不行了：

# config.py
KEYWORDS = ["3060", "显卡", "自用", "盒装"]
MIN_PRICE = 500
MAX_PRICE = 1800
POLL_INTERVAL = 30# 秒，轮询间隔，别太小，容易被当成爬虫打
HISTORY_FILE = "seen_items.json"
WEBHOOK = "YOUR_NOTIFY_WEBHOOK"# 这里放你自己的机器人地址，自己保密就行

是不是很简陋？没事，真正在生产环境里（啊不，在薅羊毛环境里）跑起来的东西，往往就是这么一坨配置先糊上去，后面再慢慢优雅。

接下来最关键的问题来了：怎么把闲鱼的列表扒下来。 这个地方我先提醒一句啊，你们要记得尊重人家平台规则，别搞那种一天几万次请求的，真的容易被封，而且也不地道，咱就小打小闹。

我当时的思路就一个： 手机端抓一下请求，看一眼搜索接口长啥样，请求头里把关键的东西（比如 token、cookie 那些）抄下来，然后在 Python 里伪装一下。 具体抓包怎么搞我就不细说了，容易被说成教唆犯罪（笑）。

于是核心拉取的函数我就写成这样：

# fetcher.py
import requests
deffetch_items(keyword: str):
    headers = {
"User-Agent": "GoofishMonitor/1.0",
# 这里正常要带上 cookie、token 之类的，不过别写死在代码里
# "Cookie": "your_cookie_here"
    }
    params = {
"q": keyword,
"page": 1,
"page_size": 20,
# 还有一堆乱七八糟的参数，抓包里有啥就带点关键的
    }
# 这里本来要写具体地址的，为了不惹事，留个占位就行
    url = "GOOFISH_SEARCH_API"# 自己抓包换成真实的
    resp = requests.get(url, headers=headers, params=params, timeout=5)
    resp.raise_for_status()
    data = resp.json()
# 假设 data 里面有个 list 叫 items，字段你自己按返回结构撸
    items = []
for raw in data.get("items", []):
        items.append({
"id": raw.get("itemId"),
"title": raw.get("title", ""),
"price": float(raw.get("price", 0)),
"url": raw.get("detailUrl", ""),  # 用来点进去看的
"desc": raw.get("desc", ""),
        })
return items

你别纠结这玩意能不能直接跑，抓包那部分每个人的结果都不太一样，我这边主要是跟你说下思路：手机 / H5 抓一抓，照着请求结构抄一遍，用 requests 伪装一下就行。

有了原始商品列表，还得过滤一遍，不然一大堆“只出盒子”“收机”“10 块挂着当广告用”的，也给你推过来，那你手机消息基本炸了。

我一开始写过滤逻辑非常原始，就是 if-else 地狱：

# filter.py
from config import KEYWORDS, MIN_PRICE, MAX_PRICE
defbasic_filter(item: dict) -> bool:
    price = item["price"]
    title = item["title"]
    desc = item.get("desc", "")
if price <= 0:
returnFalse
if price < MIN_PRICE or price > MAX_PRICE:
returnFalse
    text = (title + " " + desc).lower()
    hit = any(k.lower() in text for k in KEYWORDS)
ifnot hit:
returnFalse
# 一些简单黑名单词，防止钓鱼
    black_words = ["图片仅供参考", "帮忙点个赞", "勿拍", "租", "代充"]
if any(w in text for w in black_words):
returnFalse
returnTrue

就这个土办法，先跑起来其实已经能挡掉一大半垃圾了。 很多时候“AI”这俩字，其实可以先等会儿，你先把 if 写扎实再说，真的。

那“ai goofish monitor”里的 AI 部分呢？ 我跟你说实话，一开始我写的 AI 就是——一个函数名叫 ai_score，但里面只写了个长度判断，哈哈。

后面我稍微认真一点，搞了个“评分”的概念：

  • 标题里关键词越多，分越高
  • 越接近我们目标价格，分越高
  • 标题里出现“自用”“急出”“闲置”“盒说全”，适当加分
  • 出现“无售后”“坏的”“不包”“只要盒子”，扣分

这种东西你完全可以自己随手改，我举个例子你感受下：

# ai.py
defai_score(item: dict) -> float:
    title = item["title"]
    price = item["price"]
    desc = item.get("desc", "")
    text = (title + " " + desc)
    score = 0.0
# 关键词命中加分
for k in ["显卡", "3060", "自用"]:
if k in text:
            score += 2
# 价格越接近目标区间中间，分越高
    target_mid = (800 + 1500) / 2# 这个你可以改成配置
    diff = abs(price - target_mid)
if diff < 100:
        score += 3
elif diff < 300:
        score += 1
# 文案情绪一点点 heuristic
if"急出"in text or"换平台"in text:
        score += 1.5
if"不包邮"in text or"有问题"in text:
        score -= 2
return score

你看，这就“AI”了，起码听着像。 要真想上大模型，你把 ai_score 里面换成“丢给一个问答模型，让它判断这条是不是值得看”，再给个 0~10 的打分，也是完全可行的，注意下调用的 QPS 就行，别把你钱包烧没了。

监控这种事情有个关键点：去重。 不然你每 30 秒拉一次，一条商品从挂到卖掉，能给你推个几百次，人直接烦死。

我那会图省事，就搞了个本地 JSON 存已经看过的 id，重启脚本也不丢。 正常人肯定会上 Redis 啊、SQLite 啊这种更靠谱的东西，不过我懒。

# storage.py
import json
from pathlib import Path
from typing import Set
from config import HISTORY_FILE
defload_history() -> Set[str]:
    path = Path(HISTORY_FILE)
ifnot path.exists():
return set()
try:
        data = json.loads(path.read_text(encoding="utf-8"))
return set(data)
except Exception:
return set()
defsave_history(ids: Set[str]):
    path = Path(HISTORY_FILE)
    path.write_text(json.dumps(list(ids)), encoding="utf-8")

然后主逻辑那边就这么写：

# monitor.py
import time
from fetcher import fetch_items
from filter import basic_filter
from ai import ai_score
from storage import load_history, save_history
from notifier import notify  # 下面会写
defrun_monitor(keyword: str):
    seen = load_history()
whileTrue:
try:
            items = fetch_items(keyword)
            new_ids = set()
for item in items:
if item["id"] in seen:
continue
ifnot basic_filter(item):
continue
                score = ai_score(item)
if score < 3:  # 打个阈值，太低的就不吵醒我了
continue
                notify(item, score)
                new_ids.add(item["id"])
if new_ids:
                seen |= new_ids
                save_history(seen)
except Exception as e:
            print("监控异常：", e)
        time.sleep(30)

你看逻辑就很朴素：拉 -> 过滤 -> 打分 -> 通知 -> 记住看过的。

说到通知，这部分就比较随缘了， 有的人爱用邮箱，有的人爱用企业微信，有的人直接跑在家里树莓派上，往电视上弹窗（这个有点离谱但是真有人这么干）。

我自己是偷懒，用了最原始的“Webhook + 文本消息”的方式：

# notifier.py
import requests
from config import WEBHOOK
defnotify(item: dict, score: float):
    text = f"""发现新宝贝：
标题：{item["title"]}
价格：{item["price"]}
评分：{score:.1f}
链接：{item.get("url", "自己去应用里搜")}
"""
# 具体机器人格式每家不一样，这里随便举个 JSON 例子
    payload = {
"msgtype": "text",
"text": {
"content": text
        }
    }
# 这里别忘了给 webhook 加点简单的错误处理，避免一直报错
try:
        requests.post(WEBHOOK, json=payload, timeout=3)
except Exception as e:
        print("发送通知失败：", e)

你要是怕频繁打扰自己，还可以加个“静音时间段”，比如凌晨两点发现好东西，就先记一笔，早上八点统一推一条汇总。 这玩意我本来想写的，结果写着写着去刷了一会儿视频，就…忘了，至今还是一条一条推，手机上全是“发现新宝贝”。

这东西跑了几天之后，我最大的感受是：程序真的是很老实的打工人。 你给它一个接口，它就傻乎乎按时去问，你让它打分，它也不抱怨“不想上班”，只要别把它跑挂了，它就一直给你盯着。

当然中间也踩了些坑，比如：

  • 某天 cookie 失效了，一直报 401，我还以为是代码写挂了
  • 有一次被风控拦了一下，返回一堆奇怪的 HTML，让我以为接口改版了
  • JSON 结构偶尔变一下，某个字段没了，脚本直接 KeyError 崩掉，后来我都 .get() 了

这些小细节你真要上生产环境（比如公司里监控价格、监控库存那种），就得好好做健壮性，我这玩意纯属“薅羊毛脚本”，就不要跟我学工程质量了。

反正大概就是这么个路子：抓个接口，Python 拉数据，if-else 过滤，稍微装点“AI”的样子，再加个小通知渠道，就算是一个能用的 ai goofish monitor 了。

行了我先不说了，手机又震了下，看样子又有人上架“九成新自用显卡”，我得先看看是不是又晚了一步…

KEEP THE THREAD GOING

更多来源文章