ProjectHentai/nonebot2
6
0
Fork 0
mirror of https://github.com/nonebot/nonebot2.git synced 2025-01-19 09:38:21 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

rename session.ctx to session.event for consistency

Browse Source
This commit is contained in:
Richard Chien 2020-03-15 22:48:22 +08:00
parent c366e6c950
commit 9b54af70e6
9 changed files with 216 additions and 200 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
nonebot/__init__.py
View File

@ -2,7 +2,7 @@ import asyncio
import logging
from typing import Any, Optional
import aiocqhttp.message
import aiocqhttp
from aiocqhttp import CQHttp
from .log import logger
@ -32,16 +32,16 @@ class NoneBot(CQHttp):
from .notice_request import handle_notice_or_request
@self.on_message
async def _(ctx):
asyncio.ensure_future(handle_message(self, ctx))
async def _(event: aiocqhttp.Event):
asyncio.create_task(handle_message(self, event))
@self.on_notice
async def _(ctx):
asyncio.ensure_future(handle_notice_or_request(self, ctx))
async def _(event: aiocqhttp.Event):
asyncio.create_task(handle_notice_or_request(self, event))
@self.on_request
async def _(ctx):
asyncio.ensure_future(handle_notice_or_request(self, ctx))
async def _(event: aiocqhttp.Event):
asyncio.create_task(handle_notice_or_request(self, event))
def run(self, host: Optional[str] = None, port: Optional[int] = None,
*args, **kwargs) -> None:

106
nonebot/command/__init__.py
View File

