跳转至

chain

Ariadne 消息链的实现

MessageChain 🔗

Bases: BaseMessageChain, AriadneBaseModel

即 "消息链", 被用于承载整个消息内容的数据结构, 包含有一有序列表, 包含有元素实例.

Source code in graia/ariadne/message/chain.py
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
class MessageChain(BaseMessageChain, AriadneBaseModel):
    """
    即 "消息链", 被用于承载整个消息内容的数据结构, 包含有一有序列表, 包含有元素实例.
    """

    __root__: List[Element]
    """底层元素列表"""

    @property
    def content(self) -> List[Element]:
        """Amnesia MessageChain 的内容代理"""
        return self.__root__

    @staticmethod
    def build_chain(obj: Union[List[Dict], MessageContainer]) -> List[Element]:
        """内部接口, 会自动反序列化对象并生成.

        Args:
            obj (_Parsable): 需要反序列化的对象

        Returns:
            List[Element]: 内部承载有尽量有效的消息元素的列表
        """
        # single object
        if isinstance(obj, MessageChain):
            return deepcopy(obj.content)
        if isinstance(obj, Element):
            return [obj]
        if isinstance(obj, str):
            return [Plain(obj)]
        element_list: List[Element] = []
        for o in obj:
            if isinstance(o, dict):
                if typ := ELEMENT_MAPPING.get(o.get("type", "Unknown")):
                    element_list.append(typ.parse_obj(o))
            else:
                element_list.extend(MessageChain.build_chain(o))

        special_cnt: int = sum(element.__class__ not in ORDINARY_ELEMENT_TYPES for element in element_list)

        if special_cnt > 1:
            raise ValueError("An MessageChain can only contain *one* special element")
        return element_list

    @classmethod
    def parse_obj(cls: Type[Self], obj: Union[List[dict], List[Element]]) -> Self:
        """解析 MessageChain.

        Args:
            obj (List[T]): 需要反序列化的对象

        Returns:
            MessageChain: 内部承载有尽量有效的消息元素的消息链
        """
        return cls(cls.build_chain(obj), inline=True)

    @overload
    def __init__(self, __root__: Sequence[Element], *, inline: Literal[True]) -> None:
        ...

    @overload
    def __init__(self, *elements: MessageContainer, inline: Literal[False] = False) -> None:
        ...

    def __init__(
        self,
        __root__: MessageContainer,
        *elements: MessageContainer,
        inline: bool = False,
    ) -> None:
        """
        创建消息链.

        Args:
            *elements (Union[Iterable[Element], Element, str]): \
            元素的容器, 为承载元素的可迭代对象/单元素实例, \
            字符串会被自动不可逆的转换为 `Plain`

        Returns:
            MessageChain: 创建的消息链
        """
        if not inline:
            AriadneBaseModel.__init__(
                self,
                __root__=self.build_chain((__root__, *elements)),
            )
        else:
            AriadneBaseModel.__init__(self, __root__=[])
            self.__root__ = __root__  # type: ignore

    def __repr_args__(self) -> "ReprArgs":
        return [(None, list(self.content))]

    @overload
    def __getitem__(self, item: Tuple[Type[Element_T], int]) -> List[Element_T]:
        ...

    @overload
    def __getitem__(self, item: Type[Element_T]) -> List[Element_T]:
        ...

    @overload
    def __getitem__(self, item: int) -> Element:
        ...

    @overload
    def __getitem__(self, item: slice) -> Self:
        ...

    def __getitem__(self, item: Union[Tuple[Type[Element], int], Type[Element], int, slice]) -> Any:
        """
        可通过切片取出子消息链, 或元素.

        通过 `type, count` 型元组取出前 `count` 个 `type` 元素组成的列表

        通过 `type` 取出属于 `type` 的元素列表

        通过 `int` 取出对应位置元素.

        Args:
            item (Union[Tuple[Type[Element], int], Type[Element], int, slice]): 索引项
        Returns:
            Union[List[Element], Element, MessageChain]: 索引结果.
        """
        if isinstance(item, tuple):
            return self.get(*item)
        return super().__getitem__(item)

    def as_sendable(self) -> Self:
        """将消息链转换为可发送形式 (去除 File)

        Returns:
            MessageChain: 转换后的消息链.
        """
        return self.exclude(File)

    def get(self, element_class: Type[Element_T], count: int = -1) -> List[Element_T]:
        res = super().get(element_class, count)
        if isinstance(res, (Quote, Source)):
            raise IndexError(
                "MessageChain.get(Quote/Source) is removed in 0.10.0.\n"
                "See https://github.com/GraiaProject/Ariadne/blob/dev/CHANGELOG.md#095",
            )
        return res

    def __eq__(self, other: Union[MessageContainer, Self]) -> bool:
        if id(self) == id(other):
            return True
        if not isinstance(other, (MessageChain, list)):
            return False
        if not isinstance(other, MessageChain):
            other = MessageChain(other)
        return other.content == self.content

    def __mul__(self, time: int) -> Self:
        result = []
        for _ in range(time):
            result.extend(deepcopy(self.content))
        return MessageChain(result, inline=True)

    def __imul__(self, time: int) -> Self:
        result = []
        for _ in range(time):
            result.extend(deepcopy(self.content))
        self.content.clear()
        self.content.extend(result)
        return self

    def __len__(self) -> int:
        return len(self.content)

    def as_persistent_string(
        self,
        *,
        binary: bool = True,
        include: Iterable[Type[Element]] = (),
        exclude: Iterable[Type[Element]] = (),
    ) -> str:
        """转换为持久化字符串.

        Args:
            binary (bool, optional): 是否附带图片或声音的二进制. 默认为 True.
            include (Iterable[Type[Element]], optional): 筛选, 只包含本参数提供的元素类型.
            exclude (Iterable[Type[Element]], optional): 筛选, 排除本参数提供的元素类型.

        Raises:
            ValueError: 同时提供 include 与 exclude

        Returns:
            str: 持久化字符串. 不是 Mirai Code.
        """
        string_list = []
        include = tuple(include)
        exclude = tuple(exclude)
        if include and exclude:
            raise ValueError("Can not present include and exclude at same time!")
        for i in self.content:
            if (
                (include and isinstance(i, include))
                or (exclude and not isinstance(i, exclude))
                or not (include or exclude)
            ):
                if isinstance(i, Plain):
                    string_list.append(i.as_persistent_string().replace("[", "[_"))
                elif not isinstance(i, MultimediaElement):
                    string_list.append(i.as_persistent_string())
                else:
                    string_list.append(i.as_persistent_string(binary=binary))
        return "".join(string_list)

    async def download_binary(self) -> Self:
        """下载消息中所有的二进制数据并保存在元素实例内"""
        for elem in self.content:
            if isinstance(elem, MultimediaElement):
                await elem.get_bytes()
        return self

    @classmethod
    def from_persistent_string(cls, string: str) -> Self:
        """从持久化字符串生成消息链.

        Returns:
            MessageChain: 还原的消息链.
        """
        result: List[Element] = []
        for match in re.split(r"(\[mirai:.+?\])", string):
            if mirai := re.fullmatch(r"\[mirai:(.+?)(:(.+?))\]", match):
                j_string = mirai[3]
                element_cls = ELEMENT_MAPPING[mirai[1]]
                result.append(element_cls.parse_obj(Json.deserialize(unescape_bracket(j_string))))
            elif match:
                result.append(Plain(match.replace("[_", "[")))
        return MessageChain(result, inline=True)

    def _to_mapping_str(
        self,
        *,
        remove_source: bool = True,
        remove_quote: bool = True,
        remove_extra_space: bool = False,
    ) -> Tuple[str, Dict[str, Element]]:
        """转换消息链为映射字符串与映射字典的元组.

        Args:
            remove_source (bool, optional): 是否移除消息链中的 Source 元素. 默认为 True.
            remove_quote (bool, optional): 是否移除消息链中的 Quote 元素. 默认为 True.
            remove_extra_space (bool, optional): 是否移除 Quote At AtAll 的多余空格. 默认为 False.

        Returns:
            Tuple[str, Dict[str, Element]]: 生成的映射字符串与映射字典的元组
        """
        elem_mapping: Dict[str, Element] = {}
        elem_str_list: List[str] = []
        for i, elem in enumerate(self.content):
            if not isinstance(elem, Plain):
                if remove_quote and isinstance(elem, Quote):
                    continue
                if remove_source and isinstance(elem, Source):
                    continue
                elem_mapping[str(i)] = elem
                elem_str_list.append(f"\x02{i}_{elem.type}\x03")
            elif (
                remove_extra_space
                and i  # not first element
                and isinstance(
                    self.content[i - 1], (Quote, At, AtAll)
                )  # following elements which have an dumb trailing space
                and elem.text.startswith("  ")  # extra space (count >= 2)
            ):
                elem_str_list.append(elem.text[1:])
            else:
                elem_str_list.append(elem.text)
        return "".join(elem_str_list), elem_mapping

    __element_pattern = re.compile("(\x02\\w+\x03)")

    @classmethod
    def _from_mapping_string(cls, string: str, mapping: Dict[str, Element]) -> Self:
        """从映射字符串与映射字典的元组还原消息链.

        Args:
            string (str): 映射字符串
            mapping (Dict[int, Element]): 映射字典.

        Returns:
            MessageChain: 构建的消息链
        """
        elements: List[Element] = []
        for x in cls.__element_pattern.split(string):
            if x:
                if x[0] == "\x02" and x[-1] == "\x03":
                    index, class_name = x[1:-1].split("_")
                    if not isinstance(mapping[index], ELEMENT_MAPPING[class_name]):
                        raise ValueError("Validation failed: not matching element type!")
                    elements.append(mapping[index])
                else:
                    elements.append(Plain(x))
        return cls(elements, inline=True)

    def removeprefix(self, prefix: str, *, copy: bool = True, skip_header: bool = True) -> Self:
        """移除消息链前缀.

        Args:
            prefix (str): 要移除的前缀.
            copy (bool, optional): 是否在副本上修改, 默认为 True.
            skip_header (bool, optional): 是否要忽略 Source 与 Quote 类型查找, 默认为 True. (移除后仍会带上 Source 与 Quote)

        Returns:
            MessageChain: 修改后的消息链, 若未移除则原样返回.
        """
        header = []
        elements = []
        if not skip_header:
            elements = self.content[:]
        else:
            for element in self.content:
                if isinstance(element, (Quote, Source)):
                    header.append(element)
                else:
                    elements.append(element)
        if copy:
            header = deepcopy(header)
            elements = deepcopy(elements)
        if not elements or not isinstance(elements[0], Plain):
            return self.copy() if copy else self
        if elements[0].text.startswith(prefix):
            elements[0].text = elements[0].text[len(prefix) :]
        if copy:
            return MessageChain(header + elements, inline=True)
        self.content.clear()
        self.content.extend(header + elements)
        return self

    def removesuffix(self, suffix: str, *, copy: bool = True) -> Self:
        """移除消息链后缀.

        Args:
            suffix (str): 要移除的后缀.
            copy (bool, optional): 是否在副本上修改, 默认为 True.

        Returns:
            MessageChain: 修改后的消息链, 若未移除则原样返回.
        """
        elements = deepcopy(self.content) if copy else self.content
        if not elements or not isinstance(elements[-1], Plain):
            return self.copy() if copy else self
        last_elem: Plain = elements[-1]
        if last_elem.text.endswith(suffix):
            last_elem.text = last_elem.text[: -len(suffix)]
        if copy:
            return MessageChain(elements, inline=True)
        self.content.clear()
        self.content.extend(elements)
        return self

    def replace(
        self,
        old: MessageContainer,
        new: MessageContainer,
    ) -> Self:
        """替换消息链中的一部分. (在副本上操作)

        Args:
            old (MessageChain): 要替换的消息链.
            new (MessageChain): 替换后的消息链.

        Returns:
            MessageChain: 修改后的消息链, 若未替换则原样返回.
        """
        if not isinstance(old, MessageChain):
            old = MessageChain(old)
        if not isinstance(new, MessageChain):
            new = MessageChain(new)
        index_list: List[int] = self.index_sub(old)

        def unzip(chain: Self) -> List[Union[str, Element]]:
            unzipped: List[Union[str, Element]] = []
            for e in chain.content:
                if isinstance(e, Plain):
                    unzipped.extend(e.text)
                else:
                    unzipped.append(e)
            return unzipped

        unzipped_new: List[Union[str, Element]] = unzip(new)
        unzipped_old: List[Union[str, Element]] = unzip(old)
        unzipped_self: List[Union[str, Element]] = unzip(self)
        unzipped_result: List[Union[str, Element]] = []
        last_end: int = 0
        for start in index_list:
            unzipped_result.extend(unzipped_self[last_end:start])
            last_end = start + len(unzipped_old)
            unzipped_result.extend(unzipped_new)
        unzipped_result.extend(unzipped_self[last_end:])
        # Merge result
        result_list: List[Element] = []
        char_stk: List[str] = []
        for v in unzipped_result:
            if isinstance(v, str):
                char_stk.append(v)
            else:
                result_list.append(Plain("".join(char_stk)))
                char_stk = []
                result_list.append(v)
        if char_stk:
            result_list.append(Plain("".join(char_stk)))
        return MessageChain(result_list, inline=True)

    @property
    def display(self) -> str:
        """获取消息链的显示字符串.

        Returns:
            str: 消息链的显示字符串.
        """
        return str(self)

    @property
    def safe_display(self) -> str:
        """获取消息链的安全显示字符串. (对特殊字符进行转义了)

        Returns:
            str: 消息链的安全显示字符串.
        """
        return repr(str(self))[1:-1]

    def __hash__(self) -> int:
        return id(self)

