供应链监控验证码识别：reCAPTCHA 与 Turnstile 实战

供应商库存报表连着几天出现空行，多半不是选择器失效，而是脚本撞上了验证码。处理方式固定：请求返回后加判断分支，命中就调 CaptchaAI 拿 token 重放，解析逻辑不动——reCAPTCHA v2、Turnstile 和图片验证码正是这条链路最常见的拦截点。

安全范围： 本指南面向你自有或已获授权采集的数据源，请遵守 robots 协议与《数据安全法》《个人信息保护法》要求。

先确认：是验证码，还是页面改版

存下失败那轮的响应，搜 data-sitekey、g-recaptcha、cf-turnstile。命中任一个即验证码页；都没有就是选择器或登录态问题。

供应链链路上的验证码分布

数据源	验证码类型	采集内容	频率
供应商门户	reCAPTCHA v2	库存、价格	每天
物流承运商	Cloudflare Turnstile	轨迹、运价	每小时
制造商目录	图片验证码	规格、起订量	每周
海关与关务	reCAPTCHA v2	税率、HS 编码	每天
港口与码头	图片验证码	船期、拥堵	每 6 小时
大宗商品交易所	reCAPTCHA v3	现货、期货	准实时

上游在珠三角或长三角时，第一行要改：本土门户更常挂 GeeTest（极验）滑块。CaptchaAI 支持 v3，v4 仍标注为即将支持。reCAPTCHA 依赖 Google 托管脚本，境内加载不稳，放境外节点跑更省心。

一套采集器覆盖三类验证码

流程固定：抓页面 → 判断类型 → 提交任务 → 拿 token 重放。三个 solve 函数结构一致：向 in.php 提交，轮询 res.php。

import requests
import time
import re
import json
import base64
from datetime import datetime

CAPTCHAAI_KEY = "YOUR_API_KEY"
CAPTCHAAI_URL = "https://ocr.captchaai.com"


def solve_recaptcha(sitekey, pageurl):
    resp = requests.post(f"{CAPTCHAAI_URL}/in.php", data={
        "key": CAPTCHAAI_KEY, "method": "userrecaptcha",
        "googlekey": sitekey, "pageurl": pageurl, "json": 1,
    })
    task_id = resp.json()["request"]
    for _ in range(60):
        time.sleep(5)
        result = requests.get(f"{CAPTCHAAI_URL}/res.php", params={
            "key": CAPTCHAAI_KEY, "action": "get",
            "id": task_id, "json": 1,
        })
        data = result.json()
        if data["request"] != "CAPCHA_NOT_READY":
            return data["request"]
    raise TimeoutError("Timeout")


def solve_turnstile(sitekey, pageurl):
    resp = requests.post(f"{CAPTCHAAI_URL}/in.php", data={
        "key": CAPTCHAAI_KEY, "method": "turnstile",
        "sitekey": sitekey, "pageurl": pageurl, "json": 1,
    })
    task_id = resp.json()["request"]
    for _ in range(60):
        time.sleep(5)
        result = requests.get(f"{CAPTCHAAI_URL}/res.php", params={
            "key": CAPTCHAAI_KEY, "action": "get",
            "id": task_id, "json": 1,
        })
        data = result.json()
        if data["request"] != "CAPCHA_NOT_READY":
            return data["request"]
    raise TimeoutError("Timeout")


def solve_image(image_bytes):
    img_b64 = base64.b64encode(image_bytes).decode()
    resp = requests.post(f"{CAPTCHAAI_URL}/in.php", data={
        "key": CAPTCHAAI_KEY, "method": "base64",
        "body": img_b64, "json": 1,
    })
    task_id = resp.json()["request"]
    for _ in range(20):
        time.sleep(3)
        result = requests.get(f"{CAPTCHAAI_URL}/res.php", params={
            "key": CAPTCHAAI_KEY, "action": "get",
            "id": task_id, "json": 1,
        })
        data = result.json()
        if data["request"] != "CAPCHA_NOT_READY":
            return data["request"]
    raise TimeoutError("Timeout")


