🏭 第四課：Telegram 實戰篇
這次我們要把包裹寄到外面的世界！
徹底搞懂「非同步物流車 (aiohttp)」、「自動關門包廂」與「廠長監視器」。
這是一支具備「格式轉換」與「失敗重試」機制的企業級實戰腳本。它能將對話發送到 Telegram，就算遇到格式錯誤也能聰明應對！
from pydantic import BaseModel, Field from typing import Optional import aiohttp # 🚚 新工具：非同步物流車隊 import re class Action: # (省略 Valves 設定與 __init__) async def action(self, body: dict, __event_emitter__=None) -> Optional[dict]: # ✂️ 1. 清理氣泡紙與 Markdown 包裝轉換 message_content = re.sub(r'\*\*(.*?)\*\*', r'*\1*', message_content) # 🤝 2. 與物流車隊簽訂限時派車合約 (async with 聰明包廂) async with aiohttp.ClientSession() as session: async with session.post(api_url, json=payload) as response: result = await response.json() status = response.status # 🛡️ 3. 備用方案：如果包裝不符被 Telegram 海關退件 (400) if status == 400: del payload["parse_mode"] # 撕掉 Markdown 標籤 # 用純文字重新寄送一次！ async with session.post(...) as fallback: ...
async with：自動關門的聰明包廂
以前用的 requests 像傳統郵局，工人要死等（卡單）。
aiohttp 是非同步車隊。但開網路連線就像開門進無塵室，如果工人忘記關門 (session.close())，工廠資源很快就會耗盡爆炸。
async with 是一個「聰明包廂」。你只要在縮排區塊內做事，一旦離開這個區塊，系統會自動幫你關門、斷開連線，安全又省事！
# 🚪 進入聰明包廂 (開啟 Session) async with aiohttp.ClientSession() as session: # 🚚 派車，並設定 timeout=10 避免塞車死等 async with session.post(url, json=data, timeout=10) as res: result = await res.json() # 離開縮排區塊，包廂門自動關閉！資源釋放！
Try...Except 與 print 監視器
try 是一層防爆罩，萬一 Telegram 連線失敗，會在 except 攔截下來，保護整個工廠不當機。
遇到錯誤時，廠長通常會做兩件事：
1. 用 __event_emitter__ 廣播跟客人說抱歉。
2. 用 print(e) 廠長監視器 把真正的錯誤原因印在後台，方便自己偷偷抓蟲 (Debug)！
try: # 執行對外寄送任務 async with session.post(...) except Exception as e: # 📺 廠長監視器：把錯誤詳細印在後台給自己看 print(f"物流車禍發生原因：{e}") # 📻 廣播器：給客人的友善通知 if __event_emitter__: await __event_emitter__({"content": "系統錯誤"})