简介面向日本二手交易平台Mercari上的抢购与捡漏需求这份以Python编写的工具能够定时扫描指定关键词或价格区间的新上架商品一旦发现符合条件的目标就会通过Slack机器人向用户发送告警省去反复手动刷新的时间。使用门槛不高先按模板修改config.ini文件填入Slack API令牌和频道名称再以命令行参数指定最低价、最高价和搜索词即可运行监控任务适合有一定Python基础、希望搭建个人自动化提醒工具的买家、代购或二手交易爱好者。该压缩包约13.71MB共2000个文件类型上以Python脚本为主并混合了C头文件、文本说明、安装元数据与wheel包、动态链接库等基本涵盖项目运行所需的第三方依赖与扩展模块解压后即可查看源码、调整配置并二次开发。目前已有509人浏览学习。通过阅读源码和实际部署可以学习Mercari商品信息的请求与解析思路、Slack消息推送的完整接入方法以及命令行程序的组织方式项目还附带相关依赖库便于直接运行和调试也可在此基础上扩展搜索语法、告警规则或对接其他二手平台。1. Mercari 新物品扫描到底在解决什么事Mercari 上的好货上架之后经常按秒被拍走热门镜头、限定球鞋、绝版游戏卡带发布十几分钟就没了手动刷新搜索页根本拼不过蹲守的老手。mercari-scanner 的思路就是把这件事程序化——固定间隔去拉搜索页数据找出之前没见过的商品 ID然后推一条警报到你手机上的聊天工具或邮箱。它适合盯几个具体关键词的买家也适合个人卖家监控对手上架节奏但不适合几十个关键词以上的规模化抓取。下面按抓取、增量判断、送警报、踩坑的顺序把一套能跑的脚本拆开讲清楚。2. 先搞懂搜索页返回了什么URL、数据形态与解析路线2.1 搜索 URL 是入口但排序参数直接决定你会不会漏开头先把最容易犯的错放在前面搜索页默认排序往往不是最新上架。Mercari 这类二手平台为了让买家更容易成交默认推荐排序会混合“热度”和“接近下架”的因素你盯着推荐页刷新最新上架的商品可能根本不在第一屏。做新物品扫描第一件事就是在浏览器里打开目标关键词的搜索页把排序切成“最新上架/新着順”这类选项再把地址栏里的完整 URL 复制出来作为抓取入口。不要自己凭记忆拼 URL直接把筛选条件都选好再复制地址栏keyword、排序、类目这些参数一次带齐。排序方式页面里的表现对扫描新物品的意义推荐排序混合热度与成交可能最新上架可能不在第一屏容易漏最新上架按发布时间倒序新物品稳定出现在最前面适合扫描价格排序按价格升/降序不适合盯新物品除非同时监控价格区间排序确认之后抓取这一步就简单了。搜索页是服务端渲染加前端混合HTML 里商品详情链接都有 /item/ 这个路径特征抓住这个特征比依赖某一层 class 名更抗改版。下面这个函数把页面拉回来并用正则把所有商品 ID 抠出来后面所有解析都从这个函数出发改版时只需要调整这一处。import re import requests HEADERS { User-Agent: ( Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36 ), Accept-Language: ja,en;q0.9, Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8, } def fetch_html(url): resp requests.get(url, headersHEADERS, timeout15) resp.raise_for_status() return resp.text def fetch_item_ids(search_url): html fetch_html(search_url) ids re.findall(r/item/([A-Za-z0-9_]), html) # 同一商品在一个页面上可能出现多次保持顺序去重 return list(dict.fromkeys(ids))这段代码做了三件事带基础浏览器请求头发起 GET用 raise_for_status 把非 200 状态直接抛异常避免把错误页当正常结果从 HTML 里把所有 /item/ 后面的 ID 用正则抓出来并去重。timeout 必须给默认不设的话网络挂起时脚本会一直卡着15 秒根据你到目标站点的实际延迟调整慢网络放宽到 30。User-Agent 不要写成 python-requests 的默认值那是反爬系统最容易识别的一类特征。需要注意的是正则提取只拿到 ID标题和价格还需要从卡片结构里再抠这一步放到 2.3 处理。2.2 用 DevTools 确认数据藏在哪5 分钟定位商品标题、价格和图片很多人抓取失败后喜欢对着 HTML 盲猜 class 名我的建议是先别猜用浏览器开发工具把数据的实际位置确认清楚。打开搜索页按 F12 切到 Network 面板勾选 Fetch/XHR 后刷新页面按下 CtrlF 在返回体里搜你看到的某个商品标题或价格。实际会遇到两种情况一是搜索页 HTML 本身就带了完整的商品数据二是页面先返回框架商品数据由后续一个 XHR 接口返回。不管哪种你要找的是“包含完整商品字段的那份响应”而不是花里胡哨的 DOM 层。找到响应后做三件事把返回体存成一份 JSON 样例文件用编辑器搜索 title、price、item 这类字段确认嵌套路径把这个路径写成解析函数并保留样例文件方便日后回归测试。因为站点改版是常态把“抓取”和“解析”拆成两个函数改版时只动解析那一侧主循环完全不用碰。这个习惯比每次上线前手工抓页面核对省事得多。2.3 从 HTML 里把标题和价格一起抠出来的基础版在还没有拿到 XHR 返回结构之前可以先做一个基于 HTML 卡片文本的粗糙版本商品详情链接的文本里通常包含标题和价格取整段文本后把价格单独剥出来剩下的当标题用。这个版本不完美但能让你先把整条链路跑通后面再替换解析逻辑。from bs4 import BeautifulSoup def extract_cards(html): soup BeautifulSoup(html, html.parser) cards [] for a in soup.select(a[href*/item/]): item_id a.get(href).rsplit(/item/, 1)[-1].strip(/) text a.get_text( , stripTrue) price_match re.search(r(¥||\$)\s?([\d,]), text) cards.append({ id: item_id, raw_text: text, price: price_match.group(0) if price_match else None, }) return cards这里用 BeautifulSoup 的 select 匹配所有指向 /item/ 的链接再对每个链接取整段文本。价格用正则匹配币种开头后跟数字的模式¥//$ 三种符号都照顾到方便不同地区版本的站点通用。raw_text 保留原始文本是因为它往往还带着“新品”“免运费”这类标识后面做关键词过滤还有用。这个版本有个已知局限如果卡片文本混进了卖家昵称等多余信息raw_text 当标题会显得脏。解决办法是后面拿到 XHR JSON 后用其中的 title 字段替换代码结构上只要把 extract_cards 换掉即可外层不用改。先打印确认字段再往下接逻辑别急着一步到位。search_url https://www.mercari.com/search/... # 从2.1里复制出来的完整URL html fetch_html(search_url) items extract_cards(html) for it in items[:5]: print(it[id], it[price], it[raw_text][:30])提示上面这步只是确认解析真正跑起来前先确认你复制的 URL 确实带了“最新”排序否则你蹲到的可能只是推荐结果里翻来覆去那几件旧商品。3. 增量判定与扫描主循环只有新 ID 才值得推一次3.1 为什么必须维护一份已见 ID 清单搜索页每次返回的都是“当前所有匹配商品”里面绝大部分是历史商品。如果把整个列表每次都推一遍你的警报会淹没在旧消息里反过来如果只对比上一次的列表脚本重启一次就又全忘了。正确做法是维护一份持久化的已见 ID 清单新 ID 的定义是“不在清单里”。用 SQLite 存放最合适单文件、无独立服务、读写足够快比反复 load/dump JSON 文件更不容易写坏。import sqlite3, time def init_db(pathseen.db): conn sqlite3.connect(path) conn.execute( CREATE TABLE IF NOT EXISTS seen ( item_id TEXT PRIMARY KEY, status TEXT DEFAULT pending, first_seen_at INTEGER ) ) conn.commit() return conn def is_pending(conn, item_id): row conn.execute(SELECT status FROM seen WHERE item_id?, (item_id,)).fetchone() return row is None or row[0] failed表设计里 item_id 是主键天然去重status 字段预留了 pending、sent、failed 三种状态后面接入发送逻辑后发送失败的记录会被重新捞出来这个设计直接避免“推丢了就再也补不回来”的问题。first_seen_at 用 Unix 时间戳方便后面做按天统计。is_pending 的判断同时覆盖查无记录和上次发送失败两种情况这两件事其实代表同一个结论这条警报还没有被成功送达。3.2 扫描主循环间隔、抖动和异常兜底有了持久化存储主循环就清楚了拉取最新 ID 列表逐个判断是否 pending是则发送警报发送成功后把状态置为 sent。循环里的间隔是整个工具频率策略的核心我习惯用一个基值加随机抖动比如目标间隔 60 秒实际 sleep 在 55 到 75 秒之间随机这样不会每个周期都在同一秒打到对方服务上请求曲线的整体形状也更平滑。import random, time, logging logging.basicConfig(levellogging.INFO, format%(asctime)s %(levelname)s %(message)s) def scan_once(conn, search_url, sender): items extract_cards(fetch_html(search_url)) pending [it for it in items if is_pending(conn, it[id])] logging.info(page returned %d cards, %d pending, len(items), len(pending)) for it in pending: conn.execute( INSERT OR IGNORE INTO seen(item_id, status, first_seen_at) VALUES(?, pending, ?), (it[id], int(time.time())), ) conn.commit() if sender(it): conn.execute(UPDATE seen SET statussent WHERE item_id?, (it[id],)) conn.commit() else: logging.warning(send failed: %s, it[id])代码按顺序走先把商品插入 seen 表并置为 pending这一步相当于占坑重复 ID 会被主键挡掉再调用发送函数发送成功才更新为 sent失败则保留 pending下一轮自动重试。日志里同时打出页面卡片总数和待发送数量这个数字是判断扫描器是否健康的第一指标——总数正常但 pending 一直是 0说明没有新商品总数也是 0说明解析可能已经失效。3.3 单次运行模式为定时任务准备的入口常驻循环适合个人电脑上手动跑但更多人实际部署在服务器或 NAS 上更适合用定时任务每两分钟跑一次跑完退出。单次模式的好处是脚本崩溃后定时器会自动拉起下一轮不需要守护进程。所以我把上面代码包成 scan_once()主入口用 argparse 接收参数interval 传了就进入常驻循环不传就单次执行。import argparse def build_parser(): parser argparse.ArgumentParser(descriptionmercari scanner) parser.add_argument(--url, requiredTrue, helpsearch page url (copy from browser)) parser.add_argument(--db, defaultseen.db, helpsqlite file path) parser.add_argument(--interval, typeint, defaultNone, helprun as daemon loop, seconds between scans) return parser if __name__ __main__: args build_parser().parse_args() conn init_db(args.db) if args.interval: while True: try: scan_once(conn, args.url, send_alert_via_slack) except Exception as exc: logging.exception(scan error: %s, exc) time.sleep(random.uniform(args.interval * 0.9, args.interval * 1.2)) else: scan_once(conn, args.url, send_alert_via_slack)interval 传了就进入常驻循环不传就单次执行一个脚本兼顾两种部署方式。随机因子 0.9 到 1.2 是我用过比较稳的抖动范围太小没意义太大又会让最大间隔接近设定的 1.2 倍影响时效。单次模式配合定时任务的两分钟粒度比裸跑常驻进程更能承受偶发被杀掉的情况。前面代码里的 send_alert_via_slack 这个函数名是占位第四章就会实现它。4. 警报送达从终端到手机/聊天工具的三条路4.1 Webhook 推到聊天工具Slack 与 Discord 通用的写法警报最省事的通道是聊天工具的 Webhook建一个钩子拿到一个 URL往这个 URL POST 一段 JSON 就完成推送。Slack 的 Incoming Webhook 和 Discord 的 Webhook 都是这个模式差别只在 JSON 字段名和地址代码几乎可以共用。发送函数设计成接收一个 item 字典、返回布尔值主循环直接依据返回值更新 seen 状态。import os, requests def send_alert_via_slack(item, webhook_urlNone): url webhook_url or os.environ.get(ALERT_WEBHOOK_URL) if not url: raise RuntimeError(ALERT_WEBHOOK_URL not set) text f新上架 [{item.get(keyword, search)}] if item.get(price): text f {item[price]} text f\nhttps://www.mercari.com/item/{item[id]} payload {text: text} # Slack 字段 # payload {content: text} # Discord 换成这个字段 resp requests.post(url, jsonpayload, timeout10) return resp.status_code 200为什么要区分字段名Slack 用 text 渲染普通消息Discord 用 content。requests 传 json 参数会自动把 Content-Type 设为 application/json不需要手动加请求头。webhook URL 不要写死在脚本里放进环境变量最稳妥这类 URL 一旦泄露任何人都能往你的频道发消息等于消息通道被人接管。消息体把关键词、价格、详情链接都拼上价格保留原始货币符号不要转成数字再格式化容易丢货币单位。4.2 邮件 SMTP不需要额外依赖的兜底通道聊天工具不是哪里都顺手比如接收方更多看邮箱或者你想留一份长期可检索的记录SMTP 就是第二选择。Python 标准库 smtplib 就能搞定不需要装额外依赖。注意一个高频坑SMTP 密码在大多数服务商那里不是邮箱登录密码而是单独开的授权码弄错会一直报 authentication failed。import smtplib from email.message import EmailMessage def send_alert_via_email(item, smtp_cfgNone): cfg smtp_cfg or { host: os.environ[SMTP_HOST], port: int(os.environ.get(SMTP_PORT, 465)), user: os.environ[SMTP_USER], password: os.environ[SMTP_PASS], to: os.environ[ALERT_EMAIL_TO], } msg EmailMessage() msg[Subject] f[mercari] 新上架 {item.get(price, )} msg[From] cfg[user] msg[To] cfg[to] msg.set_content(fhttps://www.mercari.com/item/{item[id]}) with smtplib.SMTP_SSL(cfg[host], cfg[port], timeout10) as client: client.login(cfg[user], cfg[password]) client.send_message(msg) return TrueSMTP_SSL 走 465 端口加密是兼容性比较好的方式如果你的服务商只支持 587 端口 STARTTLS把 SMTP_SSL 换成 SMTP 再加一次 starttls() 即可。timeout10 防止连不上时卡死主循环。邮件适合低频场景一天几十条以内没问题如果扫的是很热的关键词消息量上来后邮箱服务容易触发限流那种场景还是聊天工具更合适。4.3 拼一条有信息量的警报关键词、价格与链接一个都不能少警报消息的内容质量直接决定这个工具好不好用。只给一个链接的警报在手机上要跳转才能看到是什么用两次就不想用了。我的消息模板是第一行关键词前缀和价格第二行标题最后是详情链接。到这里你也会发现第三章里的 item 字典需要补一个 keyword 字段在 scan_once 里把搜索关键词传进去就行。def build_message(item): lines [f【{item.get(keyword, mercari)}】] if item.get(title): lines.append(item[title]) if item.get(price): lines.append(f价格: {item[price]}) lines.append(fhttps://www.mercari.com/item/{item[id]}) return \n.join(lines)这里故意把 key 都写成 .get() 而不是直接下标访问是因为解析过程中会出现字段缺失比如图片抓不到、价格没匹配到不能让缺了某一项就导致整条警报发送失败。宁可消息少个标题也不能不发。图片 URL 如果 JSON 里能拿到可以在支持的 Webhook 里当附件发出去但对大多数文本警报来说标题、价格、链接已经足够做判断。提示发送成功与否一定要让主循环知道。我之前偷懒把发送函数写在循环里不返回结果结果某个周末 webhook 地址过期新商品全入库了但一条警报都没发出去等发现时已经漏了两天。这也是第三章里 status 字段存在的意义。5. 常见问题与避坑重复推送、漏报、限流和改版崩溃5.1 同一条商品被推两次现象同一件商品在相隔几轮扫描里各收到一次警报甚至间隔很久又冒出来。原因通常是两个发送和状态更新之间出了问题比如发送成功但更新状态那行抛了异常或者多个扫描进程盯同一个搜索 URL但各自连了不同的数据库文件去重失效。解决遵循第三章的设计发送前把记录标记为 pending发送成功置 sent状态更新代码放在 socket 异常不可能跳过的地方多个进程跑同一组关键词时必须强制共用一个 DB 文件。这个问题的本质是“去重依据”必须和数据入库是同一个事务任何拆开都会留下重复窗口。5.2 关键词明明正确却一条新警报都没有现象日志显示每轮都抓到几十个 ID但全是旧商品或者抓到 0 个 ID脚本没有任何报错。前者十有八九是排序问题——你复制的 URL 仍然是推荐排序最新上架根本不在第一屏后者是解析问题——页面结构变了正则或选择器匹配不到任何链接但 findall 天然返回空列表所以不报错。解决先用浏览器打开搜索页确认最新商品确实在结果里拿其中一个 ID 在命令行里验证抓取结果。curl -s https://www.mercari.com/search/... | grep -o /item/[a-zA-Z0-9_]* | head如果 curl 的输出里有商品 ID说明抓取链路没问题问题在解析如果输出为空先确认 URL 是不是被重定向到了登录页或验证页再考虑页面结构变化。排查这类问题最怕的是“看起来正常”的空结果所以我在 3.2 的代码里特意加了 page returned N 条日志N0 时就要立刻警惕。5.3 抓得太频繁开始收到 429/403现象请求返回 429 或 403或者是验证码页面。原因很直白频率太高或者请求特征太像程序。个人扫描工具按分钟级访问一个搜索页一般问题不大但如果你把间隔压到几秒或者同时开多线程盯几十个关键词很快就会被拦。解决间隔调到 60 秒以上并加抖动用 requests.Session 复用底层连接减少 TLS 握手次数同一时间只跑一个脚本进程单次请求失败后做指数退避重试比如第 1 次等 5 秒、第 2 次等 20 秒、第 3 次等 60 秒。这套脚本的定位是个人小流量工具不要把它改造成并发抓取器。规模化采集属于另一个工程范畴需要走平台允许的渠道。5.4 改版后解析为空脚本照样成功运行现象站点升级或灰度测试后页面从服务端渲染变成纯接口渲染或者商品链接路径变了脚本每轮都抓到 0 个 ID但进程不报错、数据库不变化看起来一切正常。解决把解析函数做成独立模块并保存一份正常时期的页面样例每次改版后先做对比测试同时在主循环里统计连续空结果次数超过阈值就给自己发一条“扫描可能失效”的提醒。下面是一段可以直接嵌进 scan_once 的守护逻辑。consecutive_empty 0 def scan_once(conn, search_url, sender): global consecutive_empty items extract_cards(fetch_html(search_url)) if not items: consecutive_empty 1 if consecutive_empty 5: send_self_alert(scanner returned empty for 5 rounds) else: consecutive_empty 0 # ... 后续发送逻辑这个设计的核心是让工具具备“自己知道坏了”的能力。连续 5 轮空结果不代表一定失效也可能是这个关键词确实没有新商品但把它当成一种信号发给自己总比默默失灵强。send_self_alert 可以直接复用第四章的 webhook 发送函数只是消息内容不同。5.5 单次超时或 5xx 导致漏一轮扫描现象某轮扫描请求超时或返回 5xx这一轮的新商品被跳过如果时间点恰好是某个商品上架后的几分钟它就漏了。原因requests 抛异常被外层 try-except 吞掉后直接进入下一轮但对异常前已处理的 ID 可能已经入库并标记 sent下一轮不会再补。解决抓取函数内部做两次重试重试只在网络异常和服务端 5xx 时进行429 一律不立刻重试而是等下一个扫描周期重试仍失败的轮次打 warning 并跳过本轮不污染数据库。def fetch_with_retry(url, retries2): for attempt in range(retries 1): try: return fetch_html(url) except requests.exceptions.HTTPError as exc: if exc.response is not None and exc.response.status_code 429: raise # 限流时不重试留给下一个周期 if attempt retries: raise time.sleep(5 * (2 ** attempt))这里对异常做了分类429 表示被限流立刻重试只会加重问题直接抛给上层交给下一轮时间窗口5xx 和连接异常属于瞬时故障用 5 秒、10 秒的退避重试两次。把“错误处理”和“业务逻辑”分开后主循环看着清爽很多排查日志时也更直接。6. 进阶让警报更聪明——过滤、黑名单与定时执行6.1 价格区间与黑名单过滤新商品警报最怕“来什么推什么”关键词稍微宽一点就全是配件、贴膜之类的周边。做法是在发送前加一道过滤价格低于下限或高于上限的直接忽略标题命中黑名单词的也略过。过滤放在 is_pending 之后、入库之前被过滤掉的商品不占 seen 表下次出现还会再过一次避免误伤后无法恢复。BLACKLIST [壳, 膜, ケース, 已售] PRICE_MIN, PRICE_MAX 3000, 80000 def should_alert(item): if item.get(raw_text) and any(w in item[raw_text] for w in BLACKLIST): return False price item.get(price) if not price: return False num int(price.replace(¥, ).replace(, ).replace(,, )) return PRICE_MIN num PRICE_MAXblacklist 里混合了中文和日文词是因为同一件商品的描述里可能出现两个语言版本的关键字。价格解析先把货币符号和逗号都剥掉再转 int比较时用纯数字展示时保留原始字符串。过滤条件返回 False 的商品不会进入 seen 表这点很重要——它保证黑名单词条一旦从商品标题里移除下次扫描还有机会推送。6.2 定时执行比常驻更省心我最终把跑法改成了单次模式加定时任务两分钟一次不用管常驻进程的崩溃和内存增长。crontab 里写成*/2 * * * * cd /path python scan.py --url ... scan.log 21日志重定向到文件定时器自动处理拉起的逻辑。这样即使某一次进程被杀下一轮调度会自动补上配合 5.5 的重试机制漏报概率很低。如果你偏爱常驻至少用 systemd 或进程管理器托管别裸跑 nohup。6.3 上线前先验证通道再上真实关键词我的习惯是用一个流量极低又相对冷门的词先跑几小时确认解析、去重、发送链路全部正常再切到真正关心的关键词。切过去之后第一轮一定会收到大量历史商品警报这是正常的因为它们第一次出现别急着改代码等第一轮消化完再观察新增是否稳定。做这个项目时我踩过最深的一个坑就是把默认推荐排序当成了最新排序蹲了三天一条没报还以为是网络问题最后点开页面才发现新商品一直在第二屏。从那以后我养成了两个习惯复制 URL 前先确认排序选项任何解析改动先跑一次样例对比。希望这篇能帮到你少走几步弯路。本文还有配套的精品资源点击获取