__root__ class-attribute 🔗

__root__: List[Element]

底层元素列表

__getitem__ 🔗

__getitem__(
    item: Union[Tuple[Type[Element], int], Type[Element], int, slice]
) -> Any

可通过切片取出子消息链, 或元素.

通过 type, count 型元组取出前 counttype 元素组成的列表

通过 type 取出属于 type 的元素列表

通过 int 取出对应位置元素.

Parameters:

Returns:

  • Any

    Union[List[Element], Element, MessageChain]: 索引结果.

Source code in graia/ariadne/message/chain.py
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def __getitem__(self, item: Union[Tuple[Type[Element], int], Type[Element], int, slice]) -> Any:
    """
    可通过切片取出子消息链, 或元素.

    通过 `type, count` 型元组取出前 `count` 个 `type` 元素组成的列表

    通过 `type` 取出属于 `type` 的元素列表

    通过 `int` 取出对应位置元素.

    Args:
        item (Union[Tuple[Type[Element], int], Type[Element], int, slice]): 索引项
    Returns:
        Union[List[Element], Element, MessageChain]: 索引结果.
    """
    if isinstance(item, tuple):
        return self.get(*item)
    return super().__getitem__(item)

__init__ 🔗

__init__(
    __root__: MessageContainer,
    *elements: MessageContainer,
    inline: bool = False
) -> None

