135 lines
4.6 KiB
Python
135 lines
4.6 KiB
Python
"""Telegram Bot API helpers for OpenClaw Bot.
|
||
|
||
Route modules should keep only request handling and command dispatch. Telegram
|
||
delivery, Markdown fallback, and file upload glue live here so they can be
|
||
reused and tested without importing the 5k-line Blueprint.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import re
|
||
|
||
import requests
|
||
|
||
from services.logger_manager import SystemLogger
|
||
|
||
|
||
BOT_TOKEN = os.getenv("OPENCLAW_BOT_TOKEN", "")
|
||
BOT_API_URL = f"https://api.telegram.org/bot{BOT_TOKEN}"
|
||
sys_log = SystemLogger("OpenClawBot").get_logger()
|
||
|
||
|
||
def _tg(method: str, payload: dict):
|
||
try:
|
||
response = requests.post(f"{BOT_API_URL}/{method}", json=payload, timeout=10)
|
||
if not response.ok:
|
||
sys_log.warning(f"[OpenClawBot] {method} failed: {response.text[:200]}")
|
||
return response.json()
|
||
except Exception as exc:
|
||
sys_log.error(f"[OpenClawBot] {method} error: {exc}")
|
||
return {}
|
||
|
||
|
||
def _strip_markdown(text: str) -> str:
|
||
"""移除 Telegram Markdown v1 格式符號，轉為純文字（send fallback 用）"""
|
||
text = re.sub(r"\*([^*]+)\*", r"\1", text)
|
||
text = re.sub(r"_([^_]+)_", r"\1", text)
|
||
text = re.sub(r"`([^`]+)`", r"\1", text)
|
||
text = re.sub(r"\[([^\]]+)\]\([^)]+\)", r"\1", text)
|
||
return text
|
||
|
||
|
||
def send_message(chat_id, text, reply_to=None, keyboard=None, parse_mode="Markdown"):
|
||
"""送出 Telegram 訊息。Markdown 解析失敗時自動降級為純文字重送。"""
|
||
|
||
def _build_payload(txt, pm):
|
||
payload = {"chat_id": chat_id, "text": txt, "disable_web_page_preview": True}
|
||
if pm:
|
||
payload["parse_mode"] = pm
|
||
if reply_to:
|
||
payload["reply_to_message_id"] = reply_to
|
||
if keyboard:
|
||
payload["reply_markup"] = {"inline_keyboard": keyboard}
|
||
return payload
|
||
|
||
result = _tg("sendMessage", _build_payload(text, parse_mode))
|
||
if result.get("ok"):
|
||
return result
|
||
|
||
if parse_mode and not result.get("ok"):
|
||
err = result.get("description", "")
|
||
if "parse" in err.lower() or "entity" in err.lower() or "can't find" in err.lower():
|
||
sys_log.warning("[OpenClawBot] Markdown parse failed, retrying as plain text")
|
||
result2 = _tg("sendMessage", _build_payload(_strip_markdown(text), None))
|
||
if result2.get("ok"):
|
||
return result2
|
||
|
||
if len(text) > 4000:
|
||
sys_log.warning(f"[OpenClawBot] Message too long ({len(text)}), truncating")
|
||
truncated = _strip_markdown(text[:3900]) + "\n...(訊息過長已截斷)"
|
||
return _tg("sendMessage", _build_payload(truncated, None))
|
||
|
||
return result
|
||
|
||
|
||
def answer_callback(cq_id, text=""):
|
||
return _tg("answerCallbackQuery", {"callback_query_id": cq_id, "text": text})
|
||
|
||
|
||
def send_typing(chat_id):
|
||
try:
|
||
requests.post(
|
||
f"{BOT_API_URL}/sendChatAction",
|
||
json={"chat_id": chat_id, "action": "typing"},
|
||
timeout=5,
|
||
)
|
||
except Exception as exc:
|
||
sys_log.debug(f"[OpenClawBot] sendChatAction skipped: {exc}")
|
||
|
||
|
||
def send_photo(chat_id, file_path, caption="", reply_to=None):
|
||
"""透過 Telegram Bot API 傳送圖片"""
|
||
try:
|
||
with open(file_path, "rb") as handle:
|
||
data = {"chat_id": chat_id}
|
||
if caption:
|
||
data["caption"] = caption
|
||
if reply_to:
|
||
data["reply_to_message_id"] = reply_to
|
||
response = requests.post(
|
||
f"{BOT_API_URL}/sendPhoto",
|
||
data=data,
|
||
files={"photo": handle},
|
||
timeout=30,
|
||
)
|
||
if not response.ok:
|
||
sys_log.warning(f"[OpenClawBot] sendPhoto failed: {response.text[:200]}")
|
||
return response.json()
|
||
except Exception as exc:
|
||
sys_log.error(f"[OpenClawBot] sendPhoto error: {exc}")
|
||
return {}
|
||
|
||
|
||
def send_document(chat_id, file_path, caption="", reply_to=None):
|
||
"""透過 Telegram Bot API 傳送文件"""
|
||
try:
|
||
with open(file_path, "rb") as handle:
|
||
data = {"chat_id": chat_id}
|
||
if caption:
|
||
data["caption"] = caption
|
||
if reply_to:
|
||
data["reply_to_message_id"] = reply_to
|
||
response = requests.post(
|
||
f"{BOT_API_URL}/sendDocument",
|
||
data=data,
|
||
files={"document": handle},
|
||
timeout=30,
|
||
)
|
||
if not response.ok:
|
||
sys_log.warning(f"[OpenClawBot] sendDocument failed: {response.text[:200]}")
|
||
return response.json()
|
||
except Exception as exc:
|
||
sys_log.error(f"[OpenClawBot] sendDocument error: {exc}")
|
||
return {}
|