nonebot/nonebot2
1
1
Fork 0
mirror of https://github.com/nonebot/nonebot2.git synced 2024-09-21 05:12:34 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

✨ cqhttp message segment support more types

Browse Source
This commit is contained in:
yanyongyu 2021-05-04 11:08:31 +08:00
parent b87971247a
commit 4b38afdcd7
1 changed files with 24 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,4 +1,7 @@
import re import re
from io import BytesIO
from pathlib import Path
from base64 import b64encode
from functools import reduce from functools import reduce
from typing import Any, Dict, Union, Tuple, Mapping, Iterable, Optional from typing import Any, Dict, Union, Tuple, Mapping, Iterable, Optional
@ -79,11 +82,17 @@ class MessageSegment(BaseMessageSegment):
return MessageSegment("forward", {"id": id_}) return MessageSegment("forward", {"id": id_})
@staticmethod @staticmethod
def image(file: str, def image(file: Union[str, bytes, BytesIO, Path],
type_: Optional[str] = None, type_: Optional[str] = None,
cache: bool = True, cache: bool = True,
proxy: bool = True, proxy: bool = True,
timeout: Optional[int] = None) -> "MessageSegment": timeout: Optional[int] = None) -> "MessageSegment":
if isinstance(file, BytesIO):
file = file.read()
if isinstance(file, bytes):
file = f"base64://{b64encode(file).decode()}"
elif isinstance(file, Path):
file = f"file:///{file.resolve()}"
return MessageSegment( return MessageSegment(
"image", { "image", {
"file": file, "file": file,
@ -148,11 +157,17 @@ class MessageSegment(BaseMessageSegment):
return MessageSegment("poke", {"type": type_, "id": id_}) return MessageSegment("poke", {"type": type_, "id": id_})
@staticmethod @staticmethod
def record(file: str, def record(file: Union[str, bytes, BytesIO, Path],
magic: Optional[bool] = None, magic: Optional[bool] = None,
cache: Optional[bool] = None, cache: Optional[bool] = None,
proxy: Optional[bool] = None, proxy: Optional[bool] = None,
timeout: Optional[int] = None) -> "MessageSegment": timeout: Optional[int] = None) -> "MessageSegment":
if isinstance(file, BytesIO):
file = file.read()
if isinstance(file, bytes):
file = f"base64://{b64encode(file).decode()}"
elif isinstance(file, Path):
file = f"file:///{file.resolve()}"
return MessageSegment( return MessageSegment(
"record", { "record", {
"file": file, "file": file,
@ -191,10 +206,16 @@ class MessageSegment(BaseMessageSegment):
return MessageSegment("text", {"text": text}) return MessageSegment("text", {"text": text})
@staticmethod @staticmethod
def video(file: str, def video(file: Union[str, bytes, BytesIO, Path],
cache: Optional[bool] = None, cache: Optional[bool] = None,
proxy: Optional[bool] = None, proxy: Optional[bool] = None,
timeout: Optional[int] = None) -> "MessageSegment": timeout: Optional[int] = None) -> "MessageSegment":
if isinstance(file, BytesIO):
file = file.read()
if isinstance(file, bytes):
file = f"base64://{b64encode(file).decode()}"
elif isinstance(file, Path):
file = f"file:///{file.resolve()}"
return MessageSegment( return MessageSegment(
"video", { "video", {
"file": file, "file": file,