kairee
71a0e8c1d6
Give getUpdates a wider HTTP timeout than the long-poll window, log transient SSL/network disconnects as warnings, apply exponential backoff with recovery logging, and only escalate to error after five consecutive failures.
129 lines
4.7 KiB
Python
129 lines
4.7 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""Telegram getUpdates long polling 与按钮回调分发。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import time
|
||
from typing import Any, Mapping
|
||
|
||
|
||
class TelegramPoller:
|
||
def __init__(self, bot, actions, *, logger=None, sleeper=time.sleep) -> None:
|
||
self.bot = bot
|
||
self.actions = actions
|
||
self.logger = logger
|
||
self._sleep = sleeper
|
||
|
||
def _log(self, level: str, event: str, **fields) -> None:
|
||
if self.logger is not None:
|
||
getattr(self.logger, level)(event, **fields)
|
||
|
||
def run_once(self, offset: int | None = None) -> int | None:
|
||
updates = self.bot.get_updates(offset)
|
||
for update in updates:
|
||
update_id = update.get("update_id")
|
||
if isinstance(update_id, int) and not isinstance(update_id, bool):
|
||
offset = max(offset or 0, update_id + 1)
|
||
callback = update.get("callback_query")
|
||
if isinstance(callback, Mapping):
|
||
self.handle_callback(callback)
|
||
return offset
|
||
|
||
def handle_callback(self, callback: Mapping[str, Any]) -> None:
|
||
callback_id = str(callback.get("id") or "")
|
||
message = callback.get("message") or {}
|
||
chat_id = (message.get("chat") or {}).get("id")
|
||
from_id = (callback.get("from") or {}).get("id")
|
||
if chat_id != self.bot.chat_id or from_id != self.bot.chat_id:
|
||
self._log(
|
||
"warning", "telegram.unauthorized",
|
||
chat_id=chat_id, from_id=from_id,
|
||
)
|
||
self._answer(callback_id, "无权限")
|
||
return
|
||
|
||
action, _, raw_aid = str(callback.get("data") or "").partition(":")
|
||
try:
|
||
aid = int(raw_aid)
|
||
except (TypeError, ValueError):
|
||
self._answer(callback_id, "未知操作")
|
||
return
|
||
|
||
if action == "fav":
|
||
result = self.actions.favorite(aid)
|
||
elif action == "ign":
|
||
result = self.actions.ignore(aid)
|
||
elif action == "blk":
|
||
result = self.actions.blacklist(aid)
|
||
else:
|
||
self._answer(callback_id, "未知操作")
|
||
return
|
||
|
||
text = result.message
|
||
if result.ok and result.position is not None:
|
||
text = f"{text}（位置 #{result.position}）"
|
||
self._log(
|
||
"info", "telegram.action",
|
||
action=action, aid=aid, ok=result.ok, state=result.state,
|
||
position=result.position,
|
||
)
|
||
if not self._answer(callback_id, text):
|
||
# 回调 ID 过期时 answerCallbackQuery 会 400；改用普通消息告知结果。
|
||
try:
|
||
self.bot.send_text(text)
|
||
except Exception as exc: # noqa: BLE001
|
||
self._log("warning", "telegram.fallback_failed", error=str(exc))
|
||
message_id = message.get("message_id")
|
||
if isinstance(message_id, int) and not isinstance(message_id, bool):
|
||
try:
|
||
self.bot.edit_reply_markup(message_id)
|
||
except Exception as exc: # noqa: BLE001
|
||
self._log("warning", "telegram.edit_failed", error=str(exc))
|
||
|
||
def _answer(self, callback_id: str, text: str) -> bool:
|
||
try:
|
||
self.bot.answer_callback(callback_id, text)
|
||
return True
|
||
except Exception as exc: # noqa: BLE001 过期回调不应升级为错误
|
||
self._log("warning", "telegram.answer_failed", error=str(exc))
|
||
return False
|
||
|
||
def run_forever(
|
||
self,
|
||
stop_event=None,
|
||
*,
|
||
initial_delay: float = 3.0,
|
||
max_delay: float = 60.0,
|
||
max_iterations: int | None = None,
|
||
) -> None:
|
||
offset: int | None = None
|
||
failures = 0
|
||
delay = initial_delay
|
||
iterations = 0
|
||
while stop_event is None or not stop_event.is_set():
|
||
try:
|
||
offset = self.run_once(offset)
|
||
except Exception as exc: # noqa: BLE001 长轮询瞬时断连后退避重试
|
||
failures += 1
|
||
level = "error" if failures >= 5 else "warning"
|
||
self._log(
|
||
level, "telegram.poll_failed",
|
||
error=str(exc), failures=failures, retry_in=delay,
|
||
)
|
||
if stop_event is not None:
|
||
if stop_event.wait(delay):
|
||
break
|
||
else:
|
||
self._sleep(delay)
|
||
delay = min(delay * 2, max_delay)
|
||
else:
|
||
if failures:
|
||
self._log(
|
||
"info", "telegram.poll_recovered", failures=failures
|
||
)
|
||
failures = 0
|
||
delay = initial_delay
|
||
iterations += 1
|
||
if max_iterations is not None and iterations >= max_iterations:
|
||
break
|