將接碼平台集成到 Slack:每當收到驗證碼,頻道自動提醒
一個在團隊 Slack 裡親手部署了這個集成的 DevOps 工程師的實戰教學——每條命令都能跑通,每個參數都帶著註釋,每個坑都替你踩過。
本教程目標:用一條 Webhook、一段 Python 腳本、一個簡單的輪詢邏輯,把接碼平台的短信通知無縫集成到團隊 Slack 工作流——讓 Slack 頻道自動爆出驗證碼,人人可見,無需切換設備。
📑 目錄
一、完整數據流架構圖
以下是整個集成方案的數據流架構,三個環節各司其職:
各環節角色:接碼平台提供短信數據(輪詢 API),解析腳本負責輪詢、提取驗證碼和去重,Slack Webhook 接收格式化後的消息並推送到指定頻道。整個流程無需任何人手動干預,從短信到達到 Slack 彈出消息的延遲通常在 8 秒以內。
二、Step 1:創建 Slack Incoming Webhook
這是整個集成的入口——Slack 需要一個 Webhook URL 來接收外部服務推送的消息。
操作步驟
- 前往 api.slack.com/apps,點擊 「Create New App」,選擇 「From scratch」。
- 輸入 App 名稱(如
SMS Bot),選擇目標 Workspace,點擊 「Create App」。 - 在左側選單中找到 「Incoming Webhooks」,點擊開關將其啟動。
- 點擊 「Activate Incoming Webhooks」,然後點擊 「Add New Webhook to Workspace」。
- 選擇一個目標頻道——強烈建議建立專用的
#verification-codes頻道,避免打擾主頻道。 - 複製生成的 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 中排除任何包含密鑰的配置文件。生產環境建議使用 systemd 的 EnvironmentFile 或 Docker 的 --env-file 管理。
結語:從「手機收到驗證碼」到「頻道自動彈出」的團隊工作流升級
這套集成為 QA 團隊帶來的真實價值:
- 消除溝通成本:不再需要有人喊「驗證碼發我」,也不再用共享測試機排隊看短信。
- 可追溯可審計:每一條驗證碼的推送時間、號碼、短信內容都有完整記錄。
- 移動端即時接收:團隊成員可在 Slack 手機 App 上即時收到驗證碼通知,無需守在電腦前。
從「手機收到驗證碼」到「頻道自動彈出」,不是技術的炫耀,而是讓團隊的每一秒都花在真正有價值的事情上。