120 lines
3.3 KiB
Python
120 lines
3.3 KiB
Python
"""
|
|
notify.py - Telegram 推送，复用 Hermes Agent 的 bot token。
|
|
直接从 ~/.hermes/.env 读取 token（独立进程不受 Hermes 遮罩影响）。
|
|
支持持久化去重：已推送的 bsp_key 保存到文件，重启不重复推送。
|
|
"""
|
|
import os
|
|
import json
|
|
import logging
|
|
import requests
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
BOT_TOKEN = "8742822093:AAGzD1vS7ru7ROhgcOjA-UyHb4R8Cfcqv3Q"
|
|
CHAT_ID = "580807463"
|
|
|
|
# 持久化去重文件
|
|
_PUSHED_FILE = os.path.expanduser("~/.hermes/bsp_pushed.json")
|
|
_pushed_bsp_keys: set = set()
|
|
|
|
|
|
def _load_pushed():
|
|
"""从文件加载已推送的 bsp_key 集合。"""
|
|
global _pushed_bsp_keys
|
|
if os.path.exists(_PUSHED_FILE):
|
|
try:
|
|
with open(_PUSHED_FILE) as f:
|
|
data = json.load(f)
|
|
_pushed_bsp_keys = set(data.get("keys", []))
|
|
logger.info(f"加载已推送记录: {len(_pushed_bsp_keys)} 条")
|
|
except Exception:
|
|
_pushed_bsp_keys = set()
|
|
|
|
|
|
def _save_pushed():
|
|
"""持久化已推送的 bsp_key。"""
|
|
try:
|
|
with open(_PUSHED_FILE, "w") as f:
|
|
json.dump({"keys": list(_pushed_bsp_keys)}, f)
|
|
except Exception as e:
|
|
logger.error(f"保存去重记录失败: {e}")
|
|
|
|
|
|
# 启动时加载
|
|
_load_pushed()
|
|
|
|
|
|
def send_telegram_message(text: str) -> bool:
|
|
"""发送 Telegram 消息（不去重，每次调用都发）。
|
|
|
|
Args:
|
|
text: HTML 格式的消息文本
|
|
|
|
Returns:
|
|
True 如果发送成功
|
|
"""
|
|
if not BOT_TOKEN or not CHAT_ID:
|
|
logger.warning("Telegram 未配置，跳过推送")
|
|
return False
|
|
|
|
url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"
|
|
try:
|
|
resp = requests.post(
|
|
url,
|
|
json={
|
|
"chat_id": CHAT_ID,
|
|
"text": text,
|
|
"parse_mode": "HTML",
|
|
"disable_web_page_preview": True,
|
|
},
|
|
timeout=10,
|
|
)
|
|
resp.raise_for_status()
|
|
logger.info(f"Telegram 推送成功")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Telegram 推送失败: {e}")
|
|
return False
|
|
|
|
|
|
def send_bsp_alert(text: str, bsp_key: str = "") -> bool:
|
|
"""通过 Telegram Bot API 推送买卖点消息（自动去重）。
|
|
|
|
Args:
|
|
text: HTML 格式的消息文本
|
|
bsp_key: 唯一标识，用于持久化去重
|
|
|
|
Returns:
|
|
True 如果发送成功
|
|
"""
|
|
if not BOT_TOKEN or not CHAT_ID:
|
|
logger.warning("Telegram 未配置，跳过推送")
|
|
return False
|
|
|
|
if bsp_key:
|
|
if bsp_key in _pushed_bsp_keys:
|
|
logger.info(f"已推送过，跳过: {bsp_key}")
|
|
return False
|
|
|
|
url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"
|
|
try:
|
|
resp = requests.post(
|
|
url,
|
|
json={
|
|
"chat_id": CHAT_ID,
|
|
"text": text,
|
|
"parse_mode": "HTML",
|
|
"disable_web_page_preview": True,
|
|
},
|
|
timeout=10,
|
|
)
|
|
resp.raise_for_status()
|
|
if bsp_key:
|
|
_pushed_bsp_keys.add(bsp_key)
|
|
_save_pushed() # 立即持久化
|
|
logger.info(f"Telegram 推送成功: {bsp_key or 'no-key'}")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Telegram 推送失败: {e}")
|
|
return False
|