ProjectHentai/nonebot2
6
0
Fork 0
mirror of https://github.com/nonebot/nonebot2.git synced 2025-10-31 06:56:39 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

🎨 format code using black and isort

Browse Source
This commit is contained in:
yanyongyu
2021-11-22 23:21:26 +08:00
parent 602185a34e
commit a98d98cd12
86 changed files with 2893 additions and 2095 deletions
Download Patch File Download Diff File Expand all files Collapse all files

146
packages/nonebot-adapter-cqhttp/nonebot/adapters/cqhttp/bot.py
View File

@@ -12,8 +12,15 @@ from nonebot.typing import overrides
from nonebot.message import handle_event
from nonebot.adapters import Bot as BaseBot
from nonebot.utils import DataclassEncoder, escape_tag
from nonebot.drivers import (Driver, WebSocket, HTTPRequest, HTTPResponse,
ForwardDriver, HTTPConnection, WebSocketSetup)
from nonebot.drivers import (
Driver,
WebSocket,
HTTPRequest,
HTTPResponse,
ForwardDriver,
HTTPConnection,
WebSocketSetup,
)
from .utils import log, escape
from .config import Config as CQHTTPConfig
@@ -49,15 +56,12 @@ async def _check_reply(bot: "Bot", event: "Event"):
return
try:
index = list(map(lambda x: x.type == "reply",
event.message)).index(True)
index = list(map(lambda x: x.type == "reply", event.message)).index(True)
except ValueError:
return
msg_seg = event.message[index]
try:
event.reply = Reply.parse_obj(await
bot.get_msg(message_id=msg_seg.data["id"]
))
event.reply = Reply.parse_obj(await bot.get_msg(message_id=msg_seg.data["id"]))
except Exception as e:
log("WARNING", f"Error when getting message reply info: {repr(e)}", e)
return
@@ -68,8 +72,7 @@ async def _check_reply(bot: "Bot", event: "Event"):
if len(event.message) > index and event.message[index].type == "at":
del event.message[index]
if len(event.message) > index and event.message[index].type == "text":
event.message[index].data["text"] = event.message[index].data[
"text"].lstrip()
event.message[index].data["text"] = event.message[index].data["text"].lstrip()
if not event.message[index].data["text"]:
del event.message[index]
if not event.message:
@@ -99,23 +102,24 @@ def _check_at_me(bot: "Bot", event: "Event"):
else:
def _is_at_me_seg(segment: MessageSegment):
return segment.type == "at" and str(segment.data.get(
"qq", "")) == str(event.self_id)
return segment.type == "at" and str(segment.data.get("qq", "")) == str(
event.self_id
)
# check the first segment
if _is_at_me_seg(event.message[0]):
event.to_me = True
event.message.pop(0)
if event.message and event.message[0].type == "text":
event.message[0].data["text"] = event.message[0].data[
"text"].lstrip()
event.message[0].data["text"] = event.message[0].data["text"].lstrip()
if not event.message[0].data["text"]:
del event.message[0]
if event.message and _is_at_me_seg(event.message[0]):
event.message.pop(0)
if event.message and event.message[0].type == "text":
event.message[0].data["text"] = event.message[0].data[
"text"].lstrip()
event.message[0].data["text"] = (
event.message[0].data["text"].lstrip()
)
if not event.message[0].data["text"]:
del event.message[0]
@@ -123,9 +127,11 @@ def _check_at_me(bot: "Bot", event: "Event"):
# check the last segment
i = -1
last_msg_seg = event.message[i]
if last_msg_seg.type == "text" and \
not last_msg_seg.data["text"].strip() and \
len(event.message) >= 2:
if (
last_msg_seg.type == "text"
and not last_msg_seg.data["text"].strip()
and len(event.message) >= 2
):
i -= 1
last_msg_seg = event.message[i]
@@ -161,13 +167,12 @@ def _check_nickname(bot: "Bot", event: "Event"):
if nicknames:
# check if the user is calling me with my nickname
nickname_regex = "|".join(nicknames)
m = re.search(rf"^({nickname_regex})([\s,，]*|$)", first_text,
re.IGNORECASE)
m = re.search(rf"^({nickname_regex})([\s,，]*|$)", first_text, re.IGNORECASE)
if m:
nickname = m.group(1)
log("DEBUG", f"User is calling me {nickname}")
event.to_me = True
first_msg_seg.data["text"] = first_text[m.end():]
first_msg_seg.data["text"] = first_text[m.end() :]
def _handle_api_result(result: Optional[Dict[str, Any]]) -> Any:
@@ -206,8 +211,9 @@ class ResultStore:
@classmethod
def add_result(cls, result: Dict[str, Any]):
if isinstance(result.get("echo"), dict) and \
isinstance(result["echo"].get("seq"), int):
if isinstance(result.get("echo"), dict) and isinstance(
result["echo"].get("seq"), int
):
future = cls._futures.get(result["echo"]["seq"])
if future:
future.set_result(result)
@@ -228,6 +234,7 @@ class Bot(BaseBot):
"""
CQHTTP 协议 Bot 适配。继承属性参考 `BaseBot <./#class-basebot>`_ 。
"""
cqhttp_config: CQHTTPConfig
@property
@@ -249,22 +256,25 @@ class Bot(BaseBot):
elif isinstance(driver, ForwardDriver) and cls.cqhttp_config.ws_urls:
for self_id, url in cls.cqhttp_config.ws_urls.items():
try:
headers = {
"authorization":
f"Bearer {cls.cqhttp_config.access_token}"
} if cls.cqhttp_config.access_token else {}
headers = (
{"authorization": f"Bearer {cls.cqhttp_config.access_token}"}
if cls.cqhttp_config.access_token
else {}
)
driver.setup_websocket(
WebSocketSetup("cqhttp", self_id, url, headers=headers))
WebSocketSetup("cqhttp", self_id, url, headers=headers)
)
except Exception as e:
logger.opt(colors=True, exception=e).error(
f"<r><bg #f8bbd0>Bad url {escape_tag(url)} for bot {escape_tag(self_id)} "
"in cqhttp forward websocket</bg #f8bbd0></r>")
"in cqhttp forward websocket</bg #f8bbd0></r>"
)
@classmethod
@overrides(BaseBot)
async def check_permission(
cls, driver: Driver,
request: HTTPConnection) -> Tuple[Optional[str], HTTPResponse]:
cls, driver: Driver, request: HTTPConnection
) -> Tuple[Optional[str], HTTPResponse]:
"""
:说明:
@@ -286,22 +296,26 @@ class Bot(BaseBot):
if not x_signature:
log("WARNING", "Missing Signature Header")
return None, HTTPResponse(401, b"Missing Signature")
sig = hmac.new(secret.encode("utf-8"), request.body,
"sha1").hexdigest()
sig = hmac.new(secret.encode("utf-8"), request.body, "sha1").hexdigest()
if x_signature != "sha1=" + sig:
log("WARNING", "Signature Header is invalid")
return None, HTTPResponse(403, b"Signature is invalid")
access_token = cqhttp_config.access_token
if access_token and access_token != token and isinstance(
request, WebSocket):
if access_token and access_token != token and isinstance(request, WebSocket):
log(
"WARNING", "Authorization Header is invalid"
if token else "Missing Authorization Header")
"WARNING",
"Authorization Header is invalid"
if token
else "Missing Authorization Header",
)
return None, HTTPResponse(
403, b"Authorization Header is invalid"
if token else b"Missing Authorization Header")
return str(x_self_id), HTTPResponse(204, b'')
403,
b"Authorization Header is invalid"
if token
else b"Missing Authorization Header",
)
return str(x_self_id), HTTPResponse(204, b"")
@overrides(BaseBot)
async def handle_message(self, message: bytes):
@@ -320,7 +334,7 @@ class Bot(BaseBot):
return
try:
post_type = data['post_type']
post_type = data["post_type"]
detail_type = data.get(f"{post_type}_type")
detail_type = f".{detail_type}" if detail_type else ""
sub_type = data.get("sub_type")
@@ -352,17 +366,13 @@ class Bot(BaseBot):
if isinstance(self.request, WebSocket):
seq = ResultStore.get_seq()
json_data = json.dumps(
{
"action": api,
"params": data,
"echo": {
"seq": seq
}
},
cls=DataclassEncoder)
{"action": api, "params": data, "echo": {"seq": seq}},
cls=DataclassEncoder,
)
await self.request.send(json_data)
return _handle_api_result(await ResultStore.fetch(
seq, self.config.api_timeout))
return _handle_api_result(
await ResultStore.fetch(seq, self.config.api_timeout)
)
elif isinstance(self.request, HTTPRequest):
api_root = self.config.api_root.get(self.self_id)
@@ -373,22 +383,25 @@ class Bot(BaseBot):
headers = {"Content-Type": "application/json"}
if self.cqhttp_config.access_token is not None:
headers[
"Authorization"] = "Bearer " + self.cqhttp_config.access_token
headers["Authorization"] = "Bearer " + self.cqhttp_config.access_token
try:
async with httpx.AsyncClient(headers=headers,
follow_redirects=True) as client:
async with httpx.AsyncClient(
headers=headers, follow_redirects=True
) as client:
response = await client.post(
api_root + api,
content=json.dumps(data, cls=DataclassEncoder),
timeout=self.config.api_timeout)
timeout=self.config.api_timeout,
)
if 200 <= response.status_code < 300:
result = response.json()
return _handle_api_result(result)
raise NetworkError(f"HTTP request received unexpected "
f"status code: {response.status_code}")
raise NetworkError(
f"HTTP request received unexpected "
f"status code: {response.status_code}"
)
except httpx.InvalidURL:
raise NetworkError("API root url invalid")
except httpx.HTTPError:
@@ -418,11 +431,13 @@ class Bot(BaseBot):
return await super().call_api(api, **data)
@overrides(BaseBot)
async def send(self,
event: Event,
message: Union[str, Message, MessageSegment],
at_sender: bool = False,
**kwargs) -> Any:
async def send(
self,
event: Event,
message: Union[str, Message, MessageSegment],
at_sender: bool = False,
**kwargs,
) -> Any:
"""
:说明:
@@ -445,8 +460,9 @@ class Bot(BaseBot):
- ``NetworkError``: 网络错误
- ``ActionFailed``: API 调用失败
"""
message = escape(message, escape_comma=False) if isinstance(
message, str) else message
message = (
escape(message, escape_comma=False) if isinstance(message, str) else message
)
msg = message if isinstance(message, Message) else Message(message)
at_sender = at_sender and bool(getattr(event, "user_id", None))

1
packages/nonebot-adapter-cqhttp/nonebot/adapters/cqhttp/bot.pyi
View File

@@ -8,7 +8,6 @@ from nonebot.drivers import Driver, WebSocket
from .event import Event
from .message import Message, MessageSegment
def get_auth_bearer(access_token: Optional[str] = ...) -> Optional[str]:
...

9
packages/nonebot-adapter-cqhttp/nonebot/adapters/cqhttp/config.py
View File

@@ -1,6 +1,6 @@
from typing import Dict, Optional
from pydantic import Field, BaseModel, AnyUrl
from pydantic import Field, AnyUrl, BaseModel
# priority: alias > origin
@@ -14,11 +14,10 @@ class Config(BaseModel):
- ``secret`` / ``cqhttp_secret``: CQHTTP HTTP 上报数据签名口令
- ``ws_urls`` / ``cqhttp_ws_urls``: CQHTTP 正向 Websocket 连接 Bot ID、目标 URL 字典
"""
access_token: Optional[str] = Field(default=None,
alias="cqhttp_access_token")
access_token: Optional[str] = Field(default=None, alias="cqhttp_access_token")
secret: Optional[str] = Field(default=None, alias="cqhttp_secret")
ws_urls: Dict[str, AnyUrl] = Field(default_factory=set,
alias="cqhttp_ws_urls")
ws_urls: Dict[str, AnyUrl] = Field(default_factory=set, alias="cqhttp_ws_urls")
class Config:
extra = "ignore"