创建消息链.

Parameters:

  • *elements (Union[Iterable[Element], Element, str]) –

    元素的容器, 为承载元素的可迭代对象/单元素实例, 字符串会被自动不可逆的转换为 Plain

Returns:

  • MessageChain( None ) –

    创建的消息链

Source code in graia/ariadne/message/chain.py
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def __init__(
    self,
    __root__: MessageContainer,
    *elements: MessageContainer,
    inline: bool = False,
) -> None:
    """
    创建消息链.

    Args:
        *elements (Union[Iterable[Element], Element, str]): \
        元素的容器, 为承载元素的可迭代对象/单元素实例, \
        字符串会被自动不可逆的转换为 `Plain`

    Returns:
        MessageChain: 创建的消息链
    """
    if not inline:
        AriadneBaseModel.__init__(
            self,
            __root__=self.build_chain((__root__, *elements)),
        )
    else:
        AriadneBaseModel.__init__(self, __root__=[])
        self.__root__ = __root__  # type: ignore

as_persistent_string 🔗

as_persistent_string(
    *,
    binary: bool = True,
    include: Iterable[Type[Element]] = (),
    exclude: Iterable[Type[Element]] = ()
) -> str

转换为持久化字符串.

Parameters:

  • binary (bool) –

    是否附带图片或声音的二进制. 默认为 True.

  • include (Iterable[Type[Element]]) –

    筛选, 只包含本参数提供的元素类型.

  • exclude (Iterable[Type[Element]]) –

    筛选, 排除本参数提供的元素类型.

