mirror of
https://github.com/nonebot/nonebot2.git
synced 2025-07-27 16:21:28 +00:00
🎨 improve cqhttp
This commit is contained in:
yanyongyu
@ -10,7 +10,7 @@ CQHTTP (OneBot) v11 协议适配
|
||||
https://github.com/howmanybots/onebot/blob/master/README.md
|
||||
"""
|
||||
|
||||
from .utils import log
|
||||
from .event import Event
|
||||
from .message import Message, MessageSegment
|
||||
from .bot import Bot, _check_at_me, _check_nickname, _check_reply
|
||||
from .utils import log, escape, unescape, _b2s
|
||||
from .bot import Bot, _check_at_me, _check_nickname, _check_reply, _handle_api_result
|
||||
|
||||
@ -10,14 +10,24 @@ from nonebot.log import logger
|
||||
from nonebot.config import Config
|
||||
from nonebot.adapters import BaseBot
|
||||
from nonebot.message import handle_event
|
||||
from nonebot.exception import RequestDenied
|
||||
from nonebot.typing import Any, Dict, Union, Optional
|
||||
from nonebot.typing import overrides, Driver, WebSocket, NoReturn
|
||||
from nonebot.exception import NetworkError, RequestDenied, ApiNotAvailable
|
||||
|
||||
from .event import Event
|
||||
from .exception import ApiError
|
||||
from .utils import log, get_auth_bearer
|
||||
from .message import Message, MessageSegment
|
||||
from .exception import NetworkError, ApiNotAvailable, ActionFailed
|
||||
from .utils import log
|
||||
|
||||
|
||||
def get_auth_bearer(
|
||||
access_token: Optional[str] = None) -> Union[Optional[str], NoReturn]:
|
||||
if not access_token:
|
||||
return None
|
||||
scheme, _, param = access_token.partition(" ")
|
||||
if scheme.lower() not in ["bearer", "token"]:
|
||||
raise RequestDenied(401, "Not authenticated")
|
||||
return param
|
||||
|
||||
|
||||
async def _check_reply(bot: "Bot", event: "Event"):
|
||||
@ -160,11 +170,11 @@ def _handle_api_result(
|
||||
|
||||
:异常:
|
||||
|
||||
- ``ApiError``: API 调用失败
|
||||
- ``ActionFailed``: API 调用失败
|
||||
"""
|
||||
if isinstance(result, dict):
|
||||
if result.get("status") == "failed":
|
||||
raise ApiError(retcode=result.get("retcode"))
|
||||
raise ActionFailed(retcode=result.get("retcode"))
|
||||
return result.get("data")
|
||||
|
||||
|
||||
@ -318,7 +328,7 @@ class Bot(BaseBot):
|
||||
:异常:
|
||||
|
||||
- ``NetworkError``: 网络错误
|
||||
- ``ApiError``: API 调用失败
|
||||
- ``ActionFailed``: API 调用失败
|
||||
"""
|
||||
if "self_id" in data:
|
||||
self_id = data.pop("self_id")
|
||||
@ -393,7 +403,7 @@ class Bot(BaseBot):
|
||||
|
||||
- ``ValueError``: 缺少 ``user_id``, ``group_id``
|
||||
- ``NetworkError``: 网络错误
|
||||
- ``ApiError``: API 调用失败
|
||||
- ``ActionFailed``: API 调用失败
|
||||
"""
|
||||
msg = message if isinstance(message, Message) else Message(message)
|
||||
|
||||
|
||||
@ -1,12 +1,11 @@
|
||||
import asyncio
|
||||
|
||||
from nonebot.config import Config
|
||||
from nonebot.adapters import BaseBot, BaseEvent, BaseMessage, BaseMessageSegment
|
||||
from nonebot.typing import Any, Dict, List, Union, Driver, Optional, NoReturn, WebSocket, Iterable
|
||||
from nonebot.adapters import BaseBot
|
||||
from nonebot.typing import Any, Dict, List, Union, Driver, Optional, NoReturn, WebSocket
|
||||
|
||||
|
||||
def log(level: str, message: str):
|
||||
...
|
||||
from .event import Event
|
||||
from .message import Message, MessageSegment
|
||||
|
||||
|
||||
def get_auth_bearer(
|
||||
@ -14,27 +13,15 @@ def get_auth_bearer(
|
||||
...
|
||||
|
||||
|
||||
def escape(s: str, *, escape_comma: bool = ...) -> str:
|
||||
async def _check_reply(bot: "Bot", event: Event):
|
||||
...
|
||||
|
||||
|
||||
def unescape(s: str) -> str:
|
||||
def _check_at_me(bot: "Bot", event: Event):
|
||||
...
|
||||
|
||||
|
||||
def _b2s(b: Optional[bool]) -> Optional[str]:
|
||||
...
|
||||
|
||||
|
||||
async def _check_reply(bot: "Bot", event: "Event"):
|
||||
...
|
||||
|
||||
|
||||
def _check_at_me(bot: "Bot", event: "Event"):
|
||||
...
|
||||
|
||||
|
||||
def _check_nickname(bot: "Bot", event: "Event"):
|
||||
def _check_nickname(bot: "Bot", event: Event):
|
||||
...
|
||||
|
||||
|
||||
@ -86,8 +73,8 @@ class Bot(BaseBot):
|
||||
async def call_api(self, api: str, **data) -> Union[Any, NoReturn]:
|
||||
...
|
||||
|
||||
async def send(self, event: "Event", message: Union[str, "Message",
|
||||
"MessageSegment"],
|
||||
async def send(self, event: Event, message: Union[str, Message,
|
||||
MessageSegment],
|
||||
**kwargs) -> Union[Any, NoReturn]:
|
||||
...
|
||||
|
||||
@ -759,242 +746,3 @@ class Bot(BaseBot):
|
||||
* ``self_id``: 机器人 QQ 号
|
||||
"""
|
||||
...
|
||||
|
||||
|
||||
class Event(BaseEvent):
|
||||
|
||||
def __init__(self, raw_event: dict):
|
||||
...
|
||||
|
||||
@property
|
||||
def id(self) -> Optional[int]:
|
||||
...
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
...
|
||||
|
||||
@property
|
||||
def self_id(self) -> str:
|
||||
...
|
||||
|
||||
@property
|
||||
def time(self) -> int:
|
||||
...
|
||||
|
||||
@property
|
||||
def type(self) -> str:
|
||||
...
|
||||
|
||||
@type.setter
|
||||
def type(self, value) -> None:
|
||||
...
|
||||
|
||||
@property
|
||||
def detail_type(self) -> str:
|
||||
...
|
||||
|
||||
@detail_type.setter
|
||||
def detail_type(self, value) -> None:
|
||||
...
|
||||
|
||||
@property
|
||||
def sub_type(self) -> Optional[str]:
|
||||
...
|
||||
|
||||
@sub_type.setter
|
||||
def sub_type(self, value) -> None:
|
||||
...
|
||||
|
||||
@property
|
||||
def user_id(self) -> Optional[int]:
|
||||
...
|
||||
|
||||
@user_id.setter
|
||||
def user_id(self, value) -> None:
|
||||
...
|
||||
|
||||
@property
|
||||
def group_id(self) -> Optional[int]:
|
||||
...
|
||||
|
||||
@group_id.setter
|
||||
def group_id(self, value) -> None:
|
||||
...
|
||||
|
||||
@property
|
||||
def to_me(self) -> Optional[bool]:
|
||||
...
|
||||
|
||||
@to_me.setter
|
||||
def to_me(self, value) -> None:
|
||||
...
|
||||
|
||||
@property
|
||||
def message(self) -> Optional["Message"]:
|
||||
...
|
||||
|
||||
@message.setter
|
||||
def message(self, value) -> None:
|
||||
...
|
||||
|
||||
@property
|
||||
def reply(self) -> Optional[dict]:
|
||||
...
|
||||
|
||||
@reply.setter
|
||||
def reply(self, value) -> None:
|
||||
...
|
||||
|
||||
@property
|
||||
def raw_message(self) -> Optional[str]:
|
||||
...
|
||||
|
||||
@raw_message.setter
|
||||
def raw_message(self, value) -> None:
|
||||
...
|
||||
|
||||
@property
|
||||
def plain_text(self) -> Optional[str]:
|
||||
...
|
||||
|
||||
@property
|
||||
def sender(self) -> Optional[dict]:
|
||||
...
|
||||
|
||||
@sender.setter
|
||||
def sender(self, value) -> None:
|
||||
...
|
||||
|
||||
|
||||
class MessageSegment(BaseMessageSegment):
|
||||
|
||||
def __init__(self, type: str, data: Dict[str, Any]) -> None:
|
||||
...
|
||||
|
||||
def __str__(self):
|
||||
...
|
||||
|
||||
def __add__(self, other) -> "Message":
|
||||
...
|
||||
|
||||
@staticmethod
|
||||
def anonymous(ignore_failure: Optional[bool] = ...) -> "MessageSegment":
|
||||
...
|
||||
|
||||
@staticmethod
|
||||
def at(user_id: Union[int, str]) -> "MessageSegment":
|
||||
...
|
||||
|
||||
@staticmethod
|
||||
def contact_group(group_id: int) -> "MessageSegment":
|
||||
...
|
||||
|
||||
@staticmethod
|
||||
def contact_user(user_id: int) -> "MessageSegment":
|
||||
...
|
||||
|
||||
@staticmethod
|
||||
def dice() -> "MessageSegment":
|
||||
...
|
||||
|
||||
@staticmethod
|
||||
def face(id_: int) -> "MessageSegment":
|
||||
...
|
||||
|
||||
@staticmethod
|
||||
def forward(id_: str) -> "MessageSegment":
|
||||
...
|
||||
|
||||
@staticmethod
|
||||
def image(file: str,
|
||||
type_: Optional[str] = ...,
|
||||
cache: bool = ...,
|
||||
proxy: bool = ...,
|
||||
timeout: Optional[int] = ...) -> "MessageSegment":
|
||||
...
|
||||
|
||||
@staticmethod
|
||||
def json(data: str) -> "MessageSegment":
|
||||
...
|
||||
|
||||
@staticmethod
|
||||
def location(latitude: float,
|
||||
longitude: float,
|
||||
title: Optional[str] = ...,
|
||||
content: Optional[str] = ...) -> "MessageSegment":
|
||||
...
|
||||
|
||||
@staticmethod
|
||||
def music(type_: str, id_: int) -> "MessageSegment":
|
||||
...
|
||||
|
||||
@staticmethod
|
||||
def music_custom(url: str,
|
||||
audio: str,
|
||||
title: str,
|
||||
content: Optional[str] = ...,
|
||||
img_url: Optional[str] = ...) -> "MessageSegment":
|
||||
...
|
||||
|
||||
@staticmethod
|
||||
def node(id_: int) -> "MessageSegment":
|
||||
...
|
||||
|
||||
@staticmethod
|
||||
def node_custom(user_id: int, nickname: str,
|
||||
content: Union[str, "Message"]) -> "MessageSegment":
|
||||
...
|
||||
|
||||
@staticmethod
|
||||
def poke(type_: str, id_: str) -> "MessageSegment":
|
||||
...
|
||||
|
||||
@staticmethod
|
||||
def record(file: str,
|
||||
magic: Optional[bool] = ...,
|
||||
cache: Optional[bool] = ...,
|
||||
proxy: Optional[bool] = ...,
|
||||
timeout: Optional[int] = ...) -> "MessageSegment":
|
||||
...
|
||||
|
||||
@staticmethod
|
||||
def reply(id_: int) -> "MessageSegment":
|
||||
...
|
||||
|
||||
@staticmethod
|
||||
def rps() -> "MessageSegment":
|
||||
...
|
||||
|
||||
@staticmethod
|
||||
def shake() -> "MessageSegment":
|
||||
...
|
||||
|
||||
@staticmethod
|
||||
def share(url: str = ...,
|
||||
title: str = ...,
|
||||
content: Optional[str] = ...,
|
||||
img_url: Optional[str] = ...) -> "MessageSegment":
|
||||
...
|
||||
|
||||
@staticmethod
|
||||
def text(text: str) -> "MessageSegment":
|
||||
...
|
||||
|
||||
@staticmethod
|
||||
def video(file: str,
|
||||
cache: Optional[bool] = ...,
|
||||
proxy: Optional[bool] = ...,
|
||||
timeout: Optional[int] = ...) -> "MessageSegment":
|
||||
...
|
||||
|
||||
@staticmethod
|
||||
def xml(data: str) -> "MessageSegment":
|
||||
...
|
||||
|
||||
|
||||
class Message(BaseMessage):
|
||||
|
||||
@staticmethod
|
||||
def _construct(msg: Union[str, dict, list]) -> Iterable[MessageSegment]:
|
||||
...
|
||||
@ -1,6 +1,5 @@
|
||||
from nonebot.typing import overrides
|
||||
from nonebot.typing import Optional
|
||||
from nonebot.adapters import BaseEvent
|
||||
from nonebot.typing import Optional, overrides
|
||||
|
||||
from .message import Message
|
||||
|
||||
|
||||
@ -1,4 +1,8 @@
|
||||
from nonebot.exception import AdapterException, ActionFailed
|
||||
from nonebot.typing import Optional
|
||||
from nonebot.exception import (AdapterException, ActionFailed as
|
||||
BaseActionFailed, NetworkError as
|
||||
BaseNetworkError, ApiNotAvailable as
|
||||
BaseApiNotAvailable)
|
||||
|
||||
|
||||
class CQHTTPAdapterException(AdapterException):
|
||||
@ -7,7 +11,7 @@ class CQHTTPAdapterException(AdapterException):
|
||||
super().__init__("cqhttp")
|
||||
|
||||
|
||||
class ApiError(CQHTTPAdapterException, ActionFailed):
|
||||
class ActionFailed(BaseActionFailed, CQHTTPAdapterException):
|
||||
"""
|
||||
:说明:
|
||||
|
||||
@ -27,3 +31,29 @@ class ApiError(CQHTTPAdapterException, ActionFailed):
|
||||
|
||||
def __str__(self):
|
||||
return self.__repr__()
|
||||
|
||||
|
||||
class NetworkError(BaseNetworkError, CQHTTPAdapterException):
|
||||
"""
|
||||
:说明:
|
||||
|
||||
网络错误。
|
||||
|
||||
:参数:
|
||||
|
||||
* ``retcode: Optional[int]``: 错误码
|
||||
"""
|
||||
|
||||
def __init__(self, msg: Optional[str] = None):
|
||||
super().__init__()
|
||||
self.msg = msg
|
||||
|
||||
def __repr__(self):
|
||||
return f"<NetWorkError message={self.msg}>"
|
||||
|
||||
def __str__(self):
|
||||
return self.__repr__()
|
||||
|
||||
|
||||
class ApiNotAvailable(BaseApiNotAvailable, CQHTTPAdapterException):
|
||||
pass
|
||||
|
||||
@ -3,6 +3,7 @@ import re
|
||||
from nonebot.typing import overrides
|
||||
from nonebot.typing import Any, Dict, Union, Tuple, Iterable, Optional
|
||||
from nonebot.adapters import BaseMessage, BaseMessageSegment
|
||||
|
||||
from .utils import log, escape, unescape, _b2s
|
||||
|
||||
|
||||
|
||||
@ -1,21 +1,9 @@
|
||||
from nonebot.typing import NoReturn
|
||||
from nonebot.typing import Union, Optional
|
||||
from nonebot.exception import RequestDenied
|
||||
from nonebot.typing import Optional
|
||||
from nonebot.utils import logger_wrapper
|
||||
|
||||
log = logger_wrapper("CQHTTP")
|
||||
|
||||
|
||||
def get_auth_bearer(
|
||||
access_token: Optional[str] = None) -> Union[Optional[str], NoReturn]:
|
||||
if not access_token:
|
||||
return None
|
||||
scheme, _, param = access_token.partition(" ")
|
||||
if scheme.lower() not in ["bearer", "token"]:
|
||||
raise RequestDenied(401, "Not authenticated")
|
||||
return param
|
||||
|
||||
|
||||
def escape(s: str, *, escape_comma: bool = True) -> str:
|
||||
"""
|
||||
:说明:
|
||||
|
||||
Reference in New Issue
Block a user