From 3e91c056882f6fe183e9c1095440e2154b825500 Mon Sep 17 00:00:00 2001 From: baby20162016 <2185823427@qq.com> Date: Thu, 1 Jan 2026 00:58:01 +0800 Subject: [PATCH] ruff --- base_plugins/__init__.py | 8 +++-- base_plugins/echo.py | 18 +++++------ core/__init__.py | 4 +-- core/command_manager.py | 44 +++++++++++++++++---------- core/config_loader.py | 9 ++++-- core/ws.py | 65 ++++++++++++++++++++++++---------------- main.py | 7 +++-- models/__init__.py | 3 +- models/base.py | 2 +- models/event.py | 21 +++++++------ models/sender.py | 13 ++++---- 11 files changed, 114 insertions(+), 80 deletions(-) diff --git a/base_plugins/__init__.py b/base_plugins/__init__.py index 3a1344d..3937c7a 100644 --- a/base_plugins/__init__.py +++ b/base_plugins/__init__.py @@ -1,7 +1,8 @@ -import os import importlib +import os import pkgutil + def load_all_plugins(): """扫描并加载当前包下所有的插件（支持文件和文件夹）""" package_name = __package__ @@ -11,7 +12,7 @@ def load_all_plugins(): for loader, module_name, is_pkg in pkgutil.iter_modules(package_path): full_module_name = f"{package_name}.{module_name}" - + try: importlib.import_module(full_module_name) type_str = "包" if is_pkg else "文件" @@ -19,4 +20,5 @@ def load_all_plugins(): except Exception as e: print(f" 加载插件 {module_name} 失败: {e}") -load_all_plugins() \ No newline at end of file + +load_all_plugins() diff --git a/base_plugins/echo.py b/base_plugins/echo.py index 5a4187c..28e287c 100644 --- a/base_plugins/echo.py +++ b/base_plugins/echo.py @@ -1,5 +1,7 @@ from core.command_manager import matcher -#TODO 把该死的这些给抽象化 + + +# TODO 把该死的这些给抽象化 @matcher.command("echo") async def handle_echo(bot, event, args): if not args: @@ -8,12 +10,10 @@ async def handle_echo(bot, event, args): reply_msg = " ".join(args) if event.message_type == "group": - await bot.call_api("send_group_msg", { - "group_id": event.group_id, - "message": reply_msg - }) + await bot.call_api( + "send_group_msg", {"group_id": event.group_id, "message": reply_msg} + ) else: - await bot.call_api("send_private_msg", { - "user_id": event.user_id, - "message": reply_msg - }) \ No newline at end of file + await bot.call_api( + "send_private_msg", {"user_id": event.user_id, "message": reply_msg} + ) diff --git a/core/__init__.py b/core/__init__.py index 62158a9..dc3ae5d 100644 --- a/core/__init__.py +++ b/core/__init__.py @@ -1,5 +1,5 @@ -from .ws import WS from .command_manager import matcher from .config_loader import global_config +from .ws import WS -__all__ = ["WS", "matcher", "global_config"] \ No newline at end of file +__all__ = ["WS", "matcher", "global_config"] diff --git a/core/command_manager.py b/core/command_manager.py index 4b67756..ff4bb40 100644 --- a/core/command_manager.py +++ b/core/command_manager.py @@ -1,39 +1,47 @@ import inspect -from typing import Any, Tuple, Dict, List, Callable +from typing import Any, Callable, Dict, List, Tuple + from .config_loader import global_config # 从配置中获取命令前缀 comm_prefixes = global_config.bot.get("command", ("/",)) + class CommandManager: def __init__(self, prefixes: Tuple[str, ...] = ("/",)): self.prefixes = prefixes - self.commands: Dict[str, Callable] = {} # 存储消息指令 - self.notice_handlers: List[Dict] = [] # 存储通知处理器 - self.request_handlers: List[Dict] = [] # 存储请求处理器 + self.commands: Dict[str, Callable] = {} # 存储消息指令 + self.notice_handlers: List[Dict] = [] # 存储通知处理器 + self.request_handlers: List[Dict] = [] # 存储请求处理器 # --- 1. 消息指令装饰器 --- def command(self, name: str): """装饰器：注册消息指令，例如 @matcher.command("echo")""" + def decorator(func): self.commands[name] = func return func + return decorator # --- 2. 通知事件装饰器 --- def on_notice(self, notice_type: str = None): """装饰器：注册通知处理器""" + def decorator(func): self.notice_handlers.append({"type": notice_type, "func": func}) return func + return decorator # --- 3. 请求事件装饰器 --- def on_request(self, request_type: str = None): """装饰器：注册请求处理器""" + def decorator(func): self.request_handlers.append({"type": request_type, "func": func}) return func + return decorator # --- 消息分发逻辑 --- @@ -41,24 +49,24 @@ class CommandManager: """解析并分发消息指令""" if not event.raw_message: return - + raw_text = event.raw_message.strip() - + # 1. 检查前缀 prefix_found = None for p in self.prefixes: if raw_text.startswith(p): prefix_found = p break - + if not prefix_found: - return + return # 2. 拆分指令和参数 - full_cmd = raw_text[len(prefix_found):].split() + full_cmd = raw_text[len(prefix_found) :].split() if not full_cmd: return - + cmd_name = full_cmd[0] args = full_cmd[1:] @@ -87,14 +95,18 @@ class CommandManager: sig = inspect.signature(func) params = sig.parameters kwargs = {} - - if "bot" in params: kwargs["bot"] = bot - if "event" in params: kwargs["event"] = event - if "args" in params and args is not None: kwargs["args"] = args - + + if "bot" in params: + kwargs["bot"] = bot + if "event" in params: + kwargs["event"] = event + if "args" in params and args is not None: + kwargs["args"] = args + # 执行函数 await func(**kwargs) + # 确保前缀是元组格式 if isinstance(comm_prefixes, list): comm_prefixes = tuple[Any, ...](comm_prefixes) @@ -102,4 +114,4 @@ elif isinstance(comm_prefixes, str): comm_prefixes = (comm_prefixes,) # 实例化全局管理器 -matcher = CommandManager(prefixes=comm_prefixes) \ No newline at end of file +matcher = CommandManager(prefixes=comm_prefixes) diff --git a/core/config_loader.py b/core/config_loader.py index b756c3c..246ba81 100644 --- a/core/config_loader.py +++ b/core/config_loader.py @@ -1,16 +1,19 @@ -import tomllib from pathlib import Path from typing import Any, Dict +import tomllib + + class Config: def __init__(self, file_path: str = "config.toml"): self.path = Path(file_path) self._data: Dict[str, Any] = {} self.load() + def load(self): if not self.path.exists(): raise FileNotFoundError(f"配置文件 {self.path} 未找到！") - + with open(self.path, "rb") as f: self._data = tomllib.load(f) @@ -27,6 +30,7 @@ class Config: def features(self) -> dict: return self._data.get("features", {}) + # 实例化全局配置对象 global_config = Config() @@ -35,4 +39,3 @@ if __name__ == "__main__": print(global_config.bot.get("command")) print(type(global_config.bot.get("command")) is list) print(global_config.features) - \ No newline at end of file diff --git a/core/ws.py b/core/ws.py index 88a3b0d..1c4a162 100644 --- a/core/ws.py +++ b/core/ws.py @@ -1,13 +1,17 @@ import asyncio import json -import uuid -import websockets import traceback -from .command_manager import matcher -from .config_loader import global_config -from models import Event +import uuid from datetime import datetime +import websockets + +from models import Event + +from .command_manager import matcher +from .config_loader import global_config + + class WS: def __init__(self): # 读取参数 @@ -15,28 +19,33 @@ class WS: self.url = cfg.get("uri") self.token = cfg.get("token") self.reconnect_interval = cfg.get("reconnect_interval", 5) - - self.ws = None - self._pending_requests = {} + + self.ws = None + self._pending_requests = {} async def connect(self): """主连接循环""" headers = {"Authorization": f"Bearer {self.token}"} if self.token else {} - + while True: try: print(f" 正在尝试连接至 NapCat: {self.url}") - async with websockets.connect(self.url, additional_headers=headers) as websocket: + async with websockets.connect( + self.url, additional_headers=headers + ) as websocket: self.ws = websocket print(" 连接成功！") await self._listen_loop(websocket) - - except (websockets.exceptions.ConnectionClosed, ConnectionRefusedError) as e: + + except ( + websockets.exceptions.ConnectionClosed, + ConnectionRefusedError, + ) as e: print(f" 连接断开或服务器拒绝访问: {e}") except Exception as e: print(f" 运行异常: {e}") traceback.print_exc() - + print(f" {self.reconnect_interval}秒后尝试重连...") await asyncio.sleep(self.reconnect_interval) @@ -45,20 +54,20 @@ class WS: async for message in websocket: try: data = json.loads(message) - + # 1. 处理 API 响应 echo_id = data.get("echo") if echo_id and echo_id in self._pending_requests: future = self._pending_requests.pop(echo_id) if not future.done(): future.set_result(data) - continue + continue # 2. 处理上报事件 if "post_type" in data: # 使用 create_task 异步执行，避免阻塞 asyncio.create_task(self.on_event(data)) - + except Exception as e: print(f" 解析消息异常: {e}") @@ -67,7 +76,7 @@ class WS: try: # 解析为 Event 对象 event = Event.from_dict(raw_data) - + # 格式化时间用于打印 t = datetime.fromtimestamp(event.time).strftime("%H:%M:%S") @@ -75,12 +84,16 @@ class WS: # A. 消息事件 (Message) if event.post_type == "message": - print(f" [{t}] [消息] {event.message_type} | {event.user_id}: {event.raw_message}") + print( + f" [{t}] [消息] {event.message_type} | {event.user_id}: {event.raw_message}" + ) await matcher.handle_message(self, event) # B. 通知事件 (Notice) elif event.post_type == "notice": - print(f" [{t}] [通知] {event.notice_type} | 来自: {event.group_id or '私聊'}") + print( + f" [{t}] [通知] {event.notice_type} | 来自: {event.group_id or '私聊'}" + ) await matcher.handle_notice(self, event) # C. 请求事件 (Request) @@ -100,23 +113,23 @@ class WS: """调用 OneBot API""" if not self.ws: return {"status": "failed", "msg": "websocket not initialized"} - + from websockets.protocol import State + if getattr(self.ws, "state", None) is not State.OPEN: - return {"status": "failed", "msg": "websocket is not open"} + return {"status": "failed", "msg": "websocket is not open"} echo_id = str(uuid.uuid4()) payload = {"action": action, "params": params or {}, "echo": echo_id} - + loop = asyncio.get_running_loop() future = loop.create_future() self._pending_requests[echo_id] = future - + await self.ws.send(json.dumps(payload)) - + try: - return await asyncio.wait_for(future, timeout=30.0) except asyncio.TimeoutError: self._pending_requests.pop(echo_id, None) - return {"status": "failed", "retcode": -1, "msg": "api timeout"} \ No newline at end of file + return {"status": "failed", "retcode": -1, "msg": "api timeout"} diff --git a/main.py b/main.py index 76729d7..3fbd9f2 100644 --- a/main.py +++ b/main.py @@ -1,10 +1,13 @@ # main.py import asyncio + from core import WS -import base_plugins + + async def main(): bot = WS() await bot.connect() + if __name__ == "__main__": - asyncio.run(main()) \ No newline at end of file + asyncio.run(main()) diff --git a/models/__init__.py b/models/__init__.py index 7791ed9..7317020 100644 --- a/models/__init__.py +++ b/models/__init__.py @@ -1,2 +1,3 @@ from .event import Event, MessageSegment -__all__ = ["Event", "MessageSegment", "Sender"] \ No newline at end of file + +__all__ = ["Event", "MessageSegment", "Sender"] diff --git a/models/base.py b/models/base.py index 6440d2a..f1d6a0e 100644 --- a/models/base.py +++ b/models/base.py @@ -1 +1 @@ -#TODO 数据类型 \ No newline at end of file +# TODO 数据类型 diff --git a/models/event.py b/models/event.py index 06b58cd..d5d5b60 100644 --- a/models/event.py +++ b/models/event.py @@ -1,7 +1,9 @@ from dataclasses import dataclass, field -from typing import List, Optional, Dict, Any +from typing import Any, Dict, List, Optional + from .sender import Sender + @dataclass class MessageSegment: type: str @@ -26,7 +28,6 @@ class MessageSegment: return f"[MS:{self.type}:{self.data}]" - @dataclass class Event: post_type: str @@ -58,27 +59,25 @@ class Event: segments = [] if isinstance(raw_msg_array, list): segments = [ - MessageSegment(type=seg["type"], data=seg["data"]) + MessageSegment(type=seg["type"], data=seg["data"]) for seg in raw_msg_array ] sender_data = data.get("sender") sender_obj = None if isinstance(sender_data, dict): - sender_obj = Sender(**{ - k: v for k, v in sender_data.items() - if k in Sender.__annotations__ - }) + sender_obj = Sender( + **{k: v for k, v in sender_data.items() if k in Sender.__annotations__} + ) # 数据整合 processed_data = data.copy() processed_data["message"] = segments processed_data["sender"] = sender_obj - + # 字段过滤：只提取 dataclass 中定义的字段 valid_data = { - k: v for k, v in processed_data.items() - if k in cls.__annotations__ + k: v for k, v in processed_data.items() if k in cls.__annotations__ } return cls(**valid_data) @@ -93,4 +92,4 @@ class Event: @property def is_request(self) -> bool: - return self.post_type == "request" \ No newline at end of file + return self.post_type == "request" diff --git a/models/sender.py b/models/sender.py index 52b2feb..b92ecaa 100644 --- a/models/sender.py +++ b/models/sender.py @@ -1,16 +1,17 @@ from dataclasses import dataclass from typing import Optional + @dataclass class Sender: user_id: int nickname: str sex: str = "unknown" age: int = 0 - + # 群聊特有字段 - card: Optional[str] = None # 群名片 - area: Optional[str] = None # 地区 - level: Optional[str] = None # 等级 - role: Optional[str] = None # 角色: owner/admin/member - title: Optional[str] = None # 专属头衔 \ No newline at end of file + card: Optional[str] = None # 群名片 + area: Optional[str] = None # 地区 + level: Optional[str] = None # 等级 + role: Optional[str] = None # 角色: owner/admin/member + title: Optional[str] = None # 专属头衔