Raises:

Returns:

  • str( str ) –

    持久化字符串. 不是 Mirai Code.

Source code in graia/ariadne/message/chain.py
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
def as_persistent_string(
    self,
    *,
    binary: bool = True,
    include: Iterable[Type[Element]] = (),
    exclude: Iterable[Type[Element]] = (),
) -> str:
    """转换为持久化字符串.

    Args:
        binary (bool, optional): 是否附带图片或声音的二进制. 默认为 True.
        include (Iterable[Type[Element]], optional): 筛选, 只包含本参数提供的元素类型.
        exclude (Iterable[Type[Element]], optional): 筛选, 排除本参数提供的元素类型.

    Raises:
        ValueError: 同时提供 include 与 exclude

    Returns:
        str: 持久化字符串. 不是 Mirai Code.
    """
    string_list = []
    include = tuple(include)
    exclude = tuple(exclude)
    if include and exclude:
        raise ValueError("Can not present include and exclude at same time!")
    for i in self.content:
        if (
            (include and isinstance(i, include))
            or (exclude and not isinstance(i, exclude))
            or not (include or exclude)
        ):
            if isinstance(i, Plain):
                string_list.append(i.as_persistent_string().replace("[", "[_"))
            elif not isinstance(i, MultimediaElement):
                string_list.append(i.as_persistent_string())
            else:
                string_list.append(i.as_persistent_string(binary=binary))
    return "".join(string_list)

