Henry_Du/uptime_kuma_puller
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(uptime_kuma_puller): 初始化Uptime Kuma状态监控AstrBot插件

Browse Source 
包含依赖配置、gitignore规则、配置Schema与使用文档，支持通过/list、/uptime、/状态、/健康指令查询监控状态，可配置多个监控项目，展示监控项在线状态、延迟及维护和事件公告。
This commit is contained in:
Henry_Du
2026-07-17 05:28:08 +08:00
commit d1b49503aa
5 changed files with 322 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+17
View File
@@ -0,0 +1,17 @@
__pycache__/
*.py[cod]
*.pyo
*.pyd
.Python
.venv/
venv/
env/
ENV/
.pytest_cache/
.mypy_cache/
.ruff_cache/
.DS_Store
Thumbs.db
README.md
+48
View File
@@ -0,0 +1,48 @@
# astrbot_plugin_uptime_kuma_puller
Uptime Kuma 状态监控查询插件，适用于 AstrBot。
## 说明
- 插件通过 AstrBot 指令 `list` 触发。
- 会从 Uptime Kuma 状态页接口拉取监控项、维护公告和事件公告。
- 支持查询单个项目，也支持一次性输出所有已配置项目。
## 安装
- 可通过 AstrBot 插件市场安装。
- 或直接下载整个仓库后放入 AstrBot 插件目录。
## 配置
请在 AstrBot 插件管理面板配置以下字段：
- `query_url`：Uptime Kuma 站点地址，例如 `https://status.example.com`
- `proj_name_list`：状态页项目名称列表
项目名称请填写状态页链接 `/status/<项目名>` 中的 `<项目名>` 部分。
例如：
- 状态页地址是 `https://status.example.com/status/game`
- 则项目名填写 `game`
## 使用方法
- `/list`
- 查询 `proj_name_list` 中的全部项目状态
- `/list <项目名>`
- 查询指定项目状态
也保留了别名：
- `/uptime`
- `/状态`
- `/健康`
## 返回内容
- 监控项在线/离线状态
- 最近一次延迟
- 维护公告
- 事件公告
_conf_schema.json
+14
View File
@@ -0,0 +1,14 @@
{
"query_url": {
"description": "Uptime Kuma 服务器地址，末尾不要带 /",
"type": "string",
"hint": "请输入完整的URL，例如 http://localhost:3225",
"default": ""
},
"proj_name_list": {
"description": "状态页项目名称列表",
"type": "list",
"hint": "填写 Uptime Kuma 状态页 /status/<项目名> 中的项目名，可配置多个",
"default": []
}
}
main.py
+242
View File
@@ -0,0 +1,242 @@
import asyncio
from datetime import datetime
from typing import Any
import aiohttp
from astrbot.api import AstrBotConfig, logger
from astrbot.api.event import AstrMessageEvent, filter
from astrbot.api.star import Context, Star, register
@register(
"UptimeKuma状态监控",
"Henry_Du",
"通过 /list 指令查询 Uptime Kuma 状态页",
"2.0.0",
"https://github.com/HenryDu8133/astrbot_plugin_uptime_kuma_puller",
)
class UptimeKumaPuller(Star):
def __init__(self, context: Context, config: AstrBotConfig):
super().__init__(context)
self.query_url = (config.get("query_url") or "").rstrip("/")
self.project_map = self._build_project_map(config.get("proj_name_list") or [])
self.timeout_seconds = 15
self.is_configured = bool(self.query_url and self.project_map)
if not self.is_configured:
logger.warning("请先在管理面板配置 query_url 和 proj_name_list")
@filter.command("list", alias={"uptime", "状态", "健康"})
async def handle_list(self, event: AstrMessageEvent):
"""查询 Uptime Kuma 状态；支持 /list 或 /list <项目名>。"""
if not self.is_configured:
yield event.plain_result(
"❌插件未配置，请先在管理面板填写 query_url 和 proj_name_list。"
)
return
raw_arg = (event.message_str or "").strip()
logger.info("收到 Uptime Kuma 查询请求：%s", raw_arg or "/list")
if raw_arg:
project_key = raw_arg.lower()
if project_key not in self.project_map:
project_names = "、".join(self.project_map.values())
yield event.plain_result(
f"❌未找到项目：{raw_arg}\n可用项目：{project_names}"
)
return
result = await self.render_project_status(
project_key,
self.project_map[project_key],
)
yield event.plain_result(result)
return
messages = []
for project_key, display_name in self.project_map.items():
messages.append(await self.render_project_status(project_key, display_name))
yield event.plain_result("\n\n====================\n\n".join(messages))
def _build_project_map(self, project_names: list[Any]) -> dict[str, str]:
project_map: dict[str, str] = {}
for item in project_names:
name = str(item).strip()
if not name:
continue
project_map[name.lower()] = name
return project_map
async def fetch_json(self, session: aiohttp.ClientSession, url: str) -> dict[str, Any]:
async with session.get(url) as response:
if response.status != 200:
raise RuntimeError(f"HTTP {response.status}: {url}")
return await response.json(content_type=None)
async def fetch_project_payload(
self, project_key: str
) -> tuple[dict[str, Any], dict[str, Any]]:
timeout = aiohttp.ClientTimeout(total=self.timeout_seconds)
status_url = f"{self.query_url}/api/status-page/{project_key}"
heartbeat_url = f"{self.query_url}/api/status-page/heartbeat/{project_key}"
async with aiohttp.ClientSession(timeout=timeout) as session:
return await asyncio.gather(
self.fetch_json(session, status_url),
self.fetch_json(session, heartbeat_url),
)
async def render_project_status(self, project_key: str, display_name: str) -> str:
try:
status_data, heartbeat_data = await self.fetch_project_payload(project_key)
except asyncio.TimeoutError:
logger.error("获取项目 %s 状态超时", display_name)
return f"❌ {display_name} 查询超时，请稍后再试。"
except Exception as exc:
logger.error("获取项目 %s 状态失败：%s", display_name, exc)
return f"❌ {display_name} 查询失败：{exc}"
project_title = (
status_data.get("config", {}).get("title")
or display_name
)
monitor_lines, up_count, down_count, unknown_count = self.build_monitor_lines(
status_data, heartbeat_data
)
notice_lines = self.build_notice_lines(status_data)
summary = [
f"📋 {project_title}",
f"✅ 正常：{up_count} ❌ 异常：{down_count} ❔ 未知：{unknown_count}",
]
if monitor_lines:
summary.extend(monitor_lines)
else:
summary.append("暂无可展示的监控项。")
if notice_lines:
summary.append("==============")
summary.extend(notice_lines)
return "\n".join(summary)
def build_monitor_lines(
self,
status_data: dict[str, Any],
heartbeat_data: dict[str, Any],
) -> tuple[list[str], int, int, int]:
heartbeat_list = heartbeat_data.get("heartbeatList", {})
monitor_items: list[tuple[str, str, int]] = []
up_count = 0
down_count = 0
unknown_count = 0
for group in status_data.get("publicGroupList", []):
for monitor in group.get("monitorList", []):
monitor_id = str(monitor.get("id"))
monitor_name = self.format_monitor_name(monitor)
latest_heartbeat = (heartbeat_list.get(monitor_id) or [None])[-1]
status_text, status_code = self.format_monitor_status(latest_heartbeat)
if status_code == 1:
up_count += 1
elif status_code == 0:
down_count += 1
else:
unknown_count += 1
monitor_items.append((monitor_name, status_text, status_code))
monitor_items.sort(key=lambda item: item[0])
monitor_lines = [f"{name} {status_text}" for name, status_text, _ in monitor_items]
return monitor_lines, up_count, down_count, unknown_count
def format_monitor_name(self, monitor: dict[str, Any]) -> str:
tags = monitor.get("tags") or []
if tags:
first_tag_name = tags[0].get("name")
if first_tag_name:
return f"[{first_tag_name}]{monitor.get('name', '未命名监控项')}"
return str(monitor.get("name", "未命名监控项"))
def format_monitor_status(
self, heartbeat: dict[str, Any] | None
) -> tuple[str, int]:
if not heartbeat:
return "⚪[未知]", -1
status = heartbeat.get("status")
ping = heartbeat.get("ping")
ping_text = f" {ping}ms" if ping is not None else ""
if status == 1:
return f"🟢[在线]{ping_text}", 1
if status in (0, 2, 3):
return f"🔴[离线]{ping_text}", 0
return f"⚪[未知]{ping_text}", -1
def build_notice_lines(self, status_data: dict[str, Any]) -> list[str]:
lines: list[str] = []
for maintenance in status_data.get("maintenanceList", []):
try:
title = maintenance.get("title", "无标题")
content = maintenance.get("description", "无描述")
start_time = self.format_timestamp(maintenance.get("startDate"))
end_time = self.format_timestamp(maintenance.get("endDate"))
lines.append(f"🔵【维护中】{title}")
lines.append(content)
if start_time and end_time:
lines.append(f"⏲️维护时间：{start_time} - {end_time}")
except Exception as exc:
logger.error("处理维护消息时出错：%s", exc)
incident_list = status_data.get("incidentList", [])
if not incident_list and status_data.get("incident"):
incident_list = [status_data["incident"]]
style_map = {
"info": "信息",
"warning": "警告",
"danger": "危险",
"primary": "主要",
"light": "明亮",
"dark": "黑暗",
}
for incident in incident_list:
try:
style = style_map.get(str(incident.get("style", "info")).lower(), "信息")
title = incident.get("title", "无标题")
content = incident.get("content", "无内容")
last_updated = self.format_datetime_string(incident.get("lastUpdatedDate"))
lines.append(f"📣【{style}】{title}")
lines.append(content)
if last_updated:
lines.append(f"⏲️本通知更新于：{last_updated}")
except Exception as exc:
logger.error("处理事件公告时出错：%s", exc)
return lines
def format_timestamp(self, value: Any) -> str:
if value in (None, ""):
return ""
try:
return datetime.fromtimestamp(int(value) / 1000).strftime("%Y-%m-%d %H:%M:%S")
except Exception:
return str(value)
def format_datetime_string(self, value: Any) -> str:
if value in (None, ""):
return ""
if isinstance(value, str):
normalized = value.replace("T", " ").replace("Z", "")
try:
return datetime.fromisoformat(normalized).strftime("%Y-%m-%d %H:%M:%S")
except ValueError:
return value
return str(value)
requirements.txt
+1
View File
@@ -0,0 +1 @@
aiohttp>=3.9,<4