背景

做 SERP 监控需要告警,Slack / Email / Webhook / SMS 各有取舍。

4 种方式对比

1. Slack

def alert_slack(text):
    requests.post(SLACK_WEBHOOK, json={
        "blocks": [
            {"type": "section", "text": {"type": "mrkdwn", "text": text}}
        ]
    })

优点:实时 (秒级)、富文本块、thread 回复、@ 提醒 缺点:依赖 Slack,团队需 Slack 账号

2. Email

def alert_email(to, subject, body):
    import smtplib
    from email.mime.text import MIMEText
    msg = MIMEText(body)
    msg["Subject"] = subject
    msg["To"] = to
    with smtplib.SMTP("smtp.gmail.com", 587) as s:
        s.starttls()
        s.login(USER, PASS)
        s.send_message(msg)

优点:通用,不需要装 app,异步不阻塞 缺点:延迟 (分钟级)、易进垃圾箱、HTML 渲染差

3. Webhook

def alert_webhook(url, payload):
    requests.post(url, json=payload, timeout=5)

优点:通用 (JSON 任意接收方)、可路由到不同系统 缺点:需要接收端,失败重试要自己做

4. SMS(短信)

def alert_sms(to, text):
    from twilio.rest import Client
    client = Client(TWILIO_SID, TWILIO_TOKEN)
    client.messages.create(
        from_="+1234567890", to=to, body=text
    )

优点:无网也能收、紧急最高优先级 缺点:贵 (每条 $0.01+)、字符限制 160、有 spam filter

4 种方式实测 (我项目 30 天)

选型

5 个工程细节

1. 降级 (主通道失败,备用)

def alert_with_fallback(text):
    try:
        alert_slack(text)
    except Exception:
        try:
            alert_webhook(BACKUP_WEBHOOK, {"text": text})
        except Exception:
            alert_email(EMERGENCY_EMAIL, "Alert", text)

2. 限流 (防刷屏)

last_alert = {}

def alert_with_dedup(key, text, cooldown=300):
    if key in last_alert and time.time() - last_alert[key] < cooldown:
        return  # 5 分钟内同 key 不重发
    last_alert[key] = time.time()
    alert_slack(text)

3. 富文本 (blocks 结构)

alert_slack_blocks(
    f"排名变化",
    blocks=[
        {"type": "section", "text": {"type": "mrkdwn", "text": "*{{kw}}* 跌了"}},
        {"type": "section", "fields": [
            {"type": "mrkdwn", "text": f"上次: #{old}"},
            {"type": "mrkdwn", "text": f"这次: #{new}"}
        ]},
        {"type": "actions", "elements": [
            {"type": "button", "text": "查看详情", "url": url}
        ]}
    ]
)

4. 异步 (不阻塞主流程)

import threading

def alert_async(text):
    threading.Thread(target=alert_slack, args=(text,)).start()

5. 限速 (Slack webhook 限速)

# Slack webhook 限速 1 条/秒
import time

last_call = 0

def alert_rate_limited(text):
    global last_call
    now = time.time()
    if now - last_call < 1:
        time.sleep(1 - (now - last_call))
    last_call = time.time()
    alert_slack(text)

4 种方式实战代码

Slack

import requests

SLACK_WEBHOOK = "https://hooks.slack.com/services/..."

def alert_slack_with_blocks(query, old_rank, new_rank, url):
    blocks = [
        {
            "type": "header",
            "text": {"type": "plain_text", "text": "📉 SERP 排名变化"}
        },
        {
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": f"*{query}* 跌了 5 位以上"
            }
        },
        {
            "type": "section",
            "fields": [
                {"type": "mrkdwn", "text": f"上次: #{old_rank}"},
                {"type": "mrkdwn", "text": f"这次: #{new_rank}"}
            ]
        },
        {
            "type": "actions",
            "elements": [
                {"type": "button", "text": {"type": "plain_text", "text": "查看"}, "url": url}
            ]
        }
    ]
    requests.post(SLACK_WEBHOOK, json={"blocks": blocks}, timeout=5)

Email

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

def alert_email_rich(to, query, old_rank, new_rank, url):
    msg = MIMEMultipart()
    msg["From"] = "alerts@yourdomain.com"
    msg["To"] = to
    msg["Subject"] = f"SERP Alert: {query} 变化"

    html = f"""
    <h2>SERP 排名变化</h2>
    <p>关键词: <b>{query}</b></p>
    <p>上次: #{old_rank} → 这次: #{new_rank}</p>
    <p><a href="{url}">查看 SERP 结果</a></p>
    """

    msg.attach(MIMEText(html, "html"))

    with smtplib.SMTP("smtp.gmail.com", 587) as s:
        s.starttls()
        s.login(USER, APP_PASS)
        s.send_message(msg)

Webhook(通用)

import requests
import hmac
import hashlib

def alert_webhook_signed(url, payload, secret):
    body = json.dumps(payload)
    sig = hmac.new(secret.encode(), body.encode(), hashlib.sha256).hexdigest()

    headers = {
        "X-Signature": f"sha256={sig}",
        "Content-Type": "application/json",
    }

    for attempt in range(3):
        try:
            r = requests.post(url, data=body, headers=headers, timeout=5)
            if r.ok:
                return
        except Exception:
            time.sleep(2 ** attempt)

SMS

from twilio.rest import Client

def alert_sms_emergency(to, query, new_rank):
    client = Client(TWILIO_SID, TWILIO_TOKEN)
    client.messages.create(
        from_="+1234567890",
        to=to,
        body=f"SEO: {query} 跌到 #{new_rank},立即查看!",
    )

实战数据 (我项目 1 个月)

5 个最佳实践

1. 多通道 fallback:Slack 主,Webhook 备,Email 三备,SMS 紧急
2. 去重防刷屏:5 分钟同 key 不重发
3. 限流防限速:Slack 1 秒/条
4. 异步不阻塞:主流程触发告警就返
5. 富文本优先:Slack 用 blocks,Email 用 HTML,展示清楚

小结

4 种告警方式选型:

• Slack:团队用 Slack(90% 团队)
• Email:非开发人员
• Webhook:接自己系统
• SMS:紧急高优 (成本高,慎用)

serpbase + 任意告警方式,5 分钟搭好。serpbase 的 1.4s P50 + auto-refund 让你告警永远及时,失败不扣 credit。