as_sendable 🔗

as_sendable() -> Self

将消息链转换为可发送形式 (去除 File)

Returns:

  • MessageChain( Self ) –

    转换后的消息链.

Source code in graia/ariadne/message/chain.py
183
184
185
186
187
188
189
def as_sendable(self) -> Self:
    """将消息链转换为可发送形式 (去除 File)

    Returns:
        MessageChain: 转换后的消息链.
    """
    return self.exclude(File)

build_chain staticmethod 🔗

build_chain(obj: Union[List[Dict], MessageContainer]) -> List[Element]

内部接口, 会自动反序列化对象并生成.

Parameters:

  • obj (_Parsable) –

    需要反序列化的对象

Returns:

  • List[Element]

    List[Element]: 内部承载有尽量有效的消息元素的列表

Source code in graia/ariadne/message/chain.py
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
@staticmethod
def build_chain(obj: Union[List[Dict], MessageContainer]) -> List[Element]:
    """内部接口, 会自动反序列化对象并生成.

    Args:
        obj (_Parsable): 需要反序列化的对象

    Returns:
        List[Element]: 内部承载有尽量有效的消息元素的列表
    """
    # single object
    if isinstance(obj, MessageChain):
        return deepcopy(obj.content)
    if isinstance(obj, Element):
        return [obj]
    if isinstance(obj, str):
        return [Plain(obj)]
    element_list: List[Element] = []
    for o in obj:
        if isinstance(o, dict):
            if typ := ELEMENT_MAPPING.get(o.get("type", "Unknown")):
                element_list.append(typ.parse_obj(o))
        else:
            element_list.extend(MessageChain.build_chain(o))

    special_cnt: int = sum(element.__class__ not in ORDINARY_ELEMENT_TYPES for element in element_list)

    if special_cnt > 1:
        raise ValueError("An MessageChain can only contain *one* special element")
    return element_list

content property 🔗

content() -> List[Element]