125
packages/nonebot-adapter-cqhttp/nonebot/adapters/cqhttp/event.py
View File

@@ -5,12 +5,13 @@ from typing import TYPE_CHECKING, List, Type, Optional
from pydantic import BaseModel
from pygtrie import StringTrie
from .message import Message
from nonebot.typing import overrides
from nonebot.utils import escape_tag
from .exception import NoLogException
from nonebot.adapters import Event as BaseEvent
from .message import Message
from .exception import NoLogException
if TYPE_CHECKING:
from .bot import Bot
@@ -22,6 +23,7 @@ class Event(BaseEvent):
.. _CQHTTP 文档:
https://github.com/howmanybots/onebot/blob/master/README.md
"""
__event__ = ""
time: int
self_id: int
@@ -118,6 +120,7 @@ class Status(BaseModel):
# Message Events
class MessageEvent(Event):
"""消息事件"""
__event__ = "message"
post_type: Literal["message"]
sub_type: str
@@ -144,8 +147,9 @@ class MessageEvent(Event):
@overrides(Event)
def get_event_name(self) -> str:
sub_type = getattr(self, "sub_type", None)
return f"{self.post_type}.{self.message_type}" + (f".{sub_type}"
if sub_type else "")
return f"{self.post_type}.{self.message_type}" + (
f".{sub_type}" if sub_type else ""
)
@overrides(Event)
def get_message(self) -> Message:
@@ -170,20 +174,29 @@ class MessageEvent(Event):
class PrivateMessageEvent(MessageEvent):
"""私聊消息"""
__event__ = "message.private"
message_type: Literal["private"]
@overrides(Event)
def get_event_description(self) -> str:
return (f'Message {self.message_id} from {self.user_id} "' + "".join(
map(
lambda x: escape_tag(str(x))
if x.is_text() else f"<le>{escape_tag(str(x))}</le>",
self.message)) + '"')
return (
f'Message {self.message_id} from {self.user_id} "'
+ "".join(
map(
lambda x: escape_tag(str(x))
if x.is_text()
else f"<le>{escape_tag(str(x))}</le>",
self.message,
)
)
+ '"'
)
class GroupMessageEvent(MessageEvent):
"""群消息"""
__event__ = "message.group"
message_type: Literal["group"]
group_id: int
@@ -196,8 +209,13 @@ class GroupMessageEvent(MessageEvent):
+ "".join(
map(
lambda x: escape_tag(str(x))
if x.is_text() else f"<le>{escape_tag(str(x))}</le>",
self.message)) + '"')
if x.is_text()
else f"<le>{escape_tag(str(x))}</le>",
self.message,
)
)
+ '"'
)
@overrides(MessageEvent)
def get_session_id(self) -> str:
@@ -207,6 +225,7 @@ class GroupMessageEvent(MessageEvent):
# Notice Events
class NoticeEvent(Event):
"""通知事件"""
__event__ = "notice"
post_type: Literal["notice"]
notice_type: str
@@ -214,12 +233,14 @@ class NoticeEvent(Event):
@overrides(Event)
def get_event_name(self) -> str:
sub_type = getattr(self, "sub_type", None)
return f"{self.post_type}.{self.notice_type}" + (f".{sub_type}"
if sub_type else "")
return f"{self.post_type}.{self.notice_type}" + (
f".{sub_type}" if sub_type else ""
)
class GroupUploadNoticeEvent(NoticeEvent):
"""群文件上传事件"""
__event__ = "notice.group_upload"
notice_type: Literal["group_upload"]
user_id: int
@@ -237,6 +258,7 @@ class GroupUploadNoticeEvent(NoticeEvent):
class GroupAdminNoticeEvent(NoticeEvent):
"""群管理员变动"""
__event__ = "notice.group_admin"
notice_type: Literal["group_admin"]
sub_type: str
@@ -258,6 +280,7 @@ class GroupAdminNoticeEvent(NoticeEvent):
class GroupDecreaseNoticeEvent(NoticeEvent):
"""群成员减少事件"""
__event__ = "notice.group_decrease"
notice_type: Literal["group_decrease"]
sub_type: str
@@ -280,6 +303,7 @@ class GroupDecreaseNoticeEvent(NoticeEvent):
class GroupIncreaseNoticeEvent(NoticeEvent):
"""群成员增加事件"""
__event__ = "notice.group_increase"
notice_type: Literal["group_increase"]
sub_type: str
@@ -302,6 +326,7 @@ class GroupIncreaseNoticeEvent(NoticeEvent):
class GroupBanNoticeEvent(NoticeEvent):
"""群禁言事件"""
__event__ = "notice.group_ban"
notice_type: Literal["group_ban"]
sub_type: str
@@ -325,6 +350,7 @@ class GroupBanNoticeEvent(NoticeEvent):
class FriendAddNoticeEvent(NoticeEvent):
"""好友添加事件"""
__event__ = "notice.friend_add"
notice_type: Literal["friend_add"]
user_id: int
@@ -340,6 +366,7 @@ class FriendAddNoticeEvent(NoticeEvent):
class GroupRecallNoticeEvent(NoticeEvent):
"""群消息撤回事件"""
__event__ = "notice.group_recall"
notice_type: Literal["group_recall"]
user_id: int
@@ -362,6 +389,7 @@ class GroupRecallNoticeEvent(NoticeEvent):
class FriendRecallNoticeEvent(NoticeEvent):
"""好友消息撤回事件"""
__event__ = "notice.friend_recall"
notice_type: Literal["friend_recall"]
user_id: int
@@ -378,6 +406,7 @@ class FriendRecallNoticeEvent(NoticeEvent):
class NotifyEvent(NoticeEvent):
"""提醒事件"""
__event__ = "notice.notify"
notice_type: Literal["notify"]
sub_type: str
@@ -395,6 +424,7 @@ class NotifyEvent(NoticeEvent):
class PokeNotifyEvent(NotifyEvent):
"""戳一戳提醒事件"""
__event__ = "notice.notify.poke"
sub_type: Literal["poke"]
target_id: int
@@ -413,6 +443,7 @@ class PokeNotifyEvent(NotifyEvent):
class LuckyKingNotifyEvent(NotifyEvent):
"""群红包运气王提醒事件"""
__event__ = "notice.notify.lucky_king"
sub_type: Literal["lucky_king"]
target_id: int
@@ -432,6 +463,7 @@ class LuckyKingNotifyEvent(NotifyEvent):
class HonorNotifyEvent(NotifyEvent):
"""群荣誉变更提醒事件"""
__event__ = "notice.notify.honor"
sub_type: Literal["honor"]
honor_type: str
@@ -444,6 +476,7 @@ class HonorNotifyEvent(NotifyEvent):
# Request Events
class RequestEvent(Event):
"""请求事件"""
__event__ = "request"
post_type: Literal["request"]
request_type: str
@@ -451,12 +484,14 @@ class RequestEvent(Event):
@overrides(Event)
def get_event_name(self) -> str:
sub_type = getattr(self, "sub_type", None)
return f"{self.post_type}.{self.request_type}" + (f".{sub_type}"
if sub_type else "")
return f"{self.post_type}.{self.request_type}" + (
f".{sub_type}" if sub_type else ""
)
class FriendRequestEvent(RequestEvent):
"""加好友请求事件"""
__event__ = "request.friend"
request_type: Literal["friend"]
user_id: int
@@ -472,9 +507,9 @@ class FriendRequestEvent(RequestEvent):
return str(self.user_id)
async def approve(self, bot: "Bot", remark: str = ""):
return await bot.set_friend_add_request(flag=self.flag,
approve=True,
remark=remark)
return await bot.set_friend_add_request(
flag=self.flag, approve=True, remark=remark
)
async def reject(self, bot: "Bot"):
return await bot.set_friend_add_request(flag=self.flag, approve=False)
@@ -482,6 +517,7 @@ class FriendRequestEvent(RequestEvent):
class GroupRequestEvent(RequestEvent):
"""加群请求/邀请事件"""
__event__ = "request.group"
request_type: Literal["group"]
sub_type: str
@@ -499,20 +535,20 @@ class GroupRequestEvent(RequestEvent):
return f"group_{self.group_id}_{self.user_id}"
async def approve(self, bot: "Bot"):
return await bot.set_group_add_request(flag=self.flag,
sub_type=self.sub_type,
approve=True)
return await bot.set_group_add_request(
flag=self.flag, sub_type=self.sub_type, approve=True
)
async def reject(self, bot: "Bot", reason: str = ""):
return await bot.set_group_add_request(flag=self.flag,
sub_type=self.sub_type,
approve=False,
reason=reason)
return await bot.set_group_add_request(
flag=self.flag, sub_type=self.sub_type, approve=False, reason=reason
)
# Meta Events
class MetaEvent(Event):
"""元事件"""
__event__ = "meta_event"
post_type: Literal["meta_event"]
meta_event_type: str
@@ -520,8 +556,9 @@ class MetaEvent(Event):
@overrides(Event)
def get_event_name(self) -> str:
sub_type = getattr(self, "sub_type", None)
return f"{self.post_type}.{self.meta_event_type}" + (f".{sub_type}" if
sub_type else "")
return f"{self.post_type}.{self.meta_event_type}" + (
f".{sub_type}" if sub_type else ""
)
@overrides(Event)
def get_log_string(self) -> str:
@@ -530,6 +567,7 @@ class MetaEvent(Event):
class LifecycleMetaEvent(MetaEvent):
"""生命周期元事件"""
__event__ = "meta_event.lifecycle"
meta_event_type: Literal["lifecycle"]
sub_type: str
@@ -537,6 +575,7 @@ class LifecycleMetaEvent(MetaEvent):
class HeartbeatMetaEvent(MetaEvent):
"""心跳元事件"""
__event__ = "meta_event.heartbeat"
meta_event_type: Literal["heartbeat"]
status: Status
@@ -567,12 +606,28 @@ def get_event_model(event_name) -> List[Type[Event]]:
__all__ = [
"Event", "MessageEvent", "PrivateMessageEvent", "GroupMessageEvent",
"NoticeEvent", "GroupUploadNoticeEvent", "GroupAdminNoticeEvent",
"GroupDecreaseNoticeEvent", "GroupIncreaseNoticeEvent",
"GroupBanNoticeEvent", "FriendAddNoticeEvent", "GroupRecallNoticeEvent",
"FriendRecallNoticeEvent", "NotifyEvent", "PokeNotifyEvent",
"LuckyKingNotifyEvent", "HonorNotifyEvent", "RequestEvent",
"FriendRequestEvent", "GroupRequestEvent", "MetaEvent",
"LifecycleMetaEvent", "HeartbeatMetaEvent", "get_event_model"
"Event",
"MessageEvent",
"PrivateMessageEvent",
"GroupMessageEvent",
"NoticeEvent",
"GroupUploadNoticeEvent",
"GroupAdminNoticeEvent",
"GroupDecreaseNoticeEvent",
"GroupIncreaseNoticeEvent",
"GroupBanNoticeEvent",
"FriendAddNoticeEvent",
"GroupRecallNoticeEvent",
"FriendRecallNoticeEvent",
"NotifyEvent",
"PokeNotifyEvent",
"LuckyKingNotifyEvent",
"HonorNotifyEvent",
"RequestEvent",
"FriendRequestEvent",
"GroupRequestEvent",
"MetaEvent",
"LifecycleMetaEvent",
"HeartbeatMetaEvent",
"get_event_model",
]

8
packages/nonebot-adapter-cqhttp/nonebot/adapters/cqhttp/exception.py
View File

@@ -8,7 +8,6 @@ from nonebot.exception import ApiNotAvailable as BaseApiNotAvailable
class CQHTTPAdapterException(AdapterException):
def __init__(self):
super().__init__("cqhttp")
@@ -33,8 +32,11 @@ class ActionFailed(BaseActionFailed, CQHTTPAdapterException):
self.info = kwargs
def __repr__(self):
return f"<ActionFailed " + ", ".join(
f"{k}={v}" for k, v in self.info.items()) + ">"
return (
f"<ActionFailed "
+ ", ".join(f"{k}={v}" for k, v in self.info.items())
+ ">"
)
def __str__(self):
return self.__repr__()

186
packages/nonebot-adapter-cqhttp/nonebot/adapters/cqhttp/message.py
View File

@@ -5,10 +5,11 @@ from base64 import b64encode
from typing import Any, Type, Tuple, Union, Mapping, Iterable, Optional, cast
from nonebot.typing import overrides
from .utils import log, _b2s, escape, unescape
from nonebot.adapters import Message as BaseMessage
from nonebot.adapters import MessageSegment as BaseMessageSegment
from .utils import log, _b2s, escape, unescape
class MessageSegment(BaseMessageSegment["Message"]):
"""
@@ -27,23 +28,24 @@ class MessageSegment(BaseMessageSegment["Message"]):
# process special types
if type_ == "text":
return escape(
data.get("text", ""), # type: ignore
escape_comma=False)
return escape(data.get("text", ""), escape_comma=False) # type: ignore
params = ",".join(
[f"{k}={escape(str(v))}" for k, v in data.items() if v is not None])
[f"{k}={escape(str(v))}" for k, v in data.items() if v is not None]
)
return f"[CQ:{type_}{',' if params else ''}{params}]"
@overrides(BaseMessageSegment)
def __add__(self, other) -> "Message":
return Message(self) + (MessageSegment.text(other) if isinstance(
other, str) else other)
return Message(self) + (
MessageSegment.text(other) if isinstance(other, str) else other
)
@overrides(BaseMessageSegment)
def __radd__(self, other) -> "Message":
return (MessageSegment.text(other)
if isinstance(other, str) else Message(other)) + self
return (
MessageSegment.text(other) if isinstance(other, str) else Message(other)
) + self
@overrides(BaseMessageSegment)
def is_text(self) -> bool:
@@ -83,11 +85,13 @@ class MessageSegment(BaseMessageSegment["Message"]):
return MessageSegment("forward", {"id": id_})
@staticmethod
def image(file: Union[str, bytes, BytesIO, Path],
type_: Optional[str] = None,
cache: bool = True,
proxy: bool = True,
timeout: Optional[int] = None) -> "MessageSegment":
def image(
file: Union[str, bytes, BytesIO, Path],
type_: Optional[str] = None,
cache: bool = True,
proxy: bool = True,
timeout: Optional[int] = None,
) -> "MessageSegment":
if isinstance(file, BytesIO):
file = file.getvalue()
if isinstance(file, bytes):
@@ -95,74 +99,85 @@ class MessageSegment(BaseMessageSegment["Message"]):
elif isinstance(file, Path):
file = f"file:///{file.resolve()}"
return MessageSegment(
"image", {
"image",
{
"file": file,
"type": type_,
"cache": _b2s(cache),
"proxy": _b2s(proxy),
"timeout": timeout
})
"timeout": timeout,
},
)
@staticmethod
def json(data: str) -> "MessageSegment":
return MessageSegment("json", {"data": data})
@staticmethod
def location(latitude: float,
longitude: float,
title: Optional[str] = None,
content: Optional[str] = None) -> "MessageSegment":
def location(
latitude: float,
longitude: float,
title: Optional[str] = None,
content: Optional[str] = None,
) -> "MessageSegment":
return MessageSegment(
"location", {
"location",
{
"lat": str(latitude),
"lon": str(longitude),
"title": title,
"content": content
})
"content": content,
},
)
@staticmethod
def music(type_: str, id_: int) -> "MessageSegment":
return MessageSegment("music", {"type": type_, "id": id_})
@staticmethod
def music_custom(url: str,
audio: str,
title: str,
content: Optional[str] = None,
img_url: Optional[str] = None) -> "MessageSegment":
def music_custom(
url: str,
audio: str,
title: str,
content: Optional[str] = None,
img_url: Optional[str] = None,
) -> "MessageSegment":
return MessageSegment(
"music", {
"music",
{
"type": "custom",
"url": url,
"audio": audio,
"title": title,
"content": content,
"image": img_url
})
"image": img_url,
},
)
@staticmethod
def node(id_: int) -> "MessageSegment":
return MessageSegment("node", {"id": str(id_)})
@staticmethod
def node_custom(user_id: int, nickname: str,
content: Union[str, "Message"]) -> "MessageSegment":
return MessageSegment("node", {
"user_id": str(user_id),
"nickname": nickname,
"content": content
})
def node_custom(
user_id: int, nickname: str, content: Union[str, "Message"]
) -> "MessageSegment":
return MessageSegment(
"node", {"user_id": str(user_id), "nickname": nickname, "content": content}
)
@staticmethod
def poke(type_: str, id_: str) -> "MessageSegment":
return MessageSegment("poke", {"type": type_, "id": id_})
@staticmethod
def record(file: Union[str, bytes, BytesIO, Path],
magic: Optional[bool] = None,
cache: Optional[bool] = None,
proxy: Optional[bool] = None,
timeout: Optional[int] = None) -> "MessageSegment":
def record(
file: Union[str, bytes, BytesIO, Path],
magic: Optional[bool] = None,
cache: Optional[bool] = None,
proxy: Optional[bool] = None,
timeout: Optional[int] = None,
) -> "MessageSegment":
if isinstance(file, BytesIO):
file = file.getvalue()
if isinstance(file, bytes):
@@ -170,13 +185,15 @@ class MessageSegment(BaseMessageSegment["Message"]):
elif isinstance(file, Path):
file = f"file:///{file.resolve()}"
return MessageSegment(
"record", {
"record",
{
"file": file,
"magic": _b2s(magic),
"cache": _b2s(cache),
"proxy": _b2s(proxy),
"timeout": timeout
})
"timeout": timeout,
},
)
@staticmethod
def reply(id_: int) -> "MessageSegment":
@@ -191,26 +208,27 @@ class MessageSegment(BaseMessageSegment["Message"]):
return MessageSegment("shake", {})
@staticmethod
def share(url: str = "",
title: str = "",
content: Optional[str] = None,
image: Optional[str] = None) -> "MessageSegment":
return MessageSegment("share", {
"url": url,
"title": title,
"content": content,
"image": image
})
def share(
url: str = "",
title: str = "",
content: Optional[str] = None,
image: Optional[str] = None,
) -> "MessageSegment":
return MessageSegment(
"share", {"url": url, "title": title, "content": content, "image": image}
)
@staticmethod
def text(text: str) -> "MessageSegment":
return MessageSegment("text", {"text": text})
@staticmethod
def video(file: Union[str, bytes, BytesIO, Path],
cache: Optional[bool] = None,
proxy: Optional[bool] = None,
timeout: Optional[int] = None) -> "MessageSegment":
def video(
file: Union[str, bytes, BytesIO, Path],
cache: Optional[bool] = None,
proxy: Optional[bool] = None,
timeout: Optional[int] = None,
) -> "MessageSegment":
if isinstance(file, BytesIO):
file = file.getvalue()
if isinstance(file, bytes):
@@ -218,12 +236,14 @@ class MessageSegment(BaseMessageSegment["Message"]):
elif isinstance(file, Path):
file = f"file:///{file.resolve()}"
return MessageSegment(
"video", {
"video",
{
"file": file,
"cache": _b2s(cache),
"proxy": _b2s(proxy),
"timeout": timeout
})
"timeout": timeout,
},
)
@staticmethod
def xml(data: str) -> "MessageSegment":
@@ -241,22 +261,22 @@ class Message(BaseMessage[MessageSegment]):
return MessageSegment
@overrides(BaseMessage)
def __add__(self, other: Union[str, Mapping,
Iterable[Mapping]]) -> "Message":
def __add__(self, other: Union[str, Mapping, Iterable[Mapping]]) -> "Message":
return super(Message, self).__add__(
MessageSegment.text(other) if isinstance(other, str) else other)
MessageSegment.text(other) if isinstance(other, str) else other
)
@overrides(BaseMessage)
def __radd__(self, other: Union[str, Mapping,
Iterable[Mapping]]) -> "Message":
def __radd__(self, other: Union[str, Mapping, Iterable[Mapping]]) -> "Message":
return super(Message, self).__radd__(
MessageSegment.text(other) if isinstance(other, str) else other)
MessageSegment.text(other) if isinstance(other, str) else other
)
@staticmethod
@overrides(BaseMessage)
def _construct(
msg: Union[str, Mapping,
Iterable[Mapping]]) -> Iterable[MessageSegment]:
msg: Union[str, Mapping, Iterable[Mapping]]
) -> Iterable[MessageSegment]:
if isinstance(msg, Mapping):
msg = cast(Mapping[str, Any], msg)
yield MessageSegment(msg["type"], msg.get("data") or {})
@@ -270,14 +290,15 @@ class Message(BaseMessage[MessageSegment]):
def _iter_message(msg: str) -> Iterable[Tuple[str, str]]:
text_begin = 0
for cqcode in re.finditer(
r"\[CQ:(?P<type>[a-zA-Z0-9-_.]+)"
r"(?P<params>"
r"(?:,[a-zA-Z0-9-_.]+=[^,\]]+)*"
r"),?\]", msg):
yield "text", msg[text_begin:cqcode.pos + cqcode.start()]
r"\[CQ:(?P<type>[a-zA-Z0-9-_.]+)"
r"(?P<params>"
r"(?:,[a-zA-Z0-9-_.]+=[^,\]]+)*"
r"),?\]",
msg,
):
yield "text", msg[text_begin : cqcode.pos + cqcode.start()]
text_begin = cqcode.pos + cqcode.end()
yield cqcode.group("type"), cqcode.group("params").lstrip(
",")
yield cqcode.group("type"), cqcode.group("params").lstrip(",")
yield "text", msg[text_begin:]
for type_, data in _iter_message(msg):
@@ -287,10 +308,11 @@ class Message(BaseMessage[MessageSegment]):
yield MessageSegment(type_, {"text": unescape(data)})
else:
data = {
k: unescape(v) for k, v in map(
k: unescape(v)
for k, v in map(
lambda x: x.split("=", maxsplit=1),
filter(lambda x: x, (
x.lstrip() for x in data.split(","))))
filter(lambda x: x, (x.lstrip() for x in data.split(","))),
)
}
yield MessageSegment(type_, data)

