K2CrO4/NeoBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ruff

Browse Source
This commit is contained in:
baby20162016
2026-01-01 00:58:01 +08:00
parent 386534c250
commit 3e91c05688
11 changed files with 114 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
base_plugins/__init__.py
View File

@@ -1,7 +1,8 @@
import os
import importlib
import os
import pkgutil
def load_all_plugins():
"""扫描并加载当前包下所有的插件（支持文件和文件夹）"""
package_name = __package__
@@ -19,4 +20,5 @@ def load_all_plugins():
except Exception as e:
print(f" 加载插件 {module_name} 失败: {e}")
load_all_plugins()

18
base_plugins/echo.py
View File

@@ -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
})
await bot.call_api(
"send_private_msg", {"user_id": event.user_id, "message": reply_msg}
)

2
core/__init__.py
View File

@@ -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"]

28
core/command_manager.py
View File

@@ -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
# --- 消息分发逻辑 ---
@@ -55,7 +63,7 @@ class CommandManager:
return
# 2. 拆分指令和参数
full_cmd = raw_text[len(prefix_found):].split()
full_cmd = raw_text[len(prefix_found) :].split()
if not full_cmd:
return
@@ -88,13 +96,17 @@ class CommandManager:
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)

7
core/config_loader.py
View File

@@ -1,12 +1,15 @@
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} 未找到！")
@@ -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)

33
core/ws.py
View File

@@ -1,12 +1,16 @@
import asyncio
import json
import uuid
import websockets
import traceback
import uuid
from datetime import datetime
import websockets
from models import Event
from .command_manager import matcher
from .config_loader import global_config
from models import Event
from datetime import datetime
class WS:
def __init__(self):
@@ -26,12 +30,17 @@ class WS:
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}")
@@ -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)
@@ -102,8 +115,9 @@ class 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}
@@ -115,7 +129,6 @@ class WS:
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)

5
main.py
View File

@@ -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())

1
models/__init__.py
View File

@@ -1,2 +1,3 @@
from .event import Event, MessageSegment
__all__ = ["Event", "MessageSegment", "Sender"]

2
models/base.py
View File

@@ -1 +1 @@
#TODO 数据类型
# TODO 数据类型

15
models/event.py
View File

@@ -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
@@ -65,10 +66,9 @@ class Event:
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()
@@ -77,8 +77,7 @@ class Event:
# 字段过滤：只提取 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)

11
models/sender.py
View File

@@ -1,6 +1,7 @@
from dataclasses import dataclass
from typing import Optional
@dataclass
class Sender:
user_id: int
@@ -9,8 +10,8 @@ class Sender:
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 # 专属头衔
card: Optional[str] = None # 群名片
area: Optional[str] = None # 地区
level: Optional[str] = None # 等级
role: Optional[str] = None # 角色: owner/admin/member
title: Optional[str] = None # 专属头衔