Amnesia MessageChain 的内容代理

Source code in graia/ariadne/message/chain.py
63
64
65
66
@property
def content(self) -> List[Element]:
    """Amnesia MessageChain 的内容代理"""
    return self.__root__

display property 🔗

display() -> str

获取消息链的显示字符串.

Returns:

  • str( str ) –

    消息链的显示字符串.

Source code in graia/ariadne/message/chain.py
463
464
465
466
467
468
469
470
@property
def display(self) -> str:
    """获取消息链的显示字符串.

    Returns:
        str: 消息链的显示字符串.
    """
    return str(self)

download_binary async 🔗

download_binary() -> Self

下载消息中所有的二进制数据并保存在元素实例内

Source code in graia/ariadne/message/chain.py
265
266
267
268
269
270
async def download_binary(self) -> Self:
    """下载消息中所有的二进制数据并保存在元素实例内"""
    for elem in self.content:
        if isinstance(elem, MultimediaElement):
            await elem.get_bytes()
    return self

from_persistent_string classmethod 🔗

from_persistent_string(string: str) -> Self

从持久化字符串生成消息链.

Returns:

  • MessageChain( Self ) –

    还原的消息链.

Source code in graia/ariadne/message/chain.py
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
@classmethod
def from_persistent_string(cls, string: str) -> Self:
    """从持久化字符串生成消息链.

    Returns:
        MessageChain: 还原的消息链.
    """
    result: List[Element] = []
    for match in re.split(r"(\[mirai:.+?\])", string):
        if mirai := re.fullmatch(r"\[mirai:(.+?)(:(.+?))\]", match):
            j_string = mirai[3]
            element_cls = ELEMENT_MAPPING[mirai[1]]
            result.append(element_cls.parse_obj(Json.deserialize(unescape_bracket(j_string))))
        elif match:
            result.append(Plain(match.replace("[_", "[")))
    return MessageChain(result, inline=True)

parse_obj classmethod 🔗

parse_obj(obj: Union[List[dict], List[Element]]) -> Self

解析 MessageChain.

Parameters:

  • obj (List[T]) –

    需要反序列化的对象

Returns:

  • MessageChain( Self ) –

    内部承载有尽量有效的消息元素的消息链

Source code in graia/ariadne/message/chain.py
 99
100
101
102
103
104
105
106
107
108
109
@classmethod
def parse_obj(cls: Type[Self], obj: Union[List[dict], List[Element]]) -> Self:
    """解析 MessageChain.

    Args:
        obj (List[T]): 需要反序列化的对象

    Returns:
        MessageChain: 内部承载有尽量有效的消息元素的消息链
    """
    return cls(cls.build_chain(obj), inline=True)

removeprefix 🔗

removeprefix(
    prefix: str, *, copy: bool = True, skip_header: bool = True
) -> Self

移除消息链前缀.

Parameters:

  • prefix (str) –

    要移除的前缀.

  • copy (bool) –

    是否在副本上修改, 默认为 True.

  • skip_header (bool) –

    是否要忽略 Source 与 Quote 类型查找, 默认为 True. (移除后仍会带上 Source 与 Quote)

Returns:

  • MessageChain( Self ) –

    修改后的消息链, 若未移除则原样返回.

Source code in graia/ariadne/message/chain.py
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
def removeprefix(self, prefix: str, *, copy: bool = True, skip_header: bool = True) -> Self:
    """移除消息链前缀.

    Args:
        prefix (str): 要移除的前缀.
        copy (bool, optional): 是否在副本上修改, 默认为 True.
        skip_header (bool, optional): 是否要忽略 Source 与 Quote 类型查找, 默认为 True. (移除后仍会带上 Source 与 Quote)

    Returns:
        MessageChain: 修改后的消息链, 若未移除则原样返回.
    """
    header = []
    elements = []
    if not skip_header:
        elements = self.content[:]
    else:
        for element in self.content:
            if isinstance(element, (Quote, Source)):
                header.append(element)
            else:
                elements.append(element)
    if copy:
        header = deepcopy(header)
        elements = deepcopy(elements)
    if not elements or not isinstance(elements[0], Plain):
        return self.copy() if copy else self
    if elements[0].text.startswith(prefix):
        elements[0].text = elements[0].text[len(prefix) :]
    if copy:
        return MessageChain(header + elements, inline=True)
    self.content.clear()
    self.content.extend(header + elements)
    return self

