验证码识别管道出问题时，最怕的不是报错本身，而是没人第一时间发现。与其为 CAPTCHA 管道单独搭一套告警面板，不如把余额、错误率、队列深度这些状态直接推送到团队已经在用的 Discord 频道——十几行 Python 代码，配合一个 webhook，就能把这套通知跑起来。
国内不少做海外电商、社媒自动化的团队，本来就习惯把 CI、监控消息统一丢进 Discord 或飞书群，CAPTCHA 管道的告警走同一条路径就好。
第一步：创建 Discord Webhook
在推送任何消息之前，先拿到一个 webhook 地址，步骤不多，5 分钟能搞定：
- 打开你的 Discord 服务器设置
- 进入 集成 → Webhooks
- 点击 New Webhook，命名为“CaptchaAI Alerts”，方便和其他机器人区分
- 复制 webhook URL——这串地址本身就是访问凭证，不要提交到代码仓库
- 存成环境变量
DISCORD_WEBHOOK_URL，代码里只读取环境变量，不要硬编码
Python 实现：完整告警系统
下面这套实现覆盖四类场景：余额不足、错误率飙升、队列积压、每日汇总，外加一条“管道已恢复”的确认消息。COLORS 字典把严重程度映射成 embed 的边框颜色——绿色恢复正常，橙色需要关注，红色要马上处理。PipelineMonitor 在内存里滚动保存最近 5 分钟的结果，并用 cooldown 控制同一类告警的最短发送间隔，避免反复刷屏。
import os
import time
import requests
from datetime import datetime
API_KEY = os.environ["CAPTCHAAI_API_KEY"]
DISCORD_WEBHOOK = os.environ["DISCORD_WEBHOOK_URL"]
session = requests.Session()
class DiscordCaptchaAlerts:
COLORS = {
"success": 0x2ECC71, # Green
"warning": 0xF39C12, # Orange
"error": 0xE74C3C, # Red
"info": 0x3498DB, # Blue
}
def __init__(self, webhook_url):
self.webhook_url = webhook_url
def send_embed(self, title, description, color_key="info", fields=None):
embed = {
"title": title,
"description": description,
"color": self.COLORS.get(color_key, self.COLORS["info"]),
"timestamp": datetime.utcnow().isoformat() + "Z",
"footer": {"text": "CaptchaAI Pipeline Monitor"}
}
if fields:
embed["fields"] = fields
payload = {"embeds": [embed]}
resp = requests.post(
self.webhook_url, json=payload, timeout=10
)
resp.raise_for_status()
def balance_alert(self, balance, threshold):
severity = "error" if balance < 2 else "warning"
self.send_embed(
title="💰 Balance Alert",
description=f"CaptchaAI balance is **${balance:.2f}**",
color_key=severity,
fields=[
{"name": "Threshold", "value": f"${threshold:.2f}", "inline": True},
{"name": "Severity", "value": severity.upper(), "inline": True},
{"name": "Action", "value": "Top up your balance at captchaai.com", "inline": False}
]
)
def error_spike(self, error_rate, error_count, total_count, top_errors):
error_list = "\n".join(
f"• `{code}`: {count}" for code, count in top_errors.items()
)
self.send_embed(
title="⚠️ Error Rate Spike",
description=f"Error rate: **{error_rate:.1%}** ({error_count}/{total_count})",
color_key="error",
fields=[
{"name": "Error Breakdown", "value": error_list or "No details", "inline": False},
{"name": "Window", "value": "Last 5 minutes", "inline": True}
]
)
def queue_alert(self, depth, workers_active):
self.send_embed(
title="📊 Queue Backup",
description=f"Queue depth: **{depth}** pending tasks",
color_key="warning",
fields=[
{"name": "Active Workers", "value": str(workers_active), "inline": True},
{"name": "Est. Drain Time", "value": f"{depth // max(workers_active, 1)} min", "inline": True}
]
)
def daily_summary(self, stats):
self.send_embed(
title="📈 Daily CAPTCHA Summary",
description=f"**{stats['total']}** tasks processed",
color_key="success" if stats["success_rate"] > 0.92 else "warning",
fields=[
{"name": "Success Rate", "value": f"{stats['success_rate']:.1%}", "inline": True},
{"name": "Avg Latency", "value": f"{stats['avg_latency']:.1f}s", "inline": True},
{"name": "Total Cost", "value": f"${stats['cost']:.2f}", "inline": True},
{"name": "Errors", "value": str(stats["errors"]), "inline": True},
{"name": "Balance", "value": f"${stats['balance']:.2f}", "inline": True},
{"name": "Peak Queue", "value": str(stats["peak_queue"]), "inline": True},
]
)
def solve_recovered(self, previous_rate, current_rate):
self.send_embed(
title="✅ Pipeline Recovered",
description=f"Solve rate recovered: {previous_rate:.1%} → {current_rate:.1%}",
color_key="success"
)
alerts = DiscordCaptchaAlerts(DISCORD_WEBHOOK)
class PipelineMonitor:
def __init__(self, check_interval=60):
self.check_interval = check_interval
self.results = [] # (timestamp, success, error_code)
self.last_balance_alert = 0
self.last_error_alert = 0
self.cooldown = 300 # 5 minutes between alerts
def record(self, success, error_code=None):
self.results.append((time.time(), success, error_code))
# Keep last 5 min
cutoff = time.time() - 300
self.results = [r for r in self.results if r[0] > cutoff]
def run_checks(self):
now = time.time()
# Balance check
if now - self.last_balance_alert > self.cooldown:
balance = self._check_balance()
if balance is not None and balance < 10:
alerts.balance_alert(balance, threshold=10)
self.last_balance_alert = now
# Error rate check
if now - self.last_error_alert > self.cooldown and len(self.results) > 10:
total = len(self.results)
errors = [r for r in self.results if not r[1]]
error_rate = len(errors) / total
if error_rate > 0.15:
error_breakdown = {}
for _, _, code in errors:
if code:
error_breakdown[code] = error_breakdown.get(code, 0) + 1
alerts.error_spike(error_rate, len(errors), total, error_breakdown)
self.last_error_alert = now
def _check_balance(self):
try:
resp = session.get("https://ocr.captchaai.com/res.php", params={
"key": API_KEY, "action": "getbalance", "json": 1
})
data = resp.json()
if data.get("status") == 1:
return float(data["request"])
except Exception:
pass
return None
monitor = PipelineMonitor()
JavaScript 实现：Webhook 客户端
如果管道跑在 Node.js 里，逻辑是一样的：用 axios 把 embed 结构 POST 到 webhook 地址，monitorBalance 每 5 分钟检查一次余额，低于阈值就触发告警。接到你现有的定时任务或 setInterval 循环里即可，不需要额外引入消息队列。
const axios = require("axios");
const API_KEY = process.env.CAPTCHAAI_API_KEY;
const DISCORD_WEBHOOK = process.env.DISCORD_WEBHOOK_URL;
const COLORS = {
success: 0x2ecc71,
warning: 0xf39c12,
error: 0xe74c3c,
info: 0x3498db,
};
async function sendDiscordEmbed(title, description, colorKey = "info", fields = []) {
await axios.post(DISCORD_WEBHOOK, {
embeds: [
{
title,
description,
color: COLORS[colorKey] || COLORS.info,
timestamp: new Date().toISOString(),
footer: { text: "CaptchaAI Pipeline Monitor" },
fields,
},
],
}, { timeout: 10000 });
}
async function alertBalance(balance, threshold = 10) {
const severity = balance < 2 ? "error" : "warning";
await sendDiscordEmbed(
"💰 Balance Alert",
`CaptchaAI balance is **$${balance.toFixed(2)}**`,
severity,
[
{ name: "Threshold", value: `$${threshold.toFixed(2)}`, inline: true },
{ name: "Severity", value: severity.toUpperCase(), inline: true },
]
);
}
async function alertErrorSpike(errorRate, details = {}) {
await sendDiscordEmbed(
"⚠️ Error Rate Spike",
`Error rate: **${(errorRate * 100).toFixed(1)}%**`,
"error",
[
{ name: "Total Tasks", value: String(details.total || 0), inline: true },
{ name: "Errors", value: String(details.errors || 0), inline: true },
]
);
}
async function sendDailySummary(stats) {
const color = stats.successRate > 0.92 ? "success" : "warning";
await sendDiscordEmbed(
"📈 Daily CAPTCHA Summary",
`**${stats.total}** tasks processed`,
color,
[
{ name: "Success Rate", value: `${(stats.successRate * 100).toFixed(1)}%`, inline: true },
{ name: "Avg Latency", value: `${stats.avgLatency.toFixed(1)}s`, inline: true },
{ name: "Balance", value: `$${stats.balance.toFixed(2)}`, inline: true },
]
);
}
// Balance monitoring loop
async function monitorBalance() {
try {
const resp = await axios.get("https://ocr.captchaai.com/res.php", {
params: { key: API_KEY, action: "getbalance", json: 1 },
});
if (resp.data.status === 1) {
const balance = parseFloat(resp.data.request);
if (balance < 10) await alertBalance(balance);
}
} catch (err) {
console.error("Balance check failed:", err.message);
}
}
setInterval(monitorBalance, 300000); // Every 5 minutes
module.exports = { alertBalance, alertErrorSpike, sendDailySummary };
Discord 告警效果预览
接入之后，频道里大致会收到这样的消息：
| 场景 | 消息内容 |
|---|---|
| 余额告警 | 💰 Balance Alert —— CaptchaAI balance is $8.42（Threshold: $10.00，Severity: WARNING） |
| 错误率飙升 | ⚠️ Error Rate Spike —— Error rate: 22.5% (45/200)，ERROR_CAPTCHA_UNSOLVABLE: 30，TIMEOUT: 15 |
| 每日汇总 | 📈 Daily CAPTCHA Summary —— 12,450 tasks processed，Success Rate: 95.2%，Avg Latency: 22.4s，Balance: $142.30 |
常见故障排查
接入后最容易踩的坑，基本逃不出这四类：
| 问题 | 原因 | 处理方式 |
|---|---|---|
| 400 Bad Request | embed 结构不合法 | 检查 fields 数组格式，确认所有取值都是字符串 |
| 触发限流 (429) | 单位时间内消息发送过多 | 在同类告警之间加冷却时间（至少 5 分钟） |
| webhook 已失效 | 有人在服务器里把它删除了 | 重新创建一个 webhook，更新环境变量 |
| embed 不显示 | 缺少 embeds 数组 |
把 embed 对象包在 {"embeds": [...]} 里 |
常见问题
Discord webhook 每分钟最多能发多少条消息？会被限流吗？
Discord 对同一个 webhook 有速率限制，短时间内发太多消息会收到 429。做法是给同类告警加冷却时间——上面的实现里 cooldown = 300，同一类告警 5 分钟内最多触发一次，基本能避开限流。
webhook URL 泄露了怎么办？
webhook URL 本身就是访问凭证，任何人拿到它都能往你的频道发消息。一旦怀疑泄露，直接在 Discord 后台删除这个 webhook、重新创建一个，再更新 DISCORD_WEBHOOK_URL 环境变量，旧地址会立刻失效。
可以在告警里 @提及 具体成员或身份组吗？
可以。在 payload 里加一个 "content": "<@USER_ID>" 或 "content": "<@&ROLE_ID>" 字段即可。建议只给余额耗尽、错误率飙升这类需要马上处理的告警加 @提及，日常汇总不用打扰人。
生产环境的关键告警，能只靠 Discord 吗？
不建议。Discord 适合做团队可见性强的辅助通知，但没有升级、排班这些机制。真正需要值班响应的告警，还是应该走 PagerDuty 或 Opsgenie，Discord 作为补充渠道保留历史记录就够了。
embed 消息的颜色和字段可以自定义吗？
可以，COLORS 字典和 fields 数组都是你自己定义的。除了教程里的余额、错误率、队列、每日汇总，你也可以按自己管道的关键指标加新的 embed 类型，比如单个 CAPTCHA 类型的识别耗时。
下一步
把管道告警接到团队本来就在用的地方——获取 CaptchaAI API Key，然后按上面的步骤连上 Discord。
相关指南：
- Slack 机器人告警集成
- PagerDuty 值班告警接入
- 错误码参考手册
