- 为bot提供配置
# neko.json { "name": "Neko", "driver": { "connect_url": "ws://url-to-server" } }
- 运行项目
async def main(): neko = Neko.from_file("neko.json") await neko.start() if __name__ == "__main__": asyncio.run(main())
- 通过继承
Command[T]
或MessageCommand
新建自定义命令。新建的命令会自动通过Commands
注册。@Commands.register(name="cmd", description="xxxx") class MyCommand(MessageCommand): @override async def handle_message(self, event: MessageEvent, message: Message) -> None: self.send_back(event, message)
- 配置
client.commands
。# neko.json { ... client: { commands: ["cmd", ...] } }
Neko
类提供了customized_setup
作为扩展点,可以通过继承的方式进行扩展。
class MyEventHandler:
@staticmethod
def apply() -> Behavior[ClientMessage]:
def on_message(_, message: ClientMessage) -> Behavior[ClientMessage]:
print(message)
return Behavior[ClientMessage].same
def setup(context: ActorContext[ClientMessage]) -> Behavior[ClientMessage]:
context.system.receptionist.tell(
ReceptionistRequest.register(EVENT_KEY, context.self)
)
return Behaviors.receive_message(on_message)
return Behaviors.setup(setup)
class Felis(Neko):
@override
def customized_setup(self, context: ActorContext[NekoMessage]) -> None:
context.spawn(MyEventHandler.apply(), "event_handler")
在上例中,我们创建了一个自定义的Actor: MyEventHandler
。它通过EVENT_KEY
订阅了从服务器推送的事件信息并打印出来;
通过继承Neko类,我们得以在customized_setup
中生成MyEventHandler
。
graph RL;
round((Receptionist))-.EVENT_KEY.->Driver--data-->Adapter
Adapter--Events/Response-->round((Receptionist))
round((Receptionist))-.ACTION_KEY.->Client--request-->Adapter;
round((Receptionist))-.xx_KEY.->...
Felis结构主要分为Driver
,Adapter
,Client
三部分:
Driver
:用于与后端交互,包括接收事件和动作响应、发送动作请求。Client
:用于处理事件,如使用Command
。Adapter
:Driver
和Client
的中间件,用于转换并广播事件。用户自定义的Actor可以通过订阅EVENT_KEY
和ACTION_KEY
获取事件和动作信息。