企业官网建设流程全解析

简介：这是一套面向爬虫与自动化爱好者的闲鱼多任务实时监控与智能分析工具，基于 Playwright 与多模态大语言模型构建，适合具备一定 Python 基础、希望研究反爬策略与 AI 筛选逻辑的开发者学习使用。资源包共 39 个文件，约 12.31MB，涵盖 11 个 py 源码、5 个 txt 说明、3 个 html 页面、2 个 yml 与 dockerfile 等部署配置，以及 png、jpg 截图和 docx 教程文档，结构上分为爬虫抓取、AI 分析、Web 服务与前端模板等模块。已有 197 人浏览学习。读者可从中获取完整的任务调度与并发监控实现、基于自然语言创建监控任务的 Prompt 生成思路、多模态商品图文与卖家画像分析流程，以及 ntfy.sh、企业微信、Bark 等多渠道通知的接入方式，同时可参考 Docker 一键部署与随机延迟反爬策略，用于研究学习与二次开发。

1. 闲鱼智能监控机器人：从“蹲不到”到“秒推送”的落地拆解

做二手交易的人都有个共识：真正的好货不是搜出来的，是蹲出来的。但人不可能 24 小时盯着手机刷新，于是“闲鱼关键词监控”成了刚需。这套闲鱼智能监控机器人，本质上就是一个任务监控分析系统——你给它关键词、价格区间、发布时间等条件，它替你轮询闲鱼搜索接口，把命中的新商品推送到你面前。它解决的不是“怎么买”，而是“怎么第一时间知道有货上架”。适合谁？做无货源、做倒卖、做收藏捡漏的从业者，以及想把自己盯盘逻辑自动化的人。我拆过不少同类脚本，大部分死在风控和去重上，这套的工程结构相对完整，值得拿出来讲透。

2. 任务监控分析系统的骨架：轮询、解析、去重、推送四件事

2.1 为什么是“任务制”而不是“关键词制”

很多人写监控脚本，第一反应是写死一个关键词列表，循环请求。这种写法跑一天就崩，因为闲鱼的关键词搜索有翻页限制、有频控，而且不同关键词的监控频率需求完全不同——热门词可能 10 秒一轮，冷门词 5 分钟一轮就够。任务制的核心是把每个监控需求抽象成一条独立任务记录，包含关键词、价格上下限、排序方式、轮询间隔、启用状态。这样调度器可以按任务粒度控制节奏，而不是一刀切。

常见做法是用 SQLite 存任务表，字段大致如下：

CREATE TABLE monitor_task ( id INTEGER PRIMARY KEY AUTOINCREMENT, keyword TEXT NOT NULL, -- 监控关键词 min_price REAL DEFAULT 0, -- 最低价过滤 max_price REAL DEFAULT 999999, -- 最高价过滤 sort_type TEXT DEFAULT 'new', -- 排序：new/time_desc/price_asc interval_sec INTEGER DEFAULT 60, -- 轮询间隔（秒） enabled INTEGER DEFAULT 1, -- 是否启用 last_run_at INTEGER DEFAULT 0, -- 上次执行时间戳 created_at INTEGER DEFAULT 0 );

这张表是整个系统的调度依据。interval_sec决定了任务多久跑一次，last_run_at配合调度器判断“这个任务现在该不该跑”。我一般会把interval_sec最小值卡在 30 秒，再低就容易触发风控。sort_type用new按发布时间倒序，能最快拿到新上架的商品，这是捡漏场景的关键参数。

2.2 请求层：签名、Cookie 与频控的三角关系

闲鱼搜索接口不是裸奔的，请求里必须带sign参数，这个签名由t（时间戳）、appKey、data等字段拼接后经过特定算法生成。不同版本的 App 签名算法有差异，这套系统里通常会把签名逻辑单独抽成一个模块，方便后续替换。请求头里还需要Cookie，里面包含_m_h5_tk和_m_h5_tk_enc，这两个值有有效期，过期后接口会返回FAIL_SYS_TOKEN_EXOIRED。

