"""
无业务状态的文本识别、清理、转换和分段能力。
"""
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)
]