1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
|
import os
import time
import psutil
from typing import Tuple
from datetime import datetime
from nonebot import get_bot
from nonebot.adapters.onebot.v11 import unescape
from ATRI.log import log
from ATRI.service import Service
from ATRI.message import MessageBuilder
from ATRI.exceptions import GetStatusError
from ATRI.utils.apscheduler import scheduler
plugin = Service("状态").document("检查自身状态")
ping = plugin.on_command("/ping", "检测bot是否存活")
@ping.handle()
async def _():
await ping.finish("I'm fine.")
status = plugin.on_command("/status", "检查bot运行资源占用")
@status.handle()
async def _():
msg, _ = get_status()
print(msg)
await status.finish(msg)
@scheduler.scheduled_job("interval", name="状态检查", minutes=15, misfire_grace_time=15)
async def _():
log.info("检查资源消耗中...")
msg, stat = get_status()
if not stat:
log.warning("资源消耗异常")
try:
bot = get_bot()
except Exception:
bot = None
if bot: await plugin.send_to_master(msg)
else:
log.info("资源消耗正常")
_STATUS_MSG = (
MessageBuilder("[Status Overview]")
.text("[CPU: {b_cpu}% of {p_cpu}%]")
.text("[Memory: {b_mem} of {p_mem}%]")
.text("[Disk usage: {p_disk}%]")
.text("")
.text("[Net sent: {inteSENT}MB]")
.text("[Net recv: {inteRECV}MB]")
.text("")
.text("[Run Duration]")
.text("[Bot: {bot_time}]")
.text("[Platform: {boot_time}]")
.text("{msg}")
.done()
)
def get_status() -> Tuple[str, bool]:
try:
cpu = psutil.cpu_percent(interval=1)
mem = psutil.virtual_memory().percent
disk = psutil.disk_usage("/").percent
inte_send = psutil.net_io_counters().bytes_sent / 1000000
inte_recv = psutil.net_io_counters().bytes_recv / 1000000
process = psutil.Process(os.getpid())
b_cpu = process.cpu_percent(interval=1)
b_mem = process.memory_percent(memtype="rss")
now = time.time()
boot = psutil.boot_time()
b = process.create_time()
boot_time = str(
datetime.utcfromtimestamp(now).replace(microsecond=0)
- datetime.utcfromtimestamp(boot).replace(microsecond=0)
)
bot_time = str(
datetime.utcfromtimestamp(now).replace(microsecond=0)
- datetime.utcfromtimestamp(b).replace(microsecond=0)
)
except Exception:
raise GetStatusError("Failed to get status.")
msg = "アトリは、高性能ですから！"
if cpu > 90:
msg = "咱感觉有些头晕..."
is_ok = False
if mem > 90:
msg = "咱感觉有点头晕并且有点累..."
is_ok = False
elif mem > 90:
msg = "咱感觉有点累..."
is_ok = False
elif disk > 90:
msg = "咱感觉身体要被塞满了..."
is_ok = False
else:
is_ok = True
msg0 = _STATUS_MSG.format(
p_cpu=cpu,
p_mem=mem,
p_disk=disk,
b_cpu=b_cpu,
b_mem="%.1f%%" % b_mem,
inteSENT=inte_send,
inteRECV=inte_recv,
bot_time=bot_time,
boot_time=boot_time,
msg=msg,
)
return unescape(msg0), is_ok
|
