diff options
40 files changed, 858 insertions, 1063 deletions
diff --git a/ATRI/__init__.py b/ATRI/__init__.py index e2c75f8..6fb9773 100644 --- a/ATRI/__init__.py +++ b/ATRI/__init__.py @@ -21,7 +21,7 @@ def driver(): def init(): nb.init(**RUNTIME_CONFIG) driver().register_adapter("cqhttp", ATRIBot) - nb.load_plugins('ATRI/plugins') + nb.load_plugins("ATRI/plugins") if RUNTIME_CONFIG["debug"]: nb.load_plugin("nonebot_plugin_test") logger.info(f"Now running: {__version__}") diff --git a/ATRI/config.py b/ATRI/config.py index cce33aa..f1aa9e6 100644 --- a/ATRI/config.py +++ b/ATRI/config.py @@ -7,50 +7,52 @@ from pydantic import BaseConfig from .utils.yaml import load_yml -CONFIG_PATH = Path('.') / 'config.yml' +CONFIG_PATH = Path(".") / "config.yml" config = load_yml(CONFIG_PATH) class Config(BaseConfig): class BotSelfConfig: - config: dict = config['BotSelfConfig'] - - host: IPv4Address = IPv4Address(config.get('host', '127.0.0.1')) - port: int = int(config.get('port', 8080)) - debug: bool = bool(config.get('debug', False)) - superusers: set = set(config.get('superusers', ['1234567890'])) + config: dict = config["BotSelfConfig"] + + host: IPv4Address = IPv4Address(config.get("host", "127.0.0.1")) + port: int = int(config.get("port", 8080)) + debug: bool = bool(config.get("debug", False)) + superusers: set = set(config.get("superusers", ["1234567890"])) nickname: set = set( - config.get('nickname', ['ATRI', 'Atri', 'atri', '亚托莉', 'アトリ'])) - command_start: set = set(config.get('command_start', [''])) - command_sep: set = set(config.get('command_sep', ['.'])) + config.get("nickname", ["ATRI", "Atri", "atri", "亚托莉", "アトリ"]) + ) + command_start: set = set(config.get("command_start", [""])) + command_sep: set = set(config.get("command_sep", ["."])) session_expire_timeout: timedelta = timedelta( - config.get('session_expire_timeout', 2)) - session_exciting_time: int = int(config.get('session_exciting_time', 60)) - + config.get("session_expire_timeout", 2) + ) + session_exciting_time: int = int(config.get("session_exciting_time", 60)) + class NetworkPost: - config: dict = config['NetworkPost'] - - host: str = config.get('host', '127.0.0.1') - port: int = int(config.get('port', 8081)) - + config: dict = config["NetworkPost"] + + host: str = config.get("host", "127.0.0.1") + port: int = int(config.get("port", 8081)) + class AdminPage: - config: dict = config['AdminPage'] - - host: str = config.get('host', '127.0.0.1') - port: int = int(config.get('port', 8082)) - + config: dict = config["AdminPage"] + + host: str = config.get("host", "127.0.0.1") + port: int = int(config.get("port", 8082)) + class NsfwCheck: - config: dict = config['NsfwCheck'] - - enabled: bool = bool(config.get('enabled', False)) - passing_rate: float = float(config.get('passing_rate', 0.8)) - host: str = config.get('host', '127.0.0.1') - port: int = int(config.get('port', 5000)) - + config: dict = config["NsfwCheck"] + + enabled: bool = bool(config.get("enabled", False)) + passing_rate: float = float(config.get("passing_rate", 0.8)) + host: str = config.get("host", "127.0.0.1") + port: int = int(config.get("port", 5000)) + class SauceNAO: - config: dict = config['SauceNAO'] - - key: str = config.get('key', '') + config: dict = config["SauceNAO"] + + key: str = config.get("key", "") RUNTIME_CONFIG = { @@ -61,5 +63,5 @@ RUNTIME_CONFIG = { "nickname": Config.BotSelfConfig.nickname, "command_start": Config.BotSelfConfig.command_start, "command_sep": Config.BotSelfConfig.command_sep, - "session_expire_timeout": Config.BotSelfConfig.session_expire_timeout + "session_expire_timeout": Config.BotSelfConfig.session_expire_timeout, } diff --git a/ATRI/exceptions.py b/ATRI/exceptions.py index d863777..6563e94 100644 --- a/ATRI/exceptions.py +++ b/ATRI/exceptions.py @@ -15,35 +15,34 @@ from nonebot.message import run_postprocessor from .log import logger -ERROR_DIR = Path('.') / 'ATRI' / 'data' / 'errors' +ERROR_DIR = Path(".") / "ATRI" / "data" / "errors" os.makedirs(ERROR_DIR, exist_ok=True) def _save_error(prompt: str, content: str) -> str: - track_id = ''.join(sample(string.ascii_letters + string.digits, 8)) + track_id = "".join(sample(string.ascii_letters + string.digits, 8)) data = { "track_id": track_id, "prompt": prompt, - "time": time.strftime('%Y-%m-%d %H:%M:%S', time.localtime()), - "content": content + "time": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), + "content": content, } - path = ERROR_DIR / f'{track_id}.json' - with open(path, 'w', encoding='utf-8') as r: + path = ERROR_DIR / f"{track_id}.json" + with open(path, "w", encoding="utf-8") as r: r.write(json.dumps(data, indent=4)) return track_id def load_error(track_id: str) -> dict: - path = ERROR_DIR / f'{track_id}.json' + path = ERROR_DIR / f"{track_id}.json" return json.loads(path.read_bytes()) class BaseBotException(BaseException): - prompt: Optional[str] = 'ignore' + prompt: Optional[str] = "ignore" def __init__(self, prompt: Optional[str]) -> None: - self.prompt = prompt or self.__class__.prompt \ - or self.__class__.__name__ + self.prompt = prompt or self.__class__.prompt or self.__class__.__name__ self.track_id = _save_error(self.prompt, format_exc()) super().__init__(self.prompt) @@ -81,14 +80,16 @@ class FormatError(BaseBotException): @run_postprocessor # type: ignore -async def _track_error(matcher: Matcher, - exception: Optional[Exception], - bot: Bot, - event: Event, - state: T_State) -> None: +async def _track_error( + matcher: Matcher, + exception: Optional[Exception], + bot: Bot, + event: Event, + state: T_State, +) -> None: if exception is None: return - + try: raise exception except BaseBotException as Error: @@ -105,5 +106,5 @@ async def _track_error(matcher: Matcher, f"触发原因: {prompt}\n" "键入 来杯红茶 以联系维护者" ) - + await bot.send(event, msg) diff --git a/ATRI/log.py b/ATRI/log.py index 560facc..b26f4e3 100644 --- a/ATRI/log.py +++ b/ATRI/log.py @@ -7,10 +7,10 @@ from nonebot.log import logger, default_format from .config import Config -LOGGER_DIR = Path('.') / 'ATRI' / 'data' / 'logs' +LOGGER_DIR = Path(".") / "ATRI" / "data" / "logs" LOGGER_DIR.parent.mkdir(exist_ok=True, parents=True) -NOW_TIME = datetime.now().strftime('%Y%m%d-%H') +NOW_TIME = datetime.now().strftime("%Y%m%d-%H") log_format = ( "\033[36mATRI\033[0m " @@ -26,7 +26,7 @@ logger.add( sys.stdout, level="DEBUG" if Config.BotSelfConfig.debug else "INFO", colorize=True, - format=log_format + format=log_format, ) logger.add( @@ -35,7 +35,7 @@ logger.add( enqueue=True, level="INFO", encoding="utf-8", - format=log_format + format=log_format, ) logger.add( @@ -44,7 +44,7 @@ logger.add( enqueue=True, level="WARNING", encoding="utf-8", - format=log_format + format=log_format, ) logger.add( @@ -53,7 +53,7 @@ logger.add( enqueue=True, level="ERROR", encoding="utf-8", - format=log_format + format=log_format, ) logger.add( @@ -62,5 +62,5 @@ logger.add( enqueue=True, level="DEBUG", encoding="utf-8", - format=log_format + format=log_format, ) diff --git a/ATRI/plugins/anime_search.py b/ATRI/plugins/anime_search.py index a62ed5e..1c7e5a3 100644 --- a/ATRI/plugins/anime_search.py +++ b/ATRI/plugins/anime_search.py @@ -24,45 +24,40 @@ __doc__ = """ 以图搜番 (pic) """ -anime_search = sv.on_command( - cmd="以图搜番", - docs=__doc__, - rule=is_in_service('以图搜番') -) +anime_search = sv.on_command(cmd="以图搜番", docs=__doc__, rule=is_in_service("以图搜番")) + @anime_search.args_parser # type: ignore async def _load_anime(bot: Bot, event: MessageEvent, state: T_State) -> None: msg = str(event.message) - quit_list = ['算了', '罢了', '不搜了', '取消'] + quit_list = ["算了", "罢了", "不搜了", "取消"] if msg in quit_list: - await anime_search.finish('好吧...') + await anime_search.finish("好吧...") if not msg: - await anime_search.reject('图呢？') + await anime_search.reject("图呢？") else: - state['pic_anime'] = msg + state["pic_anime"] = msg + @anime_search.handle() -async def _anime_search(bot: Bot, - event: MessageEvent, - state: T_State) -> None: +async def _anime_search(bot: Bot, event: MessageEvent, state: T_State) -> None: msg = str(event.message).strip() if msg: - state['pic_anime'] = msg + state["pic_anime"] = msg + -@anime_search.got('pic_anime', prompt='图呢？') -async def _deal_search(bot: Bot, - event: MessageEvent, - state: T_State) -> None: - msg = state['pic_anime'] +@anime_search.got("pic_anime", prompt="图呢？") +async def _deal_search(bot: Bot, event: MessageEvent, state: T_State) -> None: + msg = state["pic_anime"] img = re.findall(r"url=(.*?)]", msg) if not img: await anime_search.reject("请发送图片而不是其它东西！！") - + try: req = await get_bytes(URL + img[0]) except RequestTimeOut: raise RequestTimeOut("Request failed!") - + data = json.loads(req)["docs"] try: d = dict() @@ -72,28 +67,24 @@ async def _deal_search(bot: Bot, else: m = data[i]["at"] / 60 s = data[i]["at"] % 60 - + if not data[i]["episode"]: n = 1 else: n = data[i]["episode"] - + d[to_simple_string(data[i]["title_chinese"])] = [ data[i]["similarity"], f"第{n}集", - f"{int(m)}分{int(s)}秒处" + f"{int(m)}分{int(s)}秒处", ] except Exception as err: raise Exception(f"Invalid data.\n{err}") - - result = sorted( - d.items(), - key=lambda x:x[1], - reverse=True - ) - + + result = sorted(d.items(), key=lambda x: x[1], reverse=True) + t = 0 - + msg0 = f"> {event.sender.nickname}" for i in result: t += 1 @@ -104,16 +95,16 @@ async def _deal_search(bot: Bot, f"Name: {i[0]}\n" f"Time: {i[1][1]} {i[1][2]}" ) - + if len(result) == 2: await anime_search.finish(Message(msg0)) else: data = FormData() - data.add_field('poster', 'ATRI running log') - data.add_field('syntax', 'text') - data.add_field('expiration', 'day') - data.add_field('content', msg0) + data.add_field("poster", "ATRI running log") + data.add_field("syntax", "text") + data.add_field("expiration", "day") + data.add_field("content", msg0) repo = f"> {event.sender.nickname}\n" repo = repo + f"详细请移步此处~\n{await paste(data)}" - await anime_search.finish(repo)
\ No newline at end of file + await anime_search.finish(repo) diff --git a/ATRI/plugins/call_owner.py b/ATRI/plugins/call_owner.py index cd79d1c..d282824 100644 --- a/ATRI/plugins/call_owner.py +++ b/ATRI/plugins/call_owner.py @@ -1,9 +1,6 @@ from nonebot.permission import SUPERUSER from nonebot.typing import T_State -from nonebot.adapters.cqhttp import ( - Bot, - MessageEvent -) +from nonebot.adapters.cqhttp import Bot, MessageEvent from ATRI.service import Service as sv from ATRI.config import Config @@ -21,48 +18,44 @@ __doc__ = """ repo = sv.on_command(cmd="来杯红茶", docs=__doc__) + @repo.args_parser # type: ignore async def _repo_load(bot: Bot, event: MessageEvent, state: T_State) -> None: msg = str(event.message) if msg == "算了": - await repo.finish('好吧') - + await repo.finish("好吧") + if not msg: - await repo.reject('话呢？') + await repo.reject("话呢？") else: - state['msg_repo'] = msg + state["msg_repo"] = msg + @repo.handle() async def _repo(bot: Bot, event: MessageEvent, state: T_State) -> None: msg = str(event.message).strip() if msg: - state['msg_repo'] = msg + state["msg_repo"] = msg + [email protected]('msg_repo', prompt="请告诉咱需要反馈的内容~！") [email protected]("msg_repo", prompt="请告诉咱需要反馈的内容~！") async def _repo_deal(bot: Bot, event: MessageEvent, state: T_State) -> None: global repo_list - msg = state['msg_repo'] + msg = state["msg_repo"] user = event.user_id - + if count_list(repo_list, user) == 5: await repo.finish("吾辈已经喝了五杯红茶啦！明天再来吧。") - + repo_list.append(user) for sup in Config.BotSelfConfig.superusers: - await bot.send_private_msg( - user_id=sup, - message=f"来自用户[{user}]反馈：\n{msg}" - ) - + await bot.send_private_msg(user_id=sup, message=f"来自用户[{user}]反馈：\n{msg}") + await repo.finish("吾辈的心愿已由咱转告给咱的维护者了~！") [email protected]_job( - "cron", - hour=0, - misfire_grace_time=60 -) [email protected]_job("cron", hour=0, misfire_grace_time=60) async def _() -> None: global repo_list repo_list.clear() @@ -75,11 +68,8 @@ __doc__ = """ /重置红茶 """ -reset_repo = sv.on_command( - cmd="重置红茶", - docs=__doc__, - permission=SUPERUSER -) +reset_repo = sv.on_command(cmd="重置红茶", docs=__doc__, permission=SUPERUSER) + @reset_repo.handle() async def _reset_repo(bot: Bot, event: MessageEvent) -> None: diff --git a/ATRI/plugins/code_runner.py b/ATRI/plugins/code_runner.py index b8f1106..d34872f 100644 --- a/ATRI/plugins/code_runner.py +++ b/ATRI/plugins/code_runner.py @@ -10,33 +10,33 @@ from ATRI.utils.request import post_bytes from ATRI.exceptions import RequestTimeOut -RUN_API_URL_FORMAT = 'https://glot.io/run/{}?version=latest' +RUN_API_URL_FORMAT = "https://glot.io/run/{}?version=latest" SUPPORTED_LANGUAGES = { - 'assembly': {'ext': 'asm'}, - 'bash': {'ext': 'sh'}, - 'c': {'ext': 'c'}, - 'clojure': {'ext': 'clj'}, - 'coffeescript': {'ext': 'coffe'}, - 'cpp': {'ext': 'cpp'}, - 'csharp': {'ext': 'cs'}, - 'erlang': {'ext': 'erl'}, - 'fsharp': {'ext': 'fs'}, - 'go': {'ext': 'go'}, - 'groovy': {'ext': 'groovy'}, - 'haskell': {'ext': 'hs'}, - 'java': {'ext': 'java', 'name': 'Main'}, - 'javascript': {'ext': 'js'}, - 'julia': {'ext': 'jl'}, - 'kotlin': {'ext': 'kt'}, - 'lua': {'ext': 'lua'}, - 'perl': {'ext': 'pl'}, - 'php': {'ext': 'php'}, - 'python': {'ext': 'py'}, - 'ruby': {'ext': 'rb'}, - 'rust': {'ext': 'rs'}, - 'scala': {'ext': 'scala'}, - 'swift': {'ext': 'swift'}, - 'typescript': {'ext': 'ts'}, + "assembly": {"ext": "asm"}, + "bash": {"ext": "sh"}, + "c": {"ext": "c"}, + "clojure": {"ext": "clj"}, + "coffeescript": {"ext": "coffe"}, + "cpp": {"ext": "cpp"}, + "csharp": {"ext": "cs"}, + "erlang": {"ext": "erl"}, + "fsharp": {"ext": "fs"}, + "go": {"ext": "go"}, + "groovy": {"ext": "groovy"}, + "haskell": {"ext": "hs"}, + "java": {"ext": "java", "name": "Main"}, + "javascript": {"ext": "js"}, + "julia": {"ext": "jl"}, + "kotlin": {"ext": "kt"}, + "lua": {"ext": "lua"}, + "perl": {"ext": "pl"}, + "php": {"ext": "php"}, + "python": {"ext": "py"}, + "ruby": {"ext": "rb"}, + "rust": {"ext": "rs"}, + "scala": {"ext": "scala"}, + "swift": {"ext": "swift"}, + "typescript": {"ext": "ts"}, } @@ -50,54 +50,55 @@ __doc__ = """ print('Hello world!') """ -code_runner = sv.on_command( - cmd="/code", - docs=__doc__, - rule=is_in_service('code') -) +code_runner = sv.on_command(cmd="/code", docs=__doc__, rule=is_in_service("code")) + @code_runner.handle() async def _code_runner(bot: Bot, event: MessageEvent) -> None: msg = str(event.message).split("\n") - + if msg[0] == "list": msg0 = "咱现在支持的语言如下：\n" msg0 += ", ".join(map(str, SUPPORTED_LANGUAGES.keys())) - + await code_runner.finish(msg0) elif not msg[0]: await code_runner.finish("请键入/help以获取更多支持...") - + laug = msg[0].replace("\r", "") if laug not in SUPPORTED_LANGUAGES: await code_runner.finish("该语言暂不支持...") - + del msg[0] code = "\n".join(map(str, msg)) try: req = await post_bytes( RUN_API_URL_FORMAT.format(laug), json={ - "files": [{ - "name": (SUPPORTED_LANGUAGES[laug].get("name", "main") + - f".{SUPPORTED_LANGUAGES[laug]['ext']}"), - "content": code - }], + "files": [ + { + "name": ( + SUPPORTED_LANGUAGES[laug].get("name", "main") + + f".{SUPPORTED_LANGUAGES[laug]['ext']}" + ), + "content": code, + } + ], "stdin": "", - "command": "" - } + "command": "", + }, ) except RequestTimeOut: raise RequestTimeOut("Failed to request!") - + payload = json.loads(req) sent = False - for k in ['stdout', 'stderr', 'error']: + for k in ["stdout", "stderr", "error"]: v = payload.get(k) lines = v.splitlines() lines, remained_lines = lines[:10], lines[10:] - out = '\n'.join(lines) - out, remained_out = out[:60 * 10], out[60 * 10:] + out = "\n".join(lines) + out, remained_out = out[: 60 * 10], out[60 * 10 :] if remained_lines or remained_out: out += f"\n（太多了太多了...）" @@ -105,6 +106,6 @@ async def _code_runner(bot: Bot, event: MessageEvent) -> None: if out: await bot.send(event, f"{k}:\n\n{out}") sent = True - + if not sent: await code_runner.finish("Running success! Nothing print.") diff --git a/ATRI/plugins/curse/__init__.py b/ATRI/plugins/curse/__init__.py index 5085dbf..acfaa9b 100644 --- a/ATRI/plugins/curse/__init__.py +++ b/ATRI/plugins/curse/__init__.py @@ -19,23 +19,17 @@ __doc__ = """ """ curse = sv.on_command( - cmd='口臭', - docs=__doc__, - aliases={'口臭一下，骂我'}, - rule=is_in_service('口臭') + cmd="口臭", docs=__doc__, aliases={"口臭一下，骂我"}, rule=is_in_service("口臭") ) + @curse.handle() async def _curse(bot: Bot, event: MessageEvent) -> None: global sick_list user = event.get_user_id() if count_list(sick_list, user) == 3: sick_list.append(user) - repo = ( - "不是？？你这么想被咱骂的嘛？？" - "被咱骂就这么舒服的吗？！" - "该......你该不会是.....M吧！" - ) + repo = "不是？？你这么想被咱骂的嘛？？" "被咱骂就这么舒服的吗？！" "该......你该不会是.....M吧！" await curse.finish(repo) elif count_list(sick_list, user) == 6: sick_list = del_list_aim(sick_list, user) diff --git a/ATRI/plugins/essential.py b/ATRI/plugins/essential.py index 3ebaa36..7f14882 100644 --- a/ATRI/plugins/essential.py +++ b/ATRI/plugins/essential.py @@ -24,7 +24,7 @@ from nonebot.adapters.cqhttp import ( LuckyKingNotifyEvent, GroupUploadNoticeEvent, GroupRecallNoticeEvent, - FriendRecallNoticeEvent + FriendRecallNoticeEvent, ) import ATRI @@ -35,8 +35,8 @@ from ATRI.service import Service as sv from ATRI.utils.cqcode import coolq_code_check -PLUGIN_INFO_DIR = Path('.') / 'ATRI' / 'data' / 'service' / 'services' -ESSENTIAL_DIR = Path('.') / 'ATRI' / 'data' / 'database' / 'essential' +PLUGIN_INFO_DIR = Path(".") / "ATRI" / "data" / "service" / "services" +ESSENTIAL_DIR = Path(".") / "ATRI" / "data" / "database" / "essential" os.makedirs(PLUGIN_INFO_DIR, exist_ok=True) os.makedirs(ESSENTIAL_DIR, exist_ok=True) @@ -57,11 +57,7 @@ async def shutdown() -> None: shutil.rmtree(PLUGIN_INFO_DIR) logger.debug("成功！") except Exception: - repo = ( - '清理插件信息失败', - '请前往 ATRI/data/service/services 下', - '将 services 整个文件夹删除' - ) + repo = ("清理插件信息失败", "请前往 ATRI/data/service/services 下", "将 services 整个文件夹删除") time.sleep(10) raise Exception(repo) @@ -69,59 +65,49 @@ async def shutdown() -> None: @driver.on_bot_connect async def connect(bot) -> None: for superuser in Config.BotSelfConfig.superusers: - await sv.NetworkPost.send_private_msg( - int(superuser), - "WebSocket 成功连接，数据开始传输。" - ) + await sv.NetworkPost.send_private_msg(int(superuser), "WebSocket 成功连接，数据开始传输。") @driver.on_bot_disconnect async def disconnect(bot) -> None: for superuser in Config.BotSelfConfig.superusers: try: - await sv.NetworkPost.send_private_msg( - int(superuser), - "WebSocket 貌似断开了呢..." - ) + await sv.NetworkPost.send_private_msg(int(superuser), "WebSocket 貌似断开了呢...") except: logger.error("WebSocket 已断开，等待重连") @run_preprocessor # type: ignore -async def _check_block(matcher: Matcher, - bot: Bot, - event: MessageEvent, - state: T_State) -> None: +async def _check_block( + matcher: Matcher, bot: Bot, event: MessageEvent, state: T_State +) -> None: user = str(event.user_id) try: msg = str(event.message) except: msg = "" if not sv.BlockSystem.auth_user(user): - raise IgnoredException(f'Block user: {user}') - + raise IgnoredException(f"Block user: {user}") + if not sv.Dormant.is_dormant(): if "/dormant" not in msg: - raise IgnoredException('Bot has been dormant.') - + raise IgnoredException("Bot has been dormant.") + if isinstance(event, GroupMessageEvent): group = str(event.group_id) if not sv.BlockSystem.auth_group(group): - raise IgnoredException(f'Block group: {group}') + raise IgnoredException(f"Block group: {group}") @run_preprocessor # type: ignore -async def _store_message(matcher: Matcher, - bot: Bot, - event, - state: T_State) -> None: +async def _store_message(matcher: Matcher, bot: Bot, event, state: T_State) -> None: if isinstance(event, GroupMessageEvent): if event.sub_type == "normal": - now_time = datetime.now().strftime('%Y-%m-%d') - GROUP_DIR = ESSENTIAL_DIR / 'chat_history' / f'{event.group_id}' + now_time = datetime.now().strftime("%Y-%m-%d") + GROUP_DIR = ESSENTIAL_DIR / "chat_history" / f"{event.group_id}" os.makedirs(GROUP_DIR, exist_ok=True) path = GROUP_DIR / f"{now_time}.chat.json" - now_time = datetime.now().strftime('%Y%m%d-%H%M%S') + now_time = datetime.now().strftime("%Y%m%d-%H%M%S") try: data = json.loads(path.read_bytes()) @@ -147,12 +133,12 @@ async def _store_message(matcher: Matcher, "area": event.sender.area, "level": event.sender.level, "role": event.sender.role, - "title": event.sender.title + "title": event.sender.title, }, - "to_me": str(event.to_me) + "to_me": str(event.to_me), } try: - with open(path, 'w', encoding='utf-8') as r: + with open(path, "w", encoding="utf-8") as r: r.write(json.dumps(data, indent=4)) logger.debug(f"写入消息成功，id: {event.message_id}") except WriteError: @@ -168,30 +154,24 @@ async def _store_message(matcher: Matcher, # 处理：好友请求 request_friend_event = sv.on_request() + @request_friend_event.handle() async def _request_friend_event(bot, event: FriendRequestEvent) -> None: file_name = "request_friend.json" path = ESSENTIAL_DIR / file_name path.parent.mkdir(exist_ok=True, parents=True) - + try: data = json.loads(path.read_bytes()) except: data = dict() - data[event.flag] = { - "user_id": event.user_id, - "comment": event.comment - } + data[event.flag] = {"user_id": event.user_id, "comment": event.comment} try: - with open(path, 'w', encoding='utf-8') as r: - r.write( - json.dumps( - data, indent=4 - ) - ) + with open(path, "w", encoding="utf-8") as r: + r.write(json.dumps(data, indent=4)) except WriteError: raise WriteError("Writing file failed!") - + for superuser in Config.BotSelfConfig.superusers: msg = ( "主人，收到一条好友请求：\n" @@ -199,21 +179,19 @@ async def _request_friend_event(bot, event: FriendRequestEvent) -> None: f"申请信息：{event.comment}\n" f"申请码：{event.flag}" ) - await sv.NetworkPost.send_private_msg( - user_id=int(superuser), - message=msg - ) + await sv.NetworkPost.send_private_msg(user_id=int(superuser), message=msg) # 处理：邀请入群，如身为管理，还附有入群请求 request_group_event = sv.on_request() + @request_group_event.handle() async def _request_group_event(bot, event: GroupRequestEvent) -> None: file_name = "request_group.json" path = ESSENTIAL_DIR / file_name path.parent.mkdir(exist_ok=True, parents=True) - + try: data = json.loads(path.read_bytes()) except: @@ -222,18 +200,14 @@ async def _request_group_event(bot, event: GroupRequestEvent) -> None: "user_id": event.user_id, "group_id": event.group_id, "sub_type": event.sub_type, - "comment": event.comment + "comment": event.comment, } try: - with open(path, 'w', encoding='utf-8') as r: - r.write( - json.dumps( - data, indent=4 - ) - ) + with open(path, "w", encoding="utf-8") as r: + r.write(json.dumps(data, indent=4)) except WriteError: raise WriteError("Writing file failed!") - + for superuser in Config.BotSelfConfig.superusers: msg = ( "主人，收到一条入群请求：\n" @@ -241,37 +215,27 @@ async def _request_group_event(bot, event: GroupRequestEvent) -> None: f"申请信息：{event.comment}\n" f"申请码：{event.flag}" ) - await sv.NetworkPost.send_private_msg( - user_id=int(superuser), - message=msg - ) - + await sv.NetworkPost.send_private_msg(user_id=int(superuser), message=msg) + # 处理群成员变动 group_member_event = sv.on_notice() + @group_member_event.handle() async def _group_member_event(bot: Bot, event: GroupIncreaseNoticeEvent) -> None: - msg = ( - "好欸！事新人！\n" - f"在下 {choice(list(Config.BotSelfConfig.nickname))} 哒!w!" - ) + msg = "好欸！事新人！\n" f"在下 {choice(list(Config.BotSelfConfig.nickname))} 哒!w!" await group_member_event.finish(msg) + @group_member_event.handle() async def _gro(bot: Bot, event: GroupDecreaseNoticeEvent) -> None: if event.is_tome(): if event.user_id != event.self_id: return - msg = ( - "呜呜呜，主人" - f"咱被群 {event.group_id} 里的 {event.operator_id} 扔出来了..." - ) + msg = "呜呜呜，主人" f"咱被群 {event.group_id} 里的 {event.operator_id} 扔出来了..." for superuser in Config.BotSelfConfig.superusers: - await sv.NetworkPost.send_private_msg( - user_id=int(superuser), - message=msg - ) + await sv.NetworkPost.send_private_msg(user_id=int(superuser), message=msg) else: await group_member_event.finish("阿！有人离开了我们...") @@ -279,26 +243,27 @@ async def _gro(bot: Bot, event: GroupDecreaseNoticeEvent) -> None: # 处理群管理事件 group_admin_event = sv.on_notice() + @group_admin_event.handle() async def _group_admin_event(bot: Bot, event: GroupAdminNoticeEvent) -> None: if not event.is_tome(): return - + for superuser in Config.BotSelfConfig.superusers: await sv.NetworkPost.send_private_msg( - user_id=int(superuser), - message=f"好欸！主人！我在群 {event.group_id} 成为了管理！！" + user_id=int(superuser), message=f"好欸！主人！我在群 {event.group_id} 成为了管理！！" ) # 处理群禁言事件 group_ban_event = sv.on_notice() + @group_ban_event.handle() async def _group_ban_event(bot: Bot, event: GroupBanNoticeEvent) -> None: if not event.is_tome(): return - + if event.duration: msg = ( "那个..。，主人\n" @@ -306,71 +271,54 @@ async def _group_ban_event(bot: Bot, event: GroupBanNoticeEvent) -> None: f"时长...是 {event.duration} 秒" ) for superuser in Config.BotSelfConfig.superusers: - await sv.NetworkPost.send_private_msg( - user_id=int(superuser), - message=msg - ) + await sv.NetworkPost.send_private_msg(user_id=int(superuser), message=msg) else: - msg = ( - "好欸！主人\n" - f"咱在群 {event.group_id} 被 {event.operator_id} 上的口球解除了！" - ) + msg = "好欸！主人\n" f"咱在群 {event.group_id} 被 {event.operator_id} 上的口球解除了！" for superuser in Config.BotSelfConfig.superusers: - await sv.NetworkPost.send_private_msg( - user_id=int(superuser), - message=msg - ) + await sv.NetworkPost.send_private_msg(user_id=int(superuser), message=msg) # 处理群红包运气王事件 lucky_read_bag_event = sv.on_notice() + @lucky_read_bag_event.handle() async def _lucky_read_bag_event(bot, event: LuckyKingNotifyEvent) -> None: - msg = ( - "8行，这可忍？" - f"gkd [CQ:at,qq={event.user_id}] 发一个！" - ) + msg = "8行，这可忍？" f"gkd [CQ:at,qq={event.user_id}] 发一个！" await lucky_read_bag_event.finish(Message(msg)) # 处理群文件上传事件 group_file_upload_event = sv.on_notice() + @group_file_upload_event.handle() -async def _group_file_upload_event(bot, - event: GroupUploadNoticeEvent) -> None: +async def _group_file_upload_event(bot, event: GroupUploadNoticeEvent) -> None: await group_file_upload_event.finish("让我康康传了啥好东西") # 处理撤回事件 recall_event = sv.on_notice() + @recall_event.handle() async def _recall_event(bot: Bot, event: GroupRecallNoticeEvent) -> None: try: repo = await bot.get_msg(message_id=event.message_id) except: return - + group = event.group_id repo = str(repo["message"]) check = await coolq_code_check(repo, group=group) if not check: repo = repo.replace("CQ", "QC") - msg = ( - "主人，咱拿到了一条撤回信息！\n" - f"{event.user_id}@[群:{event.group_id}]\n" - "撤回了\n" - f"{repo}" - ) + msg = "主人，咱拿到了一条撤回信息！\n" f"{event.user_id}@[群:{event.group_id}]\n" "撤回了\n" f"{repo}" for superuser in Config.BotSelfConfig.superusers: - await sv.NetworkPost.send_private_msg( - user_id=int(superuser), - message=msg - ) + await sv.NetworkPost.send_private_msg(user_id=int(superuser), message=msg) + @recall_event.handle() async def _rec(bot: Bot, event: FriendRecallNoticeEvent) -> None: @@ -378,23 +326,15 @@ async def _rec(bot: Bot, event: FriendRecallNoticeEvent) -> None: repo = await bot.get_msg(message_id=event.message_id) except: return - + user = event.user_id repo = str(repo["message"]) check = await coolq_code_check(repo, user) if not check: repo = repo.replace("CQ", "QC") - msg = ( - "主人，咱拿到了一条撤回信息！\n" - f"{event.user_id}@[私聊]" - "撤回了\n" - f"{repo}" - ) + msg = "主人，咱拿到了一条撤回信息！\n" f"{event.user_id}@[私聊]" "撤回了\n" f"{repo}" await bot.send(event, "咱看到惹~！") for superuser in Config.BotSelfConfig.superusers: - await sv.NetworkPost.send_private_msg( - user_id=int(superuser), - message=msg - ) + await sv.NetworkPost.send_private_msg(user_id=int(superuser), message=msg) diff --git a/ATRI/plugins/funny.py b/ATRI/plugins/funny.py index 03b844a..72e5abb 100644 --- a/ATRI/plugins/funny.py +++ b/ATRI/plugins/funny.py @@ -22,28 +22,26 @@ __doc__ = """ 来句笑话 """ -get_laugh = sv.on_command( - cmd='来句笑话', - docs=__doc__, - rule=is_in_service('来句笑话') -) +get_laugh = sv.on_command(cmd="来句笑话", docs=__doc__, rule=is_in_service("来句笑话")) + @get_laugh.handle() async def _get_laugh(bot: Bot, event: MessageEvent) -> None: user_name = event.sender.nickname laugh_list = [] - - FILE = Path('.') / 'ATRI' / 'data' / 'database' / 'funny' / 'laugh.txt' - with open(FILE, 'r', encoding='utf-8') as r: + + FILE = Path(".") / "ATRI" / "data" / "database" / "funny" / "laugh.txt" + with open(FILE, "r", encoding="utf-8") as r: for line in r: - laugh_list.append(line.strip('\n')) - + laugh_list.append(line.strip("\n")) + result = choice(laugh_list) await get_laugh.finish(result.replace("%name", user_name)) me_to_you = sv.on_message() + @me_to_you.handle() async def _me_to_you(bot: Bot, event: MessageEvent) -> None: if randint(0, 15) == 5: @@ -59,38 +57,28 @@ __doc__ = """ 抽老婆 """ -roll_wife = sv.on_command( - cmd='抽老婆', - docs=__doc__, - rule=is_in_service('抽老婆') -) +roll_wife = sv.on_command(cmd="抽老婆", docs=__doc__, rule=is_in_service("抽老婆")) + @roll_wife.handle() async def _roll_wife(bot: Bot, event: GroupMessageEvent) -> None: user = event.user_id group = event.group_id - - user_name = await bot.get_group_member_info(group_id=group, - user_id=user) - user_name = user_name['nickname'] + + user_name = await bot.get_group_member_info(group_id=group, user_id=user) + user_name = user_name["nickname"] run = await is_too_exciting(user, group, 5, True) if not run: return - + luck_list = await bot.get_group_member_list(group_id=group) luck_user = choice(luck_list) - luck_qq = luck_user['user_id'] - luck_user = luck_user['nickname'] - msg = ( - "5秒后咱将随机抽取一位群友成为\n" - f"{user_name} 的老婆！究竟是谁呢~？" - ) + luck_qq = luck_user["user_id"] + luck_user = luck_user["nickname"] + msg = "5秒后咱将随机抽取一位群友成为\n" f"{user_name} 的老婆！究竟是谁呢~？" await bot.send(event, msg) await asyncio.sleep(5) - msg = ( - f"> {luck_user}({luck_qq})\n" - f"恭喜成为 {user_name} 的老婆~⭐" - ) + msg = f"> {luck_user}({luck_qq})\n" f"恭喜成为 {user_name} 的老婆~⭐" await bot.send(event, msg) @@ -107,43 +95,34 @@ __doc__ = """ /fakemsg 123456789*生草人*草 114514*仙贝*臭死了 """ -fake_msg = sv.on_command( - cmd="/fakemsg", - docs=__doc__, - rule=is_in_service('fakemsg') -) +fake_msg = sv.on_command(cmd="/fakemsg", docs=__doc__, rule=is_in_service("fakemsg")) + @fake_msg.handle() async def _fake_msg(bot: Bot, event: GroupMessageEvent) -> None: - msg = str(event.message).split(' ') + msg = str(event.message).split(" ") user = event.user_id group = event.group_id node = [] check = await is_too_exciting(user, group, 1, True) - + if check: for i in msg: - args = i.split('*') + args = i.split("*") qq = args[0] - name = args[1].replace('[', '[') - name = name.replace(']', ']') - repo = args[2].replace('[', '[') - repo = repo.replace(']', ']') - dic = { - "type": "node", - "data": { - "name": name, - "uin": qq, - "content": repo - } - } + name = args[1].replace("[", "[") + name = name.replace("]", "]") + repo = args[2].replace("[", "[") + repo = repo.replace("]", "]") + dic = {"type": "node", "data": {"name": name, "uin": qq, "content": repo}} node.append(dic) await bot.send_group_forward_msg(group_id=group, messages=node) EAT_URL = "https://wtf.hiigara.net/api/run/{}" -eat_wat = sv.on_regex(r"[今|明|后|大后]天(.*?)吃什么", rule=is_in_service('今天吃什么')) +eat_wat = sv.on_regex(r"[今|明|后|大后]天(.*?)吃什么", rule=is_in_service("今天吃什么")) + @eat_wat.handle() async def _eat(bot: Bot, event: MessageEvent) -> None: @@ -152,7 +131,7 @@ async def _eat(bot: Bot, event: MessageEvent) -> None: user_n = event.sender.nickname arg = re.findall(r"[今|明|后|大后]天(.*?)吃什么", msg)[0] nd = re.match(r"[今|明|后|大后][天]", msg)[0] - + if arg == "中午": a = f"LdS4K6/{randint(0, 999999)}" url = EAT_URL.format(a) @@ -160,12 +139,12 @@ async def _eat(bot: Bot, event: MessageEvent) -> None: try: data = json.loads(await post_bytes(url, params)) except RequestTimeOut: - raise RequestTimeOut('Request failed!') - - text = to_simple_string(data['text']).replace('今天', nd) + raise RequestTimeOut("Request failed!") + + text = to_simple_string(data["text"]).replace("今天", nd) get_a = re.search(r"非常(.*?)的", text)[0] - result = f"> {MessageSegment.at(user)}\n" + text.replace(get_a, '') - + result = f"> {MessageSegment.at(user)}\n" + text.replace(get_a, "") + elif arg == "晚上": a = f"KaTMS/{randint(0, 999999)}" url = EAT_URL.format(a) @@ -173,11 +152,11 @@ async def _eat(bot: Bot, event: MessageEvent) -> None: try: data = json.loads(await post_bytes(url, params)) except RequestTimeOut: - raise RequestTimeOut('Request failed!') - - text = to_simple_string(data['text']).replace('今天', '') + raise RequestTimeOut("Request failed!") + + text = to_simple_string(data["text"]).replace("今天", "") result = f"> {MessageSegment.at(user)}\n" + text - + else: rd = randint(1, 10) if rd == 5: @@ -189,10 +168,12 @@ async def _eat(bot: Bot, event: MessageEvent) -> None: try: data = json.loads(await post_bytes(url, params)) except RequestTimeOut: - raise RequestTimeOut('Request failed!') - - text = to_simple_string(data['text']).replace('今天', nd) + raise RequestTimeOut("Request failed!") + + text = to_simple_string(data["text"]).replace("今天", nd) get_a = re.match(r"(.*?)的智商", text)[0] - result = f"> {MessageSegment.at(user)}\n" + text.replace(get_a, f'{user_n}的智商') - + result = f"> {MessageSegment.at(user)}\n" + text.replace( + get_a, f"{user_n}的智商" + ) + await eat_wat.finish(Message(result)) diff --git a/ATRI/plugins/github.py b/ATRI/plugins/github.py index c238b69..e7c0d80 100644 --- a/ATRI/plugins/github.py +++ b/ATRI/plugins/github.py @@ -12,6 +12,7 @@ URL = "https://api.github.com/repos/{owner}/{repo}/issues/{issue_number}" github_issues = sv.on_message() + @github_issues.handle() async def _github_issues(bot: Bot, event: MessageEvent) -> None: msg = str(event.message) @@ -19,21 +20,19 @@ async def _github_issues(bot: Bot, event: MessageEvent) -> None: need_info = re.findall(patt, msg) if not need_info: return - + for i in need_info: need_info = list(i) owner = need_info[0] repo = need_info[1] issue_number = need_info[2] - url = URL.format(owner=owner, - repo=repo, - issue_number=issue_number) - + url = URL.format(owner=owner, repo=repo, issue_number=issue_number) + try: data = await get_bytes(url) except RequestTimeOut: return - + data = json.loads(data) msg0 = ( f"{repo}: #{issue_number} {data['state']}\n" diff --git a/ATRI/plugins/help.py b/ATRI/plugins/help.py index 32d3f65..d2754b1 100644 --- a/ATRI/plugins/help.py +++ b/ATRI/plugins/help.py @@ -7,7 +7,7 @@ from ATRI.service import SERVICE_DIR from ATRI.service import Service as sv -SERVICE_DIR = SERVICE_DIR / 'services' +SERVICE_DIR = SERVICE_DIR / "services" __doc__ = """ @@ -19,14 +19,12 @@ __doc__ = """ /help info (cmd) """ -help = sv.on_command( - cmd="/help", - docs=__doc__ -) +help = sv.on_command(cmd="/help", docs=__doc__) + @help.handle() async def _help(bot: Bot, event: MessageEvent) -> None: - msg = str(event.message).split(' ') + msg = str(event.message).split(" ") if msg[0] == "": msg = ( "呀？找不到路了？\n" @@ -41,7 +39,7 @@ async def _help(bot: Bot, event: MessageEvent) -> None: for _, _, i in os.walk(SERVICE_DIR): for a in i: f = SERVICE_DIR / a - files.append(json.loads(f.read_bytes())['command']) + files.append(json.loads(f.read_bytes())["command"]) cmds = " | ".join(map(str, files)) msg = "咱能做很多事！比如：\n" + cmds msg0 = msg + "\n没反应可能是没权限...或者为探测类型...不属于可直接触发命令..." @@ -53,13 +51,9 @@ async def _help(bot: Bot, event: MessageEvent) -> None: try: data = json.loads(path.read_bytes()) except: - await help.finish('未找到相关命令...') + await help.finish("未找到相关命令...") - msg = ( - f"{cmd} INFO:\n" - f"Enabled: {data['enabled']}\n" - f"{data['docs']}" - ) + msg = f"{cmd} INFO:\n" f"Enabled: {data['enabled']}\n" f"{data['docs']}" await help.finish(msg) else: - await help.finish('请检查输入...') + await help.finish("请检查输入...") diff --git a/ATRI/plugins/hitokoto.py b/ATRI/plugins/hitokoto.py index 2b8d81f..1078ca3 100644 --- a/ATRI/plugins/hitokoto.py +++ b/ATRI/plugins/hitokoto.py @@ -11,7 +11,7 @@ from ATRI.utils.request import get_bytes URL = [ "https://cdn.jsdelivr.net/gh/hitokoto-osc/[email protected]/sentences/a.json", "https://cdn.jsdelivr.net/gh/hitokoto-osc/[email protected]/sentences/b.json", - "https://cdn.jsdelivr.net/gh/hitokoto-osc/[email protected]/sentences/c.json" + "https://cdn.jsdelivr.net/gh/hitokoto-osc/[email protected]/sentences/c.json", ] sick_list = [] @@ -24,12 +24,10 @@ __doc__ = """ """ hitokoto = sv.on_command( - cmd='一言', - aliases={'抑郁一下', '网抑云'}, - docs=__doc__, - rule=is_in_service('一言') & to_bot() + cmd="一言", aliases={"抑郁一下", "网抑云"}, docs=__doc__, rule=is_in_service("一言") & to_bot() ) + @hitokoto.handle() async def _hitokoto(bot: Bot, event: MessageEvent) -> None: global sick_list @@ -41,10 +39,7 @@ async def _hitokoto(bot: Bot, event: MessageEvent) -> None: await hitokoto.finish("额......需要咱安慰一下嘛~？") elif count_list(sick_list, user) == 6: sick_list = del_list_aim(sick_list, user) - msg = ( - "如果心里感到难受就赶快去睡觉！别再憋自己了！\n" - "我...我会守在你身边的！...嗯..一定" - ) + msg = "如果心里感到难受就赶快去睡觉！别再憋自己了！\n" "我...我会守在你身边的！...嗯..一定" await hitokoto.finish(msg) else: sick_list.append(user) @@ -53,4 +48,4 @@ async def _hitokoto(bot: Bot, event: MessageEvent) -> None: data = json.loads(await get_bytes(url)) except RequestTimeOut: raise RequestTimeOut("Request failed!") - await hitokoto.finish(data[randint(1, len(data) - 1)]['hitokoto']) + await hitokoto.finish(data[randint(1, len(data) - 1)]["hitokoto"]) diff --git a/ATRI/plugins/manage/__init__.py b/ATRI/plugins/manage/__init__.py index d9f1ffe..b316020 100644 --- a/ATRI/plugins/manage/__init__.py +++ b/ATRI/plugins/manage/__init__.py @@ -4,5 +4,4 @@ from pathlib import Path _sub_plugins = set() -_sub_plugins |= nonebot.load_plugins( - str((Path(__file__).parent / 'modules').resolve())) +_sub_plugins |= nonebot.load_plugins(str((Path(__file__).parent / "modules").resolve())) diff --git a/ATRI/plugins/manage/modules/block.py b/ATRI/plugins/manage/modules/block.py index 65a6a86..0c2fcec 100644 --- a/ATRI/plugins/manage/modules/block.py +++ b/ATRI/plugins/manage/modules/block.py @@ -12,28 +12,24 @@ __doc__ = """ 封禁用户 QQ号 """ -block_user = sv.on_command( - cmd="封禁用户", - docs=__doc__, - permission=SUPERUSER -) +block_user = sv.on_command(cmd="封禁用户", docs=__doc__, permission=SUPERUSER) + @block_user.args_parser # type: ignore -async def _block_user_load(bot: Bot, - event: MessageEvent, - state: T_State) -> None: +async def _block_user_load(bot: Bot, event: MessageEvent, state: T_State) -> None: msg = str(event.message).strip() - cancel = ['算了', '罢了'] + cancel = ["算了", "罢了"] if msg in cancel: - await block_user.finish('好吧...') + await block_user.finish("好吧...") if not msg: - await block_user.reject('是谁呢？！GKD！') + await block_user.reject("是谁呢？！GKD！") else: - state['noob'] = msg + state["noob"] = msg + @block_user.handle() async def _block_user(bot: Bot, event: MessageEvent, state: T_State) -> None: - noob = state['noob'] + noob = state["noob"] sv.BlockSystem.control_list(True, user=noob) msg = f"用户[{noob}]已被封禁(；′⌒`)" await block_user.finish(msg) @@ -46,28 +42,24 @@ __doc__ = """ 解封用户 QQ号 """ -unblock_user = sv.on_command( - cmd="解封用户", - docs=__doc__, - permission=SUPERUSER -) +unblock_user = sv.on_command(cmd="解封用户", docs=__doc__, permission=SUPERUSER) + @unblock_user.args_parser # type: ignore -async def _unblock_user_load(bot: Bot, - event: MessageEvent, - state: T_State) -> None: +async def _unblock_user_load(bot: Bot, event: MessageEvent, state: T_State) -> None: msg = str(event.message).strip() - cancel = ['算了', '罢了'] + cancel = ["算了", "罢了"] if msg in cancel: - await unblock_user.finish('好吧...') + await unblock_user.finish("好吧...") if not msg: - await unblock_user.reject('要原谅谁呢...') + await unblock_user.reject("要原谅谁呢...") else: - state['forgive'] = msg + state["forgive"] = msg + @unblock_user.handle() async def _unblock_user(bot: Bot, event: MessageEvent, state: T_State) -> None: - forgive = state['forgive'] + forgive = state["forgive"] sv.BlockSystem.control_list(False, user=forgive) msg = f"用户[{forgive}]已被解封ヾ(´･ω･｀)ﾉ" await unblock_user.finish(msg) @@ -80,28 +72,24 @@ __doc__ = """ 封禁群 群号 """ -block_group = sv.on_command( - cmd="封禁群", - docs=__doc__, - permission=SUPERUSER -) +block_group = sv.on_command(cmd="封禁群", docs=__doc__, permission=SUPERUSER) + @block_group.args_parser # type: ignore -async def _block_group_load(bot: Bot, - event: MessageEvent, - state: T_State) -> None: +async def _block_group_load(bot: Bot, event: MessageEvent, state: T_State) -> None: msg = str(event.message).strip() - cancel = ['算了', '罢了'] + cancel = ["算了", "罢了"] if msg in cancel: - await block_user.finish('好吧...') + await block_user.finish("好吧...") if not msg: - await block_user.reject('是哪个群？！GKD！') + await block_user.reject("是哪个群？！GKD！") else: - state['noob_g'] = msg + state["noob_g"] = msg + @block_group.handle() async def _block_group(bot: Bot, event: MessageEvent, state: T_State) -> None: - noob_g = state['noob_g'] + noob_g = state["noob_g"] sv.BlockSystem.control_list(True, group=noob_g) msg = f"群[{noob_g}]已被封禁(；′⌒`)" await block_user.finish(msg) @@ -114,30 +102,24 @@ __doc__ = """ 解封 群号 """ -unblock_group = sv.on_command( - cmd="解封群", - docs=__doc__, - permission=SUPERUSER -) +unblock_group = sv.on_command(cmd="解封群", docs=__doc__, permission=SUPERUSER) + @unblock_group.args_parser # type: ignore -async def _unblock_group_load(bot: Bot, - event: MessageEvent, - state: T_State) -> None: +async def _unblock_group_load(bot: Bot, event: MessageEvent, state: T_State) -> None: msg = str(event.message).strip() - cancel = ['算了', '罢了'] + cancel = ["算了", "罢了"] if msg in cancel: - await block_user.finish('好吧...') + await block_user.finish("好吧...") if not msg: - await block_user.reject('要原谅哪个群呢...') + await block_user.reject("要原谅哪个群呢...") else: - state['forgive_g'] = msg + state["forgive_g"] = msg + @unblock_group.handle() -async def _unblock_group(bot: Bot, - event: MessageEvent, - state: T_State) -> None: - forgive_g = state['forgive_g'] +async def _unblock_group(bot: Bot, event: MessageEvent, state: T_State) -> None: + forgive_g = state["forgive_g"] sv.BlockSystem.control_list(False, group=forgive_g) msg = f"群[{forgive_g}]已被解封ヾ(´･ω･｀)ﾉ" await unblock_user.finish(msg) diff --git a/ATRI/plugins/manage/modules/broadcast.py b/ATRI/plugins/manage/modules/broadcast.py index 5086fcf..7f7816d 100644 --- a/ATRI/plugins/manage/modules/broadcast.py +++ b/ATRI/plugins/manage/modules/broadcast.py @@ -15,52 +15,46 @@ __doc__ = """ 广播 内容 """ -broadcast = sv.on_command( - cmd="广播", - docs=__doc__, - permission=SUPERUSER -) +broadcast = sv.on_command(cmd="广播", docs=__doc__, permission=SUPERUSER) + @broadcast.args_parser # type: ignore -async def _broadcast_load(bot: Bot, - event: MessageEvent, - state: T_State) -> None: +async def _broadcast_load(bot: Bot, event: MessageEvent, state: T_State) -> None: msg = str(event.message) - quit_list = ['算了', '罢了', '取消'] + quit_list = ["算了", "罢了", "取消"] if msg in quit_list: - await broadcast.finish('好吧...') + await broadcast.finish("好吧...") if not msg: - await broadcast.reject('想群发啥呢0w0') + await broadcast.reject("想群发啥呢0w0") else: - state['msg'] = msg + state["msg"] = msg + @broadcast.handle() async def _broadcast(bot: Bot, event: MessageEvent, state: T_State) -> None: msg = str(event.message).strip() if msg: - state['msg'] = msg - [email protected]('msg', prompt='请告诉咱需要群发的内容~！') -async def _deal_broadcast(bot: Bot, - event: MessageEvent, - state: T_State) -> None: - msg = state['msg'] + state["msg"] = msg + + [email protected]("msg", prompt="请告诉咱需要群发的内容~！") +async def _deal_broadcast(bot: Bot, event: MessageEvent, state: T_State) -> None: + msg = state["msg"] group_list = await bot.get_group_list() succ_list = [] err_list = [] - + for group in group_list: await asyncio.sleep(randint(0, 2)) try: - await bot.send_group_msg(group_id=group["group_id"], - message=msg) + await bot.send_group_msg(group_id=group["group_id"], message=msg) except BaseException: err_list.append(group["group_id"]) - + msg0 = "" for i in err_list: msg0 += f" {i}\n" - + repo_msg = ( f"推送消息：\n{msg}\n" "————————\n" diff --git a/ATRI/plugins/manage/modules/debug.py b/ATRI/plugins/manage/modules/debug.py index 53b75b7..7bd3992 100644 --- a/ATRI/plugins/manage/modules/debug.py +++ b/ATRI/plugins/manage/modules/debug.py @@ -11,7 +11,7 @@ from ATRI.utils.ub_paste import paste from ATRI.exceptions import load_error -level_list = ['info', 'warning', 'error', 'debug'] +level_list = ["info", "warning", "error", "debug"] __doc__ = """ @@ -25,44 +25,47 @@ __doc__ = """ get_console = sv.on_command( cmd="获取log", - aliases={'获取LOG', '获取控制台', '获取控制台信息'}, + aliases={"获取LOG", "获取控制台", "获取控制台信息"}, docs=__doc__, - permission=SUPERUSER + permission=SUPERUSER, ) + @get_console.handle() async def _get_console(bot: Bot, event: MessageEvent, state: T_State) -> None: msg = str(event.message).strip() if msg: - state['level'] = msg + state["level"] = msg + -@get_console.got('level', prompt='需要获取的等级是？') +@get_console.got("level", prompt="需要获取的等级是？") async def _got(bot: Bot, event: MessageEvent, state: T_State) -> None: - quit_list = ['算了', '罢了', '不了'] - if state['level'] in quit_list: - await get_console.finish('好吧...') + quit_list = ["算了", "罢了", "不了"] + if state["level"] in quit_list: + await get_console.finish("好吧...") -@get_console.got('line', prompt='需要获取的行数是？') + +@get_console.got("line", prompt="需要获取的行数是？") async def _deal_get(bot: Bot, event: MessageEvent, state: T_State) -> None: - level = state['level'] - line = state['line'] + level = state["level"] + line = state["line"] repo = str() - + path = LOGGER_DIR / f"{level}" / f"{NOW_TIME}.log" - logs = await open_file(path, 'readlines') - + logs = await open_file(path, "readlines") + try: - content = logs[int(line):] # type: ignore + content = logs[int(line) :] # type: ignore repo = "\n".join(content).replace("[36mATRI[0m", "ATRI") except IndexError: - await get_console.finish(f'行数错误...max: {len(logs)}') # type: ignore - + await get_console.finish(f"行数错误...max: {len(logs)}") # type: ignore + data = FormData() - data.add_field('poster', 'ATRI running log') - data.add_field('syntax', 'text') - data.add_field('expiration', 'day') - data.add_field('content', repo) - + data.add_field("poster", "ATRI running log") + data.add_field("syntax", "text") + data.add_field("expiration", "day") + data.add_field("content", repo) + msg0 = f"> {event.sender.nickname}\n" msg0 = msg0 + f"详细请移步此处~\n{await paste(data)}" await track_error.finish(msg0) @@ -76,41 +79,39 @@ __doc__ = """ """ track_error = sv.on_command( - cmd="track", - aliases={'追踪'}, - docs=__doc__, - permission=SUPERUSER + cmd="track", aliases={"追踪"}, docs=__doc__, permission=SUPERUSER ) + @track_error.args_parser # type: ignore -async def _track_error_load(bot: Bot, - event: MessageEvent, - state: T_State) -> None: +async def _track_error_load(bot: Bot, event: MessageEvent, state: T_State) -> None: msg = str(event.message).strip() - cancel = ['算了', '罢了'] + cancel = ["算了", "罢了"] if msg in cancel: - await track_error.finish('好吧...') + await track_error.finish("好吧...") if not msg: - await track_error.reject('欸？！要开始debug了吗，请提供追踪ID...Ծ‸Ծ') + await track_error.reject("欸？！要开始debug了吗，请提供追踪ID...Ծ‸Ծ") else: - state['track'] = msg + state["track"] = msg + @track_error.handle() async def _track_error(bot: Bot, event: MessageEvent, state: T_State) -> None: msg = str(event.message).strip() if msg: - state['track'] = msg + state["track"] = msg -@track_error.got('track', prompt='欸？！要开始debug了吗，请提供追踪ID...Ծ‸Ծ') + +@track_error.got("track", prompt="欸？！要开始debug了吗，请提供追踪ID...Ծ‸Ծ") async def _deal_track(bot: Bot, event: MessageEvent, state: T_State) -> None: - track_id = state['track'] + track_id = state["track"] data = dict() - + try: data = load_error(track_id) except BaseException: - await track_error.finish('未发现对应ID呢...(⇀‸↼‶)') - + await track_error.finish("未发现对应ID呢...(⇀‸↼‶)") + msg = ( f"ID: [{track_id}]\n" f"Time: [{data['time']}]\n" @@ -118,13 +119,13 @@ async def _deal_track(bot: Bot, event: MessageEvent, state: T_State) -> None: "——————\n" f"{data['content']}" ) - + data = FormData() - data.add_field('poster', track_id) - data.add_field('syntax', 'text') - data.add_field('expiration', 'day') - data.add_field('content', msg) - + data.add_field("poster", track_id) + data.add_field("syntax", "text") + data.add_field("expiration", "day") + data.add_field("content", msg) + msg0 = f"> {event.sender.nickname}\n" msg0 = msg0 + f"详细请移步此处~\n{await paste(data)}" await track_error.finish(msg0) diff --git a/ATRI/plugins/manage/modules/dormant.py b/ATRI/plugins/manage/modules/dormant.py index d5ae321..cd72d47 100644 --- a/ATRI/plugins/manage/modules/dormant.py +++ b/ATRI/plugins/manage/modules/dormant.py @@ -13,12 +13,10 @@ __doc__ = """ """ dormant_enabled = sv.on_command( - cmd='休眠', - docs=__doc__, - rule=to_bot(), - permission=SUPERUSER + cmd="休眠", docs=__doc__, rule=to_bot(), permission=SUPERUSER ) + @dormant_enabled.handle() async def _dormant_enabled(bot: Bot, event: MessageEvent) -> None: sv.Dormant.control_dormant(True) @@ -34,12 +32,10 @@ __doc__ = """ """ dormant_disabled = sv.on_command( - cmd='苏醒', - docs=__doc__, - rule=to_bot(), - permission=SUPERUSER + cmd="苏醒", docs=__doc__, rule=to_bot(), permission=SUPERUSER ) + @dormant_disabled.handle() async def _dormant_disabled(bot: Bot, event: MessageEvent) -> None: sv.Dormant.control_dormant(False) diff --git a/ATRI/plugins/manage/modules/request.py b/ATRI/plugins/manage/modules/request.py index 843a94e..762e918 100644 --- a/ATRI/plugins/manage/modules/request.py +++ b/ATRI/plugins/manage/modules/request.py @@ -9,7 +9,7 @@ from ATRI.service import Service as sv from ATRI.exceptions import ReadFileError, FormatError -ESSENTIAL_DIR = Path('.') / 'ATRI' / 'data' / 'database' / 'essential' +ESSENTIAL_DIR = Path(".") / "ATRI" / "data" / "database" / "essential" __doc__ = """ @@ -19,16 +19,13 @@ __doc__ = """ 查看申请列表 """ -req_list = sv.on_command( - cmd="申请列表", - docs=__doc__, - permission=SUPERUSER -) +req_list = sv.on_command(cmd="申请列表", docs=__doc__, permission=SUPERUSER) + @req_list.handle() async def _req_list(bot: Bot, event: MessageEvent) -> None: - path_f = ESSENTIAL_DIR / 'request_friend.json' - path_g = ESSENTIAL_DIR / 'request_group.json' + path_f = ESSENTIAL_DIR / "request_friend.json" + path_g = ESSENTIAL_DIR / "request_group.json" data_f, data_g = dict() try: data_f = json.loads(path_f.read_bytes()) @@ -38,7 +35,7 @@ async def _req_list(bot: Bot, event: MessageEvent) -> None: data_g = json.loads(path_g.read_bytes()) except ReadFileError: msg_g = "[读取文件失败]" - + msg_f = str() for i in data_f.keys(): msg_f += f"{i} | {data_f[i]['user_id']} | {data_f[i]['comment']}\n" @@ -46,75 +43,65 @@ async def _req_list(bot: Bot, event: MessageEvent) -> None: msg_g = str() for i in data_g.keys(): msg_g += f"{i} | {data_g[i]['sub_type']} | {data_g[i]['user_id']} | {data_g[i]['comment']}\n" - - msg = ( - "好友/群申请列表如下：\n" - "· 好友：\n" - f"{msg_f}" - "· 群：\n" - f"{msg_g}" - ) + + msg = "好友/群申请列表如下：\n" "· 好友：\n" f"{msg_f}" "· 群：\n" f"{msg_g}" await req_list.finish(msg) -req_deal = sv.on_regex( - r'[同意|拒绝][好友|群]申请', - permission=SUPERUSER -) +req_deal = sv.on_regex(r"[同意|拒绝][好友|群]申请", permission=SUPERUSER) + @req_deal.handle() async def _req_deal(bot: Bot, event: MessageEvent) -> None: - msg = str(event.message).split(' ') - arg = re.findall(r'[同意|拒绝][好友|群]申请', msg[0]) + msg = str(event.message).split(" ") + arg = re.findall(r"[同意|拒绝][好友|群]申请", msg[0]) app = arg[0] _type = arg[1] - + if not msg[1]: - await req_deal.finish(f'正确用法！速看\n{app}{_type}申请 (reqid)') - + await req_deal.finish(f"正确用法！速看\n{app}{_type}申请 (reqid)") + path = ESSENTIAL_DIR / "request_group.json" data_g = dict() try: data_g = json.loads(path.read_bytes()) except FileExistsError: await req_deal.finish("读取群数据失败，可能并没有请求...") - + reqid = msg[1] if app == "同意": if _type == "好友": try: - await bot.set_friend_add_request(flag=reqid, - approve=True) - await req_deal.finish('完成~！已同意申请') + await bot.set_friend_add_request(flag=reqid, approve=True) + await req_deal.finish("完成~！已同意申请") except FormatError: - await req_deal.finish('请检查输入的值是否正确——！') + await req_deal.finish("请检查输入的值是否正确——！") elif _type == "群": try: - await bot.set_group_add_request(flag=reqid, - sub_type=data_g[reqid]['sub_type'], - approve=True) - await req_deal.finish('完成~！已同意申请') + await bot.set_group_add_request( + flag=reqid, sub_type=data_g[reqid]["sub_type"], approve=True + ) + await req_deal.finish("完成~！已同意申请") except FormatError: - await req_deal.finish('请检查输入的值是否正确——！') + await req_deal.finish("请检查输入的值是否正确——！") else: - await req_deal.finish('请检查输入的值是否正确——！') + await req_deal.finish("请检查输入的值是否正确——！") elif app == "拒绝": if _type == "好友": try: - await bot.set_friend_add_request(flag=reqid, - approve=False) - await req_deal.finish('完成~！已拒绝申请') + await bot.set_friend_add_request(flag=reqid, approve=False) + await req_deal.finish("完成~！已拒绝申请") except FormatError: - await req_deal.finish('请检查输入的值是否正确——！') + await req_deal.finish("请检查输入的值是否正确——！") elif _type == "群": try: - await bot.set_group_add_request(flag=reqid, - sub_type=data_g[reqid]['sub_type'], - approve=False) - await req_deal.finish('完成~！已拒绝申请') + await bot.set_group_add_request( + flag=reqid, sub_type=data_g[reqid]["sub_type"], approve=False + ) + await req_deal.finish("完成~！已拒绝申请") except FormatError: - await req_deal.finish('请检查输入的值是否正确——！') + await req_deal.finish("请检查输入的值是否正确——！") else: - await req_deal.finish('请检查输入的值是否正确——！') + await req_deal.finish("请检查输入的值是否正确——！") else: - await req_deal.finish('请检查输入的值是否正确——！') + await req_deal.finish("请检查输入的值是否正确——！") diff --git a/ATRI/plugins/manage/modules/service.py b/ATRI/plugins/manage/modules/service.py index 6489ad6..a3624df 100644 --- a/ATRI/plugins/manage/modules/service.py +++ b/ATRI/plugins/manage/modules/service.py @@ -5,7 +5,7 @@ from nonebot.adapters.cqhttp import ( Bot, MessageEvent, GroupMessageEvent, - PrivateMessageEvent + PrivateMessageEvent, ) from ATRI.service import Service as sv @@ -21,46 +21,40 @@ __doc__ = """ """ cur_service_ena = sv.on_command( - cmd="启用功能", - docs=__doc__, - permission=SUPERUSER|GROUP_ADMIN|GROUP_OWNER + cmd="启用功能", docs=__doc__, permission=SUPERUSER | GROUP_ADMIN | GROUP_OWNER ) + @cur_service_ena.args_parser # type: ignore -async def _cur_ena_load(bot: Bot, - event: GroupMessageEvent, - state: T_State) -> None: +async def _cur_ena_load(bot: Bot, event: GroupMessageEvent, state: T_State) -> None: msg = str(event.message).strip() - quit_list = ['算了', '罢了', '取消'] + quit_list = ["算了", "罢了", "取消"] if msg in quit_list: - await cur_service_ena.finish('好吧...') + await cur_service_ena.finish("好吧...") if not msg: - await cur_service_ena.reject('请告诉咱目标命令！') + await cur_service_ena.reject("请告诉咱目标命令！") else: - state['service_e'] = msg + state["service_e"] = msg + @cur_service_ena.handle() -async def _cur_ena(bot: Bot, - event: GroupMessageEvent, - state: T_State) -> None: +async def _cur_ena(bot: Bot, event: GroupMessageEvent, state: T_State) -> None: msg = str(event.message).strip() if msg: - state['service_e'] = msg + state["service_e"] = msg -@cur_service_ena.got('service_e', prompt='请告诉咱目标命令！') -async def _deal_cur_ena(bot: Bot, - event: GroupMessageEvent, - state: T_State) -> None: - cmd = state['service_e'] + +@cur_service_ena.got("service_e", prompt="请告诉咱目标命令！") +async def _deal_cur_ena(bot: Bot, event: GroupMessageEvent, state: T_State) -> None: + cmd = state["service_e"] group = str(event.group_id) sv.control_service(cmd, False, True, group=group) - await cur_service_ena.finish(f'成功！本群已启用：{cmd}') + await cur_service_ena.finish(f"成功！本群已启用：{cmd}") + @cur_service_ena.handle() -async def _refuse_cur_ena(bot: Bot, - event: PrivateMessageEvent, - state: T_State) -> None: - await cur_service_ena.finish('只能在群聊中决定哦...') +async def _refuse_cur_ena(bot: Bot, event: PrivateMessageEvent, state: T_State) -> None: + await cur_service_ena.finish("只能在群聊中决定哦...") __doc__ = """ @@ -73,46 +67,40 @@ __doc__ = """ """ cur_service_dis = sv.on_command( - cmd="禁用功能", - docs=__doc__, - permission=SUPERUSER|GROUP_ADMIN|GROUP_OWNER + cmd="禁用功能", docs=__doc__, permission=SUPERUSER | GROUP_ADMIN | GROUP_OWNER ) + @cur_service_dis.args_parser # type: ignore -async def _cur_dis_load(bot: Bot, - event: GroupMessageEvent, - state: T_State) -> None: +async def _cur_dis_load(bot: Bot, event: GroupMessageEvent, state: T_State) -> None: msg = str(event.message).strip() - quit_list = ['算了', '罢了', '取消'] + quit_list = ["算了", "罢了", "取消"] if msg in quit_list: - await cur_service_dis.finish('好吧...') + await cur_service_dis.finish("好吧...") if not msg: - await cur_service_dis.reject('请告诉咱目标命令！') + await cur_service_dis.reject("请告诉咱目标命令！") else: - state['service_d'] = msg + state["service_d"] = msg + @cur_service_dis.handle() -async def _cur_dis(bot: Bot, - event: GroupMessageEvent, - state: T_State) -> None: +async def _cur_dis(bot: Bot, event: GroupMessageEvent, state: T_State) -> None: msg = str(event.message).strip() if msg: - state['service_d'] = msg + state["service_d"] = msg -@cur_service_dis.got('service_d', prompt='请告诉咱目标命令！') -async def _deal_cur_dis(bot: Bot, - event: GroupMessageEvent, - state: T_State) -> None: - cmd = state['service_d'] + +@cur_service_dis.got("service_d", prompt="请告诉咱目标命令！") +async def _deal_cur_dis(bot: Bot, event: GroupMessageEvent, state: T_State) -> None: + cmd = state["service_d"] group = str(event.group_id) sv.control_service(cmd, False, False, group=group) - await cur_service_dis.finish(f'成功！本群已禁用：{cmd}') + await cur_service_dis.finish(f"成功！本群已禁用：{cmd}") + @cur_service_dis.handle() -async def _refuse_cur_dis(bot: Bot, - event: PrivateMessageEvent, - state: T_State) -> None: - await cur_service_dis.finish('只能在群聊中决定哦...') +async def _refuse_cur_dis(bot: Bot, event: PrivateMessageEvent, state: T_State) -> None: + await cur_service_dis.finish("只能在群聊中决定哦...") __doc__ = """ @@ -124,40 +112,33 @@ __doc__ = """ 全局启用 以图搜图 """ -glo_service_ena = sv.on_command( - cmd="全局启用", - docs=__doc__, - permission=SUPERUSER -) +glo_service_ena = sv.on_command(cmd="全局启用", docs=__doc__, permission=SUPERUSER) + @glo_service_ena.args_parser # type: ignore -async def _glo_ena_load(bot: Bot, - event: MessageEvent, - state: T_State) -> None: +async def _glo_ena_load(bot: Bot, event: MessageEvent, state: T_State) -> None: msg = str(event.message).strip() - quit_list = ['算了', '罢了', '取消'] + quit_list = ["算了", "罢了", "取消"] if msg in quit_list: - await glo_service_ena.finish('好吧...') + await glo_service_ena.finish("好吧...") if not msg: - await glo_service_ena.reject('请告诉咱目标命令！') + await glo_service_ena.reject("请告诉咱目标命令！") else: - state['service_e_g'] = msg + state["service_e_g"] = msg + @glo_service_ena.handle() -async def _glo_ena(bot: Bot, - event: MessageEvent, - state: T_State) -> None: +async def _glo_ena(bot: Bot, event: MessageEvent, state: T_State) -> None: msg = str(event.message).strip() if msg: - state['service_e_g'] = msg + state["service_e_g"] = msg + -@glo_service_ena.got('service_e_g', prompt='请告诉咱目标命令！') -async def _deal_glo_ena(bot: Bot, - event: MessageEvent, - state: T_State) -> None: - cmd = state['service_e_g'] +@glo_service_ena.got("service_e_g", prompt="请告诉咱目标命令！") +async def _deal_glo_ena(bot: Bot, event: MessageEvent, state: T_State) -> None: + cmd = state["service_e_g"] sv.control_service(cmd, True, True) - await glo_service_ena.finish(f'成功！已全局启用：{cmd}') + await glo_service_ena.finish(f"成功！已全局启用：{cmd}") __doc__ = """ @@ -169,37 +150,30 @@ __doc__ = """ 全局禁用 以图搜图 """ -glo_service_dis = sv.on_command( - cmd="全局禁用", - docs=__doc__, - permission=SUPERUSER -) +glo_service_dis = sv.on_command(cmd="全局禁用", docs=__doc__, permission=SUPERUSER) + @glo_service_dis.args_parser # type: ignore -async def _glo_dis_load(bot: Bot, - event: MessageEvent, - state: T_State) -> None: +async def _glo_dis_load(bot: Bot, event: MessageEvent, state: T_State) -> None: msg = str(event.message).strip() - quit_list = ['算了', '罢了', '取消'] + quit_list = ["算了", "罢了", "取消"] if msg in quit_list: - await glo_service_dis.finish('好吧...') + await glo_service_dis.finish("好吧...") if not msg: - await glo_service_dis.reject('请告诉咱目标命令！') + await glo_service_dis.reject("请告诉咱目标命令！") else: - state['service_d_g'] = msg + state["service_d_g"] = msg + @glo_service_dis.handle() -async def _glo_dis(bot: Bot, - event: MessageEvent, - state: T_State) -> None: +async def _glo_dis(bot: Bot, event: MessageEvent, state: T_State) -> None: msg = str(event.message).strip() if msg: - state['service_d_g'] = msg + state["service_d_g"] = msg + -@glo_service_dis.got('service_d_g', prompt='请告诉咱目标命令！') -async def _deal_glo_dis(bot: Bot, - event: MessageEvent, - state: T_State) -> None: - cmd = state['service_d_g'] +@glo_service_dis.got("service_d_g", prompt="请告诉咱目标命令！") +async def _deal_glo_dis(bot: Bot, event: MessageEvent, state: T_State) -> None: + cmd = state["service_d_g"] sv.control_service(cmd, True, False) - await glo_service_dis.finish(f'成功！已全局禁用：{cmd}') + await glo_service_dis.finish(f"成功！已全局禁用：{cmd}") diff --git a/ATRI/plugins/manage/modules/shutdown.py b/ATRI/plugins/manage/modules/shutdown.py index 11b2b1b..78f23e8 100644 --- a/ATRI/plugins/manage/modules/shutdown.py +++ b/ATRI/plugins/manage/modules/shutdown.py @@ -12,11 +12,8 @@ __doc__ = """ @ 关机 """ -shutdown = sv.on_command( - cmd="关机", - docs=__doc__, - permission=SUPERUSER -) +shutdown = sv.on_command(cmd="关机", docs=__doc__, permission=SUPERUSER) + @shutdown.handle() async def _shutdown(bot: Bot, event: MessageEvent, state: T_State) -> None: @@ -24,9 +21,10 @@ async def _shutdown(bot: Bot, event: MessageEvent, state: T_State) -> None: if msg: state["msg"] = msg + @shutdown.got("msg", prompt="[WARNING]此项操作将强行终止bot运行，是否继续(y/n)") async def __shutdown(bot: Bot, event: MessageEvent, state: T_State) -> None: - t = ['y', 'Y', '是'] + t = ["y", "Y", "是"] if state["msg"] in t: await bot.send(event, "咱还会醒来的，一定") exit(0) diff --git a/ATRI/plugins/nsfw.py b/ATRI/plugins/nsfw.py index fe5bf78..076ce03 100644 --- a/ATRI/plugins/nsfw.py +++ b/ATRI/plugins/nsfw.py @@ -13,14 +13,12 @@ from ATRI.utils.request import get_bytes from ATRI.utils.cqcode import coolq_code_check -nsfw_url = ( - f"http://{Config.NsfwCheck.host}:" - f"{Config.NsfwCheck.port}/?url=" -) +nsfw_url = f"http://{Config.NsfwCheck.host}:" f"{Config.NsfwCheck.port}/?url=" nsfw_checking = sv.on_message() + @nsfw_checking.handle() async def _nsfw_checking(bot: Bot, event: GroupMessageEvent) -> None: if Config.NsfwCheck.enabled: @@ -28,23 +26,25 @@ async def _nsfw_checking(bot: Bot, event: GroupMessageEvent) -> None: user = event.user_id group = event.group_id check = await coolq_code_check(msg, user, group) - + if check: if "image" not in msg: return - + url = nsfw_url + re.findall(r"url=(.*?)]", msg)[0] try: data = json.loads(await get_bytes(url)) except: - log.warning('检测涩图失败，请查阅文档以获取帮助') + log.warning("检测涩图失败，请查阅文档以获取帮助") return - if round(data['score'], 4) > Config.NsfwCheck.passing_rate: - score = "{:.2%}".format(round(data['score'], 4)) - log.debug(f'截获涩图，得分：{score}') - await bot.send(event, f'好涩哦！涩值：{score}\n不行了咱要发给主人看！') + if round(data["score"], 4) > Config.NsfwCheck.passing_rate: + score = "{:.2%}".format(round(data["score"], 4)) + log.debug(f"截获涩图，得分：{score}") + await bot.send(event, f"好涩哦！涩值：{score}\n不行了咱要发给主人看！") for sup in Config.BotSelfConfig.superusers: - await bot.send_private_msg(user_id=sup, message=f"{msg}\n涩值: {score}") + await bot.send_private_msg( + user_id=sup, message=f"{msg}\n涩值: {score}" + ) else: pass @@ -60,60 +60,57 @@ __doc__ = """ /nsfw 然后Bot会向你索取图片 """ -nsfw_reading = sv.on_command( - cmd="/nsfw", - docs=__doc__, - rule=is_in_service('nsfw') -) +nsfw_reading = sv.on_command(cmd="/nsfw", docs=__doc__, rule=is_in_service("nsfw")) + @nsfw_reading.args_parser # type: ignore async def _nsfw(bot: Bot, event: GroupMessageEvent, state: T_State) -> None: msg = str(event.message) - quit_list = ['算了', '罢了', '不搜了'] + quit_list = ["算了", "罢了", "不搜了"] if msg in quit_list: - await nsfw_reading.finish('好吧') - + await nsfw_reading.finish("好吧") + if not msg: - await nsfw_reading.reject('图呢？') + await nsfw_reading.reject("图呢？") else: - state['pic_nsfw'] = msg + state["pic_nsfw"] = msg + @nsfw_reading.handle() -async def _nsfw_r(bot: Bot, - event: GroupMessageEvent, - state: T_State) -> None: +async def _nsfw_r(bot: Bot, event: GroupMessageEvent, state: T_State) -> None: user = event.user_id group = event.group_id msg = str(event.message).strip() check = await coolq_code_check(msg, user, group) if check and msg: - state['pic_nsfw'] = msg + state["pic_nsfw"] = msg + -@nsfw_reading.got('pic_nsfw', prompt='图呢？') -async def _nsfw_reading(bot: Bot, - event: GroupMessageEvent, - state: T_State) -> None: - msg = state['pic_nsfw'] +@nsfw_reading.got("pic_nsfw", prompt="图呢？") +async def _nsfw_reading(bot: Bot, event: GroupMessageEvent, state: T_State) -> None: + msg = state["pic_nsfw"] pic = re.findall(r"url=(.*?)]", msg) if not pic: - await nsfw_reading.reject('请发送图片而不是其它东西！！') - + await nsfw_reading.reject("请发送图片而不是其它东西！！") + url = nsfw_url + pic[0] try: data = json.loads(await get_bytes(url)) except RequestTimeOut: - raise RequestTimeOut('Time out!') - - score = round(data['score'], 4) - result = "{:.2%}".format(round(data['score'], 4)) + raise RequestTimeOut("Time out!") + + score = round(data["score"], 4) + result = "{:.2%}".format(round(data["score"], 4)) if score > 0.9: level = "hso! 我要发给主人看！" for sup in Config.BotSelfConfig.superusers: - await bot.send_private_msg(user_id=sup, message=f"{state['pic_nsfw']}\n涩值: {result}") + await bot.send_private_msg( + user_id=sup, message=f"{state['pic_nsfw']}\n涩值: {result}" + ) elif 0.9 > score >= 0.6: level = "嗯，可冲" else: level = "？能不能换张55完全冲不起来" - + repo = f"涩值：{result}\n{level}" await nsfw_reading.finish(repo) diff --git a/ATRI/plugins/rich/__init__.py b/ATRI/plugins/rich/__init__.py index cd4c5bd..14b9534 100644 --- a/ATRI/plugins/rich/__init__.py +++ b/ATRI/plugins/rich/__init__.py @@ -16,22 +16,23 @@ from .data_source import dec temp_list = [] img_url = [ "https://cdn.jsdelivr.net/gh/Kyomotoi/CDN@master/project/ATRI/fkrich.png", - "https://cdn.jsdelivr.net/gh/Kyomotoi/CDN@master/project/ATRI/xixi.jpg" + "https://cdn.jsdelivr.net/gh/Kyomotoi/CDN@master/project/ATRI/xixi.jpg", ] bilibili_rich = sv.on_message() + @bilibili_rich.handle() async def _bilibili_rich(bot: Bot, event: MessageEvent) -> None: global temp_list try: msg = str(event.raw_message).replace("\\", "") bv = False - + if "qqdocurl" not in msg: if "av" in msg: - av = re.findall(r"(av\d+)", msg)[0].replace('av', '') + av = re.findall(r"(av\d+)", msg)[0].replace("av", "") else: bv = re.findall(r"(BV\w+)", msg) av = str(dec(bv[0])) @@ -43,31 +44,29 @@ async def _bilibili_rich(bot: Bot, event: MessageEvent) -> None: async with session.get(url=bv_url) as r: bv = re.findall(r"(BV\w+)", str(r.url)) av = dec(bv[0]) - + if not bv: if "av" in msg: - av = re.findall(r"(av\d+)", msg)[0].replace('av', '') + av = re.findall(r"(av\d+)", msg)[0].replace("av", "") else: return - + if count_list(temp_list, av) == 4: await bot.send(event, "你是怕别人看不到么发这么多次？") temp_list = del_list_aim(temp_list, av) return - + temp_list.append(av) - + URL = f"https://api.kyomotoi.moe/api/bilibili/v2/?aid={av}" - data = json.loads(await get_bytes(URL))['data'] + data = json.loads(await get_bytes(URL))["data"] repo = ( f"{data['bvid']} INFO:\n" f"Title: {data['title']}\n" f"Link: {data['short_link']}\n" "にまねげぴのTencent rich!" ) - await bot.send( - event, - MessageSegment.image(file=choice(img_url))) + await bot.send(event, MessageSegment.image(file=choice(img_url))) await bilibili_rich.finish(repo) except BaseException: return diff --git a/ATRI/plugins/rich/data_source.py b/ATRI/plugins/rich/data_source.py index 32ac219..59474ff 100644 --- a/ATRI/plugins/rich/data_source.py +++ b/ATRI/plugins/rich/data_source.py @@ -1,4 +1,4 @@ -table = 'fZodR9XQDSUm21yCkr6zBqiveYah8bt4xsWpHnJE7jL5VG3guMTKNPAwcF' +table = "fZodR9XQDSUm21yCkr6zBqiveYah8bt4xsWpHnJE7jL5VG3guMTKNPAwcF" tr = {} for i in range(58): tr[table[i]] = i @@ -10,13 +10,13 @@ add = 8728348608 def dec(x) -> int: r = 0 for i in range(6): - r += tr[x[s[i]]] * 58**i + r += tr[x[s[i]]] * 58 ** i return (r - add) ^ xor def enc(x) -> str: x = (x ^ xor) + add - r = list('BV1 4 1 7 ') + r = list("BV1 4 1 7 ") for i in range(6): - r[s[i]] = table[x // 58**i % 58] - return ''.join(r) + r[s[i]] = table[x // 58 ** i % 58] + return "".join(r) diff --git a/ATRI/plugins/saucenao/__init__.py b/ATRI/plugins/saucenao/__init__.py index f158cf9..b4b1497 100644 --- a/ATRI/plugins/saucenao/__init__.py +++ b/ATRI/plugins/saucenao/__init__.py @@ -21,55 +21,53 @@ __doc__ = """ 以图搜图 (pic) """ -saucenao = sv.on_command( - cmd='以图搜图', - docs=__doc__, - rule=is_in_service('以图搜图') -) +saucenao = sv.on_command(cmd="以图搜图", docs=__doc__, rule=is_in_service("以图搜图")) + @saucenao.args_parser # type: ignore -async def _load_saucenao(bot: Bot, event: MessageEvent, - state: T_State) -> None: +async def _load_saucenao(bot: Bot, event: MessageEvent, state: T_State) -> None: msg = str(event.message) - quit_list = ['算了', '罢了', '不搜了'] + quit_list = ["算了", "罢了", "不搜了"] if msg in quit_list: - await saucenao.finish('好吧...') - + await saucenao.finish("好吧...") + if not msg: - await saucenao.reject('图呢？') + await saucenao.reject("图呢？") else: - state['pic_sau'] = msg + state["pic_sau"] = msg + @saucenao.handle() async def _sauce_nao(bot: Bot, event: MessageEvent, state: T_State) -> None: msg = str(event.message).strip() if msg: - state['pic_sau'] = msg + state["pic_sau"] = msg [email protected]('pic_sau', prompt='图呢？') + [email protected]("pic_sau", prompt="图呢？") async def _deal_saucenao(bot: Bot, event: MessageEvent, state: T_State) -> None: - msg = state['pic_sau'] - img = re.findall(r'url=(.*?)]', msg) + msg = state["pic_sau"] + img = re.findall(r"url=(.*?)]", msg) if not img: - await saucenao.finish('请发送图片而不是其他东西！！') - + await saucenao.finish("请发送图片而不是其他东西！！") + try: task = SauceNao(api_key=Config.SauceNAO.key) data = json.loads(await task.search(img[0])) except RequestTimeOut: - raise RequestTimeOut('Request failed!') - - res = data['results'] + raise RequestTimeOut("Request failed!") + + res = data["results"] result = list() for i in range(0, 3): data = res[i] - + _result = dict() - _result['similarity'] = data['header']['similarity'] - _result['index_name'] = data['header']['index_name'] - _result['url'] = choice(data['data'].get('ext_urls', ['None'])) + _result["similarity"] = data["header"]["similarity"] + _result["index_name"] = data["header"]["index_name"] + _result["url"] = choice(data["data"].get("ext_urls", ["None"])) result.append(_result) - + msg0 = f"> {MessageSegment.at(event.user_id)}" for i in result: msg0 = msg0 + ( @@ -78,5 +76,5 @@ async def _deal_saucenao(bot: Bot, event: MessageEvent, state: T_State) -> None: f"Name: {i['index_name']}\n" f"URL: {i['url'].replace('https://', '')}" ) - + await saucenao.finish(Message(msg0)) diff --git a/ATRI/plugins/saucenao/data_source.py b/ATRI/plugins/saucenao/data_source.py index cd88554..efe8fe1 100644 --- a/ATRI/plugins/saucenao/data_source.py +++ b/ATRI/plugins/saucenao/data_source.py @@ -5,23 +5,19 @@ URL = "https://saucenao.com/search.php" class SauceNao: - def __init__(self, - api_key: str, - output_type=2, - testmode=1, - dbmaski=32768, - db=5, - numres=5) -> None: + def __init__( + self, api_key: str, output_type=2, testmode=1, dbmaski=32768, db=5, numres=5 + ) -> None: params = dict() - params['api_key'] = api_key - params['output_type'] = output_type - params['testmode'] = testmode - params['dbmaski'] = dbmaski - params['db'] = db - params['numres'] = numres + params["api_key"] = api_key + params["output_type"] = output_type + params["testmode"] = testmode + params["dbmaski"] = dbmaski + params["db"] = db + params["numres"] = numres self.params = params - + async def search(self, url: str): - self.params['url'] = url + self.params["url"] = url res = await post_bytes(url=URL, params=self.params) return res diff --git a/ATRI/plugins/status.py b/ATRI/plugins/status.py index 5e48fc3..bd5d9c4 100644 --- a/ATRI/plugins/status.py +++ b/ATRI/plugins/status.py @@ -16,10 +16,8 @@ __doc__ = """ /ping """ -ping = sv.on_command( - cmd="/ping", - docs="测试机器人", - rule=is_in_service('ping')) +ping = sv.on_command(cmd="/ping", docs="测试机器人", rule=is_in_service("ping")) + @ping.handle() async def _ping(bot: Bot, event: MessageEvent) -> None: @@ -33,11 +31,8 @@ __doc__ = """ /status """ -status = sv.on_command( - cmd="/status", - docs=__doc__, - rule=is_in_service('status') -) +status = sv.on_command(cmd="/status", docs=__doc__, rule=is_in_service("status")) + @status.handle() async def _status(bot: Bot, event: MessageEvent) -> None: @@ -45,13 +40,13 @@ async def _status(bot: Bot, event: MessageEvent) -> None: cpu = psutil.cpu_percent(interval=1) mem = psutil.virtual_memory().percent disk = psutil.disk_usage("/").percent - inteSENT = psutil.net_io_counters().bytes_sent / 1000000 # type: ignore - inteRECV = psutil.net_io_counters().bytes_recv / 1000000 # type: ignore + inteSENT = psutil.net_io_counters().bytes_sent / 1000000 # type: ignore + inteRECV = psutil.net_io_counters().bytes_recv / 1000000 # type: ignore except GetStatusError: raise GetStatusError("Failed to get status.") - + msg = "アトリは、高性能ですから！" - + if cpu > 80: # type: ignore msg = "咱感觉有些头晕..." if mem > 80: @@ -60,7 +55,7 @@ async def _status(bot: Bot, event: MessageEvent) -> None: msg = "咱感觉有点累..." elif disk > 80: msg = "咱感觉身体要被塞满了..." - + msg0 = ( "Self status:\n" f"* CPU: {cpu}%\n" @@ -69,26 +64,22 @@ async def _status(bot: Bot, event: MessageEvent) -> None: f"* netSENT: {inteSENT}MB\n" f"* netRECV: {inteRECV}MB\n" ) + msg - + await status.finish(msg0) [email protected]_job( - 'interval', - minutes=5, - misfire_grace_time=10 -) [email protected]_job("interval", minutes=5, misfire_grace_time=10) async def _(): log.info("开始自检") try: cpu = psutil.cpu_percent(interval=1) mem = psutil.virtual_memory().percent disk = psutil.disk_usage("/").percent - inteSENT = psutil.net_io_counters().bytes_sent / 1000000 # type: ignore - inteRECV = psutil.net_io_counters().bytes_recv / 1000000 # type: ignore + inteSENT = psutil.net_io_counters().bytes_sent / 1000000 # type: ignore + inteRECV = psutil.net_io_counters().bytes_recv / 1000000 # type: ignore except GetStatusError: raise GetStatusError("Failed to get status.") - + msg = "" if cpu > 80: # type: ignore msg = "咱感觉有些头晕..." @@ -101,7 +92,7 @@ async def _(): else: log.info("运作正常") return - + msg0 = ( "Self status:\n" f"* CPU: {cpu}%\n" @@ -110,9 +101,6 @@ async def _(): f"* netSENT: {inteSENT}MB\n" f"* netRECV: {inteRECV}MB\n" ) + msg - + for sup in Config.BotSelfConfig.superusers: - await sv.NetworkPost.send_private_msg( - user_id=sup, - message=msg0 - ) + await sv.NetworkPost.send_private_msg(user_id=sup, message=msg0) diff --git a/ATRI/plugins/utils/__init__.py b/ATRI/plugins/utils/__init__.py index 5d1860a..ff28e59 100644 --- a/ATRI/plugins/utils/__init__.py +++ b/ATRI/plugins/utils/__init__.py @@ -17,26 +17,24 @@ roll一下 /roll 1d10+10d9+4d5+2d3 """ -roll = sv.on_command( - cmd="/roll", - docs=__doc__, - rule=is_in_service('roll') -) +roll = sv.on_command(cmd="/roll", docs=__doc__, rule=is_in_service("roll")) + @roll.handle() async def _roll(bot: Bot, event: MessageEvent, state: dict) -> None: args = str(event.message).strip() if args: - state['resu'] = args + state["resu"] = args + @roll.got("resu", prompt="roll 参数不能为空~！\ndemo：1d10 或 2d10+2d10") async def _(bot: Bot, event: MessageEvent, state: dict) -> None: - resu = state['resu'] - match = re.match(r'^([\dd+\s]+?)$', resu) - + resu = state["resu"] + match = re.match(r"^([\dd+\s]+?)$", resu) + if not match: await roll.finish("请输入正确的参数！！\ndemo：1d10 或 2d10+2d10") - + await roll.finish(roll_dice(resu)) @@ -52,22 +50,19 @@ __doc__ = """ /enc e アトリは高性能ですから！ """ -encrypt = sv.on_command( - cmd="/enc", - docs=__doc__, - rule=is_in_service('enc') -) +encrypt = sv.on_command(cmd="/enc", docs=__doc__, rule=is_in_service("enc")) + @encrypt.handle() async def _encrypt(bot: Bot, event: MessageEvent) -> None: - msg = str(event.message).split(' ') + msg = str(event.message).split(" ") _type = msg[0] s = msg[1] e = Encrypt() - + if _type == "e": await encrypt.finish(e.encode(s)) elif _type == "d": await encrypt.finish(e.decode(s)) else: - await encrypt.finish('请检查输入~！') + await encrypt.finish("请检查输入~！") diff --git a/ATRI/plugins/utils/data_source.py b/ATRI/plugins/utils/data_source.py index 18c0492..4ef46a3 100644 --- a/ATRI/plugins/utils/data_source.py +++ b/ATRI/plugins/utils/data_source.py @@ -6,41 +6,41 @@ from typing import Union def roll_dice(par: str) -> str: result = 0 - proc = '' + proc = "" proc_list = [] p = par.split("+") - + for i in p: args = re.findall(r"(\d{0,10})(?:(d)(\d{1,10}))", i) args = list(args[0]) - + args[0] = args[0] or 1 if int(args[0]) >= 5000 or int(args[2]) >= 5000: return "阿...好大......" - + for a in range(1, int(args[0]) + 1): rd = random.randint(1, int(args[2])) result = result + rd - + if len(proc_list) <= 10: proc_list.append(rd) - + if len(proc_list) <= 10: proc += "+".join(map(str, proc_list)) elif len(proc_list) > 10: proc += "太长了不展示了就酱w" else: proc += str(result) - + result = f"{par}=({proc})={result}" return result -class Encrypt(): - cr = 'ĀāĂ㥹ÀÁÂÃÄÅ' - cc = 'ŢţŤťŦŧṪṫṬṭṮṯṰṱ' - cn = 'ŔŕŘřṘṙŖŗȐȑȒȓṚṛṜṝṞṟɌɍⱤɽᵲᶉɼɾᵳʀＲｒ' - cb = 'ĨĩĪīĬĭĮįİı' +class Encrypt: + cr = "ĀāĂ㥹ÀÁÂÃÄÅ" + cc = "ŢţŤťŦŧṪṫṬṭṮṯṰṱ" + cn = "ŔŕŘřṘṙŖŗȐȑȒȓṚṛṜṝṞṟɌɍⱤɽᵲᶉɼɾᵳʀＲｒ" + cb = "ĨĩĪīĬĭĮįİı" sr = len(cr) sc = len(cc) @@ -55,7 +55,7 @@ class Encrypt(): def _encodeByte(self, i) -> Union[str, None]: if i > 0xFF: - raise ValueError('ERROR! at/ri overflow') + raise ValueError("ERROR! at/ri overflow") if i > 0x7F: i = i & 0x7F @@ -65,7 +65,7 @@ class Encrypt(): def _encodeShort(self, i) -> str: if i > 0xFFFF: - raise ValueError('ERROR! atri overflow') + raise ValueError("ERROR! atri overflow") reverse = False if i > 0x7FFF: @@ -75,17 +75,15 @@ class Encrypt(): char = [ self._div(i, self.scnb), self._div(i % self.scnb, self.snb), - self._div(i % self.snb, self.sb), i % self.sb - ] - char = [ - self.cr[char[0]], self.cc[char[1]], self.cn[char[2]], - self.cb[char[3]] + self._div(i % self.snb, self.sb), + i % self.sb, ] + char = [self.cr[char[0]], self.cc[char[1]], self.cn[char[2]], self.cb[char[3]]] if reverse: return char[2] + char[3] + char[0] + char[1] - return ''.join(char) + return "".join(char) def _decodeByte(self, c) -> int: nb = False @@ -93,12 +91,11 @@ class Encrypt(): if idx[0] < 0 or idx[1] < 0: idx = [self.cn.index(c[0]), self.cb.index(c[1])] nb = True - raise ValueError('ERROR! at/ri overflow') + raise ValueError("ERROR! at/ri overflow") - result = idx[0] * self.sb + idx[1] \ - if nb else idx[0] * self.sc + idx[1] + result = idx[0] * self.sb + idx[1] if nb else idx[0] * self.sc + idx[1] if result > 0x7F: - raise ValueError('ERROR! at/ri overflow') + raise ValueError("ERROR! at/ri overflow") return result | 0x80 if nb else 0 @@ -109,23 +106,22 @@ class Encrypt(): self.cr.index(c[0]), self.cc.index(c[1]), self.cn.index(c[2]), - self.cb.index(c[3]) + self.cb.index(c[3]), ] else: idx = [ self.cr.index(c[2]), self.cc.index(c[3]), self.cn.index(c[0]), - self.cb.index(c[1]) + self.cb.index(c[1]), ] if idx[0] < 0 or idx[1] < 0 or idx[2] < 0 or idx[3] < 0: - raise ValueError('ERROR! not atri') + raise ValueError("ERROR! not atri") - result = idx[0] * self.scnb + idx[1] * self.snb + idx[ - 2] * self.sb + idx[3] + result = idx[0] * self.scnb + idx[1] * self.snb + idx[2] * self.sb + idx[3] if result > 0x7FFF: - raise ValueError('ERROR! atri overflow') + raise ValueError("ERROR! atri overflow") result |= 0x8000 if reverse else 0 return result @@ -138,37 +134,36 @@ class Encrypt(): if len(b) & 1 == 1: result.append(self._encodeByte(b[-1])) - return ''.join(result) + return "".join(result) - def encode(self, s: str, encoding: str = 'utf-8'): + def encode(self, s: str, encoding: str = "utf-8"): if not isinstance(s, str): - raise ValueError('Please enter str instead of other') + raise ValueError("Please enter str instead of other") return self._encodeBytes(s.encode(encoding)) def _decodeBytes(self, s: str): if not isinstance(s, str): - raise ValueError('Please enter str instead of other') + raise ValueError("Please enter str instead of other") if len(s) & 1: - raise ValueError('ERROR length') + raise ValueError("ERROR length") result = [] for i in range(0, (len(s) >> 2)): - result.append(bytes([self._decodeShort(s[i * 4:i * 4 + 4]) >> 8])) - result.append(bytes([ - self._decodeShort(s[i * 4:i * 4 + 4]) & 0xFF])) + result.append(bytes([self._decodeShort(s[i * 4 : i * 4 + 4]) >> 8])) + result.append(bytes([self._decodeShort(s[i * 4 : i * 4 + 4]) & 0xFF])) if (len(s) & 2) == 2: result.append(bytes([self._decodeByte(s[-2:])])) - return b''.join(result) + return b"".join(result) - def decode(self, s: str, encoding: str = 'utf-8') -> str: + def decode(self, s: str, encoding: str = "utf-8") -> str: if not isinstance(s, str): - raise ValueError('Please enter str instead of other') + raise ValueError("Please enter str instead of other") try: return self._decodeBytes(s).decode(encoding) except UnicodeDecodeError: - raise ValueError('Decoding failed') + raise ValueError("Decoding failed") diff --git a/ATRI/rule.py b/ATRI/rule.py index 00a4c21..907a65d 100644 --- a/ATRI/rule.py +++ b/ATRI/rule.py @@ -11,14 +11,14 @@ def is_in_service(service: str) -> Rule: return sv.auth_service(service, user, str(event.group_id)) else: return sv.auth_service(service, user, None) - + return Rule(_is_in_service) def to_bot() -> Rule: async def _to_bot(bot, event, state) -> bool: return event.is_tome() - + return Rule(_to_bot) diff --git a/ATRI/service.py b/ATRI/service.py index 87e606c..79d3d42 100644 --- a/ATRI/service.py +++ b/ATRI/service.py @@ -3,17 +3,7 @@ import re import json from pathlib import Path from datetime import datetime -from typing import ( - Dict, - Any, - List, - Set, - Tuple, - Type, - Union, - Optional, - TYPE_CHECKING -) +from typing import Dict, Any, List, Set, Tuple, Type, Union, Optional, TYPE_CHECKING from nonebot.matcher import Matcher from nonebot.permission import Permission from nonebot.plugin import on_message @@ -28,8 +18,8 @@ if TYPE_CHECKING: from nonebot.adapters import Bot, Event -SERVICE_DIR = Path('.') / 'ATRI' / 'data' / 'service' -SERVICES_DIR = SERVICE_DIR / 'services' +SERVICE_DIR = Path(".") / "ATRI" / "data" / "service" +SERVICES_DIR = SERVICE_DIR / "services" os.makedirs(SERVICE_DIR, exist_ok=True) os.makedirs(SERVICES_DIR, exist_ok=True) @@ -44,10 +34,7 @@ def _load_block_list() -> dict: try: data = json.loads(file.read_bytes()) except: - data = { - "user": {}, - "group": {} - } + data = {"user": {}, "group": {}} with open(file, "w") as r: r.write(json.dumps(data, indent=4)) return data @@ -61,7 +48,7 @@ def _save_block_list(data: dict) -> None: def _load_service_config(service: str, docs: str = None) -> dict: - file_name = service.replace('/', '') + ".json" + file_name = service.replace("/", "") + ".json" file = SERVICES_DIR / file_name try: data = json.loads(file.read_bytes()) @@ -71,7 +58,7 @@ def _load_service_config(service: str, docs: str = None) -> dict: "docs": docs, "enabled": True, "disable_user": {}, - "disable_group": {} + "disable_group": {}, } with open(file, "w") as r: r.write(json.dumps(service_info, indent=4)) @@ -80,7 +67,7 @@ def _load_service_config(service: str, docs: str = None) -> dict: def _save_service_config(service: str, data: dict) -> None: - file_name = service.replace('/', '') + ".json" + file_name = service.replace("/", "") + ".json" file = SERVICES_DIR / file_name with open(file, "w") as r: r.write(json.dumps(data, indent=4)) @@ -91,20 +78,21 @@ class Service: 集成一套服务管理，对功能信息进行持久化 计划搭配前端使用 """ + @staticmethod def manual_reg_service(service: str, docs: str = None): - file_name = service.replace('/', '') + ".json" + file_name = service.replace("/", "") + ".json" file = SERVICES_DIR / file_name service_info = { "command": service, "docs": docs, "enabled": True, "disable_user": {}, - "disable_group": {} + "disable_group": {}, } with open(file, "w") as r: r.write(json.dumps(service_info, indent=4)) - + @staticmethod def auth_service(service: str, user: str, group: str = None) -> bool: data = _load_service_config(service) @@ -115,193 +103,215 @@ class Service: return False else: return True - + @staticmethod - def control_service(service: str, - is_global: bool, - is_enabled: int, - user: str = None, - group: str = None) -> None: + def control_service( + service: str, + is_global: bool, + is_enabled: int, + user: str = None, + group: str = None, + ) -> None: data = _load_service_config(service) is_enabled = bool(is_enabled) - + if is_global: status = "disabled" if is_enabled else "enabled" - data['enabled'] = is_enabled + data["enabled"] = is_enabled log.info(f"\033[33mService: {service} has been {status}.\033[33m") else: if user: if not is_enabled: - data['disable_user'][user] = str(datetime.now()) - log.info(f"\033[33mNew service blocked user: {user}\033[33m" - f"\033[33m | Service: {service} | Time: {datetime.now()}\033[33m") + data["disable_user"][user] = str(datetime.now()) + log.info( + f"\033[33mNew service blocked user: {user}\033[33m" + f"\033[33m | Service: {service} | Time: {datetime.now()}\033[33m" + ) else: - if user in data['disable_user']: - del data['disable_user'][user] - log.info(f"\033[33mUser: {user} has been unblock\033[33m" - f"\033[33m | Service: {service} | Time: {datetime.now()}\033[33m") + if user in data["disable_user"]: + del data["disable_user"][user] + log.info( + f"\033[33mUser: {user} has been unblock\033[33m" + f"\033[33m | Service: {service} | Time: {datetime.now()}\033[33m" + ) else: if not is_enabled: - data['disable_group'][group] = str(datetime.now()) - log.info(f"\033[33mNew service blocked group: {group}\033[33m" - f"\033[33m | Service: {service} | Time: {datetime.now()}\033[33m") + data["disable_group"][group] = str(datetime.now()) + log.info( + f"\033[33mNew service blocked group: {group}\033[33m" + f"\033[33m | Service: {service} | Time: {datetime.now()}\033[33m" + ) else: - if group in data['disable_group']: - del data['disable_group'][group] - log.info(f"\033[33mGroup: {group} has been unblock\033[33m" - f"\033[33m | Service: {service} | Time: {datetime.now()}\033[33m") + if group in data["disable_group"]: + del data["disable_group"][group] + log.info( + f"\033[33mGroup: {group} has been unblock\033[33m" + f"\033[33m | Service: {service} | Time: {datetime.now()}\033[33m" + ) _save_service_config(service, data) - + @staticmethod - def on_message(rule: Optional[Union[Rule, T_RuleChecker]] = None, - permission: Optional[Permission] = None, - *, - handlers: Optional[List[T_Handler]] = None, - temp: bool = False, - priority: int = 1, - block: bool = True, - state: Optional[T_State] = None) -> Type[Matcher]: - matcher = Matcher.new("message", - Rule() & rule, - permission or Permission(), - temp=temp, - priority=priority, - block=block, - handlers=handlers, - default_state=state) + def on_message( + rule: Optional[Union[Rule, T_RuleChecker]] = None, + permission: Optional[Permission] = None, + *, + handlers: Optional[List[T_Handler]] = None, + temp: bool = False, + priority: int = 1, + block: bool = True, + state: Optional[T_State] = None, + ) -> Type[Matcher]: + matcher = Matcher.new( + "message", + Rule() & rule, + permission or Permission(), + temp=temp, + priority=priority, + block=block, + handlers=handlers, + default_state=state, + ) return matcher @staticmethod - def on_notice(rule: Optional[Union[Rule, T_RuleChecker]] = None, - *, - handlers: Optional[List[T_Handler]] = None, - temp: bool = False, - priority: int = 1, - block: bool = False, - state: Optional[T_State] = None) -> Type[Matcher]: - matcher = Matcher.new("notice", - Rule() & rule, - Permission(), - temp=temp, - priority=priority, - block=block, - handlers=handlers, - default_state=state) + def on_notice( + rule: Optional[Union[Rule, T_RuleChecker]] = None, + *, + handlers: Optional[List[T_Handler]] = None, + temp: bool = False, + priority: int = 1, + block: bool = False, + state: Optional[T_State] = None, + ) -> Type[Matcher]: + matcher = Matcher.new( + "notice", + Rule() & rule, + Permission(), + temp=temp, + priority=priority, + block=block, + handlers=handlers, + default_state=state, + ) return matcher @staticmethod - def on_request(rule: Optional[Union[Rule, T_RuleChecker]] = None, - *, - handlers: Optional[List[T_Handler]] = None, - temp: bool = False, - priority: int = 1, - block: bool = False, - state: Optional[T_State] = None) -> Type[Matcher]: - matcher = Matcher.new("request", - Rule() & rule, - Permission(), - temp=temp, - priority=priority, - block=block, - handlers=handlers, - default_state=state) + def on_request( + rule: Optional[Union[Rule, T_RuleChecker]] = None, + *, + handlers: Optional[List[T_Handler]] = None, + temp: bool = False, + priority: int = 1, + block: bool = False, + state: Optional[T_State] = None, + ) -> Type[Matcher]: + matcher = Matcher.new( + "request", + Rule() & rule, + Permission(), + temp=temp, + priority=priority, + block=block, + handlers=handlers, + default_state=state, + ) return matcher @classmethod - def on_command(cls, - cmd: Union[str, Tuple[str, ...]], - docs: Optional[str] = None, - rule: Optional[Union[Rule, T_RuleChecker]] = None, - aliases: Optional[Set[Union[str, Tuple[str, ...]]]] = None, - **kwargs) -> Type[Matcher]: + def on_command( + cls, + cmd: Union[str, Tuple[str, ...]], + docs: Optional[str] = None, + rule: Optional[Union[Rule, T_RuleChecker]] = None, + aliases: Optional[Set[Union[str, Tuple[str, ...]]]] = None, + **kwargs, + ) -> Type[Matcher]: async def _strip_cmd(bot: "Bot", event: "Event", state: T_State): message = event.get_message() segment = message.pop(0) new_message = message.__class__( - str(segment).lstrip() - [len(state["_prefix"]["raw_command"]):].lstrip()) # type: ignore + str(segment).lstrip()[len(state["_prefix"]["raw_command"]) :].lstrip() + ) # type: ignore for new_segment in reversed(new_message): message.insert(0, new_segment) - + handlers = kwargs.pop("handlers", []) handlers.insert(0, _strip_cmd) - + commands = set([cmd]) | (aliases or set()) _load_service_config(str(cmd), docs) - return cls.on_message(command(*commands) & rule, - handlers=handlers, **kwargs) + return cls.on_message(command(*commands) & rule, handlers=handlers, **kwargs) @classmethod - def on_keyword(cls, - keywords: Set[str], - docs: Optional[str] = None, - rule: Optional[Union[Rule, T_RuleChecker]] = None, - **kwargs) -> Type[Matcher]: + def on_keyword( + cls, + keywords: Set[str], + docs: Optional[str] = None, + rule: Optional[Union[Rule, T_RuleChecker]] = None, + **kwargs, + ) -> Type[Matcher]: _load_service_config(list(keywords)[0], docs) return cls.on_message(keyword(*keywords) & rule, **kwargs) - + @classmethod - def on_regex(cls, - pattern: str, - flags: Union[int, re.RegexFlag] = 0, - rule: Optional[Union[Rule, T_RuleChecker]] = None, - **kwargs) -> Type[Matcher]: + def on_regex( + cls, + pattern: str, + flags: Union[int, re.RegexFlag] = 0, + rule: Optional[Union[Rule, T_RuleChecker]] = None, + **kwargs, + ) -> Type[Matcher]: return on_message(regex(pattern, flags) & rule, **kwargs) - - + class NetworkPost: - URL = ( - f"http://{Config.NetworkPost.host}:" - f"{Config.NetworkPost.port}/" - ) - + URL = f"http://{Config.NetworkPost.host}:" f"{Config.NetworkPost.port}/" + @classmethod - async def send_private_msg(cls, - user_id: int, - message: str, - auto_escape: bool = False) -> Dict[str, Any]: + async def send_private_msg( + cls, user_id: int, message: str, auto_escape: bool = False + ) -> Dict[str, Any]: url = cls.URL + "send_private_msg?" params = { "user_id": user_id, "message": message, - "auto_escape": f"{auto_escape}" + "auto_escape": f"{auto_escape}", } result = json.loads(await post_bytes(url, params)) log.debug(result) return result @classmethod - def send_group_msg(cls, - group_id: int, - message: Union[str], - auto_escape: Optional[bool] = ...) -> Dict[str, Any]: + def send_group_msg( + cls, group_id: int, message: Union[str], auto_escape: Optional[bool] = ... + ) -> Dict[str, Any]: ... @classmethod - async def send_msg(cls, - message_type: Optional[str] = "", - user_id: Optional[int] = None, - group_id: Optional[int] = None, - message = Union[str], - auto_escape: bool = False) -> Dict[str, Any]: + async def send_msg( + cls, + message_type: Optional[str] = "", + user_id: Optional[int] = None, + group_id: Optional[int] = None, + message=Union[str], + auto_escape: bool = False, + ) -> Dict[str, Any]: url = cls.URL + "send_msg?" params = { "message_type": "", "user_id": user_id, "group_id": group_id, "message": message, - "auto_escape": str(auto_escape) + "auto_escape": str(auto_escape), } result = json.loads(await post_bytes(url, params)) log.debug(result) return result - class Dormant: @staticmethod def is_dormant() -> bool: return False if is_sleep else True - + @staticmethod def control_dormant(is_enable: bool) -> None: global is_sleep @@ -309,37 +319,38 @@ class Service: is_sleep = True else: is_sleep = False - - + class BlockSystem: file_name = "ban.json" path = SERVICE_DIR / file_name - + @staticmethod def auth_user(user: str) -> bool: - return False if user in _load_block_list()['user'] else True - + return False if user in _load_block_list()["user"] else True + @staticmethod def auth_group(group: str) -> bool: - return False if group in _load_block_list()['group'] else True - + return False if group in _load_block_list()["group"] else True + @staticmethod - def control_list(is_enabled: bool, - user: str = None, - group: str = None) -> None: + def control_list(is_enabled: bool, user: str = None, group: str = None) -> None: data = _load_block_list() if user: if is_enabled: - data['user'][user] = str(datetime.now()) - log.info(f"\033[33mNew blocked user: {user} | Time: {datetime.now()}\033[33m") + data["user"][user] = str(datetime.now()) + log.info( + f"\033[33mNew blocked user: {user} | Time: {datetime.now()}\033[33m" + ) else: - del data['user'][str(user)] + del data["user"][str(user)] log.info(f"\033[33mUser {user} has been unblock.\033[33m") elif group: if is_enabled: - data['group'][group] = str(datetime.now()) - log.info(f"\033[33mNew blocked group: {group} | Time: {datetime.now()}\033[33m") + data["group"][group] = str(datetime.now()) + log.info( + f"\033[33mNew blocked group: {group} | Time: {datetime.now()}\033[33m" + ) else: - del data['group'][str(group)] + del data["group"][str(group)] log.info(f"\033[33mGroup {group} has been unblock.\033[33m") _save_block_list(data) diff --git a/ATRI/utils/cqcode.py b/ATRI/utils/cqcode.py index 3102a92..e4c6b5e 100644 --- a/ATRI/utils/cqcode.py +++ b/ATRI/utils/cqcode.py @@ -8,9 +8,9 @@ tencent_gchat_url = "gchat.qpic.cn" noob_code = ["record", "video", "music", "xml", "json"] -async def coolq_code_check(cq_code: str, - user: Optional[int] = None, - group: Optional[int] = None): +async def coolq_code_check( + cq_code: str, user: Optional[int] = None, group: Optional[int] = None +): _type = re.findall(r"CQ:(.*?),", cq_code) for i in _type: if i == "image": @@ -18,12 +18,10 @@ async def coolq_code_check(cq_code: str, url = "" if not result else result[0] if tencent_gchat_url not in url: msg = "你注你🐎呢" - await sv.NetworkPost.send_msg(user_id=user, - group_id=group, - message=msg) + await sv.NetworkPost.send_msg(user_id=user, group_id=group, message=msg) else: return True elif i in noob_code: return False else: - return True
\ No newline at end of file + return True diff --git a/ATRI/utils/file.py b/ATRI/utils/file.py index 7afbead..2cf03df 100644 --- a/ATRI/utils/file.py +++ b/ATRI/utils/file.py @@ -7,17 +7,17 @@ from ATRI.exceptions import RequestTimeOut, WriteError from .request import get_content -async def write_file(path: Path, text, encoding='utf-8') -> None: +async def write_file(path: Path, text, encoding="utf-8") -> None: try: - async with aiofiles.open(path, 'w', encoding=encoding) as target: + async with aiofiles.open(path, "w", encoding=encoding) as target: await target.write(text) except WriteError: raise WriteError("Writing file failed!") -async def open_file(path: Path, method, encoding='utf-8'): +async def open_file(path: Path, method, encoding="utf-8"): try: - async with aiofiles.open(path, 'r', encoding=encoding) as target: + async with aiofiles.open(path, "r", encoding=encoding) as target: if method == "read": return target.read() elif method == "readlines": diff --git a/ATRI/utils/img.py b/ATRI/utils/img.py index af2c9e9..f96f368 100644 --- a/ATRI/utils/img.py +++ b/ATRI/utils/img.py @@ -9,7 +9,7 @@ def compress_image(out_file, kb=500, quality=85, k=0.9) -> str: o_size = os.path.getsize(out_file) // 1024 if o_size <= kb: return out_file - + ImageFile.LOAD_TRUNCATED_IMAGES = True # type: ignore while o_size > kb: img = Image.open(out_file) @@ -18,6 +18,6 @@ def compress_image(out_file, kb=500, quality=85, k=0.9) -> str: try: out.save(out_file, quality=quality) except WriteError: - raise WriteError('Writing file failed!') + raise WriteError("Writing file failed!") o_size = os.path.getsize(out_file) // 1024 return out_file diff --git a/ATRI/utils/limit.py b/ATRI/utils/limit.py index 778b090..c6fae19 100644 --- a/ATRI/utils/limit.py +++ b/ATRI/utils/limit.py @@ -16,7 +16,7 @@ exciting_repo = [ "其实吧我觉得你这速度去d个vup挺适合", "我不接受！你太快了", "我有点担心，因为你太快了", - "请稍等！您冲得太快了！" + "请稍等！您冲得太快了！", ] @@ -25,34 +25,33 @@ def del_list(user: str) -> None: exciting_user = del_list_aim(exciting_user, user) -async def is_too_exciting(user: int, group: int, - times: int, repo: bool) -> bool: +async def is_too_exciting(user: int, group: int, times: int, repo: bool) -> bool: global exciting_user - + if user in exciting_user: if repo: - await sv.NetworkPost.send_msg(user_id=user, - group_id=group, - message=choice(exciting_repo)) + await sv.NetworkPost.send_msg( + user_id=user, group_id=group, message=choice(exciting_repo) + ) return False else: if count_list(exciting_user_temp, user) == times: delta = datetime.timedelta( - seconds=Config.BotSelfConfig.session_exciting_time) - trigger = DateTrigger( - run_date=datetime.datetime.now() + delta) - + seconds=Config.BotSelfConfig.session_exciting_time + ) + trigger = DateTrigger(run_date=datetime.datetime.now() + delta) + scheduler.add_job( func=del_list, trigger=trigger, args=(user,), misfire_grace_time=1, ) - + if repo: - await sv.NetworkPost.send_msg(user_id=user, - group_id=group, - message=choice(exciting_repo)) + await sv.NetworkPost.send_msg( + user_id=user, group_id=group, message=choice(exciting_repo) + ) return False else: exciting_user_temp.append(user) diff --git a/ATRI/utils/request.py b/ATRI/utils/request.py index 145b470..7923398 100644 --- a/ATRI/utils/request.py +++ b/ATRI/utils/request.py @@ -25,9 +25,9 @@ async def get_content(url: str, headers: Optional[dict] = None): return result -async def post_bytes(url: str, - params: Optional[dict] = None, - json: Optional[dict] = None) -> bytes: +async def post_bytes( + url: str, params: Optional[dict] = None, json: Optional[dict] = None +) -> bytes: """异步以 Post 方式请求 url""" async with ClientSession() as session: async with session.post(url, params=params, json=json) as r: diff --git a/ATRI/utils/translate.py b/ATRI/utils/translate.py index b183a9d..7df891c 100644 --- a/ATRI/utils/translate.py +++ b/ATRI/utils/translate.py @@ -1,5 +1,5 @@ -SIMPLE = "万与丑专业丛东丝丢两严丧个丬丰临为丽举么义乌乐乔习乡书买乱争于亏云亘亚产亩亲亵亸亿仅从仑仓仪们价众优伙会伛伞伟传伤伥伦伧伪伫体余佣佥侠侣侥侦侧侨侩侪侬俣俦俨俩俪俭债倾偬偻偾偿傥傧储傩儿兑兖党兰关兴兹养兽冁内冈册写军农冢冯冲决况冻净凄凉凌减凑凛几凤凫凭凯击凼凿刍划刘则刚创删别刬刭刽刿剀剂剐剑剥剧劝办务劢动励劲劳势勋勐勚匀匦匮区医华协单卖卢卤卧卫却卺厂厅历厉压厌厍厕厢厣厦厨厩厮县参叆叇双发变叙叠叶号叹叽吁后吓吕吗吣吨听启吴呒呓呕呖呗员呙呛呜咏咔咙咛咝咤咴咸哌响哑哒哓哔哕哗哙哜哝哟唛唝唠唡唢唣唤唿啧啬啭啮啰啴啸喷喽喾嗫呵嗳嘘嘤嘱噜噼嚣嚯团园囱围囵国图圆圣圹场坂坏块坚坛坜坝坞坟坠垄垅垆垒垦垧垩垫垭垯垱垲垴埘埙埚埝埯堑堕塆墙壮声壳壶壸处备复够头夸夹夺奁奂奋奖奥妆妇妈妩妪妫姗姜娄娅娆娇娈娱娲娴婳婴婵婶媪嫒嫔嫱嬷孙学孪宁宝实宠审宪宫宽宾寝对寻导寿将尔尘尧尴尸尽层屃屉届属屡屦屿岁岂岖岗岘岙岚岛岭岳岽岿峃峄峡峣峤峥峦崂崃崄崭嵘嵚嵛嵝嵴巅巩巯币帅师帏帐帘帜带帧帮帱帻帼幂幞干并广庄庆庐庑库应庙庞废庼廪开异弃张弥弪弯弹强归当录彟彦彻径徕御忆忏忧忾怀态怂怃怄怅怆怜总怼怿恋恳恶恸恹恺恻恼恽悦悫悬悭悯惊惧惨惩惫惬惭惮惯愍愠愤愦愿慑慭憷懑懒懔戆戋戏戗战戬户扎扑扦执扩扪扫扬扰抚抛抟抠抡抢护报担拟拢拣拥拦拧拨择挂挚挛挜挝挞挟挠挡挢挣挤挥挦捞损捡换捣据捻掳掴掷掸掺掼揸揽揿搀搁搂搅携摄摅摆摇摈摊撄撑撵撷撸撺擞攒敌敛数斋斓斗斩断无旧时旷旸昙昼昽显晋晒晓晔晕晖暂暧札术朴机杀杂权条来杨杩杰极构枞枢枣枥枧枨枪枫枭柜柠柽栀栅标栈栉栊栋栌栎栏树栖样栾桊桠桡桢档桤桥桦桧桨桩梦梼梾检棂椁椟椠椤椭楼榄榇榈榉槚槛槟槠横樯樱橥橱橹橼檐檩欢欤欧歼殁殇残殒殓殚殡殴毁毂毕毙毡毵氇气氢氩氲汇汉污汤汹沓沟没沣沤沥沦沧沨沩沪沵泞泪泶泷泸泺泻泼泽泾洁洒洼浃浅浆浇浈浉浊测浍济浏浐浑浒浓浔浕涂涌涛涝涞涟涠涡涢涣涤润涧涨涩淀渊渌渍渎渐渑渔渖渗温游湾湿溃溅溆溇滗滚滞滟滠满滢滤滥滦滨滩滪漤潆潇潋潍潜潴澜濑濒灏灭灯灵灾灿炀炉炖炜炝点炼炽烁烂烃烛烟烦烧烨烩烫烬热焕焖焘煅煳熘爱爷牍牦牵牺犊犟状犷犸犹狈狍狝狞独狭狮狯狰狱狲猃猎猕猡猪猫猬献獭玑玙玚玛玮环现玱玺珉珏珐珑珰珲琎琏琐琼瑶瑷璇璎瓒瓮瓯电画畅畲畴疖疗疟疠疡疬疮疯疱疴痈痉痒痖痨痪痫痴瘅瘆瘗瘘瘪瘫瘾瘿癞癣癫癯皑皱皲盏盐监盖盗盘眍眦眬着睁睐睑瞒瞩矫矶矾矿砀码砖砗砚砜砺砻砾础硁硅硕硖硗硙硚确硷碍碛碜碱碹磙礼祎祢祯祷祸禀禄禅离秃秆种积称秽秾稆税稣稳穑穷窃窍窑窜窝窥窦窭竖竞笃笋笔笕笺笼笾筑筚筛筜筝筹签简箓箦箧箨箩箪箫篑篓篮篱簖籁籴类籼粜粝粤粪粮糁糇紧絷纟纠纡红纣纤纥约级纨纩纪纫纬纭纮纯纰纱纲纳纴纵纶纷纸纹纺纻纼纽纾线绀绁绂练组绅细织终绉绊绋绌绍绎经绐绑绒结绔绕绖绗绘给绚绛络绝绞统绠绡绢绣绤绥绦继绨绩绪绫绬续绮绯绰绱绲绳维绵绶绷绸绹绺绻综绽绾绿缀缁缂缃缄缅缆缇缈缉缊缋缌缍缎缏缐缑缒缓缔缕编缗缘缙缚缛缜缝缞缟缠缡缢缣缤缥缦缧缨缩缪缫缬缭缮缯缰缱缲缳缴缵罂网罗罚罢罴羁羟羡翘翙翚耢耧耸耻聂聋职聍联聩聪肃肠肤肷肾肿胀胁胆胜胧胨胪胫胶脉脍脏脐脑脓脔脚脱脶脸腊腌腘腭腻腼腽腾膑臜舆舣舰舱舻艰艳艹艺节芈芗芜芦苁苇苈苋苌苍苎苏苘苹茎茏茑茔茕茧荆荐荙荚荛荜荞荟荠荡荣荤荥荦荧荨荩荪荫荬荭荮药莅莜莱莲莳莴莶获莸莹莺莼萚萝萤营萦萧萨葱蒇蒉蒋蒌蓝蓟蓠蓣蓥蓦蔷蔹蔺蔼蕲蕴薮藁藓虏虑虚虫虬虮虽虾虿蚀蚁蚂蚕蚝蚬蛊蛎蛏蛮蛰蛱蛲蛳蛴蜕蜗蜡蝇蝈蝉蝎蝼蝾螀螨蟏衅衔补衬衮袄袅袆袜袭袯装裆裈裢裣裤裥褛褴襁襕见观觃规觅视觇览觉觊觋觌觍觎觏觐觑觞触觯詟誉誊讠计订讣认讥讦讧讨让讪讫训议讯记讱讲讳讴讵讶讷许讹论讻讼讽设访诀证诂诃评诅识诇诈诉诊诋诌词诎诏诐译诒诓诔试诖诗诘诙诚诛诜话诞诟诠诡询诣诤该详诧诨诩诪诫诬语诮误诰诱诲诳说诵诶请诸诹诺读诼诽课诿谀谁谂调谄谅谆谇谈谊谋谌谍谎谏谐谑谒谓谔谕谖谗谘谙谚谛谜谝谞谟谠谡谢谣谤谥谦谧谨谩谪谫谬谭谮谯谰谱谲谳谴谵谶谷豮贝贞负贠贡财责贤败账货质贩贪贫贬购贮贯贰贱贲贳贴贵贶贷贸费贺贻贼贽贾贿赀赁赂赃资赅赆赇赈赉赊赋赌赍赎赏赐赑赒赓赔赕赖赗赘赙赚赛赜赝赞赟赠赡赢赣赪赵赶趋趱趸跃跄跖跞践跶跷跸跹跻踊踌踪踬踯蹑蹒蹰蹿躏躜躯车轧轨轩轪轫转轭轮软轰轱轲轳轴轵轶轷轸轹轺轻轼载轾轿辀辁辂较辄辅辆辇辈辉辊辋辌辍辎辏辐辑辒输辔辕辖辗辘辙辚辞辩辫边辽达迁过迈运还这进远违连迟迩迳迹适选逊递逦逻遗遥邓邝邬邮邹邺邻郁郄郏郐郑郓郦郧郸酝酦酱酽酾酿释里鉅鉴銮錾钆钇针钉钊钋钌钍钎钏钐钑钒钓钔钕钖钗钘钙钚钛钝钞钟钠钡钢钣钤钥钦钧钨钩钪钫钬钭钮钯钰钱钲钳钴钵钶钷钸钹钺钻钼钽钾钿铀铁铂铃铄铅铆铈铉铊铋铍铎铏铐铑铒铕铗铘铙铚铛铜铝铞铟铠铡铢铣铤铥铦铧铨铪铫铬铭铮铯铰铱铲铳铴铵银铷铸铹铺铻铼铽链铿销锁锂锃锄锅锆锇锈锉锊锋锌锍锎锏锐锑锒锓锔锕锖锗错锚锜锞锟锠锡锢锣锤锥锦锨锩锫锬锭键锯锰锱锲锳锴锵锶锷锸锹锺锻锼锽锾锿镀镁镂镃镆镇镈镉镊镌镍镎镏镐镑镒镕镖镗镙镚镛镜镝镞镟镠镡镢镣镤镥镦镧镨镩镪镫镬镭镮镯镰镱镲镳镴镶长门闩闪闫闬闭问闯闰闱闲闳间闵闶闷闸闹闺闻闼闽闾闿阀阁阂阃阄阅阆阇阈阉阊阋阌阍阎阏阐阑阒阓阔阕阖阗阘阙阚阛队阳阴阵阶际陆陇陈陉陕陧陨险随隐隶隽难雏雠雳雾霁霉霭靓静靥鞑鞒鞯鞴韦韧韨韩韪韫韬韵页顶顷顸项顺须顼顽顾顿颀颁颂颃预颅领颇颈颉颊颋颌颍颎颏颐频颒颓颔颕颖颗题颙颚颛颜额颞颟颠颡颢颣颤颥颦颧风飏飐飑飒飓飔飕飖飗飘飙飚飞飨餍饤饥饦饧饨饩饪饫饬饭饮饯饰饱饲饳饴饵饶饷饸饹饺饻饼饽饾饿馀馁馂馃馄馅馆馇馈馉馊馋馌馍馎馏馐馑馒馓馔馕马驭驮驯驰驱驲驳驴驵驶驷驸驹驺驻驼驽驾驿骀骁骂骃骄骅骆骇骈骉骊骋验骍骎骏骐骑骒骓骔骕骖骗骘骙骚骛骜骝骞骟骠骡骢骣骤骥骦骧髅髋髌鬓魇魉鱼鱽鱾鱿鲀鲁鲂鲄鲅鲆鲇鲈鲉鲊鲋鲌鲍鲎鲏鲐鲑鲒鲓鲔鲕鲖鲗鲘鲙鲚鲛鲜鲝鲞鲟鲠鲡鲢鲣鲤鲥鲦鲧鲨鲩鲪鲫鲬鲭鲮鲯鲰鲱鲲鲳鲴鲵鲶鲷鲸鲹鲺鲻鲼鲽鲾鲿鳀鳁鳂鳃鳄鳅鳆鳇鳈鳉鳊鳋鳌鳍鳎鳏鳐鳑鳒鳓鳔鳕鳖鳗鳘鳙鳛鳜鳝鳞鳟鳠鳡鳢鳣鸟鸠鸡鸢鸣鸤鸥鸦鸧鸨鸩鸪鸫鸬鸭鸮鸯鸰鸱鸲鸳鸴鸵鸶鸷鸸鸹鸺鸻鸼鸽鸾鸿鹀鹁鹂鹃鹄鹅鹆鹇鹈鹉鹊鹋鹌鹍鹎鹏鹐鹑鹒鹓鹔鹕鹖鹗鹘鹚鹛鹜鹝鹞鹟鹠鹡鹢鹣鹤鹥鹦鹧鹨鹩鹪鹫鹬鹭鹯鹰鹱鹲鹳鹴鹾麦麸黄黉黡黩黪黾鼋鼌鼍鼗鼹齄齐齑齿龀龁龂龃龄龅龆龇龈龉龊龋龌龙龚龛龟志制咨只里系范松没尝尝闹面准钟别闲干尽脏拼" -TRADITION = "萬與醜專業叢東絲丟兩嚴喪個爿豐臨為麗舉麼義烏樂喬習鄉書買亂爭於虧雲亙亞產畝親褻嚲億僅從侖倉儀們價眾優夥會傴傘偉傳傷倀倫傖偽佇體餘傭僉俠侶僥偵側僑儈儕儂俁儔儼倆儷儉債傾傯僂僨償儻儐儲儺兒兌兗黨蘭關興茲養獸囅內岡冊寫軍農塚馮衝決況凍淨淒涼淩減湊凜幾鳳鳧憑凱擊氹鑿芻劃劉則剛創刪別剗剄劊劌剴劑剮劍剝劇勸辦務勱動勵勁勞勢勳猛勩勻匭匱區醫華協單賣盧鹵臥衛卻巹廠廳曆厲壓厭厙廁廂厴廈廚廄廝縣參靉靆雙發變敘疊葉號歎嘰籲後嚇呂嗎唚噸聽啟吳嘸囈嘔嚦唄員咼嗆嗚詠哢嚨嚀噝吒噅鹹呱響啞噠嘵嗶噦嘩噲嚌噥喲嘜嗊嘮啢嗩唕喚呼嘖嗇囀齧囉嘽嘯噴嘍嚳囁嗬噯噓嚶囑嚕劈囂謔團園囪圍圇國圖圓聖壙場阪壞塊堅壇壢壩塢墳墜壟壟壚壘墾坰堊墊埡墶壋塏堖塒塤堝墊垵塹墮壪牆壯聲殼壺壼處備復夠頭誇夾奪奩奐奮獎奧妝婦媽嫵嫗媯姍薑婁婭嬈嬌孌娛媧嫻嫿嬰嬋嬸媼嬡嬪嬙嬤孫學孿寧寶實寵審憲宮寬賓寢對尋導壽將爾塵堯尷屍盡層屭屜屆屬屢屨嶼歲豈嶇崗峴嶴嵐島嶺嶽崠巋嶨嶧峽嶢嶠崢巒嶗崍嶮嶄嶸嶔崳嶁脊巔鞏巰幣帥師幃帳簾幟帶幀幫幬幘幗冪襆幹並廣莊慶廬廡庫應廟龐廢廎廩開異棄張彌弳彎彈強歸當錄彠彥徹徑徠禦憶懺憂愾懷態慫憮慪悵愴憐總懟懌戀懇惡慟懨愷惻惱惲悅愨懸慳憫驚懼慘懲憊愜慚憚慣湣慍憤憒願懾憖怵懣懶懍戇戔戲戧戰戩戶紮撲扡執擴捫掃揚擾撫拋摶摳掄搶護報擔擬攏揀擁攔擰撥擇掛摯攣掗撾撻挾撓擋撟掙擠揮撏撈損撿換搗據撚擄摑擲撣摻摜摣攬撳攙擱摟攪攜攝攄擺搖擯攤攖撐攆擷擼攛擻攢敵斂數齋斕鬥斬斷無舊時曠暘曇晝曨顯晉曬曉曄暈暉暫曖劄術樸機殺雜權條來楊榪傑極構樅樞棗櫪梘棖槍楓梟櫃檸檉梔柵標棧櫛櫳棟櫨櫟欄樹棲樣欒棬椏橈楨檔榿橋樺檜槳樁夢檮棶檢欞槨櫝槧欏橢樓欖櫬櫚櫸檟檻檳櫧橫檣櫻櫫櫥櫓櫞簷檁歡歟歐殲歿殤殘殞殮殫殯毆毀轂畢斃氈毿氌氣氫氬氳彙漢汙湯洶遝溝沒灃漚瀝淪滄渢溈滬濔濘淚澩瀧瀘濼瀉潑澤涇潔灑窪浹淺漿澆湞溮濁測澮濟瀏滻渾滸濃潯濜塗湧濤澇淶漣潿渦溳渙滌潤澗漲澀澱淵淥漬瀆漸澠漁瀋滲溫遊灣濕潰濺漵漊潷滾滯灩灄滿瀅濾濫灤濱灘澦濫瀠瀟瀲濰潛瀦瀾瀨瀕灝滅燈靈災燦煬爐燉煒熗點煉熾爍爛烴燭煙煩燒燁燴燙燼熱煥燜燾煆糊溜愛爺牘犛牽犧犢強狀獷獁猶狽麅獮獰獨狹獅獪猙獄猻獫獵獼玀豬貓蝟獻獺璣璵瑒瑪瑋環現瑲璽瑉玨琺瓏璫琿璡璉瑣瓊瑤璦璿瓔瓚甕甌電畫暢佘疇癤療瘧癘瘍鬁瘡瘋皰屙癰痙癢瘂癆瘓癇癡癉瘮瘞瘺癟癱癮癭癩癬癲臒皚皺皸盞鹽監蓋盜盤瞘眥矓著睜睞瞼瞞矚矯磯礬礦碭碼磚硨硯碸礪礱礫礎硜矽碩硤磽磑礄確鹼礙磧磣堿镟滾禮禕禰禎禱禍稟祿禪離禿稈種積稱穢穠穭稅穌穩穡窮竊竅窯竄窩窺竇窶豎競篤筍筆筧箋籠籩築篳篩簹箏籌簽簡籙簀篋籜籮簞簫簣簍籃籬籪籟糴類秈糶糲粵糞糧糝餱緊縶糸糾紆紅紂纖紇約級紈纊紀紉緯紜紘純紕紗綱納紝縱綸紛紙紋紡紵紖紐紓線紺絏紱練組紳細織終縐絆紼絀紹繹經紿綁絨結絝繞絰絎繪給絢絳絡絕絞統綆綃絹繡綌綏絛繼綈績緒綾緓續綺緋綽緔緄繩維綿綬繃綢綯綹綣綜綻綰綠綴緇緙緗緘緬纜緹緲緝縕繢緦綞緞緶線緱縋緩締縷編緡緣縉縛縟縝縫縗縞纏縭縊縑繽縹縵縲纓縮繆繅纈繚繕繒韁繾繰繯繳纘罌網羅罰罷羆羈羥羨翹翽翬耮耬聳恥聶聾職聹聯聵聰肅腸膚膁腎腫脹脅膽勝朧腖臚脛膠脈膾髒臍腦膿臠腳脫腡臉臘醃膕齶膩靦膃騰臏臢輿艤艦艙艫艱豔艸藝節羋薌蕪蘆蓯葦藶莧萇蒼苧蘇檾蘋莖蘢蔦塋煢繭荊薦薘莢蕘蓽蕎薈薺蕩榮葷滎犖熒蕁藎蓀蔭蕒葒葤藥蒞蓧萊蓮蒔萵薟獲蕕瑩鶯蓴蘀蘿螢營縈蕭薩蔥蕆蕢蔣蔞藍薊蘺蕷鎣驀薔蘞藺藹蘄蘊藪槁蘚虜慮虛蟲虯蟣雖蝦蠆蝕蟻螞蠶蠔蜆蠱蠣蟶蠻蟄蛺蟯螄蠐蛻蝸蠟蠅蟈蟬蠍螻蠑螿蟎蠨釁銜補襯袞襖嫋褘襪襲襏裝襠褌褳襝褲襇褸襤繈襴見觀覎規覓視覘覽覺覬覡覿覥覦覯覲覷觴觸觶讋譽謄訁計訂訃認譏訐訌討讓訕訖訓議訊記訒講諱謳詎訝訥許訛論訩訟諷設訪訣證詁訶評詛識詗詐訴診詆謅詞詘詔詖譯詒誆誄試詿詩詰詼誠誅詵話誕詬詮詭詢詣諍該詳詫諢詡譸誡誣語誚誤誥誘誨誑說誦誒請諸諏諾讀諑誹課諉諛誰諗調諂諒諄誶談誼謀諶諜謊諫諧謔謁謂諤諭諼讒諮諳諺諦謎諞諝謨讜謖謝謠謗諡謙謐謹謾謫譾謬譚譖譙讕譜譎讞譴譫讖穀豶貝貞負貟貢財責賢敗賬貨質販貪貧貶購貯貫貳賤賁貰貼貴貺貸貿費賀貽賊贄賈賄貲賃賂贓資賅贐賕賑賚賒賦賭齎贖賞賜贔賙賡賠賧賴賵贅賻賺賽賾贗讚贇贈贍贏贛赬趙趕趨趲躉躍蹌蹠躒踐躂蹺蹕躚躋踴躊蹤躓躑躡蹣躕躥躪躦軀車軋軌軒軑軔轉軛輪軟轟軲軻轤軸軹軼軤軫轢軺輕軾載輊轎輈輇輅較輒輔輛輦輩輝輥輞輬輟輜輳輻輯轀輸轡轅轄輾轆轍轔辭辯辮邊遼達遷過邁運還這進遠違連遲邇逕跡適選遜遞邐邏遺遙鄧鄺鄔郵鄒鄴鄰鬱郤郟鄶鄭鄆酈鄖鄲醞醱醬釅釃釀釋裏钜鑒鑾鏨釓釔針釘釗釙釕釷釺釧釤鈒釩釣鍆釹鍚釵鈃鈣鈈鈦鈍鈔鍾鈉鋇鋼鈑鈐鑰欽鈞鎢鉤鈧鈁鈥鈄鈕鈀鈺錢鉦鉗鈷缽鈳鉕鈽鈸鉞鑽鉬鉭鉀鈿鈾鐵鉑鈴鑠鉛鉚鈰鉉鉈鉍鈹鐸鉶銬銠鉺銪鋏鋣鐃銍鐺銅鋁銱銦鎧鍘銖銑鋌銩銛鏵銓鉿銚鉻銘錚銫鉸銥鏟銃鐋銨銀銣鑄鐒鋪鋙錸鋱鏈鏗銷鎖鋰鋥鋤鍋鋯鋨鏽銼鋝鋒鋅鋶鐦鐧銳銻鋃鋟鋦錒錆鍺錯錨錡錁錕錩錫錮鑼錘錐錦鍁錈錇錟錠鍵鋸錳錙鍥鍈鍇鏘鍶鍔鍤鍬鍾鍛鎪鍠鍰鎄鍍鎂鏤鎡鏌鎮鎛鎘鑷鐫鎳鎿鎦鎬鎊鎰鎔鏢鏜鏍鏰鏞鏡鏑鏃鏇鏐鐔钁鐐鏷鑥鐓鑭鐠鑹鏹鐙鑊鐳鐶鐲鐮鐿鑔鑣鑞鑲長門閂閃閆閈閉問闖閏闈閑閎間閔閌悶閘鬧閨聞闥閩閭闓閥閣閡閫鬮閱閬闍閾閹閶鬩閿閽閻閼闡闌闃闠闊闋闔闐闒闕闞闤隊陽陰陣階際陸隴陳陘陝隉隕險隨隱隸雋難雛讎靂霧霽黴靄靚靜靨韃鞽韉韝韋韌韍韓韙韞韜韻頁頂頃頇項順須頊頑顧頓頎頒頌頏預顱領頗頸頡頰頲頜潁熲頦頤頻頮頹頷頴穎顆題顒顎顓顏額顳顢顛顙顥纇顫顬顰顴風颺颭颮颯颶颸颼颻飀飄飆飆飛饗饜飣饑飥餳飩餼飪飫飭飯飲餞飾飽飼飿飴餌饒餉餄餎餃餏餅餑餖餓餘餒餕餜餛餡館餷饋餶餿饞饁饃餺餾饈饉饅饊饌饢馬馭馱馴馳驅馹駁驢駔駛駟駙駒騶駐駝駑駕驛駘驍罵駰驕驊駱駭駢驫驪騁驗騂駸駿騏騎騍騅騌驌驂騙騭騤騷騖驁騮騫騸驃騾驄驏驟驥驦驤髏髖髕鬢魘魎魚魛魢魷魨魯魴魺鮁鮃鯰鱸鮋鮓鮒鮊鮑鱟鮍鮐鮭鮚鮳鮪鮞鮦鰂鮜鱠鱭鮫鮮鮺鯗鱘鯁鱺鰱鰹鯉鰣鰷鯀鯊鯇鮶鯽鯒鯖鯪鯕鯫鯡鯤鯧鯝鯢鯰鯛鯨鯵鯴鯔鱝鰈鰏鱨鯷鰮鰃鰓鱷鰍鰒鰉鰁鱂鯿鰠鼇鰭鰨鰥鰩鰟鰜鰳鰾鱈鱉鰻鰵鱅鰼鱖鱔鱗鱒鱯鱤鱧鱣鳥鳩雞鳶鳴鳲鷗鴉鶬鴇鴆鴣鶇鸕鴨鴞鴦鴒鴟鴝鴛鴬鴕鷥鷙鴯鴰鵂鴴鵃鴿鸞鴻鵐鵓鸝鵑鵠鵝鵒鷳鵜鵡鵲鶓鵪鶤鵯鵬鵮鶉鶊鵷鷫鶘鶡鶚鶻鶿鶥鶩鷊鷂鶲鶹鶺鷁鶼鶴鷖鸚鷓鷚鷯鷦鷲鷸鷺鸇鷹鸌鸏鸛鸘鹺麥麩黃黌黶黷黲黽黿鼂鼉鞀鼴齇齊齏齒齔齕齗齟齡齙齠齜齦齬齪齲齷龍龔龕龜誌製谘隻裡係範鬆冇嚐嘗鬨麵準鐘彆閒乾儘臟拚" +SIMPLE = "万与丑专业丛东丝丢两严丧个丬丰临为丽举么义乌乐乔习乡书买乱争于亏云亘亚产亩亲亵亸亿仅从仑仓仪们价众优伙会伛伞伟传伤伥伦伧伪伫体余佣佥侠侣侥侦侧侨侩侪侬俣俦俨俩俪俭债倾偬偻偾偿傥傧储傩儿兑兖党兰关兴兹养兽冁内冈册写军农冢冯冲决况冻净凄凉凌减凑凛几凤凫凭凯击凼凿刍划刘则刚创删别刬刭刽刿剀剂剐剑剥剧劝办务劢动励劲劳势勋勐勚匀匦匮区医华协单卖卢卤卧卫却卺厂厅历厉压厌厍厕厢厣厦厨厩厮县参叆叇双发变叙叠叶号叹叽吁后吓吕吗吣吨听启吴呒呓呕呖呗员呙呛呜咏咔咙咛咝咤咴咸哌响哑哒哓哔哕哗哙哜哝哟唛唝唠唡唢唣唤唿啧啬啭啮啰啴啸喷喽喾嗫呵嗳嘘嘤嘱噜噼嚣嚯团园囱围囵国图圆圣圹场坂坏块坚坛坜坝坞坟坠垄垅垆垒垦垧垩垫垭垯垱垲垴埘埙埚埝埯堑堕塆墙壮声壳壶壸处备复够头夸夹夺奁奂奋奖奥妆妇妈妩妪妫姗姜娄娅娆娇娈娱娲娴婳婴婵婶媪嫒嫔嫱嬷孙学孪宁宝实宠审宪宫宽宾寝对寻导寿将尔尘尧尴尸尽层屃屉届属屡屦屿岁岂岖岗岘岙岚岛岭岳岽岿峃峄峡峣峤峥峦崂崃崄崭嵘嵚嵛嵝嵴巅巩巯币帅师帏帐帘帜带帧帮帱帻帼幂幞干并广庄庆庐庑库应庙庞废庼廪开异弃张弥弪弯弹强归当录彟彦彻径徕御忆忏忧忾怀态怂怃怄怅怆怜总怼怿恋恳恶恸恹恺恻恼恽悦悫悬悭悯惊惧惨惩惫惬惭惮惯愍愠愤愦愿慑慭憷懑懒懔戆戋戏戗战戬户扎扑扦执扩扪扫扬扰抚抛抟抠抡抢护报担拟拢拣拥拦拧拨择挂挚挛挜挝挞挟挠挡挢挣挤挥挦捞损捡换捣据捻掳掴掷掸掺掼揸揽揿搀搁搂搅携摄摅摆摇摈摊撄撑撵撷撸撺擞攒敌敛数斋斓斗斩断无旧时旷旸昙昼昽显晋晒晓晔晕晖暂暧札术朴机杀杂权条来杨杩杰极构枞枢枣枥枧枨枪枫枭柜柠柽栀栅标栈栉栊栋栌栎栏树栖样栾桊桠桡桢档桤桥桦桧桨桩梦梼梾检棂椁椟椠椤椭楼榄榇榈榉槚槛槟槠横樯樱橥橱橹橼檐檩欢欤欧歼殁殇残殒殓殚殡殴毁毂毕毙毡毵氇气氢氩氲汇汉污汤汹沓沟没沣沤沥沦沧沨沩沪沵泞泪泶泷泸泺泻泼泽泾洁洒洼浃浅浆浇浈浉浊测浍济浏浐浑浒浓浔浕涂涌涛涝涞涟涠涡涢涣涤润涧涨涩淀渊渌渍渎渐渑渔渖渗温游湾湿溃溅溆溇滗滚滞滟滠满滢滤滥滦滨滩滪漤潆潇潋潍潜潴澜濑濒灏灭灯灵灾灿炀炉炖炜炝点炼炽烁烂烃烛烟烦烧烨烩烫烬热焕焖焘煅煳熘爱爷牍牦牵牺犊犟状犷犸犹狈狍狝狞独狭狮狯狰狱狲猃猎猕猡猪猫猬献獭玑玙玚玛玮环现玱玺珉珏珐珑珰珲琎琏琐琼瑶瑷璇璎瓒瓮瓯电画畅畲畴疖疗疟疠疡疬疮疯疱疴痈痉痒痖痨痪痫痴瘅瘆瘗瘘瘪瘫瘾瘿癞癣癫癯皑皱皲盏盐监盖盗盘眍眦眬着睁睐睑瞒瞩矫矶矾矿砀码砖砗砚砜砺砻砾础硁硅硕硖硗硙硚确硷碍碛碜碱碹磙礼祎祢祯祷祸禀禄禅离秃秆种积称秽秾稆税稣稳穑穷窃窍窑窜窝窥窦窭竖竞笃笋笔笕笺笼笾筑筚筛筜筝筹签简箓箦箧箨箩箪箫篑篓篮篱簖籁籴类籼粜粝粤粪粮糁糇紧絷纟纠纡红纣纤纥约级纨纩纪纫纬纭纮纯纰纱纲纳纴纵纶纷纸纹纺纻纼纽纾线绀绁绂练组绅细织终绉绊绋绌绍绎经绐绑绒结绔绕绖绗绘给绚绛络绝绞统绠绡绢绣绤绥绦继绨绩绪绫绬续绮绯绰绱绲绳维绵绶绷绸绹绺绻综绽绾绿缀缁缂缃缄缅缆缇缈缉缊缋缌缍缎缏缐缑缒缓缔缕编缗缘缙缚缛缜缝缞缟缠缡缢缣缤缥缦缧缨缩缪缫缬缭缮缯缰缱缲缳缴缵罂网罗罚罢罴羁羟羡翘翙翚耢耧耸耻聂聋职聍联聩聪肃肠肤肷肾肿胀胁胆胜胧胨胪胫胶脉脍脏脐脑脓脔脚脱脶脸腊腌腘腭腻腼腽腾膑臜舆舣舰舱舻艰艳艹艺节芈芗芜芦苁苇苈苋苌苍苎苏苘苹茎茏茑茔茕茧荆荐荙荚荛荜荞荟荠荡荣荤荥荦荧荨荩荪荫荬荭荮药莅莜莱莲莳莴莶获莸莹莺莼萚萝萤营萦萧萨葱蒇蒉蒋蒌蓝蓟蓠蓣蓥蓦蔷蔹蔺蔼蕲蕴薮藁藓虏虑虚虫虬虮虽虾虿蚀蚁蚂蚕蚝蚬蛊蛎蛏蛮蛰蛱蛲蛳蛴蜕蜗蜡蝇蝈蝉蝎蝼蝾螀螨蟏衅衔补衬衮袄袅袆袜袭袯装裆裈裢裣裤裥褛褴襁襕见观觃规觅视觇览觉觊觋觌觍觎觏觐觑觞触觯詟誉誊讠计订讣认讥讦讧讨让讪讫训议讯记讱讲讳讴讵讶讷许讹论讻讼讽设访诀证诂诃评诅识诇诈诉诊诋诌词诎诏诐译诒诓诔试诖诗诘诙诚诛诜话诞诟诠诡询诣诤该详诧诨诩诪诫诬语诮误诰诱诲诳说诵诶请诸诹诺读诼诽课诿谀谁谂调谄谅谆谇谈谊谋谌谍谎谏谐谑谒谓谔谕谖谗谘谙谚谛谜谝谞谟谠谡谢谣谤谥谦谧谨谩谪谫谬谭谮谯谰谱谲谳谴谵谶谷豮贝贞负贠贡财责贤败账货质贩贪贫贬购贮贯贰贱贲贳贴贵贶贷贸费贺贻贼贽贾贿赀赁赂赃资赅赆赇赈赉赊赋赌赍赎赏赐赑赒赓赔赕赖赗赘赙赚赛赜赝赞赟赠赡赢赣赪赵赶趋趱趸跃跄跖跞践跶跷跸跹跻踊踌踪踬踯蹑蹒蹰蹿躏躜躯车轧轨轩轪轫转轭轮软轰轱轲轳轴轵轶轷轸轹轺轻轼载轾轿辀辁辂较辄辅辆辇辈辉辊辋辌辍辎辏辐辑辒输辔辕辖辗辘辙辚辞辩辫边辽达迁过迈运还这进远违连迟迩迳迹适选逊递逦逻遗遥邓邝邬邮邹邺邻郁郄郏郐郑郓郦郧郸酝酦酱酽酾酿释里鉅鉴銮錾钆钇针钉钊钋钌钍钎钏钐钑钒钓钔钕钖钗钘钙钚钛钝钞钟钠钡钢钣钤钥钦钧钨钩钪钫钬钭钮钯钰钱钲钳钴钵钶钷钸钹钺钻钼钽钾钿铀铁铂铃铄铅铆铈铉铊铋铍铎铏铐铑铒铕铗铘铙铚铛铜铝铞铟铠铡铢铣铤铥铦铧铨铪铫铬铭铮铯铰铱铲铳铴铵银铷铸铹铺铻铼铽链铿销锁锂锃锄锅锆锇锈锉锊锋锌锍锎锏锐锑锒锓锔锕锖锗错锚锜锞锟锠锡锢锣锤锥锦锨锩锫锬锭键锯锰锱锲锳锴锵锶锷锸锹锺锻锼锽锾锿镀镁镂镃镆镇镈镉镊镌镍镎镏镐镑镒镕镖镗镙镚镛镜镝镞镟镠镡镢镣镤镥镦镧镨镩镪镫镬镭镮镯镰镱镲镳镴镶长门闩闪闫闬闭问闯闰闱闲闳间闵闶闷闸闹闺闻闼闽闾闿阀阁阂阃阄阅阆阇阈阉阊阋阌阍阎阏阐阑阒阓阔阕阖阗阘阙阚阛队阳阴阵阶际陆陇陈陉陕陧陨险随隐隶隽难雏雠雳雾霁霉霭靓静靥鞑鞒鞯鞴韦韧韨韩韪韫韬韵页顶顷顸项顺须顼顽顾顿颀颁颂颃预颅领颇颈颉颊颋颌颍颎颏颐频颒颓颔颕颖颗题颙颚颛颜额颞颟颠颡颢颣颤颥颦颧风飏飐飑飒飓飔飕飖飗飘飙飚飞飨餍饤饥饦饧饨饩饪饫饬饭饮饯饰饱饲饳饴饵饶饷饸饹饺饻饼饽饾饿馀馁馂馃馄馅馆馇馈馉馊馋馌馍馎馏馐馑馒馓馔馕马驭驮驯驰驱驲驳驴驵驶驷驸驹驺驻驼驽驾驿骀骁骂骃骄骅骆骇骈骉骊骋验骍骎骏骐骑骒骓骔骕骖骗骘骙骚骛骜骝骞骟骠骡骢骣骤骥骦骧髅髋髌鬓魇魉鱼鱽鱾鱿鲀鲁鲂鲄鲅鲆鲇鲈鲉鲊鲋鲌鲍鲎鲏鲐鲑鲒鲓鲔鲕鲖鲗鲘鲙鲚鲛鲜鲝鲞鲟鲠鲡鲢鲣鲤鲥鲦鲧鲨鲩鲪鲫鲬鲭鲮鲯鲰鲱鲲鲳鲴鲵鲶鲷鲸鲹鲺鲻鲼鲽鲾鲿鳀鳁鳂鳃鳄鳅鳆鳇鳈鳉鳊鳋鳌鳍鳎鳏鳐鳑鳒鳓鳔鳕鳖鳗鳘鳙鳛鳜鳝鳞鳟鳠鳡鳢鳣鸟鸠鸡鸢鸣鸤鸥鸦鸧鸨鸩鸪鸫鸬鸭鸮鸯鸰鸱鸲鸳鸴鸵鸶鸷鸸鸹鸺鸻鸼鸽鸾鸿鹀鹁鹂鹃鹄鹅鹆鹇鹈鹉鹊鹋鹌鹍鹎鹏鹐鹑鹒鹓鹔鹕鹖鹗鹘鹚鹛鹜鹝鹞鹟鹠鹡鹢鹣鹤鹥鹦鹧鹨鹩鹪鹫鹬鹭鹯鹰鹱鹲鹳鹴鹾麦麸黄黉黡黩黪黾鼋鼌鼍鼗鼹齄齐齑齿龀龁龂龃龄龅龆龇龈龉龊龋龌龙龚龛龟志制咨只里系范松没尝尝闹面准钟别闲干尽脏拼" +TRADITION = "萬與醜專業叢東絲丟兩嚴喪個爿豐臨為麗舉麼義烏樂喬習鄉書買亂爭於虧雲亙亞產畝親褻嚲億僅從侖倉儀們價眾優夥會傴傘偉傳傷倀倫傖偽佇體餘傭僉俠侶僥偵側僑儈儕儂俁儔儼倆儷儉債傾傯僂僨償儻儐儲儺兒兌兗黨蘭關興茲養獸囅內岡冊寫軍農塚馮衝決況凍淨淒涼淩減湊凜幾鳳鳧憑凱擊氹鑿芻劃劉則剛創刪別剗剄劊劌剴劑剮劍剝劇勸辦務勱動勵勁勞勢勳猛勩勻匭匱區醫華協單賣盧鹵臥衛卻巹廠廳曆厲壓厭厙廁廂厴廈廚廄廝縣參靉靆雙發變敘疊葉號歎嘰籲後嚇呂嗎唚噸聽啟吳嘸囈嘔嚦唄員咼嗆嗚詠哢嚨嚀噝吒噅鹹呱響啞噠嘵嗶噦嘩噲嚌噥喲嘜嗊嘮啢嗩唕喚呼嘖嗇囀齧囉嘽嘯噴嘍嚳囁嗬噯噓嚶囑嚕劈囂謔團園囪圍圇國圖圓聖壙場阪壞塊堅壇壢壩塢墳墜壟壟壚壘墾坰堊墊埡墶壋塏堖塒塤堝墊垵塹墮壪牆壯聲殼壺壼處備復夠頭誇夾奪奩奐奮獎奧妝婦媽嫵嫗媯姍薑婁婭嬈嬌孌娛媧嫻嫿嬰嬋嬸媼嬡嬪嬙嬤孫學孿寧寶實寵審憲宮寬賓寢對尋導壽將爾塵堯尷屍盡層屭屜屆屬屢屨嶼歲豈嶇崗峴嶴嵐島嶺嶽崠巋嶨嶧峽嶢嶠崢巒嶗崍嶮嶄嶸嶔崳嶁脊巔鞏巰幣帥師幃帳簾幟帶幀幫幬幘幗冪襆幹並廣莊慶廬廡庫應廟龐廢廎廩開異棄張彌弳彎彈強歸當錄彠彥徹徑徠禦憶懺憂愾懷態慫憮慪悵愴憐總懟懌戀懇惡慟懨愷惻惱惲悅愨懸慳憫驚懼慘懲憊愜慚憚慣湣慍憤憒願懾憖怵懣懶懍戇戔戲戧戰戩戶紮撲扡執擴捫掃揚擾撫拋摶摳掄搶護報擔擬攏揀擁攔擰撥擇掛摯攣掗撾撻挾撓擋撟掙擠揮撏撈損撿換搗據撚擄摑擲撣摻摜摣攬撳攙擱摟攪攜攝攄擺搖擯攤攖撐攆擷擼攛擻攢敵斂數齋斕鬥斬斷無舊時曠暘曇晝曨顯晉曬曉曄暈暉暫曖劄術樸機殺雜權條來楊榪傑極構樅樞棗櫪梘棖槍楓梟櫃檸檉梔柵標棧櫛櫳棟櫨櫟欄樹棲樣欒棬椏橈楨檔榿橋樺檜槳樁夢檮棶檢欞槨櫝槧欏橢樓欖櫬櫚櫸檟檻檳櫧橫檣櫻櫫櫥櫓櫞簷檁歡歟歐殲歿殤殘殞殮殫殯毆毀轂畢斃氈毿氌氣氫氬氳彙漢汙湯洶遝溝沒灃漚瀝淪滄渢溈滬濔濘淚澩瀧瀘濼瀉潑澤涇潔灑窪浹淺漿澆湞溮濁測澮濟瀏滻渾滸濃潯濜塗湧濤澇淶漣潿渦溳渙滌潤澗漲澀澱淵淥漬瀆漸澠漁瀋滲溫遊灣濕潰濺漵漊潷滾滯灩灄滿瀅濾濫灤濱灘澦濫瀠瀟瀲濰潛瀦瀾瀨瀕灝滅燈靈災燦煬爐燉煒熗點煉熾爍爛烴燭煙煩燒燁燴燙燼熱煥燜燾煆糊溜愛爺牘犛牽犧犢強狀獷獁猶狽麅獮獰獨狹獅獪猙獄猻獫獵獼玀豬貓蝟獻獺璣璵瑒瑪瑋環現瑲璽瑉玨琺瓏璫琿璡璉瑣瓊瑤璦璿瓔瓚甕甌電畫暢佘疇癤療瘧癘瘍鬁瘡瘋皰屙癰痙癢瘂癆瘓癇癡癉瘮瘞瘺癟癱癮癭癩癬癲臒皚皺皸盞鹽監蓋盜盤瞘眥矓著睜睞瞼瞞矚矯磯礬礦碭碼磚硨硯碸礪礱礫礎硜矽碩硤磽磑礄確鹼礙磧磣堿镟滾禮禕禰禎禱禍稟祿禪離禿稈種積稱穢穠穭稅穌穩穡窮竊竅窯竄窩窺竇窶豎競篤筍筆筧箋籠籩築篳篩簹箏籌簽簡籙簀篋籜籮簞簫簣簍籃籬籪籟糴類秈糶糲粵糞糧糝餱緊縶糸糾紆紅紂纖紇約級紈纊紀紉緯紜紘純紕紗綱納紝縱綸紛紙紋紡紵紖紐紓線紺絏紱練組紳細織終縐絆紼絀紹繹經紿綁絨結絝繞絰絎繪給絢絳絡絕絞統綆綃絹繡綌綏絛繼綈績緒綾緓續綺緋綽緔緄繩維綿綬繃綢綯綹綣綜綻綰綠綴緇緙緗緘緬纜緹緲緝縕繢緦綞緞緶線緱縋緩締縷編緡緣縉縛縟縝縫縗縞纏縭縊縑繽縹縵縲纓縮繆繅纈繚繕繒韁繾繰繯繳纘罌網羅罰罷羆羈羥羨翹翽翬耮耬聳恥聶聾職聹聯聵聰肅腸膚膁腎腫脹脅膽勝朧腖臚脛膠脈膾髒臍腦膿臠腳脫腡臉臘醃膕齶膩靦膃騰臏臢輿艤艦艙艫艱豔艸藝節羋薌蕪蘆蓯葦藶莧萇蒼苧蘇檾蘋莖蘢蔦塋煢繭荊薦薘莢蕘蓽蕎薈薺蕩榮葷滎犖熒蕁藎蓀蔭蕒葒葤藥蒞蓧萊蓮蒔萵薟獲蕕瑩鶯蓴蘀蘿螢營縈蕭薩蔥蕆蕢蔣蔞藍薊蘺蕷鎣驀薔蘞藺藹蘄蘊藪槁蘚虜慮虛蟲虯蟣雖蝦蠆蝕蟻螞蠶蠔蜆蠱蠣蟶蠻蟄蛺蟯螄蠐蛻蝸蠟蠅蟈蟬蠍螻蠑螿蟎蠨釁銜補襯袞襖嫋褘襪襲襏裝襠褌褳襝褲襇褸襤繈襴見觀覎規覓視覘覽覺覬覡覿覥覦覯覲覷觴觸觶讋譽謄訁計訂訃認譏訐訌討讓訕訖訓議訊記訒講諱謳詎訝訥許訛論訩訟諷設訪訣證詁訶評詛識詗詐訴診詆謅詞詘詔詖譯詒誆誄試詿詩詰詼誠誅詵話誕詬詮詭詢詣諍該詳詫諢詡譸誡誣語誚誤誥誘誨誑說誦誒請諸諏諾讀諑誹課諉諛誰諗調諂諒諄誶談誼謀諶諜謊諫諧謔謁謂諤諭諼讒諮諳諺諦謎諞諝謨讜謖謝謠謗諡謙謐謹謾謫譾謬譚譖譙讕譜譎讞譴譫讖穀豶貝貞負貟貢財責賢敗賬貨質販貪貧貶購貯貫貳賤賁貰貼貴貺貸貿費賀貽賊贄賈賄貲賃賂贓資賅贐賕賑賚賒賦賭齎贖賞賜贔賙賡賠賧賴賵贅賻賺賽賾贗讚贇贈贍贏贛赬趙趕趨趲躉躍蹌蹠躒踐躂蹺蹕躚躋踴躊蹤躓躑躡蹣躕躥躪躦軀車軋軌軒軑軔轉軛輪軟轟軲軻轤軸軹軼軤軫轢軺輕軾載輊轎輈輇輅較輒輔輛輦輩輝輥輞輬輟輜輳輻輯轀輸轡轅轄輾轆轍轔辭辯辮邊遼達遷過邁運還這進遠違連遲邇逕跡適選遜遞邐邏遺遙鄧鄺鄔郵鄒鄴鄰鬱郤郟鄶鄭鄆酈鄖鄲醞醱醬釅釃釀釋裏钜鑒鑾鏨釓釔針釘釗釙釕釷釺釧釤鈒釩釣鍆釹鍚釵鈃鈣鈈鈦鈍鈔鍾鈉鋇鋼鈑鈐鑰欽鈞鎢鉤鈧鈁鈥鈄鈕鈀鈺錢鉦鉗鈷缽鈳鉕鈽鈸鉞鑽鉬鉭鉀鈿鈾鐵鉑鈴鑠鉛鉚鈰鉉鉈鉍鈹鐸鉶銬銠鉺銪鋏鋣鐃銍鐺銅鋁銱銦鎧鍘銖銑鋌銩銛鏵銓鉿銚鉻銘錚銫鉸銥鏟銃鐋銨銀銣鑄鐒鋪鋙錸鋱鏈鏗銷鎖鋰鋥鋤鍋鋯鋨鏽銼鋝鋒鋅鋶鐦鐧銳銻鋃鋟鋦錒錆鍺錯錨錡錁錕錩錫錮鑼錘錐錦鍁錈錇錟錠鍵鋸錳錙鍥鍈鍇鏘鍶鍔鍤鍬鍾鍛鎪鍠鍰鎄鍍鎂鏤鎡鏌鎮鎛鎘鑷鐫鎳鎿鎦鎬鎊鎰鎔鏢鏜鏍鏰鏞鏡鏑鏃鏇鏐鐔钁鐐鏷鑥鐓鑭鐠鑹鏹鐙鑊鐳鐶鐲鐮鐿鑔鑣鑞鑲長門閂閃閆閈閉問闖閏闈閑閎間閔閌悶閘鬧閨聞闥閩閭闓閥閣閡閫鬮閱閬闍閾閹閶鬩閿閽閻閼闡闌闃闠闊闋闔闐闒闕闞闤隊陽陰陣階際陸隴陳陘陝隉隕險隨隱隸雋難雛讎靂霧霽黴靄靚靜靨韃鞽韉韝韋韌韍韓韙韞韜韻頁頂頃頇項順須頊頑顧頓頎頒頌頏預顱領頗頸頡頰頲頜潁熲頦頤頻頮頹頷頴穎顆題顒顎顓顏額顳顢顛顙顥纇顫顬顰顴風颺颭颮颯颶颸颼颻飀飄飆飆飛饗饜飣饑飥餳飩餼飪飫飭飯飲餞飾飽飼飿飴餌饒餉餄餎餃餏餅餑餖餓餘餒餕餜餛餡館餷饋餶餿饞饁饃餺餾饈饉饅饊饌饢馬馭馱馴馳驅馹駁驢駔駛駟駙駒騶駐駝駑駕驛駘驍罵駰驕驊駱駭駢驫驪騁驗騂駸駿騏騎騍騅騌驌驂騙騭騤騷騖驁騮騫騸驃騾驄驏驟驥驦驤髏髖髕鬢魘魎魚魛魢魷魨魯魴魺鮁鮃鯰鱸鮋鮓鮒鮊鮑鱟鮍鮐鮭鮚鮳鮪鮞鮦鰂鮜鱠鱭鮫鮮鮺鯗鱘鯁鱺鰱鰹鯉鰣鰷鯀鯊鯇鮶鯽鯒鯖鯪鯕鯫鯡鯤鯧鯝鯢鯰鯛鯨鯵鯴鯔鱝鰈鰏鱨鯷鰮鰃鰓鱷鰍鰒鰉鰁鱂鯿鰠鼇鰭鰨鰥鰩鰟鰜鰳鰾鱈鱉鰻鰵鱅鰼鱖鱔鱗鱒鱯鱤鱧鱣鳥鳩雞鳶鳴鳲鷗鴉鶬鴇鴆鴣鶇鸕鴨鴞鴦鴒鴟鴝鴛鴬鴕鷥鷙鴯鴰鵂鴴鵃鴿鸞鴻鵐鵓鸝鵑鵠鵝鵒鷳鵜鵡鵲鶓鵪鶤鵯鵬鵮鶉鶊鵷鷫鶘鶡鶚鶻鶿鶥鶩鷊鷂鶲鶹鶺鷁鶼鶴鷖鸚鷓鷚鷯鷦鷲鷸鷺鸇鷹鸌鸏鸛鸘鹺麥麩黃黌黶黷黲黽黿鼂鼉鞀鼴齇齊齏齒齔齕齗齟齡齙齠齜齦齬齪齲齷龍龔龕龜誌製谘隻裡係範鬆冇嚐嘗鬨麵準鐘彆閒乾儘臟拚" def to_tradition_string(s: str): @@ -13,7 +13,7 @@ def to_tradition_string(s: str): output_str_list.append(TRADITION[found_index]) else: output_str_list.append(s[i]) - + return "".join(output_str_list) @@ -28,5 +28,5 @@ def to_simple_string(s: str): output_str_list.append(SIMPLE[found_index]) else: output_str_list.append(s[i]) - + return "".join(output_str_list) diff --git a/ATRI/utils/ub_paste.py b/ATRI/utils/ub_paste.py index c029afe..56b8261 100644 --- a/ATRI/utils/ub_paste.py +++ b/ATRI/utils/ub_paste.py @@ -7,5 +7,5 @@ URL = "https://paste.ubuntu.com/" async def paste(form_data) -> str: async with ClientSession() as session: async with session.post(url=URL, data=form_data) as r: - result = str(r.url).replace('https://', '') + result = str(r.url).replace("https://", "") return result diff --git a/ATRI/utils/yaml.py b/ATRI/utils/yaml.py index 59e8992..1ad7377 100644 --- a/ATRI/utils/yaml.py +++ b/ATRI/utils/yaml.py @@ -3,8 +3,8 @@ import yaml from pathlib import Path -def load_yml(file: Path, encoding='utf-8') -> dict: +def load_yml(file: Path, encoding="utf-8") -> dict: """加载 yml 文件""" - with open(file, 'r', encoding=encoding) as f: + with open(file, "r", encoding=encoding) as f: data = yaml.safe_load(f) return data @@ -5,5 +5,5 @@ import ATRI ATRI.init() app = ATRI.asgi() -if __name__ == '__main__': - ATRI.run('main:app') +if __name__ == "__main__": + ATRI.run("main:app") |