class SupplyChainMonitor:
    def __init__(self, suppliers, proxy=None):
        self.suppliers = suppliers
        self.session = requests.Session()
        if proxy:
            self.session.proxies = {"http": proxy, "https": proxy}
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 Chrome/126.0.0.0 Safari/537.36",
        })

    def check_all(self):
        """Check inventory and pricing across all suppliers."""
        report = {
            "timestamp": datetime.now().isoformat(),
            "suppliers": {},
        }

        for supplier in self.suppliers:
            try:
                data = self._check_supplier(supplier)
                report["suppliers"][supplier["name"]] = {
                    "status": "success",
                    "data": data,
                }
            except Exception as e:
                report["suppliers"][supplier["name"]] = {
                    "status": "error",
                    "error": str(e),
                }
            time.sleep(3)

        return report

    def _check_supplier(self, supplier):
        url = supplier["url"]
        resp = self.session.get(url, timeout=30)

        # Handle CAPTCHA based on type
        captcha_type = supplier.get("captcha_type")
        if captcha_type and self._has_captcha(resp.text):
            resp = self._solve_captcha(resp, url, supplier)

        from bs4 import BeautifulSoup
        soup = BeautifulSoup(resp.text, "html.parser")

        return {
            "products": self._extract_inventory(soup),
            "last_updated": self._extract_date(soup),
        }

    def _has_captcha(self, html):
        return any(tag in html.lower() for tag in [
            'data-sitekey', 'g-recaptcha', 'cf-turnstile', 'captcha',
        ])

    def _solve_captcha(self, resp, url, supplier):
        captcha_type = supplier.get("captcha_type", "recaptcha")
        sitekey = supplier.get("sitekey", "")

        if not sitekey:
            match = re.search(r'data-sitekey="([^"]+)"', resp.text)
            sitekey = match.group(1) if match else ""

        if captcha_type == "turnstile":
            token = solve_turnstile(sitekey, url)
            return self.session.post(url, data={"cf-turnstile-response": token})
        elif captcha_type == "image":
            match = re.search(r'src="(/captcha[^"]+)"', resp.text)
            if match:
                img_resp = self.session.get(url.rstrip("/") + match.group(1))
                answer = solve_image(img_resp.content)
                return self.session.post(url, data={"captcha": answer})
        else:
            token = solve_recaptcha(sitekey, url)
            return self.session.post(url, data={"g-recaptcha-response": token})

        return resp

    def _extract_inventory(self, soup):
        items = []
        for row in soup.select("table.inventory tr, .product-row"):
            cols = row.select("td, .col")
            if len(cols) >= 3:
                items.append({
                    "sku": cols[0].get_text(strip=True),
                    "stock": cols[1].get_text(strip=True),
                    "price": cols[2].get_text(strip=True),
                })
        return items

    def _extract_date(self, soup):
        date_el = soup.select_one(".last-updated, .update-time")
        return date_el.get_text(strip=True) if date_el else ""


# Configure suppliers
suppliers = [
    {
        "name": "Supplier A",
        "url": "https://supplier-a.example.com/inventory",
        "captcha_type": "recaptcha",
        "sitekey": "6Lc_xxxxxxx",
    },
    {
        "name": "Carrier B",
        "url": "https://carrier-b.example.com/rates",
        "captcha_type": "turnstile",
        "sitekey": "0x4AAAAAAA_xxx",
    },
    {
        "name": "Manufacturer C",
        "url": "https://manufacturer-c.example.com/catalog",
        "captcha_type": "image",
    },
]

monitor = SupplyChainMonitor(
    suppliers=suppliers,
    proxy="http://user:[email protected]:5000",
)
report = monitor.check_all()
print(json.dumps(report, indent=2))

两处易踩坑：token 字段名不能混用，reCAPTCHA 用 g-recaptcha-response，Turnstile 用 cf-turnstile-response；_has_captcha() 是总开关。

运价页的 Turnstile：最小写法

承运商运价页多半在提交表单那一刻才弹 Turnstile。先 GET 取 data-sitekey，拿到 token 再连同查询参数一起 POST。

