""" 无业务状态的文本识别、清理、转换和分段能力。 """ import html import random import re from collections.abc import Mapping from typing import Dict, Generator, List, Optional, TypedDict, Union, overload import moviepilot_rust def _escape_notification_text(value: str, text_format: str) -> str: """ 把外部文本转为指定语法的数据,避免注入格式、链接或群体提醒。 """ value = value.replace("@", "@\u200b") if text_format == "markdown": value = value.replace("<", "<").replace(">", ">") return re.sub(r"[\\`*_{}\[\]()#+.!|~-]", lambda match: "\\" + match[0], value) value = html.escape(value, quote=False) if text_format == "mrkdwn": return value.translate({ord("*"): "*", ord("_"): "_", ord("~"): "~", ord("`"): "ˋ"}) return re.sub(r"[\\`*_{}\[\]()#+.!|~-]", lambda match: f"&#{ord(match[0])};", value) def render_notification(title: Optional[str], blocks: List[Dict[str, Optional[str]]], text_format: str) -> str: """ 把通用文本段落转换为指定格式,不读取渠道配置或识别业务含义。 :param title: 通知总标题,仅在 rich_markdown 格式中输出 :param blocks: 含 title 和 text 的文本段落,原文会进行格式转义 :param text_format: 目标文本格式:markdown、mrkdwn 或 rich_markdown :return: 可交给对应渠道渲染的文本 """ bold = "*" if text_format == "mrkdwn" else "**" paragraphs = [] if title and text_format == "rich_markdown": paragraphs.append(f"## {_escape_notification_text(title, text_format)}") in_section = False for block in blocks: lines = [] block_title, block_text = block.get("title"), block.get("text") if text_format == "rich_markdown" and block_title: heading = _escape_notification_text(block_title, text_format) if not block_text: paragraphs.append(f"### {heading}") in_section = True continue if in_section: # 分节之后的标题/正文对是列表项,硬换行让续行仍缩进在当前条目内。 body = _escape_notification_text(block_text, text_format).replace("\n", " \n ") paragraphs.append(f"- **{heading}** \n {body}") continue if block_title: lines.append(f"{bold}{_escape_notification_text(block_title, text_format)}{bold}") if block_text: lines.append(_escape_notification_text(block_text, text_format)) if lines: paragraphs.append("\n".join(lines)) return "\n\n".join(paragraphs) def cut(text: str, HMM: bool = True, cut_all: bool = False) -> list[str]: """ 通过统一原生入口执行中文分词。 :param text: 待分词的中文文本 :param HMM: 是否启用隐马尔可夫模型识别未登录词 :param cut_all: 是否采用全模式分词 :return: 原生分词器产生的词语列表 """ words: list[str] = moviepilot_rust.jieba_cut(text, hmm=HMM, cut_all=cut_all) return words def convert(text: str, target: str) -> str: """ 通过统一入口执行 MediaWiki 中文简繁转换。 :param text: 待转换的中文文本 :param target: MediaWiki 转换目标代码,例如 zh-cn 或 zh-tw :return: 转换后的文本 """ converted: str = moviepilot_rust.zhconv_fast(text, target) return converted def contains_chinese(value: Union[str, list[str]]) -> bool: """ 判断文本或文本列表中是否包含中文字符。 """ if not value: return False if isinstance(value, list): value = " ".join(value) return re.search(r"[\u4e00-\u9fff]", value) is not None def contains_japanese(value: str) -> bool: """ 判断文本中是否包含平假名或片假名。 """ return re.search(r"[\u3040-\u309F\u30A0-\u30FF]", value) is not None def contains_korean(value: str) -> bool: """ 判断文本中是否包含韩文字符。 """ return re.search(r"[\uAC00-\uD7FF]", value) is not None def is_all_chinese(value: str) -> bool: """ 判断除空格外的全部字符是否都是中文。 """ return all(character == " " or "\u4e00" <= character <= "\u9fff" for character in value) def is_english_word(value: str) -> bool: """ 判断文本是否为不含空格的英文字母单词。 """ return value.encode().isalpha() def parse_int(value: str) -> int: """ 解析可能带千位分隔符的整数,无法解析时返回零。 """ if value: value = value.strip() if not value: return 0 try: return int(value.replace(",", "")) except ValueError: return 0 def parse_float(value: str) -> float: """ 解析可能带千位分隔符的浮点数,无法解析时返回零。 """ if value: value = value.strip() if not value: return 0.0 try: return float(value.replace(",", "")) except ValueError: return 0.0 @overload def remove_punctuation(value: str, replacement: str = "", allow_space: bool = False) -> str: """ 保留单段文本清理后的字符串返回类型。 """ ... @overload def remove_punctuation(value: list[str], replacement: str = "", allow_space: bool = False) -> list[str]: """ 保留文本列表清理后的列表返回类型。 """ ... def remove_punctuation( value: Union[list[str], str], replacement: str = "", allow_space: bool = False, ) -> Union[list[str], str]: """ 移除历史匹配规则使用的标点和零宽字符。 :param value: 待清理文本或文本列表,空值保持原样 :param replacement: 单段文本中标点的替换内容 :param allow_space: 单段文本是否保留合并后的空格 :return: 清理后的同类结构;列表元素沿用默认替换与空格规则 """ punctuation = r"[、.。,,·::;;!!??'’\"“”()()\[\]【】「」\-—―\+\|\\_/&#~~]" if not value: return value if isinstance(value, list): return [remove_punctuation(item) for item in value] normalized = re.sub( r"[\u200B-\u200D\uFEFF]", "", re.sub(punctuation, replacement, value, flags=re.IGNORECASE), flags=re.IGNORECASE, ) if not allow_space: return re.sub(r"\s+", "", normalized) return re.sub(r"\s+", " ", normalized).strip() def normalize_upper(value: Optional[str]) -> str: """ 移除历史匹配标点、空白并转换为大写。 """ if not value: return "" return remove_punctuation(value).upper().strip() def sanitize_filename(value: str) -> Optional[str]: """ 移除文件名中不允许使用的字符并替换英文冒号。 """ if not value: return None return re.sub(r"[*?\\/\"<>~|]", "", value, flags=re.IGNORECASE).replace(":", ":") def random_string(length: int = 16) -> str: """ 生成兼容历史字符集的指定长度随机字符串。 """ alphabet = "ABCDEFGHIGKLMNOPQRSTUVWXYZabcdefghigklmnopqrstuvwxyz0123456789" last_index = len(alphabet) - 1 return "".join(alphabet[random.randint(0, last_index)] for _index in range(length)) def parse_bool(value: object, default: bool = False) -> bool: """ 按历史配置规则把字符串或数值转换为布尔值。 :param value: 待解析值,数值大于 0 才视为真 :param default: 仅在输入为空字符串时使用的默认值 :return: 布尔值,字符串忽略大小写,仅 y、true、1、yes、on 视为真 """ if isinstance(value, str) and not value: return default if isinstance(value, bool): return value if isinstance(value, (int, float)): return value > 0 return isinstance(value, str) and value.lower() in {"y", "true", "1", "yes", "on"} def cookiejar_to_string(cookiejar: Mapping[str, str]) -> str: """ 把键值形式的 CookieJar 序列化为 Cookie 请求头文本。 """ return "; ".join("=".join(item) for item in cookiejar.items()) class NamedIdEntry(TypedDict, total=False): """ 命名条目的原始字段;缺失名称不会匹配,ID 保留原始值。 :ivar name: 用于匹配的名称;缺失或为 None 时不匹配 :ivar id: 命中后返回的原始标识,保留输入类型 """ name: str | None id: object def extract_named_ids(content: str, entries: List[NamedIdEntry]) -> tuple[list[object], str] | list[object]: """ 从空格分隔文本中提取命名条目 ID,并返回剩余内容。 :param content: 按空白分隔的待匹配文本 :param entries: 名称与原始标识条目,缺失名称时跳过 :return: 非空输入返回去重标识列表与剩余文本;空输入直接返回空列表 """ if not content: return [] identifiers: list[object] = [] content_parts = content.split() for entry in entries: name = entry.get("name") if name is not None and name in content_parts and entry.get("id") not in identifiers: identifiers.append(entry.get("id")) content = content.replace(name, "") return identifiers, re.sub(r"\s+", " ", content).strip() def format_amount(amount: object, currency: str = "$") -> str: """ 使用千位分隔符和货币前缀格式化金额。 :param amount: 支持数字格式化的金额;零值或其他假值直接显示为 0 :param currency: 非零金额前的货币符号 :return: 带千位分隔符的金额文本 """ if not amount: return "0" return currency + format(amount, ",") def count_words(value: str) -> int: """ 统计中英文混合文本中的汉字数和英文单词数。 """ if not value: return 0 chinese_words = [ word for word in re.findall(r"[\u4e00-\u9fa5]", value) if word.isalpha() ] english_words = [word for word in re.findall(r"[a-zA-Z]+", value) if word.isalpha()] return len(chinese_words) + len(english_words) def split_by_bytes(value: str, max_length: int) -> Generator[str, None, None]: """ 按 UTF-8 字节上限分段,优先保持换行和英文单词完整。 :param value: 待分段文本;空输入产生一个空字符串 :param max_length: 分段目标上限,单位为 UTF-8 字节;单个完整词语可能超过上限 :yields: 按原顺序输出的文本段,首尾空白会被去除 """ if not value: yield "" lines = re.split("\n", value) buffer = "" for line in lines: if len(line.encode("utf-8")) > max_length: separator = "" if re.match(r"^[A-Za-z0-9.\s]+", line): parts = line.split() separator = " " else: parts = line part = "" for item in parts: if len((part + item).encode("utf-8")) > max_length: yield (buffer + part).strip() buffer = "" part = f"{separator}{item}" else: part = f"{part}{separator}{item}" if part: buffer += part elif len((buffer + "\n" + line).encode("utf-8")) > max_length: yield buffer.strip() buffer = line elif buffer: buffer = f"{buffer}\n{line}" else: buffer = line if buffer: yield buffer.strip() @overload def title_case(value: str) -> str: """ 字符串输入保留字符串返回类型。 """ ... @overload def title_case(value: None) -> None: """ 空值沿用历史返回,不转换为空字符串。 """ ... def title_case(value: Optional[str]) -> Optional[str]: """ 转换为标题大小写,并兼容空值。 """ return value.title() if value else value def escape_markdown(value: str) -> str: """ 转义 Markdown 保留字符,并保持历史二次转义语义。 """ escaped = re.sub(r"([_*\[\]()~`>#+\-=|.!{}])", r"\\\1", value) return re.sub(r"\\\\([_*\[\]()~`>#+\-=|.!{}])", r"\1", escaped) def is_number(value: str) -> bool: """ 判断文本能否转换为整数或浮点数。 """ if not value: return False try: float(value) return True except ValueError: return False def common_prefix(first: str, second: str) -> str: """ 返回两个字符串从首字符开始的公共前缀。 """ if not first or not second: return "" prefix = [] for first_character, second_character in zip(first, second): if first_character != second_character: break prefix.append(first_character) return "".join(prefix) def strip_optional(value: Optional[str]) -> Optional[str]: """ 去除可空值两端空白,并保持 None。 """ return value.strip() if value is not None else None def natural_sort_key(value: str) -> List[Union[int, str]]: """ 把文本拆成数字和小写文本片段,供自然排序使用。 """ if value is None: return [] if not isinstance(value, str): value = str(value) return [ int(part) if part.isdigit() else part.lower() for part in re.split(r"(\d+)", value) ]