@ -9,6 +9,8 @@ from typing import (
Awaitable
)
from aiocqhttp import Event as CQEvent
from nonebot import NoneBot, permission as perm
from nonebot.command.argfilter import ValidateError
from nonebot.helpers import context_id, send, render_expression
@ -16,8 +18,7 @@ from nonebot.log import logger
from nonebot.message import Message
from nonebot.session import BaseSession
from nonebot.typing import (
Context_T, CommandName_T, CommandArgs_T, Message_T, State_T,
Filter_T
CommandName_T, CommandArgs_T, Message_T, State_T, Filter_T
)
# key: one segment of command name
@ -129,7 +130,7 @@ class Command:
:param session: CommandSession object
:return: the session has the permission
"""
return await perm.check_permission(session.bot, session.ctx,
return await perm.check_permission(session.bot, session.event,
self.permission)
def __repr__(self):
@ -166,6 +167,17 @@ def on_command(name: Union[str, CommandName_T], *,
cmd = Command(name=cmd_name, func=func, permission=permission,
only_to_me=only_to_me, privileged=privileged)
def args_parser(parser_func: CommandHandler_T) -> CommandHandler_T:
"""
Decorator to register a function as the arguments parser of
the corresponding command.
"""
cmd.args_parser_func = parser_func
return parser_func
func.args_parser = args_parser
if shell_like:
async def shell_like_args_parser(session):
session.args['argv'] = shlex.split(session.current_arg)
@ -190,15 +202,6 @@ def on_command(name: Union[str, CommandName_T], *,
for alias in aliases:
_aliases[alias] = cmd_name
def args_parser(parser_func: CommandHandler_T) -> CommandHandler_T:
"""
Decorator to register a function as the arguments parser of
the corresponding command.
"""
cmd.args_parser_func = parser_func
return parser_func
func.args_parser = args_parser
return func
return deco
@ -246,16 +249,16 @@ class SwitchException(Exception):
should be stopped and replaced with a new one (going through
handle_message() again).
Since the new context message will go through handle_message()
again, the later function should be notified. So this exception
is designed to be propagated to handle_message().
Since the new message will go through handle_message() again,
the later function should be notified. So this exception is
intended to be propagated to handle_message().
"""
def __init__(self, new_ctx_message: Message):
def __init__(self, new_message: Message):
"""
:param new_ctx_message: new message which should be placed in context
:param new_message: new message which should be placed in event
"""
self.new_ctx_message = new_ctx_message
self.new_message = new_message
class CommandSession(BaseSession):
@ -264,9 +267,9 @@ class CommandSession(BaseSession):
'current_arg', '_current_arg_text', '_current_arg_images',
'_state', '_last_interaction', '_running', '_run_future')
def __init__(self, bot: NoneBot, ctx: Context_T, cmd: Command, *,
def __init__(self, bot: NoneBot, event: CQEvent, cmd: Command, *,
current_arg: str = '', args: Optional[CommandArgs_T] = None):
super().__init__(bot, ctx)
super().__init__(bot, event)
self.cmd = cmd # Command object
# unique key of the argument that is currently requesting (asking)
@ -281,7 +284,7 @@ class CommandSession(BaseSession):
self.current_arg: str = '' # with potential CQ codes
self._current_arg_text = None
self._current_arg_images = None
self.refresh(ctx, current_arg=current_arg) # fill the above
self.refresh(event, current_arg=current_arg) # fill the above
self._run_future = partial(asyncio.run_coroutine_threadsafe,
loop=bot.loop)
@ -363,14 +366,14 @@ class CommandSession(BaseSession):
"""
return self.state.get('argv', [])
def refresh(self, ctx: Context_T, *, current_arg: str = '') -> None:
def refresh(self, event: CQEvent, *, current_arg: str = '') -> None:
"""
Refill the session with a new message context.
Refill the session with a new message event.
:param ctx: new message context
:param event: new message event
:param current_arg: new command argument as a string
"""
self.ctx = ctx
self.event = event
self.current_arg = current_arg
self._current_arg_text = None
self._current_arg_images = None
@ -421,9 +424,9 @@ class CommandSession(BaseSession):
self._run_future(self.send(message, **kwargs))
raise _FinishException
def switch(self, new_ctx_message: Message_T) -> None:
def switch(self, new_message: Message_T) -> None:
"""
Finish the session and switch to a new (fake) message context.
Finish the session and switch to a new (fake) message event.
The user may send another command (or another intention as natural
language) when interacting with the current session. In this case,
@ -436,9 +439,9 @@ class CommandSession(BaseSession):
# we think the command is not handled
raise _FinishException(result=False)
if not isinstance(new_ctx_message, Message):
new_ctx_message = Message(new_ctx_message)
raise SwitchException(new_ctx_message)
if not isinstance(new_message, Message):
new_message = Message(new_message)
raise SwitchException(new_message)
def parse_command(bot: NoneBot,
@ -513,26 +516,26 @@ def parse_command(bot: NoneBot,
return cmd, ''.join(cmd_remained)
async def handle_command(bot: NoneBot, ctx: Context_T) -> bool:
async def handle_command(bot: NoneBot, event: CQEvent) -> bool:
"""
Handle a message as a command.
This function is typically called by "handle_message".
:param bot: NoneBot instance
:param ctx: message context
:param event: message event
:return: the message is handled as a command
"""
cmd, current_arg = parse_command(bot, str(ctx['message']).lstrip())
cmd, current_arg = parse_command(bot, str(event.message).lstrip())
is_privileged_cmd = cmd and cmd.privileged
if is_privileged_cmd and cmd.only_to_me and not ctx['to_me']:
if is_privileged_cmd and cmd.only_to_me and not event['to_me']:
is_privileged_cmd = False
disable_interaction = is_privileged_cmd
if is_privileged_cmd:
logger.debug(f'Command {cmd.name} is a privileged command')
ctx_id = context_id(ctx)
ctx_id = context_id(event)
if not is_privileged_cmd:
# wait for 1.5 seconds (at most) if the current session is running
@ -549,7 +552,7 @@ async def handle_command(bot: NoneBot, ctx: Context_T) -> bool:
logger.warning(f'There is a session of command '
f'{session.cmd.name} running, notify the user')
asyncio.ensure_future(send(
bot, ctx,
bot, event,
render_expression(bot.config.SESSION_RUNNING_EXPRESSION)
))
# pretend we are successful, so that NLP won't handle it
@ -558,8 +561,8 @@ async def handle_command(bot: NoneBot, ctx: Context_T) -> bool:
if session.is_valid:
logger.debug(f'Session of command {session.cmd.name} exists')
# since it's in a session, the user must be talking to me
ctx['to_me'] = True
session.refresh(ctx, current_arg=str(ctx['message']))
event['to_me'] = True
session.refresh(event, current_arg=str(event['message']))
# there is no need to check permission for existing session
check_perm = False
else:
@ -573,17 +576,17 @@ async def handle_command(bot: NoneBot, ctx: Context_T) -> bool:
if not cmd:
logger.debug('Not a known command, ignored')
return False
if cmd.only_to_me and not ctx['to_me']:
if cmd.only_to_me and not event['to_me']:
logger.debug('Not to me, ignored')
return False
session = CommandSession(bot, ctx, cmd, current_arg=current_arg)
session = CommandSession(bot, event, cmd, current_arg=current_arg)
logger.debug(f'New session of command {session.cmd.name} created')
return await _real_run_command(session, ctx_id, check_perm=check_perm,
disable_interaction=disable_interaction)
async def call_command(bot: NoneBot, ctx: Context_T,
async def call_command(bot: NoneBot, event: CQEvent,
name: Union[str, CommandName_T], *,
current_arg: str = '',
args: Optional[CommandArgs_T] = None,
@ -601,7 +604,7 @@ async def call_command(bot: NoneBot, ctx: Context_T,
the user for more info).
:param bot: NoneBot instance
:param ctx: message context
:param event: message event
:param name: command name
:param current_arg: command current argument string
:param args: command args
@ -612,10 +615,13 @@ async def call_command(bot: NoneBot, ctx: Context_T,
cmd = _find_command(name)
if not cmd:
return False
session = CommandSession(bot, ctx, cmd, current_arg=current_arg, args=args)
return await _real_run_command(session, context_id(session.ctx),
session = CommandSession(bot, event, cmd,
current_arg=current_arg, args=args)
return await _real_run_command(
session, context_id(session.event),
check_perm=check_perm,
disable_interaction=disable_interaction)
disable_interaction=disable_interaction
)
async def _real_run_command(session: CommandSession,
@ -674,18 +680,18 @@ async def _real_run_command(session: CommandSession,
# make sure there is no session waiting
del _sessions[ctx_id]
logger.debug(f'Session of command {session.cmd.name} switching, '
f'new context message: {e.new_ctx_message}')
f'new message: {e.new_message}')
raise e # this is intended to be propagated to handle_message()
def kill_current_session(ctx: Context_T) -> None:
def kill_current_session(event: CQEvent) -> None:
"""
Force kill current session of the given context,
Force kill current session of the given event context,
despite whether it is running or not.
:param ctx: message context
:param event: message event
"""
ctx_id = context_id(ctx)
ctx_id = context_id(event)
if ctx_id in _sessions:
del _sessions[ctx_id]

46
nonebot/helpers.py
View File

@ -2,51 +2,53 @@ import hashlib
import random
from typing import Sequence, Callable, Any
from aiocqhttp import Event as CQEvent
from . import NoneBot
from .exceptions import CQHttpError
from .message import escape
from .typing import Context_T, Message_T, Expression_T
from .typing import Message_T, Expression_T
def context_id(ctx: Context_T, *,
def context_id(event: CQEvent, *,
mode: str = 'default', use_hash: bool = False) -> str:
"""
Calculate a unique id representing the current context.
Calculate a unique id representing the context of the given event.
mode:
default: one id for one context
group: one id for one group or discuss
user: one id for one user
:param ctx: the context dict
:param event: the event object
:param mode: unique id mode: "default", "group", or "user"
:param use_hash: use md5 to hash the id or not
"""
ctx_id = ''
if mode == 'default':
if ctx.get('group_id'):
ctx_id = f'/group/{ctx["group_id"]}'
elif ctx.get('discuss_id'):
ctx_id = f'/discuss/{ctx["discuss_id"]}'
if ctx.get('user_id'):
ctx_id += f'/user/{ctx["user_id"]}'
if event.group_id:
ctx_id = f'/group/{event.group_id}'
elif event.discuss_id:
ctx_id = f'/discuss/{event.discuss_id}'
if event.user_id:
ctx_id += f'/user/{event.user_id}'
elif mode == 'group':
if ctx.get('group_id'):
ctx_id = f'/group/{ctx["group_id"]}'
elif ctx.get('discuss_id'):
ctx_id = f'/discuss/{ctx["discuss_id"]}'
elif ctx.get('user_id'):
ctx_id = f'/user/{ctx["user_id"]}'
if event.group_id:
ctx_id = f'/group/{event.group_id}'
elif event.discuss_id:
ctx_id = f'/discuss/{event.discuss_id}'
elif event.user_id:
ctx_id = f'/user/{event.user_id}'
elif mode == 'user':
if ctx.get('user_id'):
ctx_id = f'/user/{ctx["user_id"]}'
if event.user_id:
ctx_id = f'/user/{event.user_id}'
if ctx_id and use_hash:
ctx_id = hashlib.md5(ctx_id.encode('ascii')).hexdigest()
return ctx_id
async def send(bot: NoneBot, ctx: Context_T,
async def send(bot: NoneBot, event: CQEvent,
message: Message_T, *,
ensure_private: bool = False,
ignore_failure: bool = True,
@ -54,9 +56,9 @@ async def send(bot: NoneBot, ctx: Context_T,
"""Send a message ignoring failure by default."""
try:
if ensure_private:
ctx = ctx.copy()
ctx['message_type'] = 'private'
return await bot.send(ctx, message, **kwargs)
event = event.copy()
event['message_type'] = 'private'
return await bot.send(event, message, **kwargs)
except CQHttpError:
if not ignore_failure:
raise

91
nonebot/message.py
View File

@ -1,13 +1,13 @@
import asyncio
from typing import Callable
from aiocqhttp import Event as CQEvent
from aiocqhttp.message import *
from . import NoneBot
from .command import handle_command, SwitchException
from .log import logger
from .natural_language import handle_natural_language
from .typing import Context_T
_message_preprocessors = set()
@ -17,76 +17,77 @@ def message_preprocessor(func: Callable) -> Callable:
return func
async def handle_message(bot: NoneBot, ctx: Context_T) -> None:
_log_message(ctx)
async def handle_message(bot: NoneBot, event: CQEvent) -> None:
_log_message(event)
if not ctx['message']:
ctx['message'].append(MessageSegment.text(''))
assert isinstance(event.message, Message)
if not event.message:
event.message.append(MessageSegment.text(''))
coros = []
for processor in _message_preprocessors:
coros.append(processor(bot, ctx))
for preprocessor in _message_preprocessors:
coros.append(preprocessor(bot, event))
if coros:
await asyncio.wait(coros)
raw_to_me = ctx.get('to_me', False)
_check_at_me(bot, ctx)
_check_calling_me_nickname(bot, ctx)
ctx['to_me'] = raw_to_me or ctx['to_me']
raw_to_me = event.get('to_me', False)
_check_at_me(bot, event)
_check_calling_me_nickname(bot, event)
event['to_me'] = raw_to_me or event['to_me']
while True:
try:
handled = await handle_command(bot, ctx)
handled = await handle_command(bot, event)
break
except SwitchException as e:
# we are sure that there is no session existing now
ctx['message'] = e.new_ctx_message
ctx['to_me'] = True
event['message'] = e.new_message
event['to_me'] = True
if handled:
logger.info(f'Message {ctx["message_id"]} is handled as a command')
logger.info(f'Message {event.message_id} is handled as a command')
return
handled = await handle_natural_language(bot, ctx)
handled = await handle_natural_language(bot, event)
if handled:
logger.info(f'Message {ctx["message_id"]} is handled '
logger.info(f'Message {event.message_id} is handled '
f'as natural language')
return
def _check_at_me(bot: NoneBot, ctx: Context_T) -> None:
if ctx['message_type'] == 'private':
ctx['to_me'] = True
def _check_at_me(bot: NoneBot, event: CQEvent) -> None:
if event.detail_type == 'private':
event['to_me'] = True
else:
# group or discuss
ctx['to_me'] = False
at_me_seg = MessageSegment.at(ctx['self_id'])
event['to_me'] = False
at_me_seg = MessageSegment.at(event.self_id)
# check the first segment
first_msg_seg = ctx['message'][0]
first_msg_seg = event.message[0]
if first_msg_seg == at_me_seg:
ctx['to_me'] = True
del ctx['message'][0]
event['to_me'] = True
del event.message[0]
if not ctx['to_me']:
if not event['to_me']:
# check the last segment
i = -1
last_msg_seg = ctx['message'][i]
last_msg_seg = event.message[i]
if last_msg_seg.type == 'text' and \
not last_msg_seg.data['text'].strip() and \
len(ctx['message']) >= 2:
len(event.message) >= 2:
i -= 1
last_msg_seg = ctx['message'][i]
last_msg_seg = event.message[i]
if last_msg_seg == at_me_seg:
ctx['to_me'] = True
del ctx['message'][i:]
event['to_me'] = True
del event.message[i:]
if not ctx['message']:
ctx['message'].append(MessageSegment.text(''))
if not event.message:
event.message.append(MessageSegment.text(''))
def _check_calling_me_nickname(bot: NoneBot, ctx: Context_T) -> None:
first_msg_seg = ctx['message'][0]
def _check_calling_me_nickname(bot: NoneBot, event: CQEvent) -> None:
first_msg_seg = event.message[0]
if first_msg_seg.type != 'text':
return
@ -105,16 +106,16 @@ def _check_calling_me_nickname(bot: NoneBot, ctx: Context_T) -> None:
if m:
nickname = m.group(1)
logger.debug(f'User is calling me {nickname}')
ctx['to_me'] = True
event['to_me'] = True
first_msg_seg.data['text'] = first_text[m.end():]
def _log_message(ctx: Context_T) -> None:
msg_from = str(ctx['user_id'])
if ctx['message_type'] == 'group':
msg_from += f'@[群:{ctx["group_id"]}]'
elif ctx['message_type'] == 'discuss':
msg_from += f'@[讨论组:{ctx["discuss_id"]}]'
logger.info(f'Self: {ctx["self_id"]}, '
f'Message {ctx["message_id"]} from {msg_from}: '
f'{str(ctx["message"]).__repr__()}')
def _log_message(event: CQEvent) -> None:
msg_from = str(event.user_id)
if event.detail_type == 'group':
msg_from += f'@[群:{event.group_id}]'
elif event.detail_type == 'discuss':
msg_from += f'@[讨论组:{event.discuss_id}]'
logger.info(f'Self: {event.self_id}, '
f'Message {event.message_id} from {msg_from}: '
f'{str(event.message).__repr__()}')

25
nonebot/natural_language.py
View File

@ -1,12 +1,14 @@
import asyncio
from typing import Iterable, Optional, Callable, Union, NamedTuple
from aiocqhttp import Event as CQEvent
from . import NoneBot, permission as perm
from .command import call_command
from .log import logger
from .message import Message
from .session import BaseSession
from .typing import Context_T, CommandName_T, CommandArgs_T
from .typing import CommandName_T, CommandArgs_T
_nl_processors = set()
@ -27,8 +29,9 @@ class NLProcessor:
self.allow_empty_message = allow_empty_message
def on_natural_language(keywords: Union[Optional[Iterable], str, Callable] = None,
*, permission: int = perm.EVERYBODY,
def on_natural_language(
keywords: Union[Optional[Iterable], str, Callable] = None, *,
permission: int = perm.EVERYBODY,
only_to_me: bool = True,
only_short_message: bool = True,
allow_empty_message: bool = False) -> Callable:
@ -63,8 +66,8 @@ def on_natural_language(keywords: Union[Optional[Iterable], str, Callable] = Non
class NLPSession(BaseSession):
__slots__ = ('msg', 'msg_text', 'msg_images')
def __init__(self, bot: NoneBot, ctx: Context_T, msg: str):
super().__init__(bot, ctx)
def __init__(self, bot: NoneBot, event: CQEvent, msg: str):
super().__init__(bot, event)
self.msg = msg
tmp_msg = Message(msg)
self.msg_text = tmp_msg.extract_plain_text()
@ -97,17 +100,17 @@ class IntentCommand(NamedTuple):
current_arg: str = ''
async def handle_natural_language(bot: NoneBot, ctx: Context_T) -> bool:
async def handle_natural_language(bot: NoneBot, event: CQEvent) -> bool:
"""
Handle a message as natural language.
This function is typically called by "handle_message".
:param bot: NoneBot instance
:param ctx: message context
:param event: message event
:return: the message is handled as natural language
"""
session = NLPSession(bot, ctx, str(ctx['message']))
session = NLPSession(bot, event, str(event.message))
# use msg_text here because CQ code "share" may be very long,
# at the same time some plugins may want to handle it
@ -123,10 +126,10 @@ async def handle_natural_language(bot: NoneBot, ctx: Context_T) -> bool:
msg_text_length > bot.config.SHORT_MESSAGE_MAX_LENGTH:
continue
if p.only_to_me and not ctx['to_me']:
if p.only_to_me and not event['to_me']:
continue
should_run = await perm.check_permission(bot, ctx, p.permission)
should_run = await perm.check_permission(bot, event, p.permission)
if should_run and p.keywords:
for kw in p.keywords:
if kw in session.msg_text:
@ -162,7 +165,7 @@ async def handle_natural_language(bot: NoneBot, ctx: Context_T) -> bool:
logger.debug(
f'Intent command with highest confidence: {chosen_cmd}')
return await call_command(
bot, ctx, chosen_cmd.name,
bot, event, chosen_cmd.name,
args=chosen_cmd.args,
current_arg=chosen_cmd.current_arg,
check_perm=False

51
nonebot/notice_request.py
View File

@ -1,12 +1,12 @@
from typing import Optional, Callable, Union
from aiocqhttp import Event as CQEvent
from aiocqhttp.bus import EventBus
from . import NoneBot
from .exceptions import CQHttpError
from .log import logger
from .session import BaseSession
from .typing import Context_T
_bus = EventBus()
@ -36,15 +36,15 @@ on_request = _make_event_deco('request')
class NoticeSession(BaseSession):
__slots__ = ()
def __init__(self, bot: NoneBot, ctx: Context_T):
super().__init__(bot, ctx)
def __init__(self, bot: NoneBot, event: CQEvent):
super().__init__(bot, event)
class RequestSession(BaseSession):
__slots__ = ()
def __init__(self, bot: NoneBot, ctx: Context_T):
super().__init__(bot, ctx)
def __init__(self, bot: NoneBot, event: CQEvent):
super().__init__(bot, event)
async def approve(self, remark: str = '') -> None:
"""
@ -55,8 +55,8 @@ class RequestSession(BaseSession):
try:
await self.bot.call_action(
action='.handle_quick_operation_async',
self_id=self.ctx.get('self_id'),
context=self.ctx,
self_id=self.event.self_id,
context=self.event,
operation={'approve': True, 'remark': remark}
)
except CQHttpError:
@ -71,39 +71,34 @@ class RequestSession(BaseSession):
try:
await self.bot.call_action(
action='.handle_quick_operation_async',
self_id=self.ctx.get('self_id'),
context=self.ctx,
self_id=self.event.self_id,
context=self.event,
operation={'approve': False, 'reason': reason}
)
except CQHttpError:
pass
async def handle_notice_or_request(bot: NoneBot, ctx: Context_T) -> None:
post_type = ctx['post_type'] # "notice" or "request"
detail_type = ctx[f'{post_type}_type']
event = f'{post_type}.{detail_type}'
if ctx.get('sub_type'):
event += f'.{ctx["sub_type"]}'
if post_type == 'notice':
_log_notice(ctx)
session = NoticeSession(bot, ctx)
async def handle_notice_or_request(bot: NoneBot, event: CQEvent) -> None:
if event.type == 'notice':
_log_notice(event)
session = NoticeSession(bot, event)
else: # must be 'request'
_log_request(ctx)
session = RequestSession(bot, ctx)
_log_request(event)
session = RequestSession(bot, event)
logger.debug(f'Emitting event: {event}')
ev_name = event.name
logger.debug(f'Emitting event: {ev_name}')
try:
await _bus.emit(event, session)
await _bus.emit(ev_name, session)
except Exception as e:
logger.error(f'An exception occurred while handling event {event}:')
logger.error(f'An exception occurred while handling event {ev_name}:')
logger.exception(e)
def _log_notice(ctx: Context_T) -> None:
logger.info(f'Notice: {ctx}')
def _log_notice(event: CQEvent) -> None:
logger.info(f'Notice: {event}')
def _log_request(ctx: Context_T) -> None:
logger.info(f'Request: {ctx}')
def _log_request(event: CQEvent) -> None:
logger.info(f'Request: {event}')

52
nonebot/permission.py
View File

@ -1,10 +1,10 @@
from collections import namedtuple
from aiocache import cached
from aiocqhttp import Event as CQEvent
from . import NoneBot
from .exceptions import CQHttpError
from .typing import Context_T
PRIVATE_FRIEND = 0x0001
PRIVATE_GROUP = 0x0002
@ -32,7 +32,7 @@ IS_GROUP_OWNER = GROUP_ADMIN | GROUP_OWNER
IS_GROUP = GROUP
IS_SUPERUSER = 0xFFFF
_min_context_fields = (
_min_event_fields = (
'self_id',
'message_type',
'sub_type',
@ -42,52 +42,52 @@ _min_context_fields = (
'anonymous',
)
_MinContext = namedtuple('MinContext', _min_context_fields)
_MinEvent = namedtuple('MinEvent', _min_event_fields)
async def check_permission(bot: NoneBot, ctx: Context_T,
async def check_permission(bot: NoneBot, event: CQEvent,
permission_required: int) -> bool:
"""
Check if the context has the permission required.
Check if the event context has the permission required.
:param bot: NoneBot instance
:param ctx: message context
:param event: message event
:param permission_required: permission required
:return: the context has the permission
"""
min_ctx_kwargs = {}
for field in _min_context_fields:
if field in ctx:
min_ctx_kwargs[field] = ctx[field]
min_event_kwargs = {}
for field in _min_event_fields:
if field in event:
min_event_kwargs[field] = event[field]
else:
min_ctx_kwargs[field] = None
min_ctx = _MinContext(**min_ctx_kwargs)
return await _check(bot, min_ctx, permission_required)
min_event_kwargs[field] = None
min_event = _MinEvent(**min_event_kwargs)
return await _check(bot, min_event, permission_required)
@cached(ttl=2 * 60) # cache the result for 2 minute
async def _check(bot: NoneBot, min_ctx: _MinContext,
async def _check(bot: NoneBot, min_event: _MinEvent,
permission_required: int) -> bool:
permission = 0
if min_ctx.user_id in bot.config.SUPERUSERS:
if min_event.user_id in bot.config.SUPERUSERS:
permission |= IS_SUPERUSER
if min_ctx.message_type == 'private':
if min_ctx.sub_type == 'friend':
if min_event.message_type == 'private':
if min_event.sub_type == 'friend':
permission |= IS_PRIVATE_FRIEND
elif min_ctx.sub_type == 'group':
elif min_event.sub_type == 'group':
permission |= IS_PRIVATE_GROUP
elif min_ctx.sub_type == 'discuss':
elif min_event.sub_type == 'discuss':
permission |= IS_PRIVATE_DISCUSS
elif min_ctx.sub_type == 'other':
elif min_event.sub_type == 'other':
permission |= IS_PRIVATE_OTHER
elif min_ctx.message_type == 'group':
elif min_event.message_type == 'group':
permission |= IS_GROUP_MEMBER
if not min_ctx.anonymous:
if not min_event.anonymous:
try:
member_info = await bot.get_group_member_info(
self_id=min_ctx.self_id,
group_id=min_ctx.group_id,
user_id=min_ctx.user_id,
self_id=min_event.self_id,
group_id=min_event.group_id,
user_id=min_event.user_id,
no_cache=True
)
if member_info:
@ -97,7 +97,7 @@ async def _check(bot: NoneBot, min_ctx: _MinContext,
permission |= IS_GROUP_ADMIN
except CQHttpError:
pass
elif min_ctx.message_type == 'discuss':
elif min_event.message_type == 'discuss':
permission |= IS_DISCUSS
return bool(permission & permission_required)

22
nonebot/session.py
View File

@ -1,18 +1,28 @@
from aiocqhttp import Event as CQEvent
from . import NoneBot
from .helpers import send
from .typing import Context_T, Message_T
from .typing import Message_T
class BaseSession:
__slots__ = ('bot', 'ctx')
__slots__ = ('bot', 'event')
def __init__(self, bot: NoneBot, ctx: Context_T):
def __init__(self, bot: NoneBot, event: CQEvent):
self.bot = bot
self.ctx = ctx
self.event = event
@property
def ctx(self) -> CQEvent:
return self.event
@ctx.setter
def ctx(self, val: CQEvent) -> None:
self.event = val
@property
def self_id(self) -> int:
return self.ctx['self_id']
return self.event.self_id
async def send(self, message: Message_T, *,
at_sender: bool = False,
@ -28,7 +38,7 @@ class BaseSession:
:param ignore_failure: if any CQHttpError raised, ignore it
:return: the result returned by CQHTTP
"""
return await send(self.bot, self.ctx, message,
return await send(self.bot, self.event, message,
at_sender=at_sender,
ensure_private=ensure_private,
ignore_failure=ignore_failure, **kwargs)

1
nonebot/typing.py
View File

@ -1,6 +1,5 @@
from typing import Union, List, Dict, Any, Sequence, Callable, Tuple, Awaitable
Context_T = Dict[str, Any]
Message_T = Union[str, Dict[str, Any], List[Dict[str, Any]]]
Expression_T = Union[str, Sequence[str], Callable]
CommandName_T = Tuple[str, ...]