class ShippingRateTracker:
    def __init__(self, proxy=None):
        self.session = requests.Session()
        if proxy:
            self.session.proxies = {"http": proxy, "https": proxy}
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 Chrome/126.0.0.0 Safari/537.36",
        })

    def get_rates(self, carrier_url, origin, destination, weight):
        """Fetch shipping rates, handling Turnstile CAPTCHA."""
        resp = self.session.get(carrier_url, timeout=30)

        sitekey_match = re.search(r'data-sitekey="([^"]+)"', resp.text)
        if sitekey_match:
            token = solve_turnstile(sitekey_match.group(1), carrier_url)
            resp = self.session.post(carrier_url, data={
                "origin": origin,
                "destination": destination,
                "weight": weight,
                "cf-turnstile-response": token,
            })

        if resp.status_code == 200:
            return resp.json().get("rates", [])
        return []

sitekey 每次现取，别写死在配置里。

从原始快照到可执行告警

原始快照没有价值，“SKU 跌破安全库存”才是能行动的信号。下面的循环只在穿越阈值那一刻告警。

def monitor_with_alerts(monitor, alert_thresholds, check_interval=3600):
    """Continuously monitor and alert on inventory changes."""
    previous_data = {}

    while True:
        report = monitor.check_all()

        for supplier, info in report["suppliers"].items():
            if info["status"] != "success":
                continue

            for product in info["data"].get("products", []):
                sku = product["sku"]
                stock = product.get("stock", "")

                # Parse stock level
                try:
                    stock_qty = int(re.sub(r'\D', '', stock))
                except ValueError:
                    continue

                key = f"{supplier}:{sku}"
                prev_qty = previous_data.get(key, stock_qty)

                threshold = alert_thresholds.get(sku, 10)
                if stock_qty < threshold and prev_qty >= threshold:
                    print(f"ALERT: {supplier} - {sku} dropped to {stock_qty}")

                previous_data[key] = stock_qty

        time.sleep(check_interval)

频率、降级与线程预算

数据源分两档：关键源失败升级为事件，可选源只记 warning。
写清可容忍的陈旧度：运价 6 小时前还能用，港口拥堵超 12 小时作废。
采得越勤，验证码触发越频繁；多数门户每天一次足够。
成本按并发算：CaptchaAI 按线程计费，套餐内次数不限，BASIC $15/月 5 线程、ADVANCE $90/月 50 线程。

常见问题

境内服务器跑 reCAPTCHA 任务，为什么老超时？

reCAPTCHA 脚本由 Google 托管，境内可达性不稳定，页面本身就加载不全。常规做法是调度到境外节点执行。

供应商门户挂的是极验滑块，能一起接吗？

GeeTest（极验）v3 可以，用 geetest 方法提交；v4 仍在即将支持之列。国内厂商自建的其他方案不在支持范围。

线程按什么估？供应商数量还是任务量？

按峰值并发估：同一时刻最多几个识别任务在跑，那就是线程下限。取舍见会话策略对比。

延伸阅读

注册 CaptchaAI，把识别接进监控脚本。

使用验证码处理进行供应链监控

先确认：是验证码，还是页面改版

供应链链路上的验证码分布

一套采集器覆盖三类验证码

运价页的 Turnstile：最小写法

从原始快照到可执行告警

频率、降级与线程预算

常见问题

境内服务器跑 reCAPTCHA 任务，为什么老超时？

供应商门户挂的是极验滑块，能一起接吗？

线程按什么估？供应商数量还是任务量？

延伸阅读

使用验证码处理自动提交表单

通过验证码处理监控航空公司票价

CaptchaAI API 延迟优化：更快解决问题

演出门票余票监控实战：验证码识别怎么做

电商库存与价格监控：验证码处理实战

在自有运费计算器上对 CAPTCHA 进行 QA 测试

先确认：是验证码，还是页面改版

供应链链路上的验证码分布

一套采集器覆盖三类验证码

运价页的 Turnstile：最小写法

从原始快照到可执行告警

频率、降级与线程预算

常见问题

境内服务器跑 reCAPTCHA 任务，为什么老超时？

供应商门户挂的是极验滑块，能一起接吗？

线程按什么估？供应商数量还是任务量？

延伸阅读

相关文章

使用验证码处理自动提交表单

通过验证码处理监控航空公司票价

CaptchaAI API 延迟优化：更快解决问题

演出门票余票监控实战：验证码识别怎么做

电商库存与价格监控：验证码处理实战

在自有运费计算器上对 CAPTCHA 进行 QA 测试