將接碼平台集成到 Slack:每當收到驗證碼,頻道自動提醒

一個在團隊 Slack 裡親手部署了這個集成的 DevOps 工程師的實戰教學——每條命令都能跑通,每個參數都帶著註釋,每個坑都替你踩過。

真實團隊協作痛點:QA 團隊在測試,數雙眼睛盯着一部共享測試機,或某人喊「驗證碼發我」——效率極低,訊息傳遞延遲可能長達數分鐘。
本教程目標:用一條 Webhook、一段 Python 腳本、一個簡單的輪詢邏輯,把接碼平台的短信通知無縫集成到團隊 Slack 工作流——讓 Slack 頻道自動爆出驗證碼,人人可見,無需切換設備。

一、完整數據流架構圖

以下是整個集成方案的數據流架構,三個環節各司其職:

[接碼平台 API] ──輪詢──▶ [Python 腳本] ──格式化──▶ [Slack Incoming Webhook] ──推送──▶ [Slack 頻道] │ │ │ │ 5sim / SMSPool 提取驗證碼 + 去重 Block Kit 消息卡片 #verification-codes │ │ │ │ 每 5 秒查詢 記錄已處理 ID POST 請求 團隊全員可見

各環節角色:接碼平台提供短信數據(輪詢 API),解析腳本負責輪詢、提取驗證碼和去重,Slack Webhook 接收格式化後的消息並推送到指定頻道。整個流程無需任何人手動干預,從短信到達到 Slack 彈出消息的延遲通常在 8 秒以內。

二、Step 1:創建 Slack Incoming Webhook

這是整個集成的入口——Slack 需要一個 Webhook URL 來接收外部服務推送的消息。

