RRRosmontis
34089d0042
Include utility modules, shared channels, prompts, read-only skills, wardrobe memory and regression tests. Externalize legacy credentials and exclude private runtime state.
135 lines
5.1 KiB
Python
135 lines
5.1 KiB
Python
# tg_watchdog.py — TG 长轮询自愈看门狗
|
||
# 背景: bj -> US 的 xray 隧道会僵死/重置（TCP 看似 ESTABLISHED，更新被静默吞掉，
|
||
# aiogram 侧可能只留零星报错）。检测到隧道不健康、Telegram 侧更新积压、
|
||
# 或取更新失败风暴时，停止轮询并让进程退出，由 nssm 自动拉起（新连接）。
|
||
#
|
||
# 日志: 同时写 stdout 与同目录 tg_watchdog.log（逐行 flush）。
|
||
# 强制退出走 os._exit，stdout 缓冲会丢，因此必须落文件。
|
||
import asyncio
|
||
import logging
|
||
import os
|
||
import sys
|
||
import time
|
||
|
||
WATCHDOG_INTERVAL = 120 # 检查间隔（秒）
|
||
WATCHDOG_LIMIT = 3 # 连续异常阈值（约 6 分钟）
|
||
FETCH_FAIL_LIMIT = 5 # 单个检查窗口内取更新失败次数上限（风暴判定）
|
||
API_TIMEOUT = 20 # 单次 API 检查超时（秒）
|
||
GRACE_SECONDS = 25 # 优雅停止的等待时间，超时则强制退出
|
||
|
||
LOG_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "tg_watchdog.log")
|
||
|
||
|
||
def _log(msg: str):
|
||
"""写入 stdout 与独立日志文件（即时 flush，强制退出也不丢）"""
|
||
try:
|
||
print(msg)
|
||
except Exception:
|
||
pass
|
||
try:
|
||
with open(LOG_FILE, "a", encoding="utf-8") as f:
|
||
f.write("[" + time.strftime("%Y-%m-%d %H:%M:%S") + "] " + msg + "\n")
|
||
f.flush()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
class _FetchFailureCounter(logging.Handler):
|
||
"""统计 aiogram 的取更新失败次数（长轮询被重置的直接信号）"""
|
||
|
||
def __init__(self):
|
||
super().__init__(level=logging.WARNING)
|
||
self.count = 0
|
||
|
||
def emit(self, record):
|
||
try:
|
||
if "Failed to fetch updates" in record.getMessage():
|
||
self.count += 1
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def install_fetch_counter():
|
||
"""挂到 aiogram.dispatcher logger 上，返回计数器"""
|
||
counter = _FetchFailureCounter()
|
||
logging.getLogger("aiogram.dispatcher").addHandler(counter)
|
||
return counter
|
||
|
||
|
||
async def _notify_admins(text: str):
|
||
"""尽力通知各管理员统一身份（TG 可能已死，靠 QQ 侧镜像送达）。
|
||
|
||
按 uid 扇出（TG 私聊 + 已绑定 QQ 私聊）；每个目标保留 8s 超时预算，
|
||
避免僵死连接阻塞看门狗的重建流程。
|
||
"""
|
||
try:
|
||
import identity
|
||
from channel import push_to_user
|
||
for uid in identity.admin_uids():
|
||
targets = identity.push_targets(uid)
|
||
try:
|
||
await asyncio.wait_for(
|
||
push_to_user(uid, text, tag="TG 看门狗", targets=targets),
|
||
timeout=8 * max(1, len(targets)))
|
||
except Exception:
|
||
pass
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
async def tg_watchdog(bot, dp, name: str = "utility"):
|
||
"""监控 TG 长轮询健康度；异常达阈值则重建连接（进程退出 -> nssm 拉起）"""
|
||
counter = install_fetch_counter()
|
||
_log(f"已启动: 服务={name}, 间隔 {WATCHDOG_INTERVAL}s, 连续异常阈值 {WATCHDOG_LIMIT} 次, "
|
||
f"取更新失败阈值 {FETCH_FAIL_LIMIT} 次/窗口, API 超时 {API_TIMEOUT}s")
|
||
fails = 0
|
||
backlog = 0
|
||
while True:
|
||
await asyncio.sleep(WATCHDOG_INTERVAL)
|
||
fetch_fails = counter.count
|
||
counter.count = 0
|
||
if fetch_fails:
|
||
_log(f"本窗口取更新失败 {fetch_fails} 次"
|
||
+ ("（达风暴阈值）" if fetch_fails >= FETCH_FAIL_LIMIT else ""))
|
||
|
||
try:
|
||
await asyncio.wait_for(bot.get_me(), timeout=API_TIMEOUT)
|
||
fails = 0
|
||
except Exception as e:
|
||
fails += 1
|
||
_log(f"getMe 失败 {fails}/{WATCHDOG_LIMIT}: {type(e).__name__}: {e}")
|
||
|
||
if fails == 0:
|
||
try:
|
||
info = await asyncio.wait_for(bot.get_webhook_info(), timeout=API_TIMEOUT)
|
||
pending = int(getattr(info, "pending_update_count", 0) or 0)
|
||
if pending > 0:
|
||
backlog += 1
|
||
_log(f"Telegram 侧积压 {pending} 条更新（检测 {backlog}/{WATCHDOG_LIMIT}）")
|
||
else:
|
||
backlog = 0
|
||
except Exception as e:
|
||
_log(f"getWebhookInfo 异常: {type(e).__name__}: {e}")
|
||
|
||
if (fails >= WATCHDOG_LIMIT or backlog >= WATCHDOG_LIMIT
|
||
or fetch_fails >= FETCH_FAIL_LIMIT):
|
||
_log(f"TG 通道不健康（getMe 连续失败 {fails} 次 / 积压检测 {backlog} 次 / "
|
||
f"取更新失败 {fetch_fails} 次），重建连接")
|
||
try:
|
||
sys.stdout.flush()
|
||
except Exception:
|
||
pass
|
||
await _notify_admins(f"[{name}] TG 通道异常，正在自动重建连接…")
|
||
try:
|
||
dp.stop_polling()
|
||
except Exception as e:
|
||
_log(f"stop_polling 异常: {e}")
|
||
await asyncio.sleep(GRACE_SECONDS)
|
||
_log("优雅停止超时，强制退出以触发服务重启")
|
||
try:
|
||
sys.stdout.flush()
|
||
sys.stderr.flush()
|
||
except Exception:
|
||
pass
|
||
os._exit(1)
|