mirror of
https://github.com/LiteyukiStudio/LiteyukiBot.git
synced 2025-11-04 11:06:24 +00:00
276 lines
8.7 KiB
Python
276 lines
8.7 KiB
Python
import platform
|
|
import time
|
|
|
|
import nonebot
|
|
import psutil
|
|
from cpuinfo import cpuinfo
|
|
from nonebot import require
|
|
from nonebot.adapters import satori
|
|
|
|
from liteyuki.utils import __NAME__, __VERSION__
|
|
from liteyuki.utils.base.config import get_config
|
|
from liteyuki.utils.base.data_manager import TempConfig, common_db
|
|
from liteyuki.utils.base.language import Language
|
|
from liteyuki.utils.base.resource import get_loaded_resource_packs, get_path
|
|
from liteyuki.utils.message.html_tool import template2image
|
|
from liteyuki.utils import satori_utils
|
|
from .counter_for_satori import satori_counter
|
|
|
|
require("nonebot_plugin_apscheduler")
|
|
from nonebot_plugin_apscheduler import scheduler
|
|
|
|
protocol_names = {
|
|
0: "iPad",
|
|
1: "Android Phone",
|
|
2: "Android Watch",
|
|
3: "Mac",
|
|
5: "iPad",
|
|
6: "Android Pad",
|
|
}
|
|
|
|
"""
|
|
Universal Interface
|
|
data
|
|
- bot
|
|
- name: str
|
|
icon: str
|
|
id: int
|
|
protocol_name: str
|
|
groups: int
|
|
friends: int
|
|
message_sent: int
|
|
message_received: int
|
|
app_name: str
|
|
- hardware
|
|
- cpu
|
|
- percent: float
|
|
- name: str
|
|
- mem
|
|
- percent: float
|
|
- total: int
|
|
- used: int
|
|
- free: int
|
|
- swap
|
|
- percent: float
|
|
- total: int
|
|
- used: int
|
|
- free: int
|
|
- disk: list
|
|
- name: str
|
|
- percent: float
|
|
- total: int
|
|
"""
|
|
status_card_cache = {} # lang -> bytes
|
|
|
|
|
|
# 60s刷新一次
|
|
@scheduler.scheduled_job("cron", second="*/40")
|
|
async def refresh_status_card():
|
|
nonebot.logger.debug("Refreshing status card cache...")
|
|
global status_card_cache
|
|
bot_data = await get_bots_data()
|
|
hardware_data = await get_hardware_data()
|
|
liteyuki_data = await get_liteyuki_data()
|
|
for lang in status_card_cache.keys():
|
|
status_card_cache[lang] = await generate_status_card(
|
|
bot_data,
|
|
hardware_data,
|
|
liteyuki_data,
|
|
lang=lang,
|
|
use_cache=False
|
|
)
|
|
|
|
|
|
async def generate_status_card(bot: dict, hardware: dict, liteyuki: dict, lang="zh-CN", bot_id="0",
|
|
use_cache=False) -> bytes:
|
|
if not use_cache:
|
|
return await template2image(
|
|
get_path("templates/status.html", abs_path=True),
|
|
{
|
|
"data": {
|
|
"bot": bot,
|
|
"hardware": hardware,
|
|
"liteyuki": liteyuki,
|
|
"localization": get_local_data(lang)
|
|
}
|
|
},
|
|
debug=True
|
|
)
|
|
else:
|
|
if lang not in status_card_cache:
|
|
status_card_cache[lang] = await generate_status_card(bot, hardware, liteyuki, lang=lang, bot_id=bot_id)
|
|
return status_card_cache[lang]
|
|
|
|
|
|
def get_local_data(lang_code) -> dict:
|
|
lang = Language(lang_code)
|
|
return {
|
|
"friends": lang.get("status.friends"),
|
|
"groups": lang.get("status.groups"),
|
|
"plugins": lang.get("status.plugins"),
|
|
"bots": lang.get("status.bots"),
|
|
"message_sent": lang.get("status.message_sent"),
|
|
"message_received": lang.get("status.message_received"),
|
|
"cpu": lang.get("status.cpu"),
|
|
"memory": lang.get("status.memory"),
|
|
"swap": lang.get("status.swap"),
|
|
"disk": lang.get("status.disk"),
|
|
|
|
"usage": lang.get("status.usage"),
|
|
"total": lang.get("status.total"),
|
|
"used": lang.get("status.used"),
|
|
"free": lang.get("status.free"),
|
|
|
|
"days": lang.get("status.days"),
|
|
"hours": lang.get("status.hours"),
|
|
"minutes": lang.get("status.minutes"),
|
|
"seconds": lang.get("status.seconds"),
|
|
"runtime": lang.get("status.runtime"),
|
|
"threads": lang.get("status.threads"),
|
|
"cores": lang.get("status.cores"),
|
|
"process": lang.get("status.process"),
|
|
"resources": lang.get("status.resources"),
|
|
"description": lang.get("status.description"),
|
|
}
|
|
|
|
|
|
async def get_bots_data(self_id: str = "0") -> dict:
|
|
"""获取当前所有机器人数据
|
|
Returns:
|
|
"""
|
|
result = {
|
|
"self_id": self_id,
|
|
"bots": [],
|
|
}
|
|
for bot_id, bot in nonebot.get_bots().items():
|
|
groups = 0
|
|
friends = 0
|
|
status = {}
|
|
bot_name = bot_id
|
|
version_info = {}
|
|
if isinstance(bot, satori.Bot):
|
|
try:
|
|
bot_name = (await satori_utils.user_infos.get(bot.self_id)).name
|
|
groups = str(await satori_utils.count_groups(bot))
|
|
friends = str(await satori_utils.count_friends(bot))
|
|
status = {}
|
|
version_info = await bot.get_version_info()
|
|
except Exception:
|
|
pass
|
|
else:
|
|
try:
|
|
# API fetch
|
|
bot_name = (await bot.get_login_info())["nickname"]
|
|
groups = len(await bot.get_group_list())
|
|
friends = len(await bot.get_friend_list())
|
|
status = await bot.get_status()
|
|
version_info = await bot.get_version_info()
|
|
except Exception:
|
|
pass
|
|
|
|
statistics = status.get("stat", {})
|
|
app_name = version_info.get("app_name", "UnknownImplementation")
|
|
if app_name in ["Lagrange.OneBot", "LLOneBot", "Shamrock", "NapCat.Onebot"]:
|
|
icon = f"https://q.qlogo.cn/g?b=qq&nk={bot_id}&s=640"
|
|
elif isinstance(bot, satori.Bot):
|
|
app_name = "Satori"
|
|
icon = (await bot.login_get()).user.avatar
|
|
else:
|
|
icon = None
|
|
bot_data = {
|
|
"name": bot_name,
|
|
"icon": icon,
|
|
"id": bot_id,
|
|
"protocol_name": protocol_names.get(version_info.get("protocol_name"), "Online"),
|
|
"groups": groups,
|
|
"friends": friends,
|
|
"message_sent": satori_counter.msg_sent if isinstance(bot, satori.Bot) else statistics.get("message_sent", 0),
|
|
"message_received": satori_counter.msg_received if isinstance(bot, satori.Bot) else statistics.get("message_received", 0),
|
|
"app_name": app_name
|
|
}
|
|
result["bots"].append(bot_data)
|
|
|
|
return result
|
|
|
|
|
|
async def get_hardware_data() -> dict:
|
|
mem = psutil.virtual_memory()
|
|
all_processes = psutil.Process().children(recursive=True)
|
|
all_processes.append(psutil.Process())
|
|
|
|
mem_used_process = 0
|
|
process_mem = {}
|
|
for process in all_processes:
|
|
try:
|
|
ps_name = process.name().replace(".exe", "")
|
|
if ps_name not in process_mem:
|
|
process_mem[ps_name] = 0
|
|
process_mem[ps_name] += process.memory_info().rss
|
|
mem_used_process += process.memory_info().rss
|
|
except Exception:
|
|
pass
|
|
swap = psutil.swap_memory()
|
|
cpu_brand_raw = cpuinfo.get_cpu_info().get("brand_raw", "Unknown")
|
|
if "AMD" in cpu_brand_raw:
|
|
brand = "AMD"
|
|
elif "Intel" in cpu_brand_raw:
|
|
brand = "Intel"
|
|
else:
|
|
brand = "Unknown"
|
|
result = {
|
|
"cpu": {
|
|
"percent": psutil.cpu_percent(),
|
|
"name": f"{brand} {cpuinfo.get_cpu_info().get('arch', 'Unknown')}",
|
|
"cores": psutil.cpu_count(logical=False),
|
|
"threads": psutil.cpu_count(logical=True),
|
|
"freq": psutil.cpu_freq().current # MHz
|
|
},
|
|
"memory": {
|
|
"percent": mem.percent,
|
|
"total": mem.total,
|
|
"used": mem.used,
|
|
"free": mem.free,
|
|
"usedProcess": mem_used_process,
|
|
},
|
|
"swap": {
|
|
"percent": swap.percent,
|
|
"total": swap.total,
|
|
"used": swap.used,
|
|
"free": swap.free
|
|
},
|
|
"disk": [],
|
|
}
|
|
|
|
for disk in psutil.disk_partitions(all=True):
|
|
try:
|
|
disk_usage = psutil.disk_usage(disk.mountpoint)
|
|
if disk_usage.total == 0:
|
|
continue # 虚拟磁盘
|
|
result["disk"].append({
|
|
"name": disk.mountpoint,
|
|
"percent": disk_usage.percent,
|
|
"total": disk_usage.total,
|
|
"used": disk_usage.used,
|
|
"free": disk_usage.free
|
|
})
|
|
except:
|
|
pass
|
|
|
|
return result
|
|
|
|
|
|
async def get_liteyuki_data() -> dict:
|
|
temp_data: TempConfig = common_db.where_one(TempConfig(), default=TempConfig())
|
|
result = {
|
|
"name": list(get_config("nickname", [__NAME__]))[0],
|
|
"version": __VERSION__,
|
|
"plugins": len(nonebot.get_loaded_plugins()),
|
|
"resources": len(get_loaded_resource_packs()),
|
|
"nonebot": f"{nonebot.__version__}",
|
|
"python": f"{platform.python_implementation()} {platform.python_version()}",
|
|
"system": f"{platform.system()} {platform.release()}",
|
|
"runtime": time.time() - temp_data.data.get("start_time", time.time()), # 运行时间秒数
|
|
"bots": len(nonebot.get_bots())
|
|
}
|
|
return result
|