removesuffix 🔗

removesuffix(suffix: str, *, copy: bool = True) -> Self

移除消息链后缀.

Parameters:

  • suffix (str) –

    要移除的后缀.

  • copy (bool) –

    是否在副本上修改, 默认为 True.

Returns:

  • MessageChain( Self ) –

    修改后的消息链, 若未移除则原样返回.

Source code in graia/ariadne/message/chain.py
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
def removesuffix(self, suffix: str, *, copy: bool = True) -> Self:
    """移除消息链后缀.

    Args:
        suffix (str): 要移除的后缀.
        copy (bool, optional): 是否在副本上修改, 默认为 True.

    Returns:
        MessageChain: 修改后的消息链, 若未移除则原样返回.
    """
    elements = deepcopy(self.content) if copy else self.content
    if not elements or not isinstance(elements[-1], Plain):
        return self.copy() if copy else self
    last_elem: Plain = elements[-1]
    if last_elem.text.endswith(suffix):
        last_elem.text = last_elem.text[: -len(suffix)]
    if copy:
        return MessageChain(elements, inline=True)
    self.content.clear()
    self.content.extend(elements)
    return self

replace 🔗

replace(old: MessageContainer, new: MessageContainer) -> Self

替换消息链中的一部分. (在副本上操作)

Parameters:

Returns:

  • MessageChain( Self ) –

    修改后的消息链, 若未替换则原样返回.

Source code in graia/ariadne/message/chain.py
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
def replace(
    self,
    old: MessageContainer,
    new: MessageContainer,
) -> Self:
    """替换消息链中的一部分. (在副本上操作)

    Args:
        old (MessageChain): 要替换的消息链.
        new (MessageChain): 替换后的消息链.

    Returns:
        MessageChain: 修改后的消息链, 若未替换则原样返回.
    """
    if not isinstance(old, MessageChain):
        old = MessageChain(old)
    if not isinstance(new, MessageChain):
        new = MessageChain(new)
    index_list: List[int] = self.index_sub(old)

    def unzip(chain: Self) -> List[Union[str, Element]]:
        unzipped: List[Union[str, Element]] = []
        for e in chain.content:
            if isinstance(e, Plain):
                unzipped.extend(e.text)
            else:
                unzipped.append(e)
        return unzipped

    unzipped_new: List[Union[str, Element]] = unzip(new)
    unzipped_old: List[Union[str, Element]] = unzip(old)
    unzipped_self: List[Union[str, Element]] = unzip(self)
    unzipped_result: List[Union[str, Element]] = []
    last_end: int = 0
    for start in index_list:
        unzipped_result.extend(unzipped_self[last_end:start])
        last_end = start + len(unzipped_old)
        unzipped_result.extend(unzipped_new)
    unzipped_result.extend(unzipped_self[last_end:])
    # Merge result
    result_list: List[Element] = []
    char_stk: List[str] = []
    for v in unzipped_result:
        if isinstance(v, str):
            char_stk.append(v)
        else:
            result_list.append(Plain("".join(char_stk)))
            char_stk = []
            result_list.append(v)
    if char_stk:
        result_list.append(Plain("".join(char_stk)))
    return MessageChain(result_list, inline=True)

safe_display property 🔗

safe_display() -> str

获取消息链的安全显示字符串. (对特殊字符进行转义了)

Returns:

  • str( str ) –

    消息链的安全显示字符串.

Source code in graia/ariadne/message/chain.py
472
473
474
475
476
477
478
479
@property
def safe_display(self) -> str:
    """获取消息链的安全显示字符串. (对特殊字符进行转义了)

    Returns:
        str: 消息链的安全显示字符串.
    """
    return repr(str(self))[1:-1]