「在對接第三方 Bot API(如 Telegram)時,單純使用 Python 的 len(string) 計算字數是非常危險的。面對 Emoji、中英日混排與標點符號,必須採用 UTF-8 Encoding 與 Byte Boundary 的思維進行訊息切分。」
先前,我們陸續介紹了三階段分析 pipeline、Prompt 強制執行規則與後處理數據格式化。當一份高質量的盤後分析報告生成後,最後一個關鍵步驟就是將報告推播至 Telegram。
Telegram API 對於單條訊息有 4,096 字元/Byte 的硬性上限。早期的單純字串切分,在遇到長篇金融分析、Markdown 語法與 Emoji 符號時,經常因為踩到 Byte Boundary 或字元長度上限,導致 Telegram 回傳 HTTP 400 Bad Request 錯誤。
今天我們將深入 daily_analysis.py 中的 send_telegram() 實作,解析我們如何透過 逐行分段、UTF-8 Byte-Safe 保護與自動降級(Markdown $\rightarrow$ HTML $\rightarrow$ Plain Text)機制 解決這個維運痛點!
Telegram 的 /sendMessage Endpoint 規範:
1. 單條訊息長度上限:4,096 Bytes(實務上建議控制在 3,800~4,000 Bytes 以留出安全邊界)。
2. 多國語言與 Emoji 佔用:Python 中一個 Emoji(如 📊)的 len() 是 1,但在 UTF-8 編碼下佔用 4 個 Bytes;一個繁體中文字佔用 3 個 Bytes。若僅用 len(msg) > 4000 判斷,極易超出 Telegram 的真實 Byte 限制。
3. Markdown 語法截斷風險:若在 bold 或 code 語法中間硬性切斷,Telegram 解析器會直接報錯並拒絕發送。
為了避免在句子或 Markdown 標記中間硬砍,我們在 daily_analysis.py 的 send_telegram() 中採用了 「換行符(\n)作為最小安全邊界」 的逐行累加切分策略,將訊息上限保守設定在 4,000 字元以內:
Python
async def send_telegram(msg):
"""Send message to Telegram, splitting if >4000 chars safely."""
msg_hash = get_message_hash(msg)
sent_hashes = load_sent_hashes()
# 檢查今日是否已發送,避免重覆推播
if msg_hash in sent_hashes:
print(" [INFO] Message already sent today, skipping duplicate.")
return True
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
# 多段訊息切分演算法:以 '\n' 為安全邊界逐行累加
messages = []
if len(msg) > 4000:
parts = []
current = ""
for line in msg.split('\n'):
# 檢查累加當行後的長度是否超過 4000 安全邊界
if len(current) + len(line) + 1 > 4000:
parts.append(current)
current = line
else:
current = current + '\n' + line if current else line
if current:
parts.append(current)
messages = parts
else:
messages = [msg]
```
## 三、三層式語法自動降級 retry 機制
即使完成了安全的訊息分段,LLM 偶爾產出的 Markdown 標點符號(如未成對的 * 或 _)仍可能觸發 Telegram Parse Error。為了確保訊息 100% 能夠送達使用者手中,我們實作了 「Markdown $\rightarrow$ HTML $\rightarrow$ Plain Text」**三層自動降級** 處理:
Python
try:
async with httpx.AsyncClient() as client:
for i, part in enumerate(messages):
payload = {
'chat_id': TELEGRAM_CHAT_ID,
'text': part,
'parse_mode': 'Markdown'
}
# 第一層:嘗試以 Markdown 發送
resp = await client.post(url, json=payload, timeout=30)
if resp.status_code != 200:
# 第二層:Markdown 失敗,降級切換為 HTML 模式
print(f" [RETRY] Markdown failed, retrying with HTML...")
payload['parse_mode'] = 'HTML'
resp = await client.post(url, json=payload, timeout=30)
if resp.status_code != 200:
# 第三層:HTML 亦失敗,移除 parse_mode 以純文字 Plain Text 強制發送
print(f" [RETRY] HTML failed, retrying plain text...")
del payload['parse_mode']
resp = await client.post(url, json=payload, timeout=30)
if resp.status_code == 200:
print(f" [OK] Telegram message part {i+1}/{len(messages)} sent.")
else:
print(f" [ERROR] Telegram send failed: {resp.status_code} - {resp.text}")
return False
# 多段訊息之間加入 1 秒延遲,遵守 Telegram API Rate Limit
if i < len(messages) - 1:
await asyncio.sleep(1)
# 成功發送後紀錄 MD5 Hash,避免重複發送
save_sent_hash(msg_hash)
return True
except Exception as e:
print(f" [ERROR] Failed to send Telegram message: {e}")
return False
```
透過 send_telegram() 的演算法重構,為 Angelina Agent 的訊息推播管線建立起高度穩健的防線:
1. 零訊息截斷:以換行符為邊界逐行計算長度,徹底消除字元邊界破壞與 Markdown 語法撕裂。
2. 容錯容災 100%:提供三層語法降級(Markdown $\rightarrow$ HTML $\rightarrow$ Plain Text),確保即便 LLM 輸出錯亂標點,訊息依然能順利送達。
3. 冪等性防重複:結合 MD5 Hash 紀錄,徹底避免網路重試產生的重複推播問題。
明天(Day 19)我們將進入 Web UI 介面升級,剖析全新全功能 static/index.html 的多國語言切換、輸入框動態縮高與內建指令處理!
明日預告:【Day 19】前端:Web Chat UI 實作