import time import hashlib import requests def build_sign(token, t, app_key, data): """闲鱼 h5 接口签名：token + '&' + t + '&' + appKey + '&' + data 的 MD5""" raw = f"{token}&{t}&{app_key}&{data}" return hashlib.md5(raw.encode("utf-8")).hexdigest() def fetch_search(keyword, page=1, cookie=None): t = str(int(time.time() * 1000)) app_key = "34839810" # 常见 h5 appKey，以实际抓包为准 data = f'{{"keyword":"{keyword}","page":{page},"sortType":"new"}}' token = cookie.get("_m_h5_tk", "").split("_")[0] sign = build_sign(token, t, app_key, data) params = { "t": t, "sign": sign, "appKey": app_key, "data": data, } headers = { "Cookie": "; ".join(f"{k}={v}" for k, v in cookie.items()), "User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X)", } resp = requests.get("https://h5api.m.goofish.com/h5/mtop.taobao.idlemtopsearch.pc.search/1.0/", params=params, headers=headers, timeout=10) return resp.json()

这段代码里build_sign是签名核心，token取自 Cookie 中_m_h5_tk的下划线前半段。fetch_search把签名、时间戳、appKey、data 拼进 query 参数。注意data必须是紧凑 JSON 字符串，多一个空格签名就对不上。频控方面，我一般会在请求之间加time.sleep(random.uniform(1.5, 3.5))，并且单任务连续翻页不超过 3 页，否则很容易被限流。

2.3 解析层：从 JSON 到结构化商品

接口返回的 JSON 层级很深，商品列表在data.resultList里，每个元素又嵌套了data.item.main.exContent。解析时最容易翻车的地方是字段缺失——有的商品没有价格，有的没有发布时间，直接["price"]取值会抛 KeyError。稳妥做法是用.get()逐层兜底。

def parse_items(resp_json): items = [] result_list = resp_json.get("data", {}).get("resultList", []) for entry in result_list: main = entry.get("data", {}).get("item", {}).get("main", {}) ex = main.get("exContent", {}) item_id = ex.get("id") if not item_id: continue items.append({ "item_id": item_id, "title": ex.get("title", ""), "price": ex.get("price", [{}])[0].get("text", "0"), "area": ex.get("area", ""), "user_nick": ex.get("userNickName", ""), "pic": ex.get("picUrl", ""), "publish_time": ex.get("publishTime", ""), }) return items

price字段在原始 JSON 里是个列表，取第一个元素的text才是显示价格。publish_time有时是相对时间（“3 分钟前”），有时是绝对时间，后续去重和排序要统一处理。我一般会在入库前把相对时间转成时间戳，转换函数单独放一个工具模块。

2.4 去重与推送：别让同一条商品轰炸你三次

去重是监控系统的生命线。闲鱼搜索结果里同一商品可能出现在不同页，或者因为排序变化反复出现。最稳的方案是用item_id做唯一索引，入库时INSERT OR IGNORE，只有真正新插入的记录才触发推送。

import sqlite3 def save_and_filter(conn, items): new_items = [] for it in items: cur = conn.execute( "INSERT OR IGNORE INTO seen_item (item_id, title, price, created_at) VALUES (?,?,?,?)", (it["item_id"], it["title"], it["price"], int(time.time())) ) if cur.rowcount > 0: new_items.append(it) conn.commit() return new_items

cur.rowcount > 0表示这条记录之前没出现过，是新商品。推送渠道常见的是钉钉机器人、企业微信机器人、Bark、Server 酱。我一般用钉钉，因为支持 Markdown 卡片，能把标题、价格、地区、图片链接一次推全。推送频率也要控制，同一任务 1 分钟内最多推 5 条，避免刷屏。

3. 把机器人跑起来：部署、配置与调度器实操

3.1 环境准备与依赖安装

这套系统对运行环境要求不高，Python 3.8 以上即可，核心依赖就几个：requests负责请求，sqlite3是标准库不用装，schedule或apscheduler做调度，flask可选用来做 Web 管理面板。我一般用虚拟环境隔离，避免和系统 Python 打架。