操作步驟

  1. 前往 api.slack.com/apps,點擊 「Create New App」,選擇 「From scratch」
  2. 輸入 App 名稱(如 SMS Bot),選擇目標 Workspace,點擊 「Create App」
  3. 在左側選單中找到 「Incoming Webhooks」,點擊開關將其啟動。
  4. 點擊 「Activate Incoming Webhooks」,然後點擊 「Add New Webhook to Workspace」
  5. 選擇一個目標頻道——強烈建議建立專用的 #verification-codes 頻道,避免打擾主頻道。
  6. 複製生成的 Webhook URL(格式為 https://hooks.slack.com/services/T00000000/B00000000/xxxxxxxxxxxxxxxxxxxxxxxx),保存備用。

測試連通性

用一條 cURL 命令測試 Webhook 是否正常運作:

curl -X POST -H 'Content-type: application/json' \
  --data '{"text":"✅ Webhook 連接成功!驗證碼通知將在此頻道推送。"}' \
  https://hooks.slack.com/services/YOUR/WEBHOOK/URL

如果 Slack 頻道中出現消息,說明 Webhook 已正常運作。

三、Step 2:編寫短信輪詢與推送腳本

以下是完整的 Python 腳本(約 80 行),涵蓋了輪詢接碼平台、提取驗證碼、格式化 Slack 消息和去重機制。以 5sim API 為例,可根據實際使用的平台替換 API 端點。

import requests
import time
import re
import json
import os
from datetime import datetime

API_KEY = os.getenv('FIVESIM_API_KEY', 'your_api_key_here')
SLACK_WEBHOOK = os.getenv('SLACK_WEBHOOK_URL', 'https://hooks.slack.com/services/...')
CHECK_INTERVAL = 5
MAX_WAIT_SECONDS = 120
PROCESSED_IDS_FILE = '/tmp/processed_sms_ids.txt'

def load_processed_ids():
    try:
        with open(PROCESSED_IDS_FILE, 'r') as f:
            return set(line.strip() for line in f if line.strip())
    except FileNotFoundError:
        return set()

def save_processed_id(sms_id):
    with open(PROCESSED_IDS_FILE, 'a') as f:
        f.write(f"{sms_id}\n")

def send_to_slack(phone, sms_text, otp, platform, timestamp):
    blocks = [
        {
            "type": "header",
            "text": {"type": "plain_text", "text": f"📩 新驗證碼 | {platform}"}
        },
        {
            "type": "section",
            "fields": [
                {"type": "mrkdwn", "text": f"*號碼:*\n{phone}"},
                {"type": "mrkdwn", "text": f"*時間:*\n{timestamp}"}
            ]
        },
        {
            "type": "section",
            "text": {"type": "mrkdwn", "text": f"*短信內容:*\n```{sms_text}```"}
        },
        {
            "type": "section",
            "text": {"type": "mrkdwn", "text": f"🔑 *驗證碼: `{otp}`*"}
        },
        {"type": "divider"}
    ]
    payload = {"blocks": blocks}
    try:
        resp = requests.post(SLACK_WEBHOOK, json=payload, timeout=10)
        if resp.status_code == 429:
            retry_after = int(resp.headers.get('Retry-After', 5))
            time.sleep(retry_after)
            requests.post(SLACK_WEBHOOK, json=payload, timeout=10)
    except Exception as e:
        print(f"Slack 推送失敗: {e}")

def poll_sms(max_wait=MAX_WAIT_SECONDS):
    processed = load_processed_ids()
    start_time = time.time()
    print(f"🔄 開始輪詢短信 (最長等待 {max_wait} 秒)...")

    while time.time() - start_time < max_wait:
        try:
            resp = requests.get(
                "https://5sim.net/v1/user/orders",
                headers={"Authorization": f"Bearer {API_KEY}"},
                timeout=10
            )
            data = resp.json()
        except Exception as e:
            print(f"API 請求失敗: {e},5秒後重試...")
            time.sleep(CHECK_INTERVAL)
            continue

        for order in data.get('orders', []):
            order_id = order.get('id')
            if order.get('status') == 'RECEIVED' and order_id not in processed:
                sms_list = order.get('sms', [])
                for sms in sms_list:
                    sms_id = sms.get('id')
                    sms_text = sms.get('text', '')
                    if sms_id not in processed:
                        match = re.search(r'\b\d{4,6}\b', sms_text)
                        otp = match.group(0) if match else '未識別'
                        phone = order.get('phone', '未知')
                        ts = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                        send_to_slack(phone, sms_text, otp, '5sim', ts)
                        save_processed_id(sms_id)
                        processed.add(sms_id)
                        print(f"✅ 已推送: {otp} | {sms_text[:50]}...")
        time.sleep(CHECK_INTERVAL)
    print("⏰ 輪詢超時,腳本退出。")

if __name__ == '__main__':
    poll_sms()

關鍵模組拆解

模組功能關鍵參數
API 輪詢每 5 秒查詢一次 5sim 訂單狀態CHECK_INTERVAL = 5
驗證碼提取用正則 \b\d{4,6}\b 捕獲 4-6 位數字re.search()
消息格式化使用 Slack Block Kit 生成美觀的消息卡片blocks JSON 結構
去重機制以文件記錄已處理短信 ID,避免重複推送/tmp/processed_sms_ids.txt
超時控制120 秒後自動退出,防止腳本永久掛起MAX_WAIT_SECONDS = 120
速率限制處理收到 HTTP 429 時自動讀取 Retry-After 頭並等待指數退避

四、Step 3:部署為後台服務

以下是三種部署方案,依推薦度排序。

方案一:systemd 服務(Linux 伺服器)

建立服務文件 /etc/systemd/system/sms-slack-bot.service

[Unit]
Description=SMS to Slack Bot
After=network.target

[Service]
Type=simple
User=deploy
WorkingDirectory=/opt/sms-bot
Environment="FIVESIM_API_KEY=your_key"
Environment="SLACK_WEBHOOK_URL=https://hooks.slack.com/services/..."
ExecStart=/usr/bin/python3 /opt/sms-bot/sms_to_slack.py
Restart=on-failure
RestartSec=10

[Install]
WantedBy=multi-user.target

啟動服務:

sudo systemctl daemon-reload
sudo systemctl enable sms-slack-bot
sudo systemctl start sms-slack-bot
sudo systemctl status sms-slack-bot  # 檢查運行狀態

方案二:Docker 容器

建立 Dockerfile

FROM python:3.12-slim
WORKDIR /app
COPY sms_to_slack.py .
RUN pip install requests
CMD ["python", "sms_to_slack.py"]

一行命令部署:

docker run -d --name sms-slack-bot \
  --restart always \
  -e FIVESIM_API_KEY=your_key \
  -e SLACK_WEBHOOK_URL=https://hooks.slack.com/services/... \
  -v /opt/sms-bot/processed:/tmp \
  sms-slack-bot:latest

方案三:GitHub Actions 定時任務(無伺服器方案)

建立 .github/workflows/sms-to-slack.yml

on:
  schedule:
    - cron: '*/5 * * * *'   # 每 5 分鐘觸發一次
  workflow_dispatch:

jobs:
  poll-and-notify:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      - run: pip install requests
      - run: python sms_to_slack.py
        env:
          FIVESIM_API_KEY: ${{ secrets.FIVESIM_API_KEY }}
          SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}

