🎨 impove typing

This commit is contained in:
yanyongyu
2020-12-07 00:06:09 +08:00
parent 629eed08b6
commit 6b3f6a46e6
27 changed files with 60 additions and 137 deletions

View File

@ -1,14 +1,15 @@
import asyncio
from nonebot import on_message
from nonebot.typing import State
from nonebot.permission import USER
from nonebot.typing import Bot, Event
from nonebot.adapters import Bot, Event
a = on_message(priority=0, permission=USER(123123123), temp=True)
@a.handle()
async def test_a(bot: Bot, event: Event, state: dict):
async def test_a(bot: Bot, event: Event, state: State):
print("======== A Received ========")
print("======== A Running Completed ========")
@ -17,7 +18,7 @@ b = on_message(priority=0, permission=USER(123456789), temp=True)
@b.handle()
async def test_b(bot: Bot, event: Event, state: dict):
async def test_b(bot: Bot, event: Event, state: State):
print("======== B Received ========")
await asyncio.sleep(10)
print("======== B Running Completed ========")
@ -27,5 +28,5 @@ c = on_message(priority=0, permission=USER(1111111111))
@c.handle()
async def test_c(bot: Bot, event: Event, state: dict):
async def test_c(bot: Bot, event: Event, state: State):
print("======== C Received ========")

View File

@ -1,4 +1,5 @@
from nonebot.typing import Bot, Event
from nonebot.typing import State
from nonebot.adapters import Bot, Event
from nonebot.permission import GROUP_OWNER
from . import cmd
@ -7,5 +8,5 @@ test_1 = cmd.command("1", aliases={"test"}, permission=GROUP_OWNER)
@test_1.handle()
async def test1(bot: Bot, event: Event, state: dict):
async def test1(bot: Bot, event: Event, state: State):
await test_1.finish(event.raw_message)

View File

@ -1,9 +1,10 @@
from nonebot.typing import Bot, Event
from nonebot.typing import State
from nonebot.adapters import Bot, Event
from . import match
async def heartbeat(bot: Bot, event: Event, state: dict) -> bool:
async def heartbeat(bot: Bot, event: Event, state: State) -> bool:
return event.detail_type == "heartbeat"
@ -11,5 +12,5 @@ test = match.on_metaevent(rule=heartbeat)
@test.receive()
async def handle_heartbeat(bot: Bot, event: Event, state: dict):
async def handle_heartbeat(bot: Bot, event: Event, state: State):
print("[i] Heartbeat")

View File

@ -1,8 +1,9 @@
from nonebot.typing import Bot, Event
from nonebot.typing import State
from nonebot.adapters import Bot, Event
from nonebot.plugin import on_metaevent
async def heartbeat(bot: Bot, event: Event, state: dict) -> bool:
async def heartbeat(bot: Bot, event: Event, state: State) -> bool:
return event.detail_type == "heartbeat"

View File

@ -1,13 +1,13 @@
from nonebot.rule import to_me
from nonebot.typing import Event
from nonebot.typing import State
from nonebot.plugin import on_command
from nonebot.adapters.cqhttp import Bot
from nonebot.adapters.cqhttp import Bot, Event
test_command = on_command("帮助", to_me())
@test_command.handle()
async def test_handler(bot: Bot, event: Event, state: dict):
async def test_handler(bot: Bot, event: Event, state: State):
args = str(event.message).strip()
print("[!] Command:", state["_prefix"], "Args:", args)
if args:
@ -17,7 +17,7 @@ async def test_handler(bot: Bot, event: Event, state: dict):
@test_command.got("help", prompt="你要帮助的命令是?")
async def test_handler(bot: Bot, event: Event, state: dict):
async def test_handler(bot: Bot, event: Event, state: State):
print("[!] Command 帮助:", state["help"])
if state["help"] not in ["test1", "test2"]:
await test_command.reject(f"{state['help']} 不支持,请重新输入!")

View File

@ -1,18 +1,18 @@
from nonebot.rule import to_me
from nonebot.typing import Event
from nonebot.typing import State
from nonebot.plugin import on_startswith
from nonebot.adapters.cqhttp import Bot
from nonebot.adapters.ding import Bot as DingBot, Event as DingEvent
from nonebot.permission import GROUP_ADMIN
from nonebot.adapters.ding import Bot as DingBot, Event as DingEvent
from nonebot.adapters.cqhttp import Bot as CQHTTPBot, Event as CQHTTPEvent
test_command = on_startswith("hello", to_me(), permission=GROUP_ADMIN)
@test_command.handle()
async def test_handler(bot: Bot, event: Event, state: dict):
async def test_handler(bot: CQHTTPBot, event: CQHTTPEvent, state: State):
await test_command.finish("cqhttp hello")
@test_command.handle()
async def test_handler(bot: DingBot, event: DingEvent, state: dict):
async def test_handler(bot: DingBot, event: DingEvent, state: State):
await test_command.finish("ding hello")

View File

@ -1,13 +1,15 @@
from nonebot.typing import Bot, Event, Matcher
from nonebot.typing import State
from nonebot.matcher import Matcher
from nonebot.adapters import Bot, Event
from nonebot.message import event_preprocessor, run_preprocessor
@event_preprocessor
async def handle(bot: Bot, event: Event, state: dict):
async def handle(bot: Bot, event: Event, state: State):
state["preprocessed"] = True
print(event)
@run_preprocessor
async def run(matcher: Matcher, bot: Bot, event: Event, state: dict):
async def run(matcher: Matcher, bot: Bot, event: Event, state: State):
print(matcher)

View File

@ -1,12 +1,13 @@
from nonebot import on_command
from nonebot.rule import to_me
from nonebot.typing import Bot, Event
from nonebot.typing import State
from nonebot.adapters import Bot, Event
weather = on_command("天气", rule=to_me(), priority=1)
@weather.handle()
async def handle_first_receive(bot: Bot, event: Event, state: dict):
async def handle_first_receive(bot: Bot, event: Event, state: State):
args = str(event.message).strip() # 首次发送命令时跟随的参数,例:/天气 上海则args为上海
print(f"==={args}===")
if args:
@ -14,7 +15,7 @@ async def handle_first_receive(bot: Bot, event: Event, state: dict):
@weather.got("city", prompt="你想查询哪个城市的天气呢?")
async def handle_city(bot: Bot, event: Event, state: dict):
async def handle_city(bot: Bot, event: Event, state: State):
city = state["city"]
if city not in ["上海", "北京"]:
await weather.reject("你想查询的城市暂不支持,请重新输入!")