Python库 | graia-application-mirai-0.8.0.tar.gz

import asyncio import atexit from functools import partial from typing import List, NoReturn, Optional, Tuple, Union import graia.application.event.lifecycle import graia.application.event.mirai # for init events from aiohttp import ClientSession, FormData from graia.application.event import MiraiEvent from graia.broadcast import Broadcast from graia.broadcast.entities.event import BaseEvent from graia.broadcast.entities.inject_rule import SpecialEventType from graia.broadcast.utilles import printer, run_always_await from yarl import URL from .context import enter_message_send_context from .entities import MiraiConfig, UploadMethods from .event.messages import FriendMessage, GroupMessage, TempMessage from .exceptions import InvaildArgument, InvaildSession from .friend import Friend from .group import Group, GroupConfig, Member, MemberInfo from .logger import AbstractLogger, LoggingLogger from .message import BotMessage from .message.chain import MessageChain from .message.elements import external from .message.elements.internal import Image, Source, Voice from .session import Session from .utilles import (AppMiddlewareAsDispatcher, SinceVersion, raise_for_return_code, requireAuthenticated, applicationContextManager) class GraiaMiraiApplication: """本类的实例即 应用实例(Application), 是面向 `mirai-api-http` 接口的实际功能实现. 你的应用大多都围绕着本类及本类的实例展开. Attributes: broadcast (Broadcast): 被指定的, 外置的事件系统, 即 `Broadcast Control`, 通常你不需要干涉该属性; session (ClientSession): 即 `aiohttp.ClientSession` 的实例, 用于与 `mirai-api-http` 通讯. connect_info (Session): 用于描述会话对象, 其中最重要的属性是 `sessionKey`, 用于存储当前的会话标识. logger (AbstractLogger): 日志系统实现类的实例, 默认以 `logging` 为日志驱动. """ __slots__ = ( "broadcast", "session", "connect_info", "logger", "debug" ) broadcast: Broadcast session: ClientSession connect_info: Session logger: AbstractLogger debug: bool def __init__(self, *, broadcast: Broadcast, connect_info: Session, session: Optional[ClientSession] = None, logger: Optional[AbstractLogger] = None, debug: bool = False, enable_chat_log: bool = True ): self.broadcast = broadcast self.connect_info = connect_info self.session = session or ClientSession(loop=broadcast.loop) self.logger = logger or LoggingLogger(**({ "debug": True } if debug else {})) self.debug = debug if enable_chat_log: self.broadcast.receiver("GroupMessage")(self.logger_group_message) self.broadcast.receiver("FriendMessage")(self.logger_friend_message) self.broadcast.receiver("TempMessage")(self.logger_temp_message) self.broadcast.addInjectionRule( SpecialEventType(MiraiEvent, AppMiddlewareAsDispatcher(self)) ) def logger_group_message(self, event: GroupMessage): self.logger.info("[BOT {bot_id}, GroupMessage] [{group_name}({group_id}, perm: {bot_permission})] {member_name}({member_id}, {member_permission}) -> {message_string}".format_map(dict( group_id=event.sender.group.id, group_name=event.sender.group.name, member_id=event.sender.id, member_name=event.sender.name, member_permission=event.sender.permission.name, bot_id=self.connect_info.account, bot_permission=event.sender.group.accountPerm.name, message_string=event.messageChain.asSerializationString(), ))) def logger_friend_message(self, event: FriendMessage): self.logger.info("[BOT {bot_id}, FriendMessage] {friend_name}({friend_id}) -> {message_string}".format_map(dict( bot_id=self.connect_info.account, friend_name=event.sender.nickname, friend_id=event.sender.id, message_string=event.messageChain.asSerializationString() ))) def logger_temp_message(self, event: TempMessage): self.logger.info("[BOT {bot_id}, TempMessage] [{group_name}({group_id}, perm: {bot_permission})] {member_name}({member_id}, {member_permission}) -> {message_string}".format_map(dict( group_id=event.sender.group.id, group_name=event.sender.group.name, member_id=event.sender.id, member_name=event.sender.name, member_permission=event.sender.permission.name, bot_id=self.connect_info.account, bot_permission=event.sender.group.accountPerm.name, message_string=event.messageChain.asSerializationString(), ))) def url_gen(self, path) -> str: """从 connect_info 和 path 生成接口的地址. Args: path (str): 需求的接口地址 Returns: str: 作为结果的地址 """ return str(URL(str(self.connect_info.host)).parent / path) @SinceVersion(1,6,2) @applicationContextManager async def getVersion(self, auto_set=True) -> Tuple: """从 `/about` 路由下获取当前使用的 `mirai-api-http` 版本, 注意, 该 API 并不是一开始就有的(1.6.2 版本才支持本接口). Args: auto_set (bool, optional): 是否自动将版本存入 connect_info 以判断接口是否有效. Defaults to True. Returns: Tuple: 以元组形式表示的版本信息. """ async with self.session.get(self.url_gen("about")) as response: response.raise_for_status() data = await response.json() raise_for_return_code(data) version = (int(i) for i in data['data']['version'][1:].split(".")) if auto_set: self.connect_info.current_version = version return version @applicationContextManager async def authenticate(self) -> str: """从路由 `/auth` 下获取尚未被激活的会话标识并返回; 通常的, 你还需要使用 `activeSession` 方法激活它. Returns: str: 即返回的会话标识 """ async with self.session.post(self.url_gen("auth"), json={ "authKey": self.connect_info.authKey }) as response: response.raise_for_status() data = await response.json() raise_for_return_code(data) self.connect_info.sessionKey = data['session'] return data['session'] @applicationContextManager async def activeSession(self) -> NoReturn: """激活当前已经存入 connect_info 的会话标识, 如果没有事先调用 `authenticate` 方法获取未激活的会话标识, 则会触发 `InvaildSession` 错误. Raises: InvaildSession: 没有事先调用 `authenticate` 方法获取未激活的会话标识 Returns: NoReturn: 没有有意义的返回, 或者说, 返回 `None` 就代表这个操作成功了. """ if not self.connect_info.sessionKey: raise InvaildSession("you should call 'authenticate' before this to get a sessionKey!") async with self.session.post(self.url_gen("verify"), json={ "sessionKey": self.connect_info.sessionKey, "qq": self.connect_info.account }) as response: response.raise_for_status() data = await response.json() raise_for_return_code(data) @requireAuthenticated @applicationContextManager async def signout(se