LiteyukiBot/liteyuki/plugins/liteyuki_status/api.py

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