python3 -m venv venv source venv/bin/activate pip install requests apscheduler flask

如果你打算长期跑在服务器上，建议用nohup或systemd托管。systemd的好处是崩溃自动重启，日志走journalctl方便排查。下面是一个最小化的 service 配置：

[Unit] Description=Xianyu Monitor Bot After=network.target [Service] Type=simple WorkingDirectory=/opt/xianyu-bot ExecStart=/opt/xianyu-bot/venv/bin/python main.py Restart=always RestartSec=10 [Install] WantedBy=multi-user.target

Restart=always保证进程挂了自动拉起来，RestartSec=10避免频繁重启打爆日志。这套配置我跑过几个月，稳定性没问题。

3.2 任务配置：关键词、价格区间与轮询间隔怎么设

任务配置直接决定监控效果。关键词太宽，推送量爆炸；太窄，蹲不到货。我的经验是：核心词 + 属性词组合，比如“索尼 WH-1000XM5”比“索尼耳机”精准得多。价格区间要参考近期成交价，下限设太低会混入配件，上限设太高会混入全新未拆封。轮询间隔按关键词热度分档：

配置入口一般有两种：直接改数据库，或者通过 Web 面板。我习惯先用 SQL 插一条任务测试，跑通了再上面板。

import sqlite3, time conn = sqlite3.connect("monitor.db") conn.execute( "INSERT INTO monitor_task (keyword, min_price, max_price, sort_type, interval_sec, enabled, created_at) " "VALUES (?,?,?,?,?,?,?)", ("索尼 WH-1000XM5", 800, 1800, "new", 60, 1, int(time.time())) ) conn.commit()

这条任务表示：监控“索尼 WH-1000XM5”，价格 800 到 1800 之间，按最新排序，每 60 秒跑一次。enabled=1表示立即生效。插完记得确认last_run_at初始为 0，调度器会认为它从没跑过，第一轮就会执行。

3.3 调度器：让每个任务按自己的节奏跑

调度器是整个系统的心脏。最简单的实现是主循环每秒扫一次任务表，找出enabled=1且now - last_run_at >= interval_sec的任务，丢进线程池执行。这样每个任务互不阻塞，一个任务请求慢了不影响其他任务。

import time, threading from concurrent.futures import ThreadPoolExecutor def scheduler_loop(conn, executor): while True: now = int(time.time()) rows = conn.execute( "SELECT id, keyword, min_price, max_price, interval_sec, last_run_at " "FROM monitor_task WHERE enabled=1" ).fetchall() for row in rows: task_id, kw, min_p, max_p, interval, last_run = row if now - last_run >= interval: executor.submit(run_task, task_id, kw, min_p, max_p) conn.execute("UPDATE monitor_task SET last_run_at=? WHERE id=?", (now, task_id)) conn.commit() time.sleep(1) executor = ThreadPoolExecutor(max_workers=5) threading.Thread(target=scheduler_loop, args=(conn, executor), daemon=True).start()

max_workers=5表示最多同时跑 5 个任务，太多容易触发风控。last_run_at在提交任务时就更新，而不是等任务跑完，这样避免任务执行时间过长导致重复提交。这个细节很多脚本没注意，结果同一个任务被反复触发，请求量翻倍。

3.4 推送模板：让消息一眼看清关键信息

推送内容要精简，标题、价格、地区、发布时间、链接，五样够了。钉钉 Markdown 模板大概长这样：

def build_dingtalk_msg(item): return { "msgtype": "markdown", "markdown": { "title": "闲鱼新货", "text": ( f"### 闲鱼新货提醒\n" f"- **标题**：{item['title']}\n" f"- **价格**：{item['price']}\n" f"- **地区**：{item['area']}\n" f"- **发布**：{item['publish_time']}\n" f"- [点击查看](https://www.goofish.com/item?id={item['item_id']})" ) } }