五、進階玩法:Slack 交互式驗證碼管理

在基礎推送之上,可以在 Slack 消息卡片中增加交互按鈕,讓團隊成員直接在 Slack 內完成驗證碼的狀態管理。

增加交互按鈕

在 Block Kit 消息的 blocks 中增加 actions 區塊:

{
    "type": "actions",
    "elements": [
        {
            "type": "button",
            "text": {"type": "plain_text", "text": "✅ 標記已用"},
            "style": "primary",
            "value": sms_id,
            "action_id": "mark_used"
        },
        {
            "type": "button",
            "text": {"type": "plain_text", "text": "🔄 重新獲取"},
            "style": "danger",
            "value": order_id,
            "action_id": "reacquire"
        },
        {
            "type": "button",
            "text": {"type": "plain_text", "text": "📋 查看歷史"},
            "url": f"https://your-dashboard.com/logs?phone={phone}",
            "action_id": "view_history"
        }
    ]
}

處理交互回調

需要額外部署一個小型 Flask 服務來接收 Slack 的交互回調(當用戶點擊按鈕時 Slack 會向你的服務器發送 POST 請求):

from flask import Flask, request, jsonify
app = Flask(__name__)

@app.route('/slack/interactive', methods=['POST'])
def handle_interactive():
    payload = json.loads(request.form['payload'])
    action_id = payload['actions'][0]['action_id']
    value = payload['actions'][0]['value']

    if action_id == 'mark_used':
        mark_as_used(value)
        return jsonify({"text": f"✅ 驗證碼 {value} 已標記為已用"})
    elif action_id == 'reacquire':
        trigger_new_number(value)
        return jsonify({"text": "🔄 已觸發重新獲取號碼"})
    return jsonify({"text": "未知操作"})

if __name__ == '__main__':
    app.run(port=5000)

建立使用日誌

每次驗證碼被標記為「已用」時,寫入一條日誌記錄,方便團隊追溯:

def mark_as_used(sms_id):
    with open('/var/log/sms_usage.log', 'a') as f:
        f.write(f"{datetime.now().isoformat()} | {sms_id} | marked_used\n")
    # 同步更新 Redis 或數據庫中的狀態

六、排坑指南

坑一:Slack 消息推送頻率限制

症狀:腳本運行正常,但 Slack 頻道中偶爾出現消息缺失,後台日誌顯示 HTTP 429 Too Many Requests。

解法:Slack 對單個頻道的 Incoming Webhook 限制為每秒最多 1 條消息。輪詢間隔不應低於 3 秒。當收到 429 回應時,必須讀取回應頭中的 Retry-After 字段並暫停對應秒數。代碼中已內建指數退避重試邏輯。

坑二:接碼平台 API 回應慢或超時

症狀:輪詢循環卡在某次 API 請求上,整個腳本凍結,後續短信無法被推送。

解法:每次 API 請求必須設置超時(timeout=10 秒),捕獲 requests.exceptions.Timeout 異常後自動重試。重試最多 3 次,失敗後跳過本輪並在 5 秒後繼續下一輪。

坑三:消息重複推送

症狀:同一條短信在 Slack 頻道中出現多次,團隊成員抱怨被刷屏。

解法:使用文件或 Redis 記錄已處理的短信 ID。腳本中使用了 /tmp/processed_sms_ids.txt 文件方案(適合單實例部署)。如果是多實例部署,應改用 Redis 的 SADD 命令進行原子去重。

坑四:環境變數洩漏

症狀:API Key 或 Webhook URL 被硬編碼在腳本中,推送 GitHub 後被惡意掃描機器人抓取。

解法:全部敏感資訊通過環境變數傳入,腳本中只使用 os.getenv() 讀取。在 .gitignore 中排除任何包含密鑰的配置文件。生產環境建議使用 systemdEnvironmentFile 或 Docker 的 --env-file 管理。


結語:從「手機收到驗證碼」到「頻道自動彈出」的團隊工作流升級

這套集成為 QA 團隊帶來的真實價值:

行動建議:先用 10 分鐘完成 Step 1(創建 Webhook)和 Step 2(運行腳本),搭建基礎版本體驗效果。當團隊習慣了頻道自動彈出驗證碼後,再按需增加進階的交互按鈕和使用日誌功能。
從「手機收到驗證碼」到「頻道自動彈出」,不是技術的炫耀,而是讓團隊的每一秒都花在真正有價值的事情上。