mirror of
https://github.com/nonebot/nonebot2.git
synced 2024-09-21 13:22:35 +00:00
95 lines
2.6 KiB
Python
95 lines
2.6 KiB
Python
from typing import Dict, Any, Optional, Callable, Union
|
|
|
|
from aiocqhttp import Error as CQHttpError
|
|
from aiocqhttp.bus import EventBus
|
|
|
|
from . import NoneBot
|
|
from .log import logger
|
|
from .session import BaseSession
|
|
|
|
_bus = EventBus()
|
|
|
|
|
|
def _make_event_deco(post_type: str) -> Callable:
|
|
def deco_deco(arg: Optional[Union[str, Callable]] = None,
|
|
*events: str) -> Callable:
|
|
def deco(func: Callable) -> Callable:
|
|
if isinstance(arg, str):
|
|
for e in [arg] + list(events):
|
|
_bus.subscribe(f'{post_type}.{e}', func)
|
|
else:
|
|
_bus.subscribe(post_type, func)
|
|
return func
|
|
|
|
if isinstance(arg, Callable):
|
|
return deco(arg)
|
|
return deco
|
|
|
|
return deco_deco
|
|
|
|
|
|
on_notice = _make_event_deco('notice')
|
|
on_request = _make_event_deco('request')
|
|
|
|
|
|
class NoticeSession(BaseSession):
|
|
__slots__ = ()
|
|
|
|
def __init__(self, bot: NoneBot, ctx: Dict[str, Any]):
|
|
super().__init__(bot, ctx)
|
|
|
|
|
|
class RequestSession(BaseSession):
|
|
__slots__ = ()
|
|
|
|
def __init__(self, bot: NoneBot, ctx: Dict[str, Any]):
|
|
super().__init__(bot, ctx)
|
|
|
|
async def approve(self, remark: str = ''):
|
|
try:
|
|
await self.bot.call_action(
|
|
action='.handle_quick_operation_async',
|
|
self_id=self.ctx.get('self_id'),
|
|
context=self.ctx,
|
|
operation={'approve': True, 'remark': remark}
|
|
)
|
|
except CQHttpError:
|
|
pass
|
|
|
|
async def reject(self, reason: str = ''):
|
|
try:
|
|
await self.bot.call_action(
|
|
action='.handle_quick_operation_async',
|
|
self_id=self.ctx.get('self_id'),
|
|
context=self.ctx,
|
|
operation={'approve': False, 'reason': reason}
|
|
)
|
|
except CQHttpError:
|
|
pass
|
|
|
|
|
|
async def handle_notice_or_request(bot: NoneBot, ctx: Dict[str, Any]) -> 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)
|
|
else: # must be 'request'
|
|
_log_request(ctx)
|
|
session = RequestSession(bot, ctx)
|
|
|
|
logger.debug(f'Emitting event: {event}')
|
|
await _bus.emit(event, session)
|
|
|
|
|
|
def _log_notice(ctx: Dict[str, Any]) -> None:
|
|
logger.info(f'Notice: {ctx}')
|
|
|
|
|
|
def _log_request(ctx: Dict[str, Any]) -> None:
|
|
logger.info(f'Request: {ctx}')
|