"""Telegram Bot 消息发送(同步、带超时;失败不抛异常,返回 False)。""" from __future__ import annotations import logging import requests logger = logging.getLogger("monitor.telegram") API_BASE = "https://api.telegram.org/bot{token}/sendMessage" def send_message(bot_token: str, chat_id: str, text: str) -> bool: """发送一条消息。配置缺失或失败时记日志并返回 False。""" if not bot_token or not chat_id: logger.warning("Telegram 未配置(bot_token/chat_id 为空),跳过发送: %s", text[:60]) return False try: resp = requests.post( API_BASE.format(token=bot_token), json={"chat_id": chat_id, "text": text}, timeout=10, ) if resp.status_code == 200: logger.info("Telegram 消息已发送(%d 字符)", len(text)) return True logger.error("Telegram 发送失败 status=%s body=%s", resp.status_code, resp.text[:200]) except requests.RequestException as exc: logger.error("Telegram 请求异常: %s", exc) return False