item_id拼进链接就能直接跳转商品页。注意钉钉机器人有频率限制，每分钟最多 20 条，超过会被限流。我一般会在推送层加一个令牌桶，控制发送速率。

4. 避坑与排查：那些让我半夜爬起来改代码的瞬间

4.1 现象：接口返回FAIL_SYS_TOKEN_EXOIRED，任务全部失效

原因：Cookie 里的_m_h5_tk过期了，通常有效期几小时到一天不等。解决：在请求层捕获这个错误码，自动重新获取 Cookie。获取方式可以是手动抓包更新，也可以用一个独立的浏览器自动化模块定时刷新。我一般会写一个refresh_cookie()函数，检测到 token 过期就调用，并把新 Cookie 写回配置文件。

4.2 现象：同一商品被推送了七八次

原因：去重表只存了item_id，但闲鱼同一商品可能因为重新编辑而生成新的item_id，或者标题微调后被当成新商品。解决：去重时加一层标题相似度判断，用difflib.SequenceMatcher对比标题，相似度超过 0.9 视为同一商品，跳过推送。

4.3 现象：跑了一晚上，一条推送都没有

原因：调度器线程挂了，或者last_run_at更新逻辑有 bug，导致任务永远不满足执行条件。解决：加一个心跳日志，每 5 分钟打印一次“调度器存活，当前任务数 X，待执行 Y”。另外last_run_at更新后要commit，否则下次查询还是旧值。

4.4 现象：请求频繁返回 429 或空列表

原因：触发风控了。可能是轮询间隔太短、单 IP 请求量太大、或者 User-Agent 太单一。解决：加大间隔、随机化 User-Agent、控制并发数不超过 3。如果还不行，考虑加一层请求队列，把请求均匀分散到时间轴上。

4.5 现象：价格解析出来是“￥”开头，没法比较

原因：price字段带货币符号，直接转 float 会报错。解决：解析时用正则去掉非数字字符，re.sub(r"[^\d.]", "", price_text)，再转 float。这个坑很隐蔽，因为不报错，只是过滤逻辑失效，导致高价商品也被推过来。

5. 进阶技巧：用历史数据反推最佳监控策略

跑了一段时间后，数据库里会积累大量商品记录。这些数据不只是日志，还能用来优化监控策略。我一般会做两件事：一是统计每个关键词的“上架频率”和“成交速度”，二是根据统计结果动态调整轮询间隔。

import sqlite3 from collections import Counter conn = sqlite3.connect("monitor.db") rows = conn.execute( "SELECT keyword, COUNT(*) as cnt FROM seen_item " "WHERE created_at > ? GROUP BY keyword", (int(time.time()) - 86400 * 7,) ).fetchall() for kw, cnt in rows: avg_per_hour = cnt / (24 * 7) if avg_per_hour > 10: print(f"{kw}: 高频，建议间隔 30 秒") elif avg_per_hour > 2: print(f"{kw}: 中频，建议间隔 60 秒") else: print(f"{kw}: 低频，建议间隔 300 秒")

这段代码统计过去 7 天每个关键词的新增商品数，换算成每小时均值，据此推荐轮询间隔。高频词说明竞争激烈，必须快；低频词说明上架少，慢一点省资源。我还会进一步统计“从推送到被拍下”的时间差，但这个需要额外记录商品状态变化，实现成本高一些。

另一个技巧是关键词扩展。跑一段时间后，你会发现某些商品标题里反复出现你没监控到的词，比如“日版”“限定”“未拆”。把这些词加进监控任务，能覆盖更多长尾货源。我一般每周花 10 分钟做一次关键词复盘，把推送记录里标题的高频词提取出来，补充到任务表里。

提示：动态调整间隔时，不要一次性把所有任务都改成 30 秒，先挑一两个高频词试跑一天，确认风控没反应再推广。

从那以后我每次部署新监控任务，都会先跑一轮“只记录不推送”的观察模式，确认解析和去重逻辑没问题，再打开推送开关。这个习惯帮我省了很多半夜被无效推送吵醒的麻烦。希望帮到你。

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