In this Article
没抢到 Labubu 盲盒？这并不奇怪。限量款如今备受粉丝、收藏家和转售商追捧，而线上购买往往是唯一途径。在这种情况下，靠手动操作几乎不可能抢到心仪款式，专用机器人便成了有效的解决方案。本指南将介绍用机器人购买 Labubu 时需要注意的要点，并提供可直接使用的方案。
为什么 Labubu 机器人不可或缺？
Labubu 是一只毛绒小怪兽。自其创作者 Kasing Lung 与 Pop Mart 合作以来，它在 2019 年开始走红。这家中国零售商是其独家制造商，主要以盲盒形式发售 Labubu。盲盒会组成主题系列，其中包括与其他品牌联名的限量款，系列里通常还会有一款未公开的稀有”隐藏款”。此外，Pop Mart 的补货并非固定时间进行，粉丝必须留意规律并持续关注。这掀起了抢购热潮，买玩具也变成了一场争夺。这并非夸张，由于曾发生顾客受伤事件，Pop Mart 已在部分线下门店停售 Labubu 盲盒。线上购买虽然更安全，但同样需要速度和精准度，哪怕误点一个链接这样的小失误，也可能让您与玩具失之交臂。
换个角度看，
专用机器人能帮您省去不少麻烦：
- 监控新品补货，无需手动查看是否有库存；
- 自动执行购买流程，您只需完成最后的结账；
- 即时执行操作，速度是抢购 Labubu 的关键；
- 减少人为失误，例如误点错误按钮。
使用机器人公平吗？
并不禁止使用机器人，但您应遵守 Pop Mart 的服务条款，且不得从事违法行为。此外，您不应利用机器人获取敏感数据或损害网站。 DataImpulse 鼓励您仅以合乎道德的方式进行抓取，且不对任何非法活动负责。
那么，让我们为下一次补货做好准备。
使用 Python 构建 Labubu 机器人
开始编写代码前，先了解决定机器人能否成功的关键因素：
- 设计整体逻辑；
- 处理验证提示、弹窗和基于地理位置的限制；
- 模拟真实用户行为，例如设置请求之间的时间间隔；
- 选择辅助工具；
- 定期检查并更新机器人，因为 Pop Mart 会持续更新网站，您的机器人也应能适应这些变化。
下面我们将逐步处理这些问题。
入门
- 访问 Python 官方网站，下载最新版本或进行更新。安装时，请务必将 Python 添加到 PATH，以便能在终端中运行它。
- 运行以下命令来安装必要的库：
pip install playwright httpx python-dotenv
这会安装 3 个 Python 软件包：
- Playwright，用于浏览器自动化；
- httpx，Python 的 HTTP 客户端，支持 HTTP/1.1 和 HTTP/2、异步请求、代理等；
- python-dotenv，用于从 .env 文件加载环境变量，适合将密码、令牌等敏感信息存放在代码之外。
- 使用以下命令下载浏览器二进制文件：
playwright install
创建 Telegram 机器人和频道
我们需要一个 Telegram 机器人，以便在新品上架时接收通知。这样您可以照常忙自己的事，也不会错过任何一次补货。
这里提供了创建机器人的说明，以及获取 API 密钥的方法。
接着，创建一个公开频道，并将机器人设为管理员。
开发机器人
机器人项目的结构大致如下：
labububot/
├─ main.py
├─ runner.py
├─ config.py
├─ scraping.py
├─ telegram_utils.py
├─ state.py
├─ labubu_state.json
├─ .env
下面说明各个文件的作用。
runner.py 负责协调机器人的运行逻辑。它会按设定的时间间隔持续循环：加载当前状态、收集产品信息、检查库存、发送 Telegram 通知，并将更新保存到 labubu_state.json 文件中。该文件会自动创建。
import asyncio
from playwright.async_api import async_playwright
from state import load_state, save_state
from scraping import gather_products_once
from telegram_utils import render_message, send_telegram, now_iso
from config import INTERVAL_SECONDS
async def run():
state = load_state()
async with async_playwright() as pw:
print("[START] Labubu stock watcher started.")
while True:
try:
items = await gather_products_once(pw)
for it in items:
url = it["url"]
prev = state.get(url, {}).get("in_stock")
state[url] = {"in_stock": it["in_stock"], "name": it["name"], "price": it["price"]}
if it["in_stock"] and prev is not True:
msg = render_message(it)
await send_telegram(msg)
save_state(state)
print(f"[OK] Cycle finished: {len(items)} item(s). {now_iso()}")
except Exception as e:
print(f"[FATAL] {e}")
await asyncio.sleep(INTERVAL_SECONDS)
config.py 集中管理常量和相关配置。它定义要扫描的 Pop Mart 页面、数据的存储位置，以及从 .env 文件中读取代理详情和 Telegram 机器人 API 密钥等敏感信息。
import os
from pathlib import Path
from dotenv import load_dotenv
SEARCH_PAGES = [
"https://www.popmart.com/us/search/LABUBU",
"https://www.popmart.com/us/new-arrivals",
"https://www.popmart.com/us/collection/11/the-monsters",
]
load_dotenv()
INTERVAL_SECONDS = int(os.getenv("INTERVAL_SECONDS", "120"))
TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID")
PROXY_SERVER = os.getenv("PROXY_SERVER")
PROXY_USERNAME = os.getenv("PROXY_USERNAME")
PROXY_PASSWORD = os.getenv("PROXY_PASSWORD")
HEADLESS = os.getenv("HEADLESS", "1") != "0"
STATE_FILE = Path("labubu_state.json")
.env 用于存储敏感数据。将这些信息放在单独文件中而非硬编码到程序里，既更安全，也便于在需要时更改密码等内容。该文件还设置浏览器的运行模式（有头或无头）及抓取间隔。请务必填入您的真实凭据和数据。您可以从所选套餐的标签页复制您的 DataImpulse 凭据 。
TELEGRAM_BOT_TOKEN=your_token
TELEGRAM_CHAT_ID=@name
INTERVAL_SECONDS=90
PROXY_SERVER=proxy_server
PROXY_USERNAME=proxy_login
PROXY_PASSWORD=proxy_password
HEADLESS=1
scraping.py 是抓取模块的核心。它使用 Playwright 监控指定 Pop Mart 页面的产品库存、提取所需信息，并将结构化数据解析为 .json 文件。
import re
from urllib.parse import urljoin
from playwright.async_api import async_playwright
from config import SEARCH_PAGES, PROXY_SERVER, PROXY_USERNAME, PROXY_PASSWORD, HEADLESS
async def extract_products_from_search(page) -> list[dict]:
products, seen = [], set()
anchors = await page.locator('a[href*="/products/"]').all()
for a in anchors:
href = await a.get_attribute("href")
if not href or "/products/" not in href:
continue
url = urljoin(page.url, href)
if url in seen:
continue
name = (await a.text_content() or "").strip()
if not name or len(name) < 3:
try:
name = (await a.locator("..").text_content() or "").strip()
except Exception:
pass
if "labubu" in (name or "").lower() or "labubu" in url.lower() or "the-monsters" in url.lower():
products.append({"name": name, "url": url})
seen.add(url)
return products
async def parse_ld_json(page):
try:
content = await page.locator('script[type="application/ld+json"]').nth(0).text_content()
import json as _json
data = _json.loads(content)
if isinstance(data, list):
for item in data:
if isinstance(item, dict) and "offers" in item:
offers = item["offers"]
if isinstance(offers, dict):
return {
"availability": offers.get("availability"),
"price": offers.get("price"),
"priceCurrency": offers.get("priceCurrency"),
}
elif isinstance(data, dict) and "offers" in data:
offers = data["offers"]
return {
"availability": offers.get("availability"),
"price": offers.get("price"),
"priceCurrency": offers.get("priceCurrency"),
}
except Exception:
pass
return {}
async def check_product_stock(page, url: str) -> dict:
await page.goto(url, wait_until="domcontentloaded")
await page.wait_for_timeout(1200)
title = (await page.title()) or ""
title = title.replace(" - POP MART", "").strip()
meta = await parse_ld_json(page)
availability = (meta.get("availability") or "") if meta else ""
price = meta.get("price") if meta else None
if not price:
try:
price_text = await page.locator('[class*="price"], .price, [data-test*="price"]').first.text_content()
if price_text:
price_clean = re.sub(r"[^\d\.,]", "", price_text)
price = price_clean if price_clean else None
except Exception:
pass
in_stock = None
try:
if await page.locator("text=ADD TO BAG").count():
in_stock = True
elif await page.locator("text=SOLD OUT").count():
in_stock = False
except Exception:
pass
if in_stock is None and availability:
in_stock = "InStock" in availability or "instock" in availability.lower()
if in_stock is None:
try:
btn_texts = [t.lower() for t in await page.locator("button, div[role=button]").all_text_contents()]
joined = " | ".join(btn_texts)
if "add to bag" in joined or "add to cart" in joined:
in_stock = True
elif "sold out" in joined or "out of stock" in joined:
in_stock = False
except Exception:
pass
return {
"url": url,
"name": title or "LABUBU",
"price": price,
"in_stock": bool(in_stock),
}
async def gather_products_once(pw) -> list[dict]:
browser = await pw.chromium.launch(headless=HEADLESS)
context_args = {}
if PROXY_SERVER:
context_args["proxy"] = {"server": PROXY_SERVER, "username":PROXY_USERNAME, "password":PROXY_PASSWORD}
context = await browser.new_context(**context_args)
page = await context.new_page()
pool = {}
for url in SEARCH_PAGES:
try:
await page.goto(url, wait_until="domcontentloaded")
await page.wait_for_timeout(1000)
await page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
await page.wait_for_timeout(800)
prods = await extract_products_from_search(page)
for p in prods:
pool[p["url"]] = p
except Exception as e:
print(f"[WARN] Failed to open {url}: {e}")
results = []
for url in list(pool.keys()):
try:
res = await check_product_stock(page, url)
results.append(res)
except Exception as e:
print(f"[WARN] Failed to check {url}: {e}")
await context.close()
await browser.close()
return results
telegram_utils.py 负责在目标商品有货时发送 Telegram 通知。
import html
import httpx
from datetime import datetime, timezone
from config import TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID
def now_iso() -> str:
return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")
async def send_telegram(text: str, disable_preview: bool = False) -> None:
if not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID:
print("[WARN] TELEGRAM_BOT_TOKEN / TELEGRAM_CHAT_ID not set — message not sent.")
return
api_url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
payload = {
"chat_id": TELEGRAM_CHAT_ID,
"text": text,
"parse_mode": "HTML",
"disable_web_page_preview": disable_preview,
}
async with httpx.AsyncClient(timeout=20) as client:
r = await client.post(api_url, data=payload)
if r.status_code != 200:
print(f"[TG ERROR] {r.status_code}: {r.text}")
def render_message(item: dict) -> str:
title = html.escape(item.get("name") or "Labubu item")
price = item.get("price")
price_line = f"\n💵 Price: {price}" if price else ""
return (
f"🟢 LABUBU in stock\n"
f"{title}{price_line}\n"
f"🔗 {item['url']}\n"
f"🕒 {now_iso()}\n"
f"#Labubu #PopMart"
)
state.py 负责加载和保存产品状态信息到 labubu_state.json 文件。
import json
from config import STATE_FILE
def load_state() -> dict:
if STATE_FILE.exists():
try:
return json.loads(STATE_FILE.read_text("utf-8"))
except Exception:
return {}
return {}
def save_state(state: dict) -> None:
STATE_FILE.write_text(
json.dumps(state, ensure_ascii=False, indent=2),
encoding="utf-8"
)
main.py 是整个应用程序的入口。就像汽车的”启动引擎”按钮一样，这段代码会启动机器人。
import asyncio
from runner import run
if __name__ == "__main__":
try:
asyncio.run(run())
except KeyboardInterrupt:
print("\n[STOP] Stopped by user.")
要运行该文件并启动机器人，请在终端中输入以下命令：
python main.py
让机器人稳定运行的辅助工具
还可以使用以下工具：
- 日志框架，用于跟踪错误和监控机器人的整体活动，对调试很有帮助。
- 错误跟踪工具，可自动捕获并实时报告错误，帮助您快速解决问题。
- 监控与告警服务，用于监测机器人性能，并在异常活动或停机时发出警报。
- 环境隔离工具，容器可创建隔离环境，从而简化部署。
在示例代码中，我们还使用 dotenv 保护敏感数据。在攻击日益增多的今天，这一点尤为重要。代理同样关键，因为它们可将流量分配到不同的 IP 地址，帮助您模拟真实用户行为。它们还能让您同时使用多个连接，提高抢到 Labubu 的机会。如需代理支持，DataImpulse 随时为您服务。请发送电子邮件至 [email protected]，或点击”立即尝试“按钮开始使用。