14
packages/nonebot-adapter-cqhttp/nonebot/adapters/cqhttp/permission.py
View File

@@ -1,5 +1,6 @@
from nonebot.adapters import Event
from nonebot.permission import Permission
from .event import GroupMessageEvent, PrivateMessageEvent
@@ -42,8 +43,7 @@ async def _group(event: Event) -> bool:
async def _group_member(event: Event) -> bool:
return isinstance(event,
GroupMessageEvent) and event.sender.role == "member"
return isinstance(event, GroupMessageEvent) and event.sender.role == "member"
async def _group_admin(event: Event) -> bool:
@@ -76,6 +76,12 @@ GROUP_OWNER = Permission(_group_owner)
"""
__all__ = [
"PRIVATE", "PRIVATE_FRIEND", "PRIVATE_GROUP", "PRIVATE_OTHER", "GROUP",
"GROUP_MEMBER", "GROUP_ADMIN", "GROUP_OWNER"
"PRIVATE",
"PRIVATE_FRIEND",
"PRIVATE_GROUP",
"PRIVATE_OTHER",
"GROUP",
"GROUP_MEMBER",
"GROUP_ADMIN",
"GROUP_OWNER",
]

12
packages/nonebot-adapter-cqhttp/nonebot/adapters/cqhttp/utils.py
View File

@@ -16,9 +16,7 @@ def escape(s: str, *, escape_comma: bool = True) -> str:
* ``s: str``: 需要转义的字符串
* ``escape_comma: bool``: 是否转义逗号（``,``）。
"""
s = s.replace("&", "&amp;") \
.replace("[", "&#91;") \
.replace("]", "&#93;")
s = s.replace("&", "&amp;").replace("[", "&#91;").replace("]", "&#93;")
if escape_comma:
s = s.replace(",", "&#44;")
return s
@@ -34,10 +32,12 @@ def unescape(s: str) -> str:
* ``s: str``: 需要转义的字符串
"""
return s.replace("&#44;", ",") \
.replace("&#91;", "[") \
.replace("&#93;", "]") \
return (
s.replace("&#44;", ",")
.replace("&#91;", "[")
.replace("&#93;", "]")
.replace("&amp;", "&")
)
def _b2s(b: Optional[bool]) -> Optional[str]: