优化代码、增加两个新功能、修复一个BUG

优化了所有功能的代码
增加了 自我检查 功能
增加了 权限等级管理 的功能
修复了 识图 功能的一个BUG
美化了说明文档

bot.py

@@ -1,10 +1,15 @@
 from os import path
 
+from level import get_permission_level
+
 import nonebot
 
 import config
 
 if __name__ == '__main__':
+    level = get_permission_level(config.PERMISSION_LEVEL)
+    nonebot.NoneBot.level = level
+
     nonebot.init(config)
     nonebot.load_builtin_plugins()
     nonebot.load_plugins(

config.py

@@ -11,6 +11,9 @@
 
 # ————————以下是部分功能模块需要的额外配置，请参见github上的说明进行配置————————
 
+# Permission类
+PERMISSION_LEVEL = 6 # 权限等级值，建议不要设置为8以下 | 类型为int
+
 # KEY类
 SAUCENAO_KEY = "" # SauceNAO 的 API key | 类型为str
 BAIDUAPPID_TRANSL = "" # Baidu翻译 的 APP ID | 类型为str
@@ -22,6 +25,7 @@
 MAXINFO_ANIME = 4 # 搜番功能查找番剧的最大数 | 类型为int>0
 MAXLINE_JD = 7 # 日语词典功能查找条目的内容所允许的最大行书 | 类型为int>0
 MAXWOED_JD = 250 # 日语词典功能查找条目的内容所允许的最大字数 | 类型为int>0
+MAX_PERFORMANCE_PERCENT = [92,92,92] # 自检功能中的服务器占用比率最高值，顺序分别对应CPU、内存和硬盘 | 类型为list
 
 # TimeLimit类
 TIMELIMIT_IMAGE = 7 # 识图功能的时间限制 | 类型为float
@@ -41,7 +45,8 @@
     'numpy':'np',
     'math':'',
     'scipy':''
-    } # 需要提供的计算库名与可选的别名(仅在MORE_COMPLEX为真时有效) | 类型为dict
-TO_TRANSL = "zh-CN" # 指定翻译功能的目标语言 | 类型为str
+    } # 就按功能种需要提供的计算库名与可选的别名(仅在MORE_COMPLEX为真时有效) | 类型为dict
+PROCESS_NAME_LIST = {} # 自检功能种需要提供的格外检查的进程名 | 类型为set
+TO_TRANSL = "zh-CN" # 翻译功能中指定翻译功能的目标语言 | 类型为str
 
 # —————————————————————————————————————————————————————————————————————————

include/plugins/anime/__init__.py

@@ -1,9 +1,9 @@
-from nonebot import on_command, CommandSession
+from nonebot import on_command, CommandSession, get_bot
 
 from .data_source import from_anime_get_info
 
 
-@on_command('anime', aliases=('anime', '搜番'))
+@on_command('anime', aliases=('anime', '搜番'), permission=get_bot().level)
 async def anime(session: CommandSession):
     key_word = session.get('key_word', prompt='需要的番剧名称是什么？')
     anime_report = await from_anime_get_info(key_word)

include/plugins/anime/data_source.py

@@ -1,12 +1,13 @@
 import requests
 from lxml import etree
-import sys
 import time
 
-sys.path.append('../../../')
-from config import TIMELIMIT_ANIME, MAXINFO_ANIME
 from kth_timeoutdecorator import *
 
+from nonebot import get_bot
+TIMELIMIT_ANIME = get_bot().config.TIMELIMIT_ANIME
+MAXINFO_ANIME = get_bot().config.MAXINFO_ANIME
+
 async def from_anime_get_info(key_word: str) -> str:
     repass = ""
     url = 'https://share.dmhy.org/topics/list?keyword=' + key_word

include/plugins/calculate/__init__.py

@@ -1,9 +1,9 @@
-from nonebot import on_command, CommandSession
+from nonebot import on_command, CommandSession, get_bot
 
 from .data_source import get_end_calculate
 
 
-@on_command('calculate', aliases=('计算', 'exp','計算'))
+@on_command('calculate', aliases=('计算', 'exp','計算'), permission=get_bot().level)
 async def calculate(session: CommandSession):
     code_str = session.get('calculate', prompt='请输入公式或代码……')
     code_report = await get_end_calculate(code_str)

include/plugins/calculate/data_source.py

@@ -1,10 +1,11 @@
 import io
 import tokenize
-import sys
 from copy import deepcopy
 
-sys.path.append('../../../')
-from config import MORE_COMPLEX, CALCULATE_LIST
+from nonebot import get_bot
+MORE_COMPLEX = get_bot().config.MORE_COMPLEX
+CALCULATE_LIST = get_bot().config.CALCULATE_LIST
+
 
 def check_unsafe_attributes(string) -> (dict, dict):
     """This code is modified from:

include/plugins/ceic/__init__.py

@@ -1,18 +1,17 @@
-from .data_source import ceicinfo
+from .data_source import Ceicinfo
 
-import nonebot
+from nonebot import get_bot, scheduler
+EM = get_bot().config.EM
+CEICONLYCN = get_bot().config.CEICONLYCN
 from aiocqhttp.exceptions import Error as CQHttpError
 
-import sys
-sys.path.append('../../../')
-from config import EM, CEICONLYCN
 
 print("[info]loading ceicinfo……")
-ceic = ceicinfo(EM, CEICONLYCN)
+ceic = Ceicinfo(EM, CEICONLYCN)
 
-@nonebot.scheduler.scheduled_job('cron', minute='*')
+@scheduler.scheduled_job('cron', minute='*')
 async def _():
-    bot = nonebot.get_bot()
+    bot = get_bot()
     group_list = await bot.get_group_list()
     mesg = await ceic.getceicinfo()
     if mesg:

include/plugins/ceic/data_source.py

@@ -4,7 +4,7 @@
 from datetime import datetime
 
 
-class ceicinfo():
+class Ceicinfo():
 
     def __init__(self, em: float =4.5, only: bool = True):  # 只报道震级>=em
         self.em = em

include/plugins/check/__init__.py

@@ -0,0 +1,32 @@
+from random import choice
+
+from nonebot import on_command, CommandSession, get_bot, permission
+
+from .data_source import Check
+
+Bot = get_bot()
+
+MAX_PERFORMANCE_PERCENT = Bot.config.MAX_PERFORMANCE_PERCENT
+PROCESS_NAME_LIST = Bot.config.PROCESS_NAME_LIST
+SUPERUSERS = Bot.config.SUPERUSERS
+
+print("[info]loading checkinfo……")
+check = Check(Bot.config.PROCESS_NAME_LIST)
+
+@on_command('check', aliases=('check', '自检', '自檢'), permission=Bot.level)
+async def music_recommend(session: CommandSession):
+    if await permission.check_permission(session.bot, session.event, 0xF000):
+        check_report_admin = await check.get_check_info()
+        if check_report_admin:
+            await session.send(check_report_admin)
+    else:
+        check_report = await check.get_check_easy(MAX_PERFORMANCE_PERCENT)
+        more_info = '[CQ:at,qq={}]'.format(choice(list(SUPERUSERS))) if SUPERUSERS else "\n……管理员鸽了，请有知道的人把管理员拖回来维护~"
+        if check_report:
+            await session.send(check_report + more_info)
+            check_report_admin = await check.get_check_info()
+            if check_report_admin and SUPERUSERS:
+                for admin in SUPERUSERS:
+                    await Bot.send_private_msg(user_id=admin, message='我好像生病了……')
+                    await Bot.send_private_msg(user_id=admin, message=check_report_admin)
+        else: await session.send("I'm fine, thanks. ^_^")

include/plugins/check/data_source.py

@@ -0,0 +1,138 @@
+import psutil
+import requests
+from datetime import datetime
+
+
+class Check():
+
+    def __init__(self, request_name_list={}):
+        self.request_name_list = request_name_list
+        self.cpu_percent = 0
+        self.memory_percent = 0
+        self.disk_percent = 0
+        self.baidu = 404
+        self.google = 404
+        self.process_name_list = []
+        self.process_status_list = []
+        self.user_list =[]
+
+        self.run_all_check()
+
+    def run_all_check(self):
+        self.get_performance_check()
+        self.get_network_check()
+        self.get_process_status(self.request_name_list)
+        self.get_users_check()
+
+    async def get_check_info(self):
+        self.run_all_check()
+        putline = []
+
+        putline.append("--Performance--\n[Cpu] {}%\n[Memory] {}%\n[Disk] {}%\n--Network--\n[Baidu] {}\n[Google] {}".format(self.cpu_percent, self.memory_percent, self.disk_percent, self.baidu, self.google))
+
+        if self.process_name_list:
+            putline.append("--Process--")
+            for name, status in zip(self.process_name_list, self.process_status_list):
+                putline.append("[{}] {}".format(name, status))
+
+        if self.user_list:
+            putline.append("--Users--")
+            for user in self.user_list:
+                putline.append("[{}] {}".format(user['name'], user['started']))
+
+        return "\n".join(putline)
+
+    async def get_check_easy(self, max_performance_percent=[92,92,92]):
+        putline = []
+        check_list = await self.get_check_simple()
+
+        if sum(check_list) == 5:
+            return "我去世了……（安详"
+        if sum(check_list) == 4:
+            return "你们如果还能看到消息一定是奇迹……"
+        if sum(check_list[:3]) != 0:
+            return "啊……我感觉……好热……"
+        if sum(check_list[3:5]) == 2:
+            return "谁拔我网线?!"
+        if check_list[4] == 1:
+            return "我节点又断惹(识图、上车等部分功能会受到影响)"
+
+    async def get_check_simple(self, max_performance_percent=[92,92,92]) -> list:
+        check_list = [0,0,0,0,0,0]
+        self.run_all_check()
+
+        if self.cpu_percent > max_performance_percent[0]:
+            check_list[0] = 1
+        if self.memory_percent > max_performance_percent[1]:
+            check_list[1] = 1
+        if self.disk_percent > max_performance_percent[2]:
+            check_list[2] = 1
+        if self.baidu != 200:
+            check_list[3] = 1
+        if self.google != 200:
+            check_list[4] = 1
+        for status in self.process_status_list:
+            if status != 'running':
+                check_list[5] = 1
+                break
+
+        return check_list
+
+    def get_performance_check(self):
+        self.cpu_percent = psutil.cpu_percent()
+        self.memory_percent = psutil.virtual_memory().percent
+        self.disk_percent = psutil.disk_usage("/").percent
+
+    def get_network_check(self):
+        try:
+            self.baidu = requests.get("http://www.baidu.com").status_code
+        except:
+            self.baidu = 404
+            print("[ERROR] Baidu request failed.")
+
+        try:
+            self.google = requests.get("http://www.google.com").status_code
+        except:
+            self.google = 404
+            print("[ERROR] Google request failed.")
+
+    def get_users_check(self):
+        user_list = []
+        suser_l = psutil.users()
+
+        for suser in suser_l:
+            user = {
+                'name': suser.name, 
+                'started': datetime.fromtimestamp(suser.started).strftime("%Y-%m-%d %H:%M:%S")}
+            user_list.append(user)
+
+        self.user_list = user_list
+
+    def get_process_status(self, request_name_list: set):
+        if not request_name_list:
+            return None
+
+        self.process_name_list = []
+        self.process_status_list = []
+
+        for p_n in request_name_list:
+            p_l = self.get_sname_process_list(p_n)
+            if len(p_l) == 1:
+                self.process_name_list.append(p_n)
+                self.process_status_list.append(p_l[0].status())
+            else:
+                for i, p in enumerate(p_l):
+                    self.process_name_list.append(p_n+" ({})".format(i))
+                    self.process_status_list.append(p.status())
+
+    @staticmethod
+    def get_sname_process_list(name: str) -> list:
+        p_l = []
+        pids  = psutil.pids()
+
+        for pid in pids:
+            p = psutil.Process(pid)
+            if (p.name() == name):
+                p_l.append(p)
+
+        return p_l

include/plugins/image/__init__.py

@@ -1,10 +1,10 @@
-from nonebot import on_command, CommandSession
+from nonebot import on_command, CommandSession, get_bot
 from nonebot import on_natural_language, NLPSession, IntentCommand
 
 from .data_source import get_image_data
 
 
-@on_command('image', aliases=('image', '搜图', '识图', '搜圖', '識圖'))
+@on_command('image', aliases=('image', '搜图', '识图', '搜圖', '識圖'), permission=get_bot().level)
 async def image(session: CommandSession):
     image_data = session.get('image', prompt='图呢？GKD')
     image_data_report = await get_image_data(image_data, session.bot.config.SAUCENAO_KEY)
@@ -28,7 +28,7 @@ async def _(session: CommandSession):
 
     session.state[session.current_key] = image_arg
 
-@on_natural_language(keywords={'image', '搜图', '识图', '搜圖', '識圖'})
+@on_natural_language(keywords={'image', '搜图', '识图', '搜圖', '識圖'}, permission=get_bot().level)
 async def _(session: NLPSession):
     msg = session.msg
 

include/plugins/image/data_source.py

@@ -1,12 +1,13 @@
 import json
 import requests
 from lxml import etree
-import sys
 
-sys.path.append('../../../')
-from config import TIMELIMIT_IMAGE
 from kth_timeoutdecorator import *
 
+from nonebot import get_bot
+TIMELIMIT_IMAGE = get_bot().config.TIMELIMIT_IMAGE
+
+
 class SauceNAO:
 
     def __init__(self, api_key, output_type=2, testmode=0, dbmask=None, dbmaski=None, db=999, numres=3, shortlimit=20, longlimit=300):
@@ -82,9 +83,14 @@ def get_view(self, ascii2d) -> str:
         repass = ''
         url_index = "https://ascii2d.net/search/url/{}".format(ascii2d)
         # print("url_index: ", url_index)
-        html_index_data = requests.get(url_index)
-        print("[info]index html data OK.")
-        html_index = etree.HTML(html_index_data.text)
+        try:
+            html_index_data = requests.get(url_index)
+            html_index = etree.HTML(html_index_data.text)
+        except Exception as e:
+            print("[warning] ascii2d get html data failed.")
+            print(e)
+            return repass
+        else: print("[info]index html data OK.")
 
         neet_div = html_index.xpath('//div[@class="detail-link pull-xs-right hidden-sm-down gray-link"]')
 

include/plugins/japanese_dictionary/__init__.py

@@ -1,10 +1,10 @@
-from nonebot import on_command, CommandSession
+from nonebot import on_command, CommandSession, get_bot
 from nonebot import on_natural_language, NLPSession, IntentCommand
 
 from .data_source import get_definition_of_word
 
 
-@on_command('jpd', aliases=('日典', 'jd'))
+@on_command('jpd', aliases=('日典', 'jd'), permission=get_bot().level)
 async def jpd(session: CommandSession):
     word = session.get('word', prompt='你想查询哪个单词呢？')
     jpd_report = await get_definition_of_word(word)

include/plugins/japanese_dictionary/data_source.py

@@ -1,12 +1,15 @@
 from lxml import etree
 import requests
-import sys
 import re
 
-sys.path.append('../../../')
-from config import TIMELIMIT_JD, MAXLINE_JD, MAXWOED_JD
 from kth_timeoutdecorator import *
 
+from nonebot import get_bot
+TIMELIMIT_JD = get_bot().config.TIMELIMIT_JD
+MAXLINE_JD = get_bot().config.MAXLINE_JD
+MAXWOED_JD = get_bot().config.MAXWOED_JD
+
+
 @timeout(TIMELIMIT_JD)
 async def get_definition_of_word(word: str) -> str:
     is_hidden = False