简介：面向日本二手交易平台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;q=0.9", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", } def fetch_html(url): resp = requests.get(url, headers=HEADERS, timeout=15) 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 后刷新页面，按下 Ctrl+F 在返回体里搜你看到的某个商品标题或价格。实际会遇到两种情况：一是搜索页 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(" ", strip=True) 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(path="seen.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(level=logging.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 status='sent' 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(description="mercari scanner") parser.add_argument("--url", required=True, help="search page url (copy from browser)") parser.add_argument("--db", default="seen.db", help="sqlite file path") parser.add_argument("--interval", type=int, default=None, help="run 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_url=None): 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, json=payload, timeout=10) 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_cfg=None): 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(f"https://www.mercari.com/item/{item['id']}") with smtplib.SMTP_SSL(cfg["host"], cfg["port"], timeout=10) as client: client.login(cfg["user"], cfg["password"]) client.send_message(msg) return True

SMTP_SSL 走 465 端口加密，是兼容性比较好的方式；如果你的服务商只支持 587 端口 STARTTLS，把 SMTP_SSL 换成 SMTP 再加一次 starttls() 即可。timeout=10 防止连不上时卡死主循环。邮件适合低频场景，一天几十条以内没问题；如果扫的是很热的关键词，消息量上来后邮箱服务容易触发限流，那种场景还是聊天工具更合适。

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(f"https://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 条日志，N=0 时就要立刻警惕。

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, retries=2): 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_MAX

blacklist 里混合了中文和日文词，是因为同一件商品的描述里可能出现两个语言版本的关键字。价格解析先把货币符号和逗号都剥掉再转 int，比较时用纯数字，展示时保留原始字符串。过滤条件返回 False 的商品不会进入 seen 表，这点很重要——它保证黑名单词条一旦从商品标题里移除，下次扫描还有机会推送。

6.2 定时执行比常驻更省心

我最终把跑法改成了单次模式加定时任务，两分钟一次，不用管常驻进程的崩溃和内存增长。crontab 里写成*/2 * * * * cd /path && python scan.py --url "..." >> scan.log 2>&1，日志重定向到文件，定时器自动处理拉起的逻辑。这样即使某一次进程被杀，下一轮调度会自动补上，配合 5.5 的重试机制，漏报概率很低。如果你偏爱常驻，至少用 systemd 或进程管理器托管，别裸跑 nohup。

6.3 上线前先验证通道，再上真实关键词

我的习惯是用一个流量极低又相对冷门的词先跑几小时，确认解析、去重、发送链路全部正常，再切到真正关心的关键词。切过去之后第一轮一定会收到大量历史商品警报，这是正常的，因为它们第一次出现；别急着改代码，等第一轮消化完，再观察新增是否稳定。

做这个项目时我踩过最深的一个坑，就是把默认推荐排序当成了最新排序，蹲了三天一条没报，还以为是网络问题，最后点开页面才发现新商品一直在第二屏。从那以后我养成了两个习惯：复制 URL 前先确认排序选项；任何解析改动先跑一次样例对比。希望这篇能帮到你，少走几步弯路。

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