LiteyukiBot-TriM/liteyuki/plugins/liteyuki_sign_status.py

166 lines
4.8 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import datetime
import time
import aiohttp
from nonebot import require
from nonebot.plugin import PluginMetadata
from liteyuki.utils.base.config import get_config
from liteyuki.utils.base.data import Database, LiteModel
from liteyuki.utils.base.resource import get_path
from liteyuki.utils.message.html_tool import template2image
require("nonebot_plugin_alconna")
require("nonebot_plugin_apscheduler")
from nonebot_plugin_apscheduler import scheduler
from nonebot_plugin_alconna import Alconna, AlconnaResult, CommandResult, Subcommand, UniMessage, on_alconna, Args
__author__ = "snowykami"
__plugin_meta__ = PluginMetadata(
name="签名服务器状态",
description="适用于ntqq的签名状态查看",
usage=(
"sign count 查看当前签名数\n"
"sign data 查看签名数变化\n"
"sign chart [limit] 查看签名数变化图表\n"
),
type="application",
homepage="https://github.com/snowykami/LiteyukiBot",
extra={
"liteyuki" : True,
"toggleable" : True,
"default_enable": True,
}
)
SIGN_COUNT_URLS: dict[str, str] = get_config("sign_count_urls", None)
SIGN_COUNT_DURATION = get_config("sign_count_duration", 10)
class SignCount(LiteModel):
TABLE_NAME: str = "sign_count"
time: float = 0.0
count: int = 0
sid: str = ""
sign_db = Database("data/liteyuki/ntqq_sign.ldb")
sign_db.auto_migrate(SignCount())
sign_status = on_alconna(Alconna(
"sign",
Subcommand(
"chart",
Args["limit", int, 10000]
),
Subcommand(
"count"
),
Subcommand(
"data"
)
))
cache_img: bytes = None
@sign_status.assign("count")
async def _():
reply = "Current sign count:"
for name, count in (await get_now_sign()).items():
reply += f"\n{name}: {count[1]}"
await sign_status.send(reply)
@sign_status.assign("data")
async def _():
query_stamp = [1, 5, 10, 15]
reply = "Count from last " + ", ".join([str(i) for i in query_stamp]) + "mins"
for name, url in SIGN_COUNT_URLS.items():
count_data = []
for stamp in query_stamp:
count_rows = sign_db.all(SignCount(), "sid = ? and time > ?", url, time.time() - 60 * stamp)
if len(count_rows) < 2:
count_data.append(-1)
else:
count_data.append(count_rows[-1].count - count_rows[0].count)
reply += f"\n{name}: " + ", ".join([str(i) for i in count_data])
await sign_status.send(reply)
@sign_status.assign("chart")
async def _(arp: CommandResult = AlconnaResult()):
limit = arp.result.subcommands.get("chart").args.get("limit")
if limit == 10000:
if cache_img:
await sign_status.send(UniMessage.image(raw=cache_img))
return
img = await generate_chart(limit)
await sign_status.send(UniMessage.image(raw=img))
@scheduler.scheduled_job("interval", seconds=SIGN_COUNT_DURATION, next_run_time=datetime.datetime.now())
async def update_sign_count():
global cache_img
if not SIGN_COUNT_URLS:
return
data = await get_now_sign()
for name, count in data.items():
await save_sign_count(count[0], count[1], SIGN_COUNT_URLS[name])
cache_img = await generate_chart(10000)
async def get_now_sign() -> dict[str, tuple[float, int]]:
"""
Get the sign count and the time of the latest sign
Returns:
tuple[float, int] | None: (time, count)
"""
data = {}
now = time.time()
async with aiohttp.ClientSession() as client:
for name, url in SIGN_COUNT_URLS.items():
async with client.get(url) as resp:
count = (await resp.json())["count"]
data[name] = (now, count)
return data
async def save_sign_count(timestamp: float, count: int, sid: str):
"""
Save the sign count to the database
Args:
sid: the sign id use url as the id
count:
timestamp (float): the time of the sign count (int): the count of the sign
"""
sign_db.save(SignCount(time=timestamp, count=count, sid=sid))
async def generate_chart(limit):
data = []
for name, url in SIGN_COUNT_URLS.items():
count_rows = sign_db.all(SignCount(), "sid = ? ORDER BY id DESC LIMIT ?", url, limit)
count_rows.reverse()
data.append(
{
"name" : name,
# "data": [[row.time, row.count] for row in count_rows]
"times" : [row.time for row in count_rows],
"counts": [row.count for row in count_rows]
}
)
print(len(count_rows))
img = await template2image(
template=get_path("templates/sign_status.html", debug=True),
templates={
"data": data
},
debug=True
)
return img