K2cr2O1
caf5b06097
feat(ws_pool): 新增 WebSocket 连接池实现 perf(json): 使用 orjson 替代标准 json 库提升性能 style: 清理未使用的导入和冗余代码 docs: 更新架构文档和开发规范 test: 添加 WebSocket 连接池测试用例 fix(plugins): 修复自动审批插件 API 调用参数格式
143 lines
3.5 KiB
Python
143 lines
3.5 KiB
Python
# -*- coding: utf-8 -*-
|
||
import re
|
||
from typing import Dict, Any, List
|
||
|
||
from models import MessageEvent
|
||
|
||
|
||
def format_duration(seconds: int) -> str:
|
||
"""
|
||
将秒数格式化为 MM:SS 的形式
|
||
|
||
Args:
|
||
seconds (int): 秒数
|
||
|
||
Returns:
|
||
str: 格式化后的时间字符串
|
||
"""
|
||
if not isinstance(seconds, int) or seconds < 0:
|
||
return "00:00"
|
||
minutes, seconds = divmod(seconds, 60)
|
||
return f"{minutes:02d}:{seconds:02d}"
|
||
|
||
|
||
def clean_url(url: str) -> str:
|
||
"""
|
||
清理URL，去掉不必要的查询参数
|
||
|
||
Args:
|
||
url (str): 原始URL
|
||
|
||
Returns:
|
||
str: 清理后的URL
|
||
"""
|
||
clean_url = url.split('?')[0]
|
||
if '#/' in clean_url:
|
||
clean_url = clean_url.split('#/')[0]
|
||
return clean_url
|
||
|
||
|
||
def extract_original_text(segments: List[Any], url_pattern: re.Pattern) -> str:
|
||
"""
|
||
从消息段中提取原始文本（去除链接）
|
||
|
||
Args:
|
||
segments (List[Any]): 消息段列表
|
||
url_pattern (re.Pattern): URL正则表达式模式
|
||
|
||
Returns:
|
||
str: 提取的原始文本
|
||
"""
|
||
for segment in segments:
|
||
if segment.type == "text":
|
||
text_content = segment.data.get("text", "")
|
||
# 移除链接
|
||
cleaned_text = re.sub(url_pattern, '', text_content)
|
||
# 移除常见的分享提示
|
||
cleaned_text = re.sub(r'复制此链接.*?打开.*?搜索.*?直接观看视频！', '', cleaned_text)
|
||
cleaned_text = cleaned_text.strip()
|
||
if cleaned_text:
|
||
return cleaned_text
|
||
return ""
|
||
|
||
|
||
def build_forward_nodes(event: MessageEvent, nickname: str, messages: List[Any]) -> List[Any]:
|
||
"""
|
||
构建转发消息节点
|
||
|
||
Args:
|
||
event (MessageEvent): 消息事件对象
|
||
nickname (str): 发送者昵称
|
||
messages (List[Any]): 消息内容列表
|
||
|
||
Returns:
|
||
List[Any]: 转发消息节点列表
|
||
"""
|
||
nodes = []
|
||
for msg in messages:
|
||
if isinstance(msg, str):
|
||
node = event.bot.build_forward_node(
|
||
user_id=event.self_id,
|
||
nickname=nickname,
|
||
message=msg
|
||
)
|
||
nodes.append(node)
|
||
elif isinstance(msg, list):
|
||
node = event.bot.build_forward_node(
|
||
user_id=event.self_id,
|
||
nickname=nickname,
|
||
message=msg
|
||
)
|
||
nodes.append(node)
|
||
return nodes
|
||
|
||
|
||
def safe_get(data: Dict[str, Any], keys: List[str], default: Any = None) -> Any:
|
||
"""
|
||
安全地从嵌套字典中获取值
|
||
|
||
Args:
|
||
data (Dict[str, Any]): 嵌套字典
|
||
keys (List[str]): 键路径列表
|
||
default (Any, optional): 默认值. Defaults to None.
|
||
|
||
Returns:
|
||
Any: 获取的值或默认值
|
||
"""
|
||
result = data
|
||
for key in keys:
|
||
if isinstance(result, dict) and key in result:
|
||
result = result[key]
|
||
else:
|
||
return default
|
||
return result
|
||
|
||
|
||
def normalize_url(url: str) -> str:
|
||
"""
|
||
规范化URL
|
||
|
||
Args:
|
||
url (str): 原始URL
|
||
|
||
Returns:
|
||
str: 规范化后的URL
|
||
"""
|
||
if not url.startswith('http'):
|
||
url = 'https://' + url
|
||
return url
|
||
|
||
|
||
def validate_url(url: str) -> bool:
|
||
"""
|
||
验证URL格式是否正确
|
||
|
||
Args:
|
||
url (str): URL
|
||
|
||
Returns:
|
||
bool: URL格式是否正确
|
||
"""
|
||
url_pattern = re.compile(r'https?://[^]+')
|
||
return bool(url_pattern.match(url))
|