Lib.utils.QQRichText
QQ富文本
1""" 2QQ富文本 3""" 4import inspect 5import json 6import traceback 7from typing import Any 8from urllib.parse import urlparse 9 10from Lib.constants import * 11from Lib.common import save_exc_dump 12from Lib.core import ConfigManager 13from Lib.utils import QQDataCacher, Logger 14 15 16def cq_decode(text, in_cq: bool = False) -> str: 17 """ 18 CQ解码 19 Args: 20 text: 文本(CQ) 21 in_cq: 该文本是否是在CQ内的 22 Returns: 23 解码后的文本 24 """ 25 text = str(text) 26 if in_cq: 27 return text.replace("&", "&").replace("[", "["). \ 28 replace("]", "]").replace(",", ",") 29 else: 30 return text.replace("&", "&").replace("[", "["). \ 31 replace("]", "]") 32 33 34def cq_encode(text, in_cq: bool = False) -> str: 35 """ 36 CQ编码 37 Args: 38 text: 文本 39 in_cq: 该文本是否是在CQ内的 40 Returns: 41 编码后的文本 42 """ 43 text = str(text) 44 if in_cq: 45 return text.replace("&", "&").replace("[", "["). \ 46 replace("]", "]").replace(",", ",") 47 else: 48 return text.replace("&", "&").replace("[", "["). \ 49 replace("]", "]") 50 51 52def cq_2_array(cq: str) -> list[dict[str, dict[str, Any]]]: 53 """ 54 将CQCode格式的字符串转换为消息段数组。 55 56 Args: 57 cq (str): CQCode字符串。 58 59 Returns: 60 list[dict[str, dict[str, Any]]]: 解析后的消息段数组。 61 """ 62 if not isinstance(cq, str): 63 raise TypeError("cq_2_array: 输入类型错误") 64 65 cq_array = [] # 存储解析后的消息段 66 now_state = 0 # 当前解析状态 67 # 0: 不在CQ码内 68 # 1: 在CQ码的类型部分 69 # 2: 在CQ码的参数键部分 70 # 3: 在CQ码的参数值部分 71 72 segment_data = {"text": ""} # 用于存储当前解析的消息段 73 now_key = "" # 当前解析的参数键 74 now_segment_type = "" # 当前解析的CQ码类型 75 76 for c in cq: 77 if now_state == 0: # 解析普通文本 78 if c == "[": # 进入CQ码解析 79 now_state = 1 80 if len(segment_data["text"]): # 先存储之前的普通文本 81 cq_array.append({"type": "text", "data": {"text": cq_decode(segment_data["text"])}}) 82 segment_data = {} # 重置segment_data用于存储新的CQ码 83 else: 84 segment_data["text"] += c # 继续拼接普通文本 85 86 elif now_state == 1: # 解析CQ码类型 87 if c == ",": # 类型解析结束,进入参数键解析 88 now_state = 2 89 now_segment_type = now_segment_type[3:] # 移除CQ:前缀 90 else: 91 now_segment_type += c # 继续拼接类型字符串 92 93 elif now_state == 2: # 解析参数键 94 if c == "=": # 键名解析结束,进入值解析 95 now_state = 3 96 now_key = cq_decode(now_key, in_cq=True) # 解码键名 97 if now_key not in segment_data: 98 segment_data[now_key] = "" # 初始化键值 99 else: 100 raise ValueError("cq_2_array: CQ码键名称重复") 101 else: 102 now_key += c # 继续拼接键名 103 104 elif now_state == 3: # 解析参数值 105 if c == "]": # CQ码结束 106 now_state = 0 107 segment_data[now_key] = cq_decode(segment_data[now_key], in_cq=True) # 解码值 108 cq_array.append({"type": now_segment_type, "data": segment_data}) # 存入解析结果 109 segment_data = {"text": ""} # 重置segment_data 110 now_segment_type = "" # 清空类型 111 now_key = "" # 清空键名 112 elif c == ",": # 进入下一个参数键解析 113 segment_data[now_key] = cq_decode(segment_data[now_key], in_cq=True) # 解码值 114 now_state = 2 115 now_key = "" # 清空键名,准备解析下一个键 116 else: 117 segment_data[now_key] += c # 继续拼接参数值 118 119 if "text" in segment_data and len(segment_data["text"]): # 处理末尾可能存在的文本内容 120 cq_array.append({"type": "text", "data": {"text": cq_decode(segment_data["text"])}}) 121 122 return cq_array 123 124 125def array_2_cq(cq_array: list | dict) -> str: 126 """ 127 array消息段转CQCode 128 Args: 129 cq_array: array消息段数组 130 Returns: 131 CQCode 132 """ 133 # 特判 134 if isinstance(cq_array, dict): 135 cq_array = [cq_array] 136 137 if not isinstance(cq_array, (list, tuple)): 138 raise TypeError("array_2_cq: 输入类型错误") 139 140 # 将json形式的富文本转换为CQ码 141 text = "" 142 for segment in cq_array: 143 # 纯文本 144 if segment.get("type") == "text": 145 text += cq_encode(segment.get("data").get("text")) 146 # CQ码 147 else: 148 if segment.get("data"): # 特判 149 text += f"[CQ:{segment.get('type')}," + ",".join( 150 [cq_encode(x, in_cq=True) + "=" + cq_encode(segment.get("data")[x], in_cq=True) 151 for x in segment.get("data").keys()]) + "]" 152 else: 153 text += f"[CQ:{segment.get('type')}]" 154 return text 155 156 157def convert_to_fileurl(input_str: str) -> str: 158 """ 159 自动将输入的路径转换成fileurl 160 Args: 161 input_str: 输入的路径 162 163 Returns: 164 转换后的 fileurl 165 """ 166 # 检查是否已经是 file:// 格式 167 if input_str.startswith("file://"): 168 return input_str 169 170 # 检查输入是否是有效的 URL 171 parsed_url = urlparse(input_str) 172 if parsed_url.scheme in ['http', 'https', 'ftp', 'file', 'data']: 173 return input_str # 已经是 URL 格式,直接返回 174 175 # 检查输入是否是有效的本地文件路径 176 if os.path.isfile(input_str): 177 # 转换为 file:// 格式 178 return f"file://{os.path.abspath(input_str)}" 179 180 # 如果是相对路径或其他文件类型,则尝试转换 181 if os.path.exists(input_str): 182 return f"file://{os.path.abspath(input_str)}" 183 184 raise ValueError("输入的路径无效,无法转换为 fileurl 格式") 185 186 187segments = [] 188segments_map = {} 189 190 191class SegmentMeta(type): 192 """ 193 元类用于自动注册 Segment 子类到全局列表 segments 和映射 segments_map 中。 194 """ 195 196 def __init__(cls, name, bases, dct): 197 super().__init__(name, bases, dct) 198 if 'Segment' in globals() and issubclass(cls, Segment): 199 segments.append(cls) # 将子类添加到全局列表中 200 segments_map[cls.segment_type] = cls 201 202 203class Segment(metaclass=SegmentMeta): 204 """ 205 消息段 206 """ 207 segment_type = None 208 209 def __init__(self, cq): 210 self.cq = cq 211 if isinstance(cq, str): 212 self.array = cq_2_array(cq)[0] 213 self.type, self.data = list(self.array.values()) 214 elif isinstance(cq, dict): 215 self.array = cq 216 self.cq = array_2_cq(self.array) 217 self.type, self.data = list(self.array.values()) 218 else: 219 for segment in segments: 220 if isinstance(cq, segment): 221 self.array = cq.array 222 self.cq = str(self.cq) 223 # print(self.array.values(), list(self.array.values())) 224 self.type, self.data = list(self.array.values()) 225 break 226 else: 227 # print(cq, str(cq), type(cq)) 228 raise TypeError("Segment: 输入类型错误") 229 230 def __str__(self): 231 return self.__repr__() 232 233 def __repr__(self): 234 self.cq = array_2_cq(self.array) 235 return self.cq 236 237 def __setitem__(self, key, value): 238 self.array[key] = value 239 240 def __getitem__(self, key): 241 return self.array.get(key) 242 243 def get(self, key, default=None): 244 """ 245 获取消息段中的数据 246 Args: 247 key: key 248 default: 默认值(默认为None) 249 250 Returns: 251 获取到的数据 252 """ 253 return self.array.get(key, default) 254 255 def __delitem__(self, key): 256 del self.array[key] 257 258 def __eq__(self, other): 259 other = Segment(other) 260 return self.array == other.array 261 262 def __contains__(self, other): 263 if isinstance(other, Segment): 264 return all(item in self.array for item in other.array) 265 else: 266 try: 267 return str(other) in str(self) 268 except (TypeError, AttributeError): 269 return False 270 271 def render(self, group_id: int | None = None): 272 """ 273 渲染消息段为字符串 274 Args: 275 group_id: 群号(选填) 276 Returns: 277 渲染完毕的消息段 278 """ 279 return f"[{self.array.get('type', 'unknown')}: {self.cq}]" 280 281 def set_data(self, k, v): 282 """ 283 设置消息段的Data项 284 Args: 285 k: 要修改的key 286 v: 要修改成的value 287 """ 288 self.array["data"][k] = v 289 290 291segments.append(Segment) 292 293 294class Text(Segment): 295 """ 296 文本消息段 297 """ 298 segment_type = "text" 299 300 def __init__(self, text): 301 """ 302 Args: 303 text: 文本 304 """ 305 self.text = text 306 super().__init__({"type": "text", "data": {"text": text}}) 307 308 def __add__(self, other): 309 other = Text(other) 310 return self.text + other.text 311 312 def __eq__(self, other): 313 other = Text(other) 314 return self.text == other.text 315 316 def __contains__(self, other): 317 if isinstance(other, Text): 318 return other.text in self.text 319 else: 320 try: 321 return str(other) in str(self.text) 322 except (TypeError, AttributeError): 323 return False 324 325 def set_text(self, text): 326 """ 327 设置文本 328 Args: 329 text: 文本 330 """ 331 self.text = text 332 self["data"]["text"] = text 333 334 def render(self, group_id: int | None = None): 335 return self.text 336 337 338class Face(Segment): 339 """ 340 表情消息段 341 """ 342 segment_type = "face" 343 344 def __init__(self, face_id): 345 """ 346 Args: 347 face_id: 表情id 348 """ 349 self.face_id = face_id 350 super().__init__({"type": "face", "data": {"id": str(face_id)}}) 351 352 def set_id(self, face_id): 353 """ 354 设置表情id 355 Args: 356 face_id: 表情id 357 """ 358 self.face_id = face_id 359 self.array["data"]["id"] = str(face_id) 360 361 def render(self, group_id: int | None = None): 362 return "[表情: %s]" % self.face_id 363 364 365class At(Segment): 366 """ 367 At消息段 368 """ 369 segment_type = "at" 370 371 def __init__(self, qq): 372 """ 373 Args: 374 qq: qq号 375 """ 376 self.qq = qq 377 super().__init__({"type": "at", "data": {"qq": str(qq)}}) 378 379 def set_id(self, qq_id): 380 """ 381 设置At的id 382 Args: 383 qq_id: qq号 384 """ 385 self.qq = qq_id 386 self.array["data"]["qq"] = str(qq_id) 387 388 def render(self, group_id: int | None = None): 389 if group_id: 390 if str(self.qq) == "all" or str(self.qq) == "0": 391 return f"@全体成员" 392 return f"@{QQDataCacher.get_group_member_info(group_id, self.qq).get_nickname()}: {self.qq}" 393 else: 394 return f"@{QQDataCacher.get_user_info(self.qq).nickname}: {self.qq}" 395 396 397class Image(Segment): 398 """ 399 图片消息段 400 """ 401 segment_type = "image" 402 403 def __init__(self, file: str): 404 """ 405 Args: 406 file: 图片文件(url,对于文件使用file url格式) 407 """ 408 file = convert_to_fileurl(file) 409 self.file = file 410 super().__init__({"type": "image", "data": {"file": str(file)}}) 411 412 def set_file(self, file: str): 413 """ 414 设置图片文件 415 Args: 416 file: 图片文件 417 """ 418 file = convert_to_fileurl(file) 419 self.file = file 420 self.array["data"]["file"] = str(file) 421 422 def render(self, group_id: int | None = None): 423 return "[图片: %s]" % self.file 424 425 426class Record(Segment): 427 """ 428 语音消息段 429 """ 430 segment_type = "record" 431 432 def __init__(self, file: str): 433 """ 434 Args: 435 file: 语音文件(url,对于文件使用file url格式) 436 """ 437 file = convert_to_fileurl(file) 438 self.file = file 439 super().__init__({"type": "record", "data": {"file": str(file)}}) 440 441 def set_file(self, file: str): 442 """ 443 设置语音文件 444 Args: 445 file: 语音文件(url,对于文件使用file url格式) 446 """ 447 file = convert_to_fileurl(file) 448 self.file = file 449 self.array["data"]["file"] = str(file) 450 451 def render(self, group_id: int | None = None): 452 return "[语音: %s]" % self.file 453 454 455class Video(Segment): 456 """ 457 视频消息段 458 """ 459 segment_type = "video" 460 461 def __init__(self, file: str): 462 """ 463 Args: 464 file: 视频文件(url,对于文件使用file url格式) 465 """ 466 file = convert_to_fileurl(file) 467 self.file = file 468 super().__init__({"type": "video", "data": {"file": str(file)}}) 469 470 def set_file(self, file: str): 471 """ 472 设置视频文件 473 Args: 474 file: 视频文件(url,对于文件使用file url格式) 475 """ 476 file = convert_to_fileurl(file) 477 self.file = file 478 self.array["data"]["file"] = str(file) 479 480 def render(self, group_id: int | None = None): 481 return f"[视频: {self.file}]" 482 483 484class Rps(Segment): 485 """ 486 猜拳消息段 487 """ 488 segment_type = "rps" 489 490 def __init__(self): 491 super().__init__({"type": "rps"}) 492 493 494class Dice(Segment): 495 segment_type = "dice" 496 497 def __init__(self): 498 super().__init__({"type": "dice"}) 499 500 501class Shake(Segment): 502 """ 503 窗口抖动消息段 504 (相当于戳一戳最基本类型的快捷方式。) 505 """ 506 segment_type = "shake" 507 508 def __init__(self): 509 super().__init__({"type": "shake"}) 510 511 512class Poke(Segment): 513 """ 514 戳一戳消息段 515 """ 516 segment_type = "poke" 517 518 def __init__(self, type_, poke_id): 519 """ 520 Args: 521 type_: 见https://github.com/botuniverse/onebot-11/blob/master/message/segment.md#%E6%88%B3%E4%B8%80%E6%88%B3 522 poke_id: 同上 523 """ 524 self.type = type_ 525 self.poke_id = poke_id 526 super().__init__({"type": "poke", "data": {"type": str(self.type)}, "id": str(self.poke_id)}) 527 528 def set_type(self, qq_type): 529 """ 530 设置戳一戳类型 531 Args: 532 qq_type: qq类型 533 """ 534 self.type = qq_type 535 self.array["data"]["type"] = str(qq_type) 536 537 def set_id(self, poke_id): 538 """ 539 设置戳一戳id 540 Args: 541 poke_id: 戳一戳id 542 """ 543 self.poke_id = poke_id 544 self.array["data"]["id"] = str(poke_id) 545 546 def render(self, group_id: int | None = None): 547 return f"[戳一戳: {self.type}]" 548 549 550class Anonymous(Segment): 551 """ 552 匿名消息段 553 """ 554 segment_type = "anonymous" 555 556 def __init__(self, ignore=False): 557 """ 558 Args: 559 ignore: 是否忽略 560 """ 561 self.ignore = 0 if ignore else 1 562 super().__init__({"type": "anonymous", "data": {"ignore": str(self.ignore)}}) 563 564 def set_ignore(self, ignore): 565 """ 566 设置是否忽略 567 Args: 568 ignore: 是否忽略 569 """ 570 self.ignore = 0 if ignore else 1 571 self.array["data"]["ignore"] = str(self.ignore) 572 573 574class Share(Segment): 575 """ 576 链接分享消息段 577 """ 578 segment_type = "share" 579 580 def __init__(self, url, title, content="", image=""): 581 """ 582 Args: 583 url: URL 584 title: 标题 585 content: 发送时可选,内容描述 586 image: 发送时可选,图片 URL 587 """ 588 self.url = url 589 self.title = title 590 self.content = content 591 self.image = image 592 super().__init__({"type": "share", "data": {"url": str(self.url), "title": str(self.title)}}) 593 594 if content != "": 595 self.array["data"]["content"] = str(self.content) 596 597 if image != "": 598 self.array["data"]["image"] = str(self.image) 599 600 def set_url(self, url): 601 """ 602 设置URL 603 Args: 604 url: URL 605 """ 606 self.array["data"]["url"] = str(url) 607 self.url = url 608 609 def set_title(self, title): 610 """ 611 设置标题 612 Args: 613 title: 标题 614 """ 615 self.title = title 616 self.array["data"]["title"] = str(title) 617 618 def set_content(self, content): 619 """ 620 设置内容描述 621 Args: 622 content: 内容描述 623 """ 624 self.content = content 625 self.array["data"]["content"] = str(content) 626 627 def set_image(self, image): 628 """ 629 设置图片 URL 630 Args: 631 image: 图片 URL 632 """ 633 self.image = image 634 self.array["data"]["image"] = str(image) 635 636 637class Contact(Segment): 638 """ 639 推荐好友/推荐群 640 """ 641 segment_type = "contact" 642 643 def __init__(self, type_, id_): 644 """ 645 Args: 646 type_: 推荐的类型(friend/group) 647 id_: 推荐的qqid 648 """ 649 self.type = type_ 650 self.id = id_ 651 super().__init__({"type": "contact", "data": {"type": str(self.type), "id": str(self.id)}}) 652 653 def set_type(self, type_): 654 """ 655 设置推荐类型 656 Args: 657 type_: 推荐的类型(friend/group) 658 """ 659 self.type = type_ 660 self.array["data"]["type"] = str(type_) 661 662 def set_id(self, id_): 663 """ 664 设置推荐的qqid 665 Args: 666 id_: qqid 667 """ 668 self.id = id_ 669 self.array["data"]["id"] = str(id_) 670 671 672class Location(Segment): 673 """ 674 位置消息段 675 """ 676 segment_type = "location" 677 678 def __init__(self, lat, lon, title="", content=""): 679 """ 680 Args: 681 lat: 纬度 682 lon: 经度 683 title: 发送时可选,标题 684 content: 发送时可选,内容描述 685 """ 686 self.lat = lat 687 self.lon = lon 688 self.title = title 689 self.content = content 690 super().__init__({"type": "location", "data": {"lat": str(self.lat), "lon": str(self.lon)}}) 691 692 if title != "": 693 self.array["data"]["title"] = str(self.title) 694 695 if content != "": 696 self.array["data"]["content"] = str(self.content) 697 698 def set_lat(self, lat): 699 """ 700 设置纬度 701 Args: 702 lat: 纬度 703 """ 704 self.lat = lat 705 self.array["data"]["lat"] = str(lat) 706 707 def set_lon(self, lon): 708 """ 709 设置经度 710 Args: 711 lon: 经度 712 """ 713 self.lon = lon 714 self.array["data"]["lon"] = str(lon) 715 716 def set_title(self, title): 717 """ 718 设置标题 719 Args: 720 title: 标题 721 """ 722 self.title = title 723 self.array["data"]["title"] = str(title) 724 725 def set_content(self, content): 726 """ 727 设置内容描述 728 Args: 729 content: 内容描述 730 """ 731 self.content = content 732 self.array["data"]["content"] = str(content) 733 734 735class Node(Segment): 736 """ 737 合并转发消息节点 738 接收时,此消息段不会直接出现在消息事件的 message 中,需通过 get_forward_msg API 获取。 739 """ 740 segment_type = "node" 741 742 def __init__(self, name: str, user_id: int, message, message_id: int = None): 743 """ 744 Args: 745 name: 发送者昵称 746 user_id: 发送者 QQ 号 747 message: 消息内容 748 message_id: 消息 ID(选填,若设置,上面三者失效) 749 """ 750 if message_id is None: 751 self.name = name 752 self.user_id = user_id 753 self.message = QQRichText(message).get_array() 754 super().__init__({"type": "node", "data": {"nickname": str(self.name), "user_id": str(self.user_id), 755 "content": self.message}}) 756 else: 757 self.message_id = message_id 758 super().__init__({"type": "node", "data": {"id": str(message_id)}}) 759 760 def set_message(self, message): 761 """ 762 设置消息 763 Args: 764 message: 消息内容 765 """ 766 self.message = QQRichText(message).get_array() 767 self.array["data"]["content"] = self.message 768 769 def set_name(self, name): 770 """ 771 设置发送者昵称 772 Args: 773 name: 发送者昵称 774 """ 775 self.name = name 776 self.array["data"]["name"] = str(name) 777 778 def set_user_id(self, user_id): 779 """ 780 设置发送者 QQ 号 781 Args: 782 user_id: 发送者 QQ 号 783 """ 784 self.user_id = user_id 785 self.array["data"]["uin"] = str(user_id) 786 787 def render(self, group_id: int | None = None): 788 if self.message_id is not None: 789 return f"[合并转发节点: {self.name}({self.user_id}): {self.message}]" 790 else: 791 return f"[合并转发节点: {self.message_id}]" 792 793 794class Music(Segment): 795 """ 796 音乐消息段 797 """ 798 segment_type = "music" 799 800 def __init__(self, type_, id_): 801 """ 802 Args: 803 type_: 音乐类型(可为qq 163 xm) 804 id_: 音乐 ID 805 """ 806 self.type = type_ 807 self.id = id_ 808 super().__init__({"type": "music", "data": {"type": str(self.type), "id": str(self.id)}}) 809 810 def set_type(self, type_): 811 """ 812 设置音乐类型 813 Args: 814 type_: 音乐类型(qq 163 xm) 815 """ 816 self.type = type_ 817 self.array["data"]["type"] = str(type_) 818 819 def set_id(self, id_): 820 """ 821 设置音乐 ID 822 Args: 823 id_: 音乐 ID 824 """ 825 self.id = id_ 826 self.array["data"]["id"] = str(id_) 827 828 829class CustomizeMusic(Segment): 830 """ 831 自定义音乐消息段 832 """ 833 segment_type = "music" 834 835 def __init__(self, url, audio, image, title="", content=""): 836 """ 837 Args: 838 url: 点击后跳转目标 URL 839 audio: 音乐 URL 840 image: 标题 841 title: 发送时可选,内容描述 842 content: 发送时可选,图片 URL 843 """ 844 self.url = url 845 self.audio = audio 846 self.image = image 847 self.title = title 848 self.content = content 849 super().__init__({"type": "music", "data": {"type": "custom", "url": str(self.url), "audio": str(self.audio), 850 "image": str(self.image)}}) 851 if title != "": 852 self.array["data"]["title"] = str(self.title) 853 854 if content != "": 855 self.array["data"]["content"] = str(self.content) 856 857 def set_url(self, url): 858 """ 859 设置 URL 860 Args: 861 url: 点击后跳转目标 URL 862 """ 863 self.url = url 864 self.array["data"]["url"] = str(url) 865 866 def set_audio(self, audio): 867 """ 868 设置音乐 URL 869 Args: 870 audio: 音乐 URL 871 """ 872 self.audio = audio 873 self.array["data"]["audio"] = str(audio) 874 875 def set_image(self, image): 876 """ 877 设置图片 URL 878 Args: 879 image: 图片 URL 880 """ 881 self.image = image 882 self.array["data"]["image"] = str(image) 883 884 def set_title(self, title): 885 """ 886 设置标题 887 Args: 888 title: 标题 889 """ 890 self.title = title 891 self.array["data"]["title"] = str(title) 892 893 def set_content(self, content): 894 """ 895 设置内容描述 896 Args: 897 content: 898 """ 899 self.content = content 900 self.array["data"]["content"] = str(content) 901 902 903class Reply(Segment): 904 """ 905 回复消息段 906 """ 907 segment_type = "reply" 908 909 def __init__(self, message_id): 910 """ 911 Args: 912 message_id: 回复消息 ID 913 """ 914 self.message_id = message_id 915 super().__init__({"type": "reply", "data": {"id": str(self.message_id)}}) 916 917 def set_message_id(self, message_id): 918 """ 919 设置消息 ID 920 Args: 921 message_id: 消息 ID 922 """ 923 self.message_id = message_id 924 self.array["data"]["id"] = str(self.message_id) 925 926 def render(self, group_id: int | None = None): 927 return f"[回复: {self.message_id}]" 928 929 930class Forward(Segment): 931 """ 932 合并转发消息段 933 """ 934 segment_type = "forward" 935 936 def __init__(self, forward_id): 937 """ 938 Args: 939 forward_id: 合并转发消息 ID 940 """ 941 self.forward_id = forward_id 942 super().__init__({"type": "forward", "data": {"id": str(self.forward_id)}}) 943 944 def set_forward_id(self, forward_id): 945 """ 946 设置合并转发消息 ID 947 Args: 948 forward_id: 合并转发消息 ID 949 """ 950 self.forward_id = forward_id 951 self.array["data"]["id"] = str(self.forward_id) 952 953 def render(self, group_id: int | None = None): 954 return f"[合并转发: {self.forward_id}]" 955 956 957class XML(Segment): 958 """ 959 XML消息段 960 """ 961 segment_type = "xml" 962 963 def __init__(self, data): 964 self.xml_data = data 965 super().__init__({"type": "xml", "data": {"data": str(self.xml_data)}}) 966 967 def set_xml_data(self, data): 968 """ 969 设置xml数据 970 Args: 971 data: xml数据 972 """ 973 self.xml_data = data 974 self.array["data"]["data"] = str(self.xml_data) 975 976 977class JSON(Segment): 978 """ 979 JSON消息段 980 """ 981 segment_type = "json" 982 983 def __init__(self, data): 984 """ 985 Args: 986 data: JSON 内容 987 """ 988 self.json_data = data 989 super().__init__({"type": "json", "data": {"data": str(self.json_data)}}) 990 991 def set_json(self, data): 992 """ 993 设置json数据 994 Args: 995 data: json 内容 996 """ 997 self.json_data = data 998 self.array["data"]["data"] = str(self.json_data) 999 1000 def get_json(self): 1001 """ 1002 获取json数据(自动序列化) 1003 Returns: 1004 json: json数据 1005 """ 1006 return json.loads(self.json_data) 1007 1008 1009class QQRichText: 1010 """ 1011 QQ富文本 1012 """ 1013 1014 def __init__(self, *rich: str | dict | list | tuple | Segment): 1015 """ 1016 Args: 1017 *rich: 富文本内容,可为 str、dict、list、tuple、Segment、QQRichText 1018 """ 1019 1020 # 特判 1021 self.rich_array: list[Segment] = [] 1022 if len(rich) == 1: 1023 rich = rich[0] 1024 1025 # 识别输入的是CQCode or json形式的富文本 1026 # 如果输入的是CQCode,则转换为json形式的富文本 1027 1028 # 处理CQCode 1029 if isinstance(rich, str): 1030 rich_string = rich 1031 rich = cq_2_array(rich_string) 1032 1033 elif isinstance(rich, dict): 1034 rich = [rich] 1035 elif isinstance(rich, (list, tuple)): 1036 array = [] 1037 for item in rich: 1038 if isinstance(item, dict): 1039 array.append(item) 1040 elif isinstance(item, str): 1041 array += cq_2_array(item) 1042 else: 1043 for segment in segments: 1044 if isinstance(item, segment): 1045 array.append(item.array) 1046 break 1047 else: 1048 if isinstance(rich, QQRichText): 1049 array += rich.rich_array 1050 else: 1051 raise TypeError("QQRichText: 输入类型错误") 1052 rich = array 1053 else: 1054 for segment in segments: 1055 if isinstance(rich, segment): 1056 rich = [rich.array] 1057 break 1058 else: 1059 if isinstance(rich, QQRichText): 1060 rich = rich.rich_array 1061 else: 1062 raise TypeError("QQRichText: 输入类型错误") 1063 1064 # 将rich转换为的Segment 1065 for i in range(len(rich)): 1066 if rich[i]["type"] in segments_map: 1067 try: 1068 params = inspect.signature(segments_map[rich[i]["type"]]).parameters 1069 kwargs = {} 1070 for param in params: 1071 if param in rich[i]["data"]: 1072 kwargs[param] = rich[i]["data"][param] 1073 else: 1074 if rich[i]["type"] == "reply" and param == "message_id": 1075 kwargs[param] = rich[i]["data"].get("id") 1076 elif rich[i]["type"] == "face" and param == "face_id": 1077 kwargs[param] = rich[i]["data"].get("id") 1078 elif rich[i]["type"] == "forward" and param == "forward_id": 1079 kwargs[param] = rich[i]["data"].get("id") 1080 elif rich[i]["type"] == "poke" and param == "poke_id": 1081 kwargs[param] = rich[i]["data"].get("id") 1082 elif param == "id_": 1083 kwargs[param] = rich[i]["data"].get("id") 1084 elif param == "type_": 1085 kwargs[param] = rich[i]["data"].get("type") 1086 else: 1087 if params[param].default != params[param].empty: 1088 kwargs[param] = params[param].default 1089 segment = segments_map[rich[i]["type"]](**kwargs) 1090 # 检查原cq中是否含有不在segment的data中的参数 1091 for k, v in rich[i]["data"].items(): 1092 if k not in segment["data"]: 1093 segment.set_data(k, v) 1094 rich[i] = segment 1095 except Exception as e: 1096 if ConfigManager.GlobalConfig().debug.save_dump: 1097 dump_path = save_exc_dump(f"转换{rich[i]}时失败") 1098 else: 1099 dump_path = None 1100 Logger.get_logger().warning(f"转换{rich[i]}时失败,报错信息: {repr(e)}\n" 1101 f"{traceback.format_exc()}" 1102 f"{f"\n已保存异常到 {dump_path}" if dump_path else ""}") 1103 rich[i] = Segment(rich[i]) 1104 else: 1105 rich[i] = Segment(rich[i]) 1106 1107 self.rich_array: list[Segment] = rich 1108 1109 def render(self, group_id: int | None = None): 1110 """ 1111 渲染消息(调用rich_array下所有消息段的render方法拼接) 1112 Args: 1113 group_id: 群号,选填,可优化效果 1114 """ 1115 text = "" 1116 for rich in self.rich_array: 1117 text += rich.render(group_id=group_id) 1118 return text 1119 1120 def __str__(self): 1121 self.rich_string = array_2_cq(self.rich_array) 1122 return self.rich_string 1123 1124 def __repr__(self): 1125 return self.__str__() 1126 1127 def __getitem__(self, index): 1128 return self.rich_array[index] 1129 1130 def __add__(self, other): 1131 other = QQRichText(other) 1132 return QQRichText(self.rich_array + other.rich_array) 1133 1134 def __eq__(self, other): 1135 other = QQRichText(other) 1136 1137 return self.rich_array == other.rich_array 1138 1139 def __contains__(self, other): 1140 if isinstance(other, QQRichText): 1141 return all(item in self.rich_array for item in other.rich_array) 1142 else: 1143 try: 1144 return str(other) in str(self) 1145 except (TypeError, AttributeError): 1146 return False 1147 1148 def get_array(self): 1149 """ 1150 获取rich_array(非抽象类,可用于API调用等) 1151 Returns: 1152 rich_array 1153 """ 1154 return [array.array for array in self.rich_array] 1155 1156 def add(self, *segments): 1157 """ 1158 添加消息段 1159 Args: 1160 *segments: 消息段 1161 1162 Returns: 1163 self 1164 """ 1165 for segment in segments: 1166 if isinstance(segment, Segment): 1167 self.rich_array.append(segment) 1168 else: 1169 self.rich_array += QQRichText(segment).rich_array 1170 return self 1171 1172 1173# 使用示例 1174if __name__ == "__main__": 1175 # 测试CQ解码 1176 print(cq_decode(" - [x] 使用 `&data` 获取地址")) 1177 1178 # 测试CQ编码 1179 print(cq_encode(" - [x] 使用 `&data` 获取地址")) 1180 1181 # 测试QQRichText 1182 rich = QQRichText( 1183 "[CQ:reply,id=123][CQ:share,title=标题,url=https://baidu.com] [CQ:at,qq=1919810,abc=123] - [x] 使用 " 1184 " `&data` 获取地址") 1185 print(rich.get_array()) 1186 print(rich) 1187 print(rich.render()) 1188 1189 print(QQRichText(At(114514))) 1190 print(Segment(At(1919810))) 1191 print(QQRichText([{"type": "text", "data": {"text": "1919810"}}])) 1192 print(QQRichText().add(At(114514)).add(Text("我吃柠檬")) + QQRichText(At(1919810)).rich_array) 1193 rich_array = [{'type': 'at', 'data': {'qq': '123'}}, {'type': 'text', 'data': {'text': '[期待]'}}] 1194 rich = QQRichText(rich_array) 1195 print(rich) 1196 print(rich.get_array())
def
cq_decode(text, in_cq: bool = False) -> str:
17def cq_decode(text, in_cq: bool = False) -> str: 18 """ 19 CQ解码 20 Args: 21 text: 文本(CQ) 22 in_cq: 该文本是否是在CQ内的 23 Returns: 24 解码后的文本 25 """ 26 text = str(text) 27 if in_cq: 28 return text.replace("&", "&").replace("[", "["). \ 29 replace("]", "]").replace(",", ",") 30 else: 31 return text.replace("&", "&").replace("[", "["). \ 32 replace("]", "]")
CQ解码
Arguments:
- text: 文本(CQ)
- in_cq: 该文本是否是在CQ内的
Returns:
解码后的文本
def
cq_encode(text, in_cq: bool = False) -> str:
35def cq_encode(text, in_cq: bool = False) -> str: 36 """ 37 CQ编码 38 Args: 39 text: 文本 40 in_cq: 该文本是否是在CQ内的 41 Returns: 42 编码后的文本 43 """ 44 text = str(text) 45 if in_cq: 46 return text.replace("&", "&").replace("[", "["). \ 47 replace("]", "]").replace(",", ",") 48 else: 49 return text.replace("&", "&").replace("[", "["). \ 50 replace("]", "]")
CQ编码
Arguments:
- text: 文本
- in_cq: 该文本是否是在CQ内的
Returns:
编码后的文本
def
cq_2_array(cq: str) -> list[dict[str, dict[str, typing.Any]]]:
53def cq_2_array(cq: str) -> list[dict[str, dict[str, Any]]]: 54 """ 55 将CQCode格式的字符串转换为消息段数组。 56 57 Args: 58 cq (str): CQCode字符串。 59 60 Returns: 61 list[dict[str, dict[str, Any]]]: 解析后的消息段数组。 62 """ 63 if not isinstance(cq, str): 64 raise TypeError("cq_2_array: 输入类型错误") 65 66 cq_array = [] # 存储解析后的消息段 67 now_state = 0 # 当前解析状态 68 # 0: 不在CQ码内 69 # 1: 在CQ码的类型部分 70 # 2: 在CQ码的参数键部分 71 # 3: 在CQ码的参数值部分 72 73 segment_data = {"text": ""} # 用于存储当前解析的消息段 74 now_key = "" # 当前解析的参数键 75 now_segment_type = "" # 当前解析的CQ码类型 76 77 for c in cq: 78 if now_state == 0: # 解析普通文本 79 if c == "[": # 进入CQ码解析 80 now_state = 1 81 if len(segment_data["text"]): # 先存储之前的普通文本 82 cq_array.append({"type": "text", "data": {"text": cq_decode(segment_data["text"])}}) 83 segment_data = {} # 重置segment_data用于存储新的CQ码 84 else: 85 segment_data["text"] += c # 继续拼接普通文本 86 87 elif now_state == 1: # 解析CQ码类型 88 if c == ",": # 类型解析结束,进入参数键解析 89 now_state = 2 90 now_segment_type = now_segment_type[3:] # 移除CQ:前缀 91 else: 92 now_segment_type += c # 继续拼接类型字符串 93 94 elif now_state == 2: # 解析参数键 95 if c == "=": # 键名解析结束,进入值解析 96 now_state = 3 97 now_key = cq_decode(now_key, in_cq=True) # 解码键名 98 if now_key not in segment_data: 99 segment_data[now_key] = "" # 初始化键值 100 else: 101 raise ValueError("cq_2_array: CQ码键名称重复") 102 else: 103 now_key += c # 继续拼接键名 104 105 elif now_state == 3: # 解析参数值 106 if c == "]": # CQ码结束 107 now_state = 0 108 segment_data[now_key] = cq_decode(segment_data[now_key], in_cq=True) # 解码值 109 cq_array.append({"type": now_segment_type, "data": segment_data}) # 存入解析结果 110 segment_data = {"text": ""} # 重置segment_data 111 now_segment_type = "" # 清空类型 112 now_key = "" # 清空键名 113 elif c == ",": # 进入下一个参数键解析 114 segment_data[now_key] = cq_decode(segment_data[now_key], in_cq=True) # 解码值 115 now_state = 2 116 now_key = "" # 清空键名,准备解析下一个键 117 else: 118 segment_data[now_key] += c # 继续拼接参数值 119 120 if "text" in segment_data and len(segment_data["text"]): # 处理末尾可能存在的文本内容 121 cq_array.append({"type": "text", "data": {"text": cq_decode(segment_data["text"])}}) 122 123 return cq_array
将CQCode格式的字符串转换为消息段数组。
Arguments:
- cq (str): CQCode字符串。
Returns:
list[dict[str, dict[str, Any]]]: 解析后的消息段数组。
def
array_2_cq(cq_array: list | dict) -> str:
126def array_2_cq(cq_array: list | dict) -> str: 127 """ 128 array消息段转CQCode 129 Args: 130 cq_array: array消息段数组 131 Returns: 132 CQCode 133 """ 134 # 特判 135 if isinstance(cq_array, dict): 136 cq_array = [cq_array] 137 138 if not isinstance(cq_array, (list, tuple)): 139 raise TypeError("array_2_cq: 输入类型错误") 140 141 # 将json形式的富文本转换为CQ码 142 text = "" 143 for segment in cq_array: 144 # 纯文本 145 if segment.get("type") == "text": 146 text += cq_encode(segment.get("data").get("text")) 147 # CQ码 148 else: 149 if segment.get("data"): # 特判 150 text += f"[CQ:{segment.get('type')}," + ",".join( 151 [cq_encode(x, in_cq=True) + "=" + cq_encode(segment.get("data")[x], in_cq=True) 152 for x in segment.get("data").keys()]) + "]" 153 else: 154 text += f"[CQ:{segment.get('type')}]" 155 return text
array消息段转CQCode
Arguments:
- cq_array: array消息段数组
Returns:
CQCode
def
convert_to_fileurl(input_str: str) -> str:
158def convert_to_fileurl(input_str: str) -> str: 159 """ 160 自动将输入的路径转换成fileurl 161 Args: 162 input_str: 输入的路径 163 164 Returns: 165 转换后的 fileurl 166 """ 167 # 检查是否已经是 file:// 格式 168 if input_str.startswith("file://"): 169 return input_str 170 171 # 检查输入是否是有效的 URL 172 parsed_url = urlparse(input_str) 173 if parsed_url.scheme in ['http', 'https', 'ftp', 'file', 'data']: 174 return input_str # 已经是 URL 格式,直接返回 175 176 # 检查输入是否是有效的本地文件路径 177 if os.path.isfile(input_str): 178 # 转换为 file:// 格式 179 return f"file://{os.path.abspath(input_str)}" 180 181 # 如果是相对路径或其他文件类型,则尝试转换 182 if os.path.exists(input_str): 183 return f"file://{os.path.abspath(input_str)}" 184 185 raise ValueError("输入的路径无效,无法转换为 fileurl 格式")
自动将输入的路径转换成fileurl
Arguments:
- input_str: 输入的路径
Returns:
转换后的 fileurl
segments =
[<class 'Segment'>, <class 'Text'>, <class 'Face'>, <class 'At'>, <class 'Image'>, <class 'Record'>, <class 'Video'>, <class 'Rps'>, <class 'Dice'>, <class 'Shake'>, <class 'Poke'>, <class 'Anonymous'>, <class 'Share'>, <class 'Contact'>, <class 'Location'>, <class 'Node'>, <class 'Music'>, <class 'CustomizeMusic'>, <class 'Reply'>, <class 'Forward'>, <class 'XML'>, <class 'JSON'>]
segments_map =
{'text': <class 'Text'>, 'face': <class 'Face'>, 'at': <class 'At'>, 'image': <class 'Image'>, 'record': <class 'Record'>, 'video': <class 'Video'>, 'rps': <class 'Rps'>, 'dice': <class 'Dice'>, 'shake': <class 'Shake'>, 'poke': <class 'Poke'>, 'anonymous': <class 'Anonymous'>, 'share': <class 'Share'>, 'contact': <class 'Contact'>, 'location': <class 'Location'>, 'node': <class 'Node'>, 'music': <class 'CustomizeMusic'>, 'reply': <class 'Reply'>, 'forward': <class 'Forward'>, 'xml': <class 'XML'>, 'json': <class 'JSON'>}
class
SegmentMeta(builtins.type):
192class SegmentMeta(type): 193 """ 194 元类用于自动注册 Segment 子类到全局列表 segments 和映射 segments_map 中。 195 """ 196 197 def __init__(cls, name, bases, dct): 198 super().__init__(name, bases, dct) 199 if 'Segment' in globals() and issubclass(cls, Segment): 200 segments.append(cls) # 将子类添加到全局列表中 201 segments_map[cls.segment_type] = cls
元类用于自动注册 Segment 子类到全局列表 segments 和映射 segments_map 中。
class
Segment:
204class Segment(metaclass=SegmentMeta): 205 """ 206 消息段 207 """ 208 segment_type = None 209 210 def __init__(self, cq): 211 self.cq = cq 212 if isinstance(cq, str): 213 self.array = cq_2_array(cq)[0] 214 self.type, self.data = list(self.array.values()) 215 elif isinstance(cq, dict): 216 self.array = cq 217 self.cq = array_2_cq(self.array) 218 self.type, self.data = list(self.array.values()) 219 else: 220 for segment in segments: 221 if isinstance(cq, segment): 222 self.array = cq.array 223 self.cq = str(self.cq) 224 # print(self.array.values(), list(self.array.values())) 225 self.type, self.data = list(self.array.values()) 226 break 227 else: 228 # print(cq, str(cq), type(cq)) 229 raise TypeError("Segment: 输入类型错误") 230 231 def __str__(self): 232 return self.__repr__() 233 234 def __repr__(self): 235 self.cq = array_2_cq(self.array) 236 return self.cq 237 238 def __setitem__(self, key, value): 239 self.array[key] = value 240 241 def __getitem__(self, key): 242 return self.array.get(key) 243 244 def get(self, key, default=None): 245 """ 246 获取消息段中的数据 247 Args: 248 key: key 249 default: 默认值(默认为None) 250 251 Returns: 252 获取到的数据 253 """ 254 return self.array.get(key, default) 255 256 def __delitem__(self, key): 257 del self.array[key] 258 259 def __eq__(self, other): 260 other = Segment(other) 261 return self.array == other.array 262 263 def __contains__(self, other): 264 if isinstance(other, Segment): 265 return all(item in self.array for item in other.array) 266 else: 267 try: 268 return str(other) in str(self) 269 except (TypeError, AttributeError): 270 return False 271 272 def render(self, group_id: int | None = None): 273 """ 274 渲染消息段为字符串 275 Args: 276 group_id: 群号(选填) 277 Returns: 278 渲染完毕的消息段 279 """ 280 return f"[{self.array.get('type', 'unknown')}: {self.cq}]" 281 282 def set_data(self, k, v): 283 """ 284 设置消息段的Data项 285 Args: 286 k: 要修改的key 287 v: 要修改成的value 288 """ 289 self.array["data"][k] = v
消息段
Segment(cq)
210 def __init__(self, cq): 211 self.cq = cq 212 if isinstance(cq, str): 213 self.array = cq_2_array(cq)[0] 214 self.type, self.data = list(self.array.values()) 215 elif isinstance(cq, dict): 216 self.array = cq 217 self.cq = array_2_cq(self.array) 218 self.type, self.data = list(self.array.values()) 219 else: 220 for segment in segments: 221 if isinstance(cq, segment): 222 self.array = cq.array 223 self.cq = str(self.cq) 224 # print(self.array.values(), list(self.array.values())) 225 self.type, self.data = list(self.array.values()) 226 break 227 else: 228 # print(cq, str(cq), type(cq)) 229 raise TypeError("Segment: 输入类型错误")
def
get(self, key, default=None):
244 def get(self, key, default=None): 245 """ 246 获取消息段中的数据 247 Args: 248 key: key 249 default: 默认值(默认为None) 250 251 Returns: 252 获取到的数据 253 """ 254 return self.array.get(key, default)
获取消息段中的数据
Arguments:
- key: key
- default: 默认值(默认为None)
Returns:
获取到的数据
def
render(self, group_id: int | None = None):
272 def render(self, group_id: int | None = None): 273 """ 274 渲染消息段为字符串 275 Args: 276 group_id: 群号(选填) 277 Returns: 278 渲染完毕的消息段 279 """ 280 return f"[{self.array.get('type', 'unknown')}: {self.cq}]"
渲染消息段为字符串
Arguments:
- group_id: 群号(选填)
Returns:
渲染完毕的消息段
295class Text(Segment): 296 """ 297 文本消息段 298 """ 299 segment_type = "text" 300 301 def __init__(self, text): 302 """ 303 Args: 304 text: 文本 305 """ 306 self.text = text 307 super().__init__({"type": "text", "data": {"text": text}}) 308 309 def __add__(self, other): 310 other = Text(other) 311 return self.text + other.text 312 313 def __eq__(self, other): 314 other = Text(other) 315 return self.text == other.text 316 317 def __contains__(self, other): 318 if isinstance(other, Text): 319 return other.text in self.text 320 else: 321 try: 322 return str(other) in str(self.text) 323 except (TypeError, AttributeError): 324 return False 325 326 def set_text(self, text): 327 """ 328 设置文本 329 Args: 330 text: 文本 331 """ 332 self.text = text 333 self["data"]["text"] = text 334 335 def render(self, group_id: int | None = None): 336 return self.text
文本消息段
Text(text)
301 def __init__(self, text): 302 """ 303 Args: 304 text: 文本 305 """ 306 self.text = text 307 super().__init__({"type": "text", "data": {"text": text}})
Arguments:
- text: 文本
def
set_text(self, text):
326 def set_text(self, text): 327 """ 328 设置文本 329 Args: 330 text: 文本 331 """ 332 self.text = text 333 self["data"]["text"] = text
设置文本
Arguments:
- text: 文本
339class Face(Segment): 340 """ 341 表情消息段 342 """ 343 segment_type = "face" 344 345 def __init__(self, face_id): 346 """ 347 Args: 348 face_id: 表情id 349 """ 350 self.face_id = face_id 351 super().__init__({"type": "face", "data": {"id": str(face_id)}}) 352 353 def set_id(self, face_id): 354 """ 355 设置表情id 356 Args: 357 face_id: 表情id 358 """ 359 self.face_id = face_id 360 self.array["data"]["id"] = str(face_id) 361 362 def render(self, group_id: int | None = None): 363 return "[表情: %s]" % self.face_id
表情消息段
Face(face_id)
345 def __init__(self, face_id): 346 """ 347 Args: 348 face_id: 表情id 349 """ 350 self.face_id = face_id 351 super().__init__({"type": "face", "data": {"id": str(face_id)}})
Arguments:
- face_id: 表情id
def
set_id(self, face_id):
353 def set_id(self, face_id): 354 """ 355 设置表情id 356 Args: 357 face_id: 表情id 358 """ 359 self.face_id = face_id 360 self.array["data"]["id"] = str(face_id)
设置表情id
Arguments:
- face_id: 表情id
366class At(Segment): 367 """ 368 At消息段 369 """ 370 segment_type = "at" 371 372 def __init__(self, qq): 373 """ 374 Args: 375 qq: qq号 376 """ 377 self.qq = qq 378 super().__init__({"type": "at", "data": {"qq": str(qq)}}) 379 380 def set_id(self, qq_id): 381 """ 382 设置At的id 383 Args: 384 qq_id: qq号 385 """ 386 self.qq = qq_id 387 self.array["data"]["qq"] = str(qq_id) 388 389 def render(self, group_id: int | None = None): 390 if group_id: 391 if str(self.qq) == "all" or str(self.qq) == "0": 392 return f"@全体成员" 393 return f"@{QQDataCacher.get_group_member_info(group_id, self.qq).get_nickname()}: {self.qq}" 394 else: 395 return f"@{QQDataCacher.get_user_info(self.qq).nickname}: {self.qq}"
At消息段
At(qq)
372 def __init__(self, qq): 373 """ 374 Args: 375 qq: qq号 376 """ 377 self.qq = qq 378 super().__init__({"type": "at", "data": {"qq": str(qq)}})
Arguments:
- qq: qq号
def
set_id(self, qq_id):
380 def set_id(self, qq_id): 381 """ 382 设置At的id 383 Args: 384 qq_id: qq号 385 """ 386 self.qq = qq_id 387 self.array["data"]["qq"] = str(qq_id)
设置At的id
Arguments:
- qq_id: qq号
def
render(self, group_id: int | None = None):
389 def render(self, group_id: int | None = None): 390 if group_id: 391 if str(self.qq) == "all" or str(self.qq) == "0": 392 return f"@全体成员" 393 return f"@{QQDataCacher.get_group_member_info(group_id, self.qq).get_nickname()}: {self.qq}" 394 else: 395 return f"@{QQDataCacher.get_user_info(self.qq).nickname}: {self.qq}"
渲染消息段为字符串
Arguments:
- group_id: 群号(选填)
Returns:
渲染完毕的消息段
398class Image(Segment): 399 """ 400 图片消息段 401 """ 402 segment_type = "image" 403 404 def __init__(self, file: str): 405 """ 406 Args: 407 file: 图片文件(url,对于文件使用file url格式) 408 """ 409 file = convert_to_fileurl(file) 410 self.file = file 411 super().__init__({"type": "image", "data": {"file": str(file)}}) 412 413 def set_file(self, file: str): 414 """ 415 设置图片文件 416 Args: 417 file: 图片文件 418 """ 419 file = convert_to_fileurl(file) 420 self.file = file 421 self.array["data"]["file"] = str(file) 422 423 def render(self, group_id: int | None = None): 424 return "[图片: %s]" % self.file
图片消息段
Image(file: str)
404 def __init__(self, file: str): 405 """ 406 Args: 407 file: 图片文件(url,对于文件使用file url格式) 408 """ 409 file = convert_to_fileurl(file) 410 self.file = file 411 super().__init__({"type": "image", "data": {"file": str(file)}})
Arguments:
- file: 图片文件(url,对于文件使用file url格式)
def
set_file(self, file: str):
413 def set_file(self, file: str): 414 """ 415 设置图片文件 416 Args: 417 file: 图片文件 418 """ 419 file = convert_to_fileurl(file) 420 self.file = file 421 self.array["data"]["file"] = str(file)
设置图片文件
Arguments:
- file: 图片文件
427class Record(Segment): 428 """ 429 语音消息段 430 """ 431 segment_type = "record" 432 433 def __init__(self, file: str): 434 """ 435 Args: 436 file: 语音文件(url,对于文件使用file url格式) 437 """ 438 file = convert_to_fileurl(file) 439 self.file = file 440 super().__init__({"type": "record", "data": {"file": str(file)}}) 441 442 def set_file(self, file: str): 443 """ 444 设置语音文件 445 Args: 446 file: 语音文件(url,对于文件使用file url格式) 447 """ 448 file = convert_to_fileurl(file) 449 self.file = file 450 self.array["data"]["file"] = str(file) 451 452 def render(self, group_id: int | None = None): 453 return "[语音: %s]" % self.file
语音消息段
Record(file: str)
433 def __init__(self, file: str): 434 """ 435 Args: 436 file: 语音文件(url,对于文件使用file url格式) 437 """ 438 file = convert_to_fileurl(file) 439 self.file = file 440 super().__init__({"type": "record", "data": {"file": str(file)}})
Arguments:
- file: 语音文件(url,对于文件使用file url格式)
def
set_file(self, file: str):
442 def set_file(self, file: str): 443 """ 444 设置语音文件 445 Args: 446 file: 语音文件(url,对于文件使用file url格式) 447 """ 448 file = convert_to_fileurl(file) 449 self.file = file 450 self.array["data"]["file"] = str(file)
设置语音文件
Arguments:
- file: 语音文件(url,对于文件使用file url格式)
456class Video(Segment): 457 """ 458 视频消息段 459 """ 460 segment_type = "video" 461 462 def __init__(self, file: str): 463 """ 464 Args: 465 file: 视频文件(url,对于文件使用file url格式) 466 """ 467 file = convert_to_fileurl(file) 468 self.file = file 469 super().__init__({"type": "video", "data": {"file": str(file)}}) 470 471 def set_file(self, file: str): 472 """ 473 设置视频文件 474 Args: 475 file: 视频文件(url,对于文件使用file url格式) 476 """ 477 file = convert_to_fileurl(file) 478 self.file = file 479 self.array["data"]["file"] = str(file) 480 481 def render(self, group_id: int | None = None): 482 return f"[视频: {self.file}]"
视频消息段
Video(file: str)
462 def __init__(self, file: str): 463 """ 464 Args: 465 file: 视频文件(url,对于文件使用file url格式) 466 """ 467 file = convert_to_fileurl(file) 468 self.file = file 469 super().__init__({"type": "video", "data": {"file": str(file)}})
Arguments:
- file: 视频文件(url,对于文件使用file url格式)
def
set_file(self, file: str):
471 def set_file(self, file: str): 472 """ 473 设置视频文件 474 Args: 475 file: 视频文件(url,对于文件使用file url格式) 476 """ 477 file = convert_to_fileurl(file) 478 self.file = file 479 self.array["data"]["file"] = str(file)
设置视频文件
Arguments:
- file: 视频文件(url,对于文件使用file url格式)
485class Rps(Segment): 486 """ 487 猜拳消息段 488 """ 489 segment_type = "rps" 490 491 def __init__(self): 492 super().__init__({"type": "rps"})
猜拳消息段
495class Dice(Segment): 496 segment_type = "dice" 497 498 def __init__(self): 499 super().__init__({"type": "dice"})
消息段
502class Shake(Segment): 503 """ 504 窗口抖动消息段 505 (相当于戳一戳最基本类型的快捷方式。) 506 """ 507 segment_type = "shake" 508 509 def __init__(self): 510 super().__init__({"type": "shake"})
窗口抖动消息段 (相当于戳一戳最基本类型的快捷方式。)
513class Poke(Segment): 514 """ 515 戳一戳消息段 516 """ 517 segment_type = "poke" 518 519 def __init__(self, type_, poke_id): 520 """ 521 Args: 522 type_: 见https://github.com/botuniverse/onebot-11/blob/master/message/segment.md#%E6%88%B3%E4%B8%80%E6%88%B3 523 poke_id: 同上 524 """ 525 self.type = type_ 526 self.poke_id = poke_id 527 super().__init__({"type": "poke", "data": {"type": str(self.type)}, "id": str(self.poke_id)}) 528 529 def set_type(self, qq_type): 530 """ 531 设置戳一戳类型 532 Args: 533 qq_type: qq类型 534 """ 535 self.type = qq_type 536 self.array["data"]["type"] = str(qq_type) 537 538 def set_id(self, poke_id): 539 """ 540 设置戳一戳id 541 Args: 542 poke_id: 戳一戳id 543 """ 544 self.poke_id = poke_id 545 self.array["data"]["id"] = str(poke_id) 546 547 def render(self, group_id: int | None = None): 548 return f"[戳一戳: {self.type}]"
戳一戳消息段
Poke(type_, poke_id)
519 def __init__(self, type_, poke_id): 520 """ 521 Args: 522 type_: 见https://github.com/botuniverse/onebot-11/blob/master/message/segment.md#%E6%88%B3%E4%B8%80%E6%88%B3 523 poke_id: 同上 524 """ 525 self.type = type_ 526 self.poke_id = poke_id 527 super().__init__({"type": "poke", "data": {"type": str(self.type)}, "id": str(self.poke_id)})
Arguments:
- type_: 见https://github.com/botuniverse/onebot-11/blob/master/message/segment.md#%E6%88%B3%E4%B8%80%E6%88%B3
- poke_id: 同上
def
set_type(self, qq_type):
529 def set_type(self, qq_type): 530 """ 531 设置戳一戳类型 532 Args: 533 qq_type: qq类型 534 """ 535 self.type = qq_type 536 self.array["data"]["type"] = str(qq_type)
设置戳一戳类型
Arguments:
- qq_type: qq类型
def
set_id(self, poke_id):
538 def set_id(self, poke_id): 539 """ 540 设置戳一戳id 541 Args: 542 poke_id: 戳一戳id 543 """ 544 self.poke_id = poke_id 545 self.array["data"]["id"] = str(poke_id)
设置戳一戳id
Arguments:
- poke_id: 戳一戳id
551class Anonymous(Segment): 552 """ 553 匿名消息段 554 """ 555 segment_type = "anonymous" 556 557 def __init__(self, ignore=False): 558 """ 559 Args: 560 ignore: 是否忽略 561 """ 562 self.ignore = 0 if ignore else 1 563 super().__init__({"type": "anonymous", "data": {"ignore": str(self.ignore)}}) 564 565 def set_ignore(self, ignore): 566 """ 567 设置是否忽略 568 Args: 569 ignore: 是否忽略 570 """ 571 self.ignore = 0 if ignore else 1 572 self.array["data"]["ignore"] = str(self.ignore)
匿名消息段
Anonymous(ignore=False)
557 def __init__(self, ignore=False): 558 """ 559 Args: 560 ignore: 是否忽略 561 """ 562 self.ignore = 0 if ignore else 1 563 super().__init__({"type": "anonymous", "data": {"ignore": str(self.ignore)}})
Arguments:
- ignore: 是否忽略
638class Contact(Segment): 639 """ 640 推荐好友/推荐群 641 """ 642 segment_type = "contact" 643 644 def __init__(self, type_, id_): 645 """ 646 Args: 647 type_: 推荐的类型(friend/group) 648 id_: 推荐的qqid 649 """ 650 self.type = type_ 651 self.id = id_ 652 super().__init__({"type": "contact", "data": {"type": str(self.type), "id": str(self.id)}}) 653 654 def set_type(self, type_): 655 """ 656 设置推荐类型 657 Args: 658 type_: 推荐的类型(friend/group) 659 """ 660 self.type = type_ 661 self.array["data"]["type"] = str(type_) 662 663 def set_id(self, id_): 664 """ 665 设置推荐的qqid 666 Args: 667 id_: qqid 668 """ 669 self.id = id_ 670 self.array["data"]["id"] = str(id_)
推荐好友/推荐群
Contact(type_, id_)
644 def __init__(self, type_, id_): 645 """ 646 Args: 647 type_: 推荐的类型(friend/group) 648 id_: 推荐的qqid 649 """ 650 self.type = type_ 651 self.id = id_ 652 super().__init__({"type": "contact", "data": {"type": str(self.type), "id": str(self.id)}})
Arguments:
- type_: 推荐的类型(friend/group)
- id_: 推荐的qqid
def
set_type(self, type_):
654 def set_type(self, type_): 655 """ 656 设置推荐类型 657 Args: 658 type_: 推荐的类型(friend/group) 659 """ 660 self.type = type_ 661 self.array["data"]["type"] = str(type_)
设置推荐类型
Arguments:
- type_: 推荐的类型(friend/group)
673class Location(Segment): 674 """ 675 位置消息段 676 """ 677 segment_type = "location" 678 679 def __init__(self, lat, lon, title="", content=""): 680 """ 681 Args: 682 lat: 纬度 683 lon: 经度 684 title: 发送时可选,标题 685 content: 发送时可选,内容描述 686 """ 687 self.lat = lat 688 self.lon = lon 689 self.title = title 690 self.content = content 691 super().__init__({"type": "location", "data": {"lat": str(self.lat), "lon": str(self.lon)}}) 692 693 if title != "": 694 self.array["data"]["title"] = str(self.title) 695 696 if content != "": 697 self.array["data"]["content"] = str(self.content) 698 699 def set_lat(self, lat): 700 """ 701 设置纬度 702 Args: 703 lat: 纬度 704 """ 705 self.lat = lat 706 self.array["data"]["lat"] = str(lat) 707 708 def set_lon(self, lon): 709 """ 710 设置经度 711 Args: 712 lon: 经度 713 """ 714 self.lon = lon 715 self.array["data"]["lon"] = str(lon) 716 717 def set_title(self, title): 718 """ 719 设置标题 720 Args: 721 title: 标题 722 """ 723 self.title = title 724 self.array["data"]["title"] = str(title) 725 726 def set_content(self, content): 727 """ 728 设置内容描述 729 Args: 730 content: 内容描述 731 """ 732 self.content = content 733 self.array["data"]["content"] = str(content)
位置消息段
Location(lat, lon, title='', content='')
679 def __init__(self, lat, lon, title="", content=""): 680 """ 681 Args: 682 lat: 纬度 683 lon: 经度 684 title: 发送时可选,标题 685 content: 发送时可选,内容描述 686 """ 687 self.lat = lat 688 self.lon = lon 689 self.title = title 690 self.content = content 691 super().__init__({"type": "location", "data": {"lat": str(self.lat), "lon": str(self.lon)}}) 692 693 if title != "": 694 self.array["data"]["title"] = str(self.title) 695 696 if content != "": 697 self.array["data"]["content"] = str(self.content)
Arguments:
- lat: 纬度
- lon: 经度
- title: 发送时可选,标题
- content: 发送时可选,内容描述
def
set_lat(self, lat):
699 def set_lat(self, lat): 700 """ 701 设置纬度 702 Args: 703 lat: 纬度 704 """ 705 self.lat = lat 706 self.array["data"]["lat"] = str(lat)
设置纬度
Arguments:
- lat: 纬度
def
set_lon(self, lon):
708 def set_lon(self, lon): 709 """ 710 设置经度 711 Args: 712 lon: 经度 713 """ 714 self.lon = lon 715 self.array["data"]["lon"] = str(lon)
设置经度
Arguments:
- lon: 经度
def
set_title(self, title):
717 def set_title(self, title): 718 """ 719 设置标题 720 Args: 721 title: 标题 722 """ 723 self.title = title 724 self.array["data"]["title"] = str(title)
设置标题
Arguments:
- title: 标题
736class Node(Segment): 737 """ 738 合并转发消息节点 739 接收时,此消息段不会直接出现在消息事件的 message 中,需通过 get_forward_msg API 获取。 740 """ 741 segment_type = "node" 742 743 def __init__(self, name: str, user_id: int, message, message_id: int = None): 744 """ 745 Args: 746 name: 发送者昵称 747 user_id: 发送者 QQ 号 748 message: 消息内容 749 message_id: 消息 ID(选填,若设置,上面三者失效) 750 """ 751 if message_id is None: 752 self.name = name 753 self.user_id = user_id 754 self.message = QQRichText(message).get_array() 755 super().__init__({"type": "node", "data": {"nickname": str(self.name), "user_id": str(self.user_id), 756 "content": self.message}}) 757 else: 758 self.message_id = message_id 759 super().__init__({"type": "node", "data": {"id": str(message_id)}}) 760 761 def set_message(self, message): 762 """ 763 设置消息 764 Args: 765 message: 消息内容 766 """ 767 self.message = QQRichText(message).get_array() 768 self.array["data"]["content"] = self.message 769 770 def set_name(self, name): 771 """ 772 设置发送者昵称 773 Args: 774 name: 发送者昵称 775 """ 776 self.name = name 777 self.array["data"]["name"] = str(name) 778 779 def set_user_id(self, user_id): 780 """ 781 设置发送者 QQ 号 782 Args: 783 user_id: 发送者 QQ 号 784 """ 785 self.user_id = user_id 786 self.array["data"]["uin"] = str(user_id) 787 788 def render(self, group_id: int | None = None): 789 if self.message_id is not None: 790 return f"[合并转发节点: {self.name}({self.user_id}): {self.message}]" 791 else: 792 return f"[合并转发节点: {self.message_id}]"
合并转发消息节点 接收时,此消息段不会直接出现在消息事件的 message 中,需通过 get_forward_msg API 获取。
Node(name: str, user_id: int, message, message_id: int = None)
743 def __init__(self, name: str, user_id: int, message, message_id: int = None): 744 """ 745 Args: 746 name: 发送者昵称 747 user_id: 发送者 QQ 号 748 message: 消息内容 749 message_id: 消息 ID(选填,若设置,上面三者失效) 750 """ 751 if message_id is None: 752 self.name = name 753 self.user_id = user_id 754 self.message = QQRichText(message).get_array() 755 super().__init__({"type": "node", "data": {"nickname": str(self.name), "user_id": str(self.user_id), 756 "content": self.message}}) 757 else: 758 self.message_id = message_id 759 super().__init__({"type": "node", "data": {"id": str(message_id)}})
Arguments:
- name: 发送者昵称
- user_id: 发送者 QQ 号
- message: 消息内容
- message_id: 消息 ID(选填,若设置,上面三者失效)
def
set_message(self, message):
761 def set_message(self, message): 762 """ 763 设置消息 764 Args: 765 message: 消息内容 766 """ 767 self.message = QQRichText(message).get_array() 768 self.array["data"]["content"] = self.message
设置消息
Arguments:
- message: 消息内容
def
set_name(self, name):
770 def set_name(self, name): 771 """ 772 设置发送者昵称 773 Args: 774 name: 发送者昵称 775 """ 776 self.name = name 777 self.array["data"]["name"] = str(name)
设置发送者昵称
Arguments:
- name: 发送者昵称
def
set_user_id(self, user_id):
779 def set_user_id(self, user_id): 780 """ 781 设置发送者 QQ 号 782 Args: 783 user_id: 发送者 QQ 号 784 """ 785 self.user_id = user_id 786 self.array["data"]["uin"] = str(user_id)
设置发送者 QQ 号
Arguments:
- user_id: 发送者 QQ 号
def
render(self, group_id: int | None = None):
788 def render(self, group_id: int | None = None): 789 if self.message_id is not None: 790 return f"[合并转发节点: {self.name}({self.user_id}): {self.message}]" 791 else: 792 return f"[合并转发节点: {self.message_id}]"
渲染消息段为字符串
Arguments:
- group_id: 群号(选填)
Returns:
渲染完毕的消息段
795class Music(Segment): 796 """ 797 音乐消息段 798 """ 799 segment_type = "music" 800 801 def __init__(self, type_, id_): 802 """ 803 Args: 804 type_: 音乐类型(可为qq 163 xm) 805 id_: 音乐 ID 806 """ 807 self.type = type_ 808 self.id = id_ 809 super().__init__({"type": "music", "data": {"type": str(self.type), "id": str(self.id)}}) 810 811 def set_type(self, type_): 812 """ 813 设置音乐类型 814 Args: 815 type_: 音乐类型(qq 163 xm) 816 """ 817 self.type = type_ 818 self.array["data"]["type"] = str(type_) 819 820 def set_id(self, id_): 821 """ 822 设置音乐 ID 823 Args: 824 id_: 音乐 ID 825 """ 826 self.id = id_ 827 self.array["data"]["id"] = str(id_)
音乐消息段
Music(type_, id_)
801 def __init__(self, type_, id_): 802 """ 803 Args: 804 type_: 音乐类型(可为qq 163 xm) 805 id_: 音乐 ID 806 """ 807 self.type = type_ 808 self.id = id_ 809 super().__init__({"type": "music", "data": {"type": str(self.type), "id": str(self.id)}})
Arguments:
- type_: 音乐类型(可为qq 163 xm)
- id_: 音乐 ID
def
set_type(self, type_):
811 def set_type(self, type_): 812 """ 813 设置音乐类型 814 Args: 815 type_: 音乐类型(qq 163 xm) 816 """ 817 self.type = type_ 818 self.array["data"]["type"] = str(type_)
设置音乐类型
Arguments:
- type_: 音乐类型(qq 163 xm)
830class CustomizeMusic(Segment): 831 """ 832 自定义音乐消息段 833 """ 834 segment_type = "music" 835 836 def __init__(self, url, audio, image, title="", content=""): 837 """ 838 Args: 839 url: 点击后跳转目标 URL 840 audio: 音乐 URL 841 image: 标题 842 title: 发送时可选,内容描述 843 content: 发送时可选,图片 URL 844 """ 845 self.url = url 846 self.audio = audio 847 self.image = image 848 self.title = title 849 self.content = content 850 super().__init__({"type": "music", "data": {"type": "custom", "url": str(self.url), "audio": str(self.audio), 851 "image": str(self.image)}}) 852 if title != "": 853 self.array["data"]["title"] = str(self.title) 854 855 if content != "": 856 self.array["data"]["content"] = str(self.content) 857 858 def set_url(self, url): 859 """ 860 设置 URL 861 Args: 862 url: 点击后跳转目标 URL 863 """ 864 self.url = url 865 self.array["data"]["url"] = str(url) 866 867 def set_audio(self, audio): 868 """ 869 设置音乐 URL 870 Args: 871 audio: 音乐 URL 872 """ 873 self.audio = audio 874 self.array["data"]["audio"] = str(audio) 875 876 def set_image(self, image): 877 """ 878 设置图片 URL 879 Args: 880 image: 图片 URL 881 """ 882 self.image = image 883 self.array["data"]["image"] = str(image) 884 885 def set_title(self, title): 886 """ 887 设置标题 888 Args: 889 title: 标题 890 """ 891 self.title = title 892 self.array["data"]["title"] = str(title) 893 894 def set_content(self, content): 895 """ 896 设置内容描述 897 Args: 898 content: 899 """ 900 self.content = content 901 self.array["data"]["content"] = str(content)
自定义音乐消息段
CustomizeMusic(url, audio, image, title='', content='')
836 def __init__(self, url, audio, image, title="", content=""): 837 """ 838 Args: 839 url: 点击后跳转目标 URL 840 audio: 音乐 URL 841 image: 标题 842 title: 发送时可选,内容描述 843 content: 发送时可选,图片 URL 844 """ 845 self.url = url 846 self.audio = audio 847 self.image = image 848 self.title = title 849 self.content = content 850 super().__init__({"type": "music", "data": {"type": "custom", "url": str(self.url), "audio": str(self.audio), 851 "image": str(self.image)}}) 852 if title != "": 853 self.array["data"]["title"] = str(self.title) 854 855 if content != "": 856 self.array["data"]["content"] = str(self.content)
Arguments:
- url: 点击后跳转目标 URL
- audio: 音乐 URL
- image: 标题
- title: 发送时可选,内容描述
- content: 发送时可选,图片 URL
def
set_url(self, url):
858 def set_url(self, url): 859 """ 860 设置 URL 861 Args: 862 url: 点击后跳转目标 URL 863 """ 864 self.url = url 865 self.array["data"]["url"] = str(url)
设置 URL
Arguments:
- url: 点击后跳转目标 URL
def
set_audio(self, audio):
867 def set_audio(self, audio): 868 """ 869 设置音乐 URL 870 Args: 871 audio: 音乐 URL 872 """ 873 self.audio = audio 874 self.array["data"]["audio"] = str(audio)
设置音乐 URL
Arguments:
- audio: 音乐 URL
def
set_image(self, image):
876 def set_image(self, image): 877 """ 878 设置图片 URL 879 Args: 880 image: 图片 URL 881 """ 882 self.image = image 883 self.array["data"]["image"] = str(image)
设置图片 URL
Arguments:
- image: 图片 URL
def
set_title(self, title):
885 def set_title(self, title): 886 """ 887 设置标题 888 Args: 889 title: 标题 890 """ 891 self.title = title 892 self.array["data"]["title"] = str(title)
设置标题
Arguments:
- title: 标题
904class Reply(Segment): 905 """ 906 回复消息段 907 """ 908 segment_type = "reply" 909 910 def __init__(self, message_id): 911 """ 912 Args: 913 message_id: 回复消息 ID 914 """ 915 self.message_id = message_id 916 super().__init__({"type": "reply", "data": {"id": str(self.message_id)}}) 917 918 def set_message_id(self, message_id): 919 """ 920 设置消息 ID 921 Args: 922 message_id: 消息 ID 923 """ 924 self.message_id = message_id 925 self.array["data"]["id"] = str(self.message_id) 926 927 def render(self, group_id: int | None = None): 928 return f"[回复: {self.message_id}]"
回复消息段
Reply(message_id)
910 def __init__(self, message_id): 911 """ 912 Args: 913 message_id: 回复消息 ID 914 """ 915 self.message_id = message_id 916 super().__init__({"type": "reply", "data": {"id": str(self.message_id)}})
Arguments:
- message_id: 回复消息 ID
def
set_message_id(self, message_id):
918 def set_message_id(self, message_id): 919 """ 920 设置消息 ID 921 Args: 922 message_id: 消息 ID 923 """ 924 self.message_id = message_id 925 self.array["data"]["id"] = str(self.message_id)
设置消息 ID
Arguments:
- message_id: 消息 ID
931class Forward(Segment): 932 """ 933 合并转发消息段 934 """ 935 segment_type = "forward" 936 937 def __init__(self, forward_id): 938 """ 939 Args: 940 forward_id: 合并转发消息 ID 941 """ 942 self.forward_id = forward_id 943 super().__init__({"type": "forward", "data": {"id": str(self.forward_id)}}) 944 945 def set_forward_id(self, forward_id): 946 """ 947 设置合并转发消息 ID 948 Args: 949 forward_id: 合并转发消息 ID 950 """ 951 self.forward_id = forward_id 952 self.array["data"]["id"] = str(self.forward_id) 953 954 def render(self, group_id: int | None = None): 955 return f"[合并转发: {self.forward_id}]"
合并转发消息段
Forward(forward_id)
937 def __init__(self, forward_id): 938 """ 939 Args: 940 forward_id: 合并转发消息 ID 941 """ 942 self.forward_id = forward_id 943 super().__init__({"type": "forward", "data": {"id": str(self.forward_id)}})
Arguments:
- forward_id: 合并转发消息 ID
def
set_forward_id(self, forward_id):
945 def set_forward_id(self, forward_id): 946 """ 947 设置合并转发消息 ID 948 Args: 949 forward_id: 合并转发消息 ID 950 """ 951 self.forward_id = forward_id 952 self.array["data"]["id"] = str(self.forward_id)
设置合并转发消息 ID
Arguments:
- forward_id: 合并转发消息 ID
958class XML(Segment): 959 """ 960 XML消息段 961 """ 962 segment_type = "xml" 963 964 def __init__(self, data): 965 self.xml_data = data 966 super().__init__({"type": "xml", "data": {"data": str(self.xml_data)}}) 967 968 def set_xml_data(self, data): 969 """ 970 设置xml数据 971 Args: 972 data: xml数据 973 """ 974 self.xml_data = data 975 self.array["data"]["data"] = str(self.xml_data)
XML消息段
978class JSON(Segment): 979 """ 980 JSON消息段 981 """ 982 segment_type = "json" 983 984 def __init__(self, data): 985 """ 986 Args: 987 data: JSON 内容 988 """ 989 self.json_data = data 990 super().__init__({"type": "json", "data": {"data": str(self.json_data)}}) 991 992 def set_json(self, data): 993 """ 994 设置json数据 995 Args: 996 data: json 内容 997 """ 998 self.json_data = data 999 self.array["data"]["data"] = str(self.json_data) 1000 1001 def get_json(self): 1002 """ 1003 获取json数据(自动序列化) 1004 Returns: 1005 json: json数据 1006 """ 1007 return json.loads(self.json_data)
JSON消息段
JSON(data)
984 def __init__(self, data): 985 """ 986 Args: 987 data: JSON 内容 988 """ 989 self.json_data = data 990 super().__init__({"type": "json", "data": {"data": str(self.json_data)}})
Arguments:
- data: JSON 内容
def
set_json(self, data):
992 def set_json(self, data): 993 """ 994 设置json数据 995 Args: 996 data: json 内容 997 """ 998 self.json_data = data 999 self.array["data"]["data"] = str(self.json_data)
设置json数据
Arguments:
- data: json 内容
class
QQRichText:
1010class QQRichText: 1011 """ 1012 QQ富文本 1013 """ 1014 1015 def __init__(self, *rich: str | dict | list | tuple | Segment): 1016 """ 1017 Args: 1018 *rich: 富文本内容,可为 str、dict、list、tuple、Segment、QQRichText 1019 """ 1020 1021 # 特判 1022 self.rich_array: list[Segment] = [] 1023 if len(rich) == 1: 1024 rich = rich[0] 1025 1026 # 识别输入的是CQCode or json形式的富文本 1027 # 如果输入的是CQCode,则转换为json形式的富文本 1028 1029 # 处理CQCode 1030 if isinstance(rich, str): 1031 rich_string = rich 1032 rich = cq_2_array(rich_string) 1033 1034 elif isinstance(rich, dict): 1035 rich = [rich] 1036 elif isinstance(rich, (list, tuple)): 1037 array = [] 1038 for item in rich: 1039 if isinstance(item, dict): 1040 array.append(item) 1041 elif isinstance(item, str): 1042 array += cq_2_array(item) 1043 else: 1044 for segment in segments: 1045 if isinstance(item, segment): 1046 array.append(item.array) 1047 break 1048 else: 1049 if isinstance(rich, QQRichText): 1050 array += rich.rich_array 1051 else: 1052 raise TypeError("QQRichText: 输入类型错误") 1053 rich = array 1054 else: 1055 for segment in segments: 1056 if isinstance(rich, segment): 1057 rich = [rich.array] 1058 break 1059 else: 1060 if isinstance(rich, QQRichText): 1061 rich = rich.rich_array 1062 else: 1063 raise TypeError("QQRichText: 输入类型错误") 1064 1065 # 将rich转换为的Segment 1066 for i in range(len(rich)): 1067 if rich[i]["type"] in segments_map: 1068 try: 1069 params = inspect.signature(segments_map[rich[i]["type"]]).parameters 1070 kwargs = {} 1071 for param in params: 1072 if param in rich[i]["data"]: 1073 kwargs[param] = rich[i]["data"][param] 1074 else: 1075 if rich[i]["type"] == "reply" and param == "message_id": 1076 kwargs[param] = rich[i]["data"].get("id") 1077 elif rich[i]["type"] == "face" and param == "face_id": 1078 kwargs[param] = rich[i]["data"].get("id") 1079 elif rich[i]["type"] == "forward" and param == "forward_id": 1080 kwargs[param] = rich[i]["data"].get("id") 1081 elif rich[i]["type"] == "poke" and param == "poke_id": 1082 kwargs[param] = rich[i]["data"].get("id") 1083 elif param == "id_": 1084 kwargs[param] = rich[i]["data"].get("id") 1085 elif param == "type_": 1086 kwargs[param] = rich[i]["data"].get("type") 1087 else: 1088 if params[param].default != params[param].empty: 1089 kwargs[param] = params[param].default 1090 segment = segments_map[rich[i]["type"]](**kwargs) 1091 # 检查原cq中是否含有不在segment的data中的参数 1092 for k, v in rich[i]["data"].items(): 1093 if k not in segment["data"]: 1094 segment.set_data(k, v) 1095 rich[i] = segment 1096 except Exception as e: 1097 if ConfigManager.GlobalConfig().debug.save_dump: 1098 dump_path = save_exc_dump(f"转换{rich[i]}时失败") 1099 else: 1100 dump_path = None 1101 Logger.get_logger().warning(f"转换{rich[i]}时失败,报错信息: {repr(e)}\n" 1102 f"{traceback.format_exc()}" 1103 f"{f"\n已保存异常到 {dump_path}" if dump_path else ""}") 1104 rich[i] = Segment(rich[i]) 1105 else: 1106 rich[i] = Segment(rich[i]) 1107 1108 self.rich_array: list[Segment] = rich 1109 1110 def render(self, group_id: int | None = None): 1111 """ 1112 渲染消息(调用rich_array下所有消息段的render方法拼接) 1113 Args: 1114 group_id: 群号,选填,可优化效果 1115 """ 1116 text = "" 1117 for rich in self.rich_array: 1118 text += rich.render(group_id=group_id) 1119 return text 1120 1121 def __str__(self): 1122 self.rich_string = array_2_cq(self.rich_array) 1123 return self.rich_string 1124 1125 def __repr__(self): 1126 return self.__str__() 1127 1128 def __getitem__(self, index): 1129 return self.rich_array[index] 1130 1131 def __add__(self, other): 1132 other = QQRichText(other) 1133 return QQRichText(self.rich_array + other.rich_array) 1134 1135 def __eq__(self, other): 1136 other = QQRichText(other) 1137 1138 return self.rich_array == other.rich_array 1139 1140 def __contains__(self, other): 1141 if isinstance(other, QQRichText): 1142 return all(item in self.rich_array for item in other.rich_array) 1143 else: 1144 try: 1145 return str(other) in str(self) 1146 except (TypeError, AttributeError): 1147 return False 1148 1149 def get_array(self): 1150 """ 1151 获取rich_array(非抽象类,可用于API调用等) 1152 Returns: 1153 rich_array 1154 """ 1155 return [array.array for array in self.rich_array] 1156 1157 def add(self, *segments): 1158 """ 1159 添加消息段 1160 Args: 1161 *segments: 消息段 1162 1163 Returns: 1164 self 1165 """ 1166 for segment in segments: 1167 if isinstance(segment, Segment): 1168 self.rich_array.append(segment) 1169 else: 1170 self.rich_array += QQRichText(segment).rich_array 1171 return self
QQ富文本
QQRichText(*rich: str | dict | list | tuple | Segment)
1015 def __init__(self, *rich: str | dict | list | tuple | Segment): 1016 """ 1017 Args: 1018 *rich: 富文本内容,可为 str、dict、list、tuple、Segment、QQRichText 1019 """ 1020 1021 # 特判 1022 self.rich_array: list[Segment] = [] 1023 if len(rich) == 1: 1024 rich = rich[0] 1025 1026 # 识别输入的是CQCode or json形式的富文本 1027 # 如果输入的是CQCode,则转换为json形式的富文本 1028 1029 # 处理CQCode 1030 if isinstance(rich, str): 1031 rich_string = rich 1032 rich = cq_2_array(rich_string) 1033 1034 elif isinstance(rich, dict): 1035 rich = [rich] 1036 elif isinstance(rich, (list, tuple)): 1037 array = [] 1038 for item in rich: 1039 if isinstance(item, dict): 1040 array.append(item) 1041 elif isinstance(item, str): 1042 array += cq_2_array(item) 1043 else: 1044 for segment in segments: 1045 if isinstance(item, segment): 1046 array.append(item.array) 1047 break 1048 else: 1049 if isinstance(rich, QQRichText): 1050 array += rich.rich_array 1051 else: 1052 raise TypeError("QQRichText: 输入类型错误") 1053 rich = array 1054 else: 1055 for segment in segments: 1056 if isinstance(rich, segment): 1057 rich = [rich.array] 1058 break 1059 else: 1060 if isinstance(rich, QQRichText): 1061 rich = rich.rich_array 1062 else: 1063 raise TypeError("QQRichText: 输入类型错误") 1064 1065 # 将rich转换为的Segment 1066 for i in range(len(rich)): 1067 if rich[i]["type"] in segments_map: 1068 try: 1069 params = inspect.signature(segments_map[rich[i]["type"]]).parameters 1070 kwargs = {} 1071 for param in params: 1072 if param in rich[i]["data"]: 1073 kwargs[param] = rich[i]["data"][param] 1074 else: 1075 if rich[i]["type"] == "reply" and param == "message_id": 1076 kwargs[param] = rich[i]["data"].get("id") 1077 elif rich[i]["type"] == "face" and param == "face_id": 1078 kwargs[param] = rich[i]["data"].get("id") 1079 elif rich[i]["type"] == "forward" and param == "forward_id": 1080 kwargs[param] = rich[i]["data"].get("id") 1081 elif rich[i]["type"] == "poke" and param == "poke_id": 1082 kwargs[param] = rich[i]["data"].get("id") 1083 elif param == "id_": 1084 kwargs[param] = rich[i]["data"].get("id") 1085 elif param == "type_": 1086 kwargs[param] = rich[i]["data"].get("type") 1087 else: 1088 if params[param].default != params[param].empty: 1089 kwargs[param] = params[param].default 1090 segment = segments_map[rich[i]["type"]](**kwargs) 1091 # 检查原cq中是否含有不在segment的data中的参数 1092 for k, v in rich[i]["data"].items(): 1093 if k not in segment["data"]: 1094 segment.set_data(k, v) 1095 rich[i] = segment 1096 except Exception as e: 1097 if ConfigManager.GlobalConfig().debug.save_dump: 1098 dump_path = save_exc_dump(f"转换{rich[i]}时失败") 1099 else: 1100 dump_path = None 1101 Logger.get_logger().warning(f"转换{rich[i]}时失败,报错信息: {repr(e)}\n" 1102 f"{traceback.format_exc()}" 1103 f"{f"\n已保存异常到 {dump_path}" if dump_path else ""}") 1104 rich[i] = Segment(rich[i]) 1105 else: 1106 rich[i] = Segment(rich[i]) 1107 1108 self.rich_array: list[Segment] = rich
Arguments:
- *rich: 富文本内容,可为 str、dict、list、tuple、Segment、QQRichText
rich_array: list[Segment]
def
render(self, group_id: int | None = None):
1110 def render(self, group_id: int | None = None): 1111 """ 1112 渲染消息(调用rich_array下所有消息段的render方法拼接) 1113 Args: 1114 group_id: 群号,选填,可优化效果 1115 """ 1116 text = "" 1117 for rich in self.rich_array: 1118 text += rich.render(group_id=group_id) 1119 return text
渲染消息(调用rich_array下所有消息段的render方法拼接)
Arguments:
- group_id: 群号,选填,可优化效果
def
get_array(self):
1149 def get_array(self): 1150 """ 1151 获取rich_array(非抽象类,可用于API调用等) 1152 Returns: 1153 rich_array 1154 """ 1155 return [array.array for array in self.rich_array]
获取rich_array(非抽象类,可用于API调用等)
Returns:
rich_array
def
add(self, *segments):
1157 def add(self, *segments): 1158 """ 1159 添加消息段 1160 Args: 1161 *segments: 消息段 1162 1163 Returns: 1164 self 1165 """ 1166 for segment in segments: 1167 if isinstance(segment, Segment): 1168 self.rich_array.append(segment) 1169 else: 1170 self.rich_array += QQRichText(segment).rich_array 1171 return self
添加消息段
Arguments:
- *segments: 消息段
Returns:
self