summaryrefslogtreecommitdiff
path: root/ATRI/plugins/status/data_source.py
blob: ddd356834e32f49fb7ac03458e721a00acf682a3 (plain)
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
import os
import time
import psutil
from datetime import datetime

from ATRI.service import Service
from ATRI.rule import is_in_service
from ATRI.exceptions import GetStatusError


_status_msg = """
> Status Overview

[CPU: {p_cpu}% of {b_cpu}]
[Memory: {p_mem}% of {b_mem}]
[Disk usage: {p_disk}%]

[Net sent: {inteSENT}MB]
[Net recv: {inteRECV}MB]

[Bot runtime: {bot_time}]
[Platform runtime: {boot_time}]
{msg}
""".strip()


class Status(Service):
    def __init__(self):
        Service.__init__(self, "状态", "检查自身状态", rule=is_in_service("状态"))

    @staticmethod
    def ping() -> str:
        return "I'm fine."

    @staticmethod
    def get_status() -> tuple:
        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  # type: ignore
            inte_recv = psutil.net_io_counters().bytes_recv / 1000000  # type: ignore

            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 GetStatusError:
            raise GetStatusError("Failed to get status.")

        msg = "アトリは、高性能ですから!"
        if cpu > 90:  # type: ignore
            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=f"{b_cpu}%",
            b_mem="%.1f%%" % b_mem,
            inteSENT=inte_send,
            inteRECV=inte_recv,
            bot_time=bot_time,
            boot_time=boot_time,
            msg=msg,
        )

        return msg0, is_ok