针对 CAPTCHA 事件的 Slack 机器人通知

当您的 CAPTCHA 管道开始大规模失败时，您需要立即知道，而不是数小时后抓取结果为空。本指南针对失败、低余额和异常错误率设置 Slack 通知。

设置：创建 Slack Webhook

1. 前往api.slack.com/apps→ 创建新应用程序
2. 选择 传入 Webhook → 激活
3. 单击 将新 Webhook 添加到工作区 → 选择频道
4. 复制网络钩子 URL

Python：Slack 通知助手

import requests
import json
from datetime import datetime

SLACK_WEBHOOK_URL = "https://hooks.slack.com/services/T00/B00/xxx"


def send_slack_alert(title, message, color="#ff0000", fields=None):
    """Send a formatted Slack alert."""
    attachment = {
        "color": color,
        "title": title,
        "text": message,
        "ts": int(datetime.now().timestamp()),
    }
    if fields:
        attachment["fields"] = [
            {"title": k, "value": str(v), "short": True}
            for k, v in fields.items()
        ]

    payload = {"attachments": [attachment]}
    resp = requests.post(SLACK_WEBHOOK_URL, json=payload, timeout=10)
    return resp.status_code == 200

警报1：解决失败

def notify_solve_failure(task_id, captcha_type, error_code, site_url):
    send_slack_alert(
        title="CAPTCHA Solve Failed",
        message=f"Task `{task_id}` failed with `{error_code}`",
        color="#ff0000",
        fields={
            "Type": captcha_type,
            "Error": error_code,
            "Site": site_url,
            "Time": datetime.now().strftime("%H:%M:%S"),
        },
    )

# Use after a failed solve
result = poll_for_result(task_id)
if result.get("error"):
    notify_solve_failure(task_id, "recaptcha_v2", result["error"], "https://example.com")

警报 2：余额低

def check_balance_alert(api_key, threshold=5.0):
    """Alert when balance drops below threshold."""
    resp = requests.get("https://ocr.captchaai.com/res.php", params={
        "key": api_key, "action": "getbalance", "json": "1"
    }).json()

    balance = float(resp.get("request", 0))

    if balance < threshold:
        send_slack_alert(
            title="Low CaptchaAI Balance",
            message=f"Balance is ${balance:.2f} (threshold: ${threshold:.2f})",
            color="#ff9900",
            fields={
                "Current Balance": f"${balance:.2f}",
                "Threshold": f"${threshold:.2f}",
            },
        )
    return balance

# Run periodically
import threading

def balance_monitor(api_key, interval=300):
    """Check balance every 5 minutes."""
    check_balance_alert(api_key)
    timer = threading.Timer(interval, balance_monitor, args=[api_key, interval])
    timer.daemon = True
    timer.start()

balance_monitor("YOUR_API_KEY")

警报 3：错误率高

from collections import deque

class ErrorRateNotifier:
    def __init__(self, window=50, threshold=0.3, cooldown=300):
        self.results = deque(maxlen=window)
        self.threshold = threshold
        self.cooldown = cooldown
        self.last_alert = 0

    def record(self, success):
        self.results.append(success)

        if len(self.results) < 20:
            return

        error_rate = 1 - sum(self.results) / len(self.results)

        import time
        now = time.time()
        if error_rate > self.threshold and (now - self.last_alert) > self.cooldown:
            self.last_alert = now
            send_slack_alert(
                title="High CAPTCHA Error Rate",
                message=f"Error rate: {error_rate:.0%} over last {len(self.results)} tasks",
                color="#ff0000",
                fields={
                    "Error Rate": f"{error_rate:.1%}",
                    "Window": f"{len(self.results)} tasks",
                    "Threshold": f"{self.threshold:.0%}",
                },
            )

notifier = ErrorRateNotifier()

# After each solve attempt
notifier.record(success=True)   # solved
notifier.record(success=False)  # failed

Node.js 实现

const axios = require('axios');

const SLACK_WEBHOOK = 'https://hooks.slack.com/services/T00/B00/xxx';

async function sendSlackAlert(title, message, color = '#ff0000', fields = {}) {
  const attachment = {
    color,
    title,
    text: message,
    ts: Math.floor(Date.now() / 1000),
    fields: Object.entries(fields).map(([k, v]) => ({
      title: k, value: String(v), short: true,
    })),
  };

  await axios.post(SLACK_WEBHOOK, { attachments: [attachment] });
}

// Failure alert
async function notifySolveFailure(taskId, type, error) {
  await sendSlackAlert(
    'CAPTCHA Solve Failed',
    `Task \`${taskId}\` failed: \`${error}\``,
    '#ff0000',
    { Type: type, Error: error }
  );
}

// Balance alert
async function checkBalance(apiKey, threshold = 5.0) {
  const resp = await axios.get('https://ocr.captchaai.com/res.php', {
    params: { key: apiKey, action: 'getbalance', json: 1 },
  });
  const balance = parseFloat(resp.data.request);

  if (balance < threshold) {
    await sendSlackAlert(
      'Low CaptchaAI Balance',
      `Balance: $${balance.toFixed(2)}`,
      '#ff9900',
      { Balance: `$${balance.toFixed(2)}`, Threshold: `$${threshold.toFixed(2)}` }
    );
  }
  return balance;
}

// Periodic check
setInterval(() => checkBalance('YOUR_API_KEY'), 5 * 60 * 1000);

每日总结

def send_daily_summary(stats):
    """Send a daily digest to Slack."""
    send_slack_alert(
        title="Daily CAPTCHA Summary",
        message=f"{stats['total']} tasks processed",
        color="#36a64f",
        fields={
            "Solved": stats["solved"],
            "Failed": stats["failed"],
            "Avg Solve Time": f"{stats['avg_time_ms']}ms",
            "Total Cost": f"${stats['total_cost']:.2f}",
            "Success Rate": f"{stats['success_rate']:.1%}",
        },
    )

故障排除

常问问题

如何避免警觉疲劳？

使用冷却时间（最少 5 分钟）、将错误批量记录到摘要中，并且仅针对错误率阈值发出警报，而不是个别故障。

我可以使用 Discord 代替 Slack 吗？

是的。 Discord webhooks 接受类似的 JSON 格式。更改有效负载结构以使用 embeds 而不是 attachments。

使用 CaptchaAI 监控您的验证码工作流程

获取您的 API 密钥：验证码网站。

相关指南

• CAPTCHA 操作的结构化日志记录
• CaptchaAI 余额检查和自动充值
• 构建 CaptchaAI 使用仪表板