CQ码支持以及视频解析

This commit is contained in:
2026-01-05 21:36:03 +08:00
parent 2aa438bc1e
commit 723f7b9724
2 changed files with 362 additions and 32 deletions

View File

@@ -42,6 +42,40 @@ class MessageSegment:
"""
return self.data.get("url", "") if self.type == "image" else ""
@property
def share_url(self) -> str:
"""
当消息段类型为 'share' 时,快速获取其分享 URL。
Returns:
str: 分享的 URL。如果类型不是 'share' 或数据中不含 'url',则返回空字符串。
"""
return self.data.get("url", "") if self.type == "share" else ""
@property
def music_url(self) -> str:
"""
当消息段类型为 'music' 且为 'custom' 类型时,快速获取其 URL。
Returns:
str: 音乐的 URL。如果类型不匹配则返回空字符串。
"""
if self.type == "music" and self.data.get("type") == "custom":
return self.data.get("url", "")
return ""
@property
def file_url(self) -> str:
"""
当消息段类型为 'record', 'video', 'file' 时,快速获取其文件 URL。
Returns:
str: 文件的 URL 或路径。如果类型不匹配,则返回空字符串。
"""
if self.type in ("record", "video", "file"):
return self.data.get("file", "")
return ""
def is_at(self, user_id: int = None) -> bool:
"""
检查当前消息段是否是一个 'at' (提及) 消息段。
@@ -81,30 +115,46 @@ class MessageSegment:
return MessageSegment(type="text", data={"text": text})
@staticmethod
def at(user_id: int | str) -> "MessageSegment":
def at(user_id: int | str, name: str = None) -> "MessageSegment":
"""
创建一个 @某人 的消息段。
Args:
user_id (int | str): 要提及的 QQ 号。若为 "all",则表示 @全体成员。
name (str, optional): 当在群中找不到对应的QQ号时显示的名称。Defaults to None.
Returns:
MessageSegment: 一个类型为 'at' 的消息段对象。
"""
return MessageSegment(type="at", data={"qq": str(user_id)})
data = {"qq": str(user_id)}
if name:
data["name"] = name
return MessageSegment(type="at", data=data)
@staticmethod
def image(file: str) -> "MessageSegment":
def image(file: str, image_type: str = None, cache: bool = True, proxy: bool = True, timeout: int = None, sub_type: int = None) -> "MessageSegment":
"""
创建一个图片消息段。
Args:
file (str): 图片的路径、URL 或 Base64 编码的字符串。
image_type (str, optional): 图片类型,'flash' 表示闪照。Defaults to None.
cache (bool, optional): 是否使用缓存。Defaults to True.
proxy (bool, optional): 是否通过代理下载。Defaults to True.
timeout (int, optional): 下载超时时间。Defaults to None.
sub_type (int, optional): 图片子类型用于特殊图片。Defaults to None.
Returns:
MessageSegment: 一个类型为 'image' 的消息段对象。
"""
return MessageSegment(type="image", data={"file": file})
data = {"file": file, "cache": "1" if cache else "0", "proxy": "1" if proxy else "0"}
if image_type:
data["type"] = image_type
if timeout:
data["timeout"] = str(timeout)
if sub_type:
data["subType"] = str(sub_type)
return MessageSegment(type="image", data=data)
@staticmethod
def face(id: int) -> "MessageSegment":
@@ -119,3 +169,220 @@ class MessageSegment:
"""
return MessageSegment(type="face", data={"id": str(id)})
@staticmethod
def json(data: str) -> "MessageSegment":
"""
创建一个 JSON 消息段。
Args:
data (str): JSON 字符串。
Returns:
MessageSegment: 一个类型为 'json' 的消息段对象。
"""
return MessageSegment(type="json", data={"data": data})
@staticmethod
def xml(data: str) -> "MessageSegment":
"""
创建一个 XML 消息段。
Args:
data (str): XML 字符串。
Returns:
MessageSegment: 一个类型为 'xml' 的消息段对象。
"""
return MessageSegment(type="xml", data={"data": data})
@staticmethod
def share(url: str, title: str, content: str = None, image: str = None) -> "MessageSegment":
"""
创建一个分享消息段。
Args:
url (str): 分享的 URL。
title (str): 分享的标题。
content (str, optional): 分享的描述内容。Defaults to None.
image (str, optional): 分享的图片 URL。Defaults to None.
Returns:
MessageSegment: 一个类型为 'share' 的消息段对象。
"""
data = {"url": url, "title": title}
if content:
data["content"] = content
if image:
data["image"] = image
return MessageSegment(type="share", data=data)
@staticmethod
def music(type: str, id: str) -> "MessageSegment":
"""
创建一个音乐消息段。
Args:
type (str): 音乐平台类型,如 "qq""xiami" 等。
id (str): 音乐在平台上的唯一标识符。
Returns:
MessageSegment: 一个类型为 'music' 的消息段对象。
"""
return MessageSegment(type="music", data={"type": type, "id": id})
@staticmethod
def music_custom(url: str, audio: str, title: str, content: str = None, image: str = None) -> "MessageSegment":
"""
创建一个自定义音乐消息段。
Args:
url (str): 音乐的 URL。
audio (str): 音乐的音频 URL。
title (str): 音乐的标题。
content (str, optional): 音乐的描述内容。Defaults to None.
image (str, optional): 音乐的图片 URL。Defaults to None.
Returns:
MessageSegment: 一个类型为 'music_custom' 的消息段对象。
"""
data = {"url": url, "audio": audio, "title": title}
if content:
data["content"] = content
if image:
data["image"] = image
return MessageSegment(type="music", data={"type": "custom", **data})
@staticmethod
def record(file: str, magic: bool = False, cache: bool = True, proxy: bool = True, timeout: int = None) -> "MessageSegment":
"""
创建一个语音消息段。
Args:
file (str): 语音的路径、URL 或 Base64 编码的字符串。
magic (bool, optional): 是否为变声。Defaults to False.
cache (bool, optional): 是否使用缓存。Defaults to True.
proxy (bool, optional): 是否通过代理下载。Defaults to True.
timeout (int, optional): 下载超时时间。Defaults to None.
Returns:
MessageSegment: 一个类型为 'record' 的消息段对象。
"""
data = {"file": file, "magic": "1" if magic else "0", "cache": "1" if cache else "0", "proxy": "1" if proxy else "0"}
if timeout:
data["timeout"] = str(timeout)
return MessageSegment(type="record", data=data)
@staticmethod
def video(file: str, cover: str = None, c: int = 2) -> "MessageSegment":
"""
创建一个视频消息段。
Args:
file (str): 视频的路径、URL 或 Base64 编码的字符串。
cover (str, optional): 视频封面支持http, file和base64。Defaults to None.
c (int, optional): 下载线程数。Defaults to 2.
Returns:
MessageSegment: 一个类型为 'video' 的消息段对象。
"""
data = {"file": file, "c": str(c)}
if cover:
data["cover"] = cover
return MessageSegment(type="video", data=data)
@staticmethod
def file(file: str) -> "MessageSegment":
"""
创建一个文件消息段。
Args:
file (str): 文件的路径、URL 或 Base64 编码的字符串。
Returns:
MessageSegment: 一个类型为 'file' 的消息段对象。
"""
return MessageSegment(type="file", data={"file": file})
@staticmethod
def reply(message_id: str) -> "MessageSegment":
"""
创建一个回复消息段。
Args:
message_id (str): 被回复的消息 ID。
Returns:
MessageSegment: 一个类型为 'reply' 的消息段对象。
"""
return MessageSegment(type="reply", data={"id": message_id})
@staticmethod
def rps() -> "MessageSegment":
"""
创建一个猜拳魔法表情消息段。
Returns:
MessageSegment: 一个类型为 'rps' 的消息段对象。
"""
return MessageSegment(type="rps", data={})
@staticmethod
def dice() -> "MessageSegment":
"""
创建一个掷骰子魔法表情消息段。
Returns:
MessageSegment: 一个类型为 'dice' 的消息段对象。
"""
return MessageSegment(type="dice", data={})
@staticmethod
def shake() -> "MessageSegment":
"""
创建一个戳一戳消息段。
Returns:
MessageSegment: 一个类型为 'shake' 的消息段对象。
"""
return MessageSegment(type="shake", data={})
@staticmethod
def anonymous(ignore: bool = False) -> "MessageSegment":
"""
创建一个匿名消息段。
Args:
ignore (bool, optional): 发送失败时是否忽略。Defaults to False.
Returns:
MessageSegment: 一个类型为 'anonymous' 的消息段对象。
"""
return MessageSegment(type="anonymous", data={"ignore": "1" if ignore else "0"})
@staticmethod
def contact(contact_type: str, contact_id: int) -> "MessageSegment":
"""
创建一个推荐好友/群消息段。
Args:
contact_type (str): 推荐类型,'qq''group'
contact_id (int): 被推荐的 QQ 号或群号。
Returns:
MessageSegment: 一个类型为 'contact' 的消息段对象。
"""
return MessageSegment(type="contact", data={"type": contact_type, "id": str(contact_id)})
@staticmethod
def location(lat: float, lon: float, title: str = "", content: str = "") -> "MessageSegment":
"""
创建一个位置消息段。
Args:
lat (float): 纬度。
lon (float): 经度。
title (str, optional): 标题。Defaults to "".
content (str, optional): 内容描述。Defaults to "".
Returns:
MessageSegment: 一个类型为 'location' 的消息段对象。
"""
data = {"lat": str(lat), "lon": str(lon)}
if title:
data["title"] = title
if content:
data["content"] = content
return MessageSegment(type="location", data=data)