mirror of
https://github.com/LiteyukiStudio/LiteyukiBot.git
synced 2025-07-28 05:40:54 +00:00
fix: 插件列表显示错误问题
This commit is contained in:
37
liteyuki/utils/__init__.py
Normal file
37
liteyuki/utils/__init__.py
Normal file
@ -0,0 +1,37 @@
|
||||
import nonebot
|
||||
|
||||
from .log import logger
|
||||
import sys
|
||||
|
||||
__NAME__ = "LiteyukiBot"
|
||||
__VERSION__ = "6.2.1" # 60201
|
||||
major, minor, patch = map(int, __VERSION__.split("."))
|
||||
__VERSION_I__ = major * 10000 + minor * 100 + patch
|
||||
|
||||
|
||||
def init():
|
||||
"""
|
||||
初始化
|
||||
Returns:
|
||||
|
||||
"""
|
||||
# 检测python版本是否高于3.10
|
||||
if sys.version_info < (3, 10):
|
||||
nonebot.logger.error("This project requires Python3.10+ to run, please upgrade your Python Environment.")
|
||||
exit(1)
|
||||
|
||||
print("\033[34m" + r""" __ ______ ________ ________ __ __ __ __ __ __ ______
|
||||
/ | / |/ |/ |/ \ / |/ | / |/ | / |/ |
|
||||
$$ | $$$$$$/ $$$$$$$$/ $$$$$$$$/ $$ \ /$$/ $$ | $$ |$$ | /$$/ $$$$$$/
|
||||
$$ | $$ | $$ | $$ |__ $$ \/$$/ $$ | $$ |$$ |/$$/ $$ |
|
||||
$$ | $$ | $$ | $$ | $$ $$/ $$ | $$ |$$ $$< $$ |
|
||||
$$ | $$ | $$ | $$$$$/ $$$$/ $$ | $$ |$$$$$ \ $$ |
|
||||
$$ |_____ _$$ |_ $$ | $$ |_____ $$ | $$ \__$$ |$$ |$$ \ _$$ |_
|
||||
$$ |/ $$ | $$ | $$ | $$ | $$ $$/ $$ | $$ |/ $$ |
|
||||
$$$$$$$$/ $$$$$$/ $$/ $$$$$$$$/ $$/ $$$$$$/ $$/ $$/ $$$$$$/ """ + "\033[0m")
|
||||
nonebot.logger.info(
|
||||
f"Run Liteyuki with Python{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro} "
|
||||
f"at {sys.executable}"
|
||||
)
|
||||
|
||||
nonebot.logger.info(f"{__NAME__} {__VERSION__}({__VERSION_I__}) is running")
|
||||
32
liteyuki/utils/config.py
Normal file
32
liteyuki/utils/config.py
Normal file
@ -0,0 +1,32 @@
|
||||
import os
|
||||
|
||||
import nonebot
|
||||
import yaml
|
||||
from pydantic import BaseModel
|
||||
|
||||
config = {}
|
||||
|
||||
|
||||
class BasicConfig(BaseModel):
|
||||
host: str = "127.0.0.1"
|
||||
port: int = 20216
|
||||
superusers: list[str] = []
|
||||
command_start: list[str] = ["/", ""]
|
||||
nickname: list[str] = ["liteyuki"]
|
||||
|
||||
|
||||
def load_from_yaml(file: str) -> dict:
|
||||
nonebot.logger.debug("Loading config from %s" % file)
|
||||
global config
|
||||
if not os.path.exists(file):
|
||||
nonebot.logger.warning(f'Config file {file} not found, created default config, please modify it and restart')
|
||||
with open(file, 'w', encoding='utf-8') as f:
|
||||
yaml.dump(BasicConfig().dict(), f, default_flow_style=False)
|
||||
|
||||
with open(file, 'r', encoding='utf-8') as f:
|
||||
conf = yaml.load(f, Loader=yaml.FullLoader)
|
||||
config = conf
|
||||
if conf is None:
|
||||
nonebot.logger.warning(f'Config file {file} is empty, use default config. please modify it and restart')
|
||||
conf = BasicConfig().dict()
|
||||
return conf
|
||||
373
liteyuki/utils/data.py
Normal file
373
liteyuki/utils/data.py
Normal file
@ -0,0 +1,373 @@
|
||||
import json
|
||||
import os
|
||||
import sqlite3
|
||||
import types
|
||||
from abc import ABC
|
||||
from collections.abc import Iterable
|
||||
|
||||
import nonebot
|
||||
from pydantic import BaseModel
|
||||
from typing import Any
|
||||
|
||||
BaseIterable = list | tuple | set | dict
|
||||
|
||||
|
||||
class LiteModel(BaseModel):
|
||||
"""轻量级模型基类
|
||||
类型注解统一使用Python3.9的PEP585标准，如需使用泛型请使用typing模块的泛型类型
|
||||
"""
|
||||
id: int = None
|
||||
|
||||
|
||||
class BaseORMAdapter(ABC):
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
def auto_migrate(self, *args, **kwargs):
|
||||
"""自动迁移
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def upsert(self, *args, **kwargs):
|
||||
"""存储数据
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def first(self, *args, **kwargs):
|
||||
"""查询第一条数据
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def all(self, *args, **kwargs):
|
||||
"""查询所有数据
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def delete(self, *args, **kwargs):
|
||||
"""删除数据
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def update(self, *args, **kwargs):
|
||||
"""更新数据
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class Database(BaseORMAdapter):
|
||||
"""SQLiteORM适配器，严禁使用`FORIEGNID`和`JSON`作为主键前缀，严禁使用`$ID:`作为字符串值前缀
|
||||
|
||||
Attributes:
|
||||
|
||||
"""
|
||||
type_map = {
|
||||
# default: TEXT
|
||||
str : 'TEXT',
|
||||
int : 'INTEGER',
|
||||
float: 'REAL',
|
||||
bool : 'INTEGER',
|
||||
list : 'TEXT'
|
||||
}
|
||||
|
||||
DEFAULT_VALUE = {
|
||||
'TEXT' : '',
|
||||
'INTEGER': 0,
|
||||
'REAL' : 0.0
|
||||
}
|
||||
|
||||
FOREIGNID = 'FOREIGNID'
|
||||
JSON = 'JSON'
|
||||
LIST = 'LIST'
|
||||
DICT = 'DICT'
|
||||
ID = '$ID'
|
||||
|
||||
def __init__(self, db_name: str):
|
||||
super().__init__()
|
||||
if not os.path.exists(os.path.dirname(db_name)):
|
||||
os.makedirs(os.path.dirname(db_name))
|
||||
self.conn = sqlite3.connect(db_name)
|
||||
self.conn.row_factory = sqlite3.Row
|
||||
self.cursor = self.conn.cursor()
|
||||
|
||||
def auto_migrate(self, *args: type(LiteModel)):
|
||||
"""自动迁移，检测新模型字段和原有表字段的差异，如有差异自动增删新字段
|
||||
|
||||
Args:
|
||||
*args: 模型类
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
table_name = ''
|
||||
for model in args:
|
||||
model: type(LiteModel)
|
||||
# 检测并创建表，若模型未定义id字段则使用自增主键，有定义的话使用id字段，且id有可能为字符串
|
||||
table_name = model.__name__
|
||||
if 'id' in model.__annotations__ and model.__annotations__['id'] is not None:
|
||||
# 如果模型定义了id字段，那么使用模型的id字段
|
||||
id_type = self.type_map.get(model.__annotations__['id'], 'TEXT')
|
||||
self.cursor.execute(f'CREATE TABLE IF NOT EXISTS {table_name} (id {id_type} PRIMARY KEY)')
|
||||
else:
|
||||
# 如果模型未定义id字段，那么使用自增主键
|
||||
self.cursor.execute(f'CREATE TABLE IF NOT EXISTS {table_name} (id INTEGER PRIMARY KEY AUTOINCREMENT)')
|
||||
# 获取表字段
|
||||
self.cursor.execute(f'PRAGMA table_info({table_name})')
|
||||
table_fields = self.cursor.fetchall()
|
||||
table_fields = [field[1] for field in table_fields]
|
||||
|
||||
raw_fields, raw_types = zip(*model.__annotations__.items())
|
||||
# 获取模型字段，若有模型则添加FOREIGNID前缀，若为BaseIterable则添加JSON前缀，用多行if判断
|
||||
model_fields = []
|
||||
model_types = []
|
||||
for field, r_type in zip(raw_fields, raw_types):
|
||||
if isinstance(r_type, type(LiteModel)):
|
||||
model_fields.append(f'{self.FOREIGNID}{field}')
|
||||
model_types.append('TEXT')
|
||||
elif r_type in [list[str], list[int], list[float], list[bool], list]:
|
||||
model_fields.append(f'{self.LIST}{field}')
|
||||
model_types.append('TEXT')
|
||||
elif r_type in [dict[str, str], dict[str, int], dict[str, float], dict[str, bool], dict]:
|
||||
model_fields.append(f'{self.DICT}{field}')
|
||||
model_types.append('TEXT')
|
||||
elif isinstance(r_type, types.GenericAlias):
|
||||
model_fields.append(f'{self.JSON}{field}')
|
||||
model_types.append('TEXT')
|
||||
else:
|
||||
model_fields.append(field)
|
||||
model_types.append(self.type_map.get(r_type, 'TEXT'))
|
||||
|
||||
# 检测新字段或字段类型是否有变化，有则增删字段，已经加了前缀类型
|
||||
for field_changed, type_, r_type in zip(model_fields, model_types, raw_types):
|
||||
if field_changed not in table_fields:
|
||||
nonebot.logger.debug(f'ALTER TABLE {table_name} ADD COLUMN {field_changed} {type_}')
|
||||
self.cursor.execute(f'ALTER TABLE {table_name} ADD COLUMN {field_changed} {type_}')
|
||||
# 在原有的行中添加新字段对应类型的默认值，从DEFAULT_TYPE中获取
|
||||
self.cursor.execute(f'UPDATE {table_name} SET {field_changed} = ? WHERE {field_changed} IS NULL', (self.DEFAULT_VALUE.get(type_, ""),))
|
||||
|
||||
# 检测多余字段，除了id字段
|
||||
for field in table_fields:
|
||||
if field not in model_fields and field != 'id':
|
||||
nonebot.logger.debug(f'ALTER TABLE {table_name} DROP COLUMN {field}')
|
||||
self.cursor.execute(f'ALTER TABLE {table_name} DROP COLUMN {field}')
|
||||
|
||||
self.conn.commit()
|
||||
nonebot.logger.debug(f'Table {table_name} migrated successfully')
|
||||
|
||||
def upsert(self, *models: LiteModel) -> int | tuple:
|
||||
"""存储数据，检查id字段，如果有id字段则更新，没有则插入
|
||||
|
||||
Args:
|
||||
models: 数据
|
||||
|
||||
Returns:
|
||||
id: 数据id，如果有多个数据则返回id元组
|
||||
"""
|
||||
|
||||
ids = []
|
||||
for model in models:
|
||||
table_name = model.__class__.__name__
|
||||
if not self._detect_for_table(table_name):
|
||||
raise ValueError(f'表{table_name}不存在，请先迁移')
|
||||
key_list = []
|
||||
value_list = []
|
||||
# 处理外键，添加前缀'$IDFieldName'
|
||||
for field, value in model.__dict__.items():
|
||||
if isinstance(value, LiteModel):
|
||||
key_list.append(f'{self.FOREIGNID}{field}')
|
||||
value_list.append(f'{self.ID}:{value.__class__.__name__}:{self.upsert(value)}')
|
||||
elif isinstance(value, list):
|
||||
key_list.append(f'{self.LIST}{field}')
|
||||
value_list.append(self._flat(value))
|
||||
elif isinstance(value, dict):
|
||||
key_list.append(f'{self.DICT}{field}')
|
||||
value_list.append(self._flat(value))
|
||||
elif isinstance(value, BaseIterable):
|
||||
key_list.append(f'{self.JSON}{field}')
|
||||
value_list.append(self._flat(value))
|
||||
else:
|
||||
key_list.append(field)
|
||||
value_list.append(value)
|
||||
# 更新或插入数据，用?占位
|
||||
nonebot.logger.debug(f'INSERT OR REPLACE INTO {table_name} ({",".join(key_list)}) VALUES ({",".join(["?" for _ in key_list])})')
|
||||
self.cursor.execute(f'INSERT OR REPLACE INTO {table_name} ({",".join(key_list)}) VALUES ({",".join(["?" for _ in key_list])})', value_list)
|
||||
|
||||
ids.append(self.cursor.lastrowid)
|
||||
self.conn.commit()
|
||||
return ids[0] if len(ids) == 1 else tuple(ids)
|
||||
|
||||
def _flat(self, data: Iterable) -> str:
|
||||
"""扁平化数据，返回扁平化对象
|
||||
|
||||
Args:
|
||||
data: 数据，可迭代对象
|
||||
|
||||
Returns: json字符串
|
||||
"""
|
||||
if isinstance(data, dict):
|
||||
return_data = {}
|
||||
for k, v in data.items():
|
||||
if isinstance(v, LiteModel):
|
||||
return_data[f'{self.FOREIGNID}{k}'] = f'{self.ID}:{v.__class__.__name__}:{self.upsert(v)}'
|
||||
elif isinstance(v, list):
|
||||
return_data[f'{self.LIST}{k}'] = self._flat(v)
|
||||
elif isinstance(v, dict):
|
||||
return_data[f'{self.DICT}{k}'] = self._flat(v)
|
||||
elif isinstance(v, BaseIterable):
|
||||
return_data[f'{self.JSON}{k}'] = self._flat(v)
|
||||
else:
|
||||
return_data[k] = v
|
||||
|
||||
elif isinstance(data, list | tuple | set):
|
||||
return_data = []
|
||||
for v in data:
|
||||
if isinstance(v, LiteModel):
|
||||
return_data.append(f'{self.ID}:{v.__class__.__name__}:{self.upsert(v)}')
|
||||
elif isinstance(v, list):
|
||||
return_data.append(self._flat(v))
|
||||
elif isinstance(v, dict):
|
||||
return_data.append(self._flat(v))
|
||||
elif isinstance(v, BaseIterable):
|
||||
return_data.append(self._flat(v))
|
||||
else:
|
||||
return_data.append(v)
|
||||
else:
|
||||
raise ValueError('数据类型错误')
|
||||
|
||||
return json.dumps(return_data)
|
||||
|
||||
def _detect_for_table(self, table_name: str) -> bool:
|
||||
"""在进行增删查改前检测表是否存在
|
||||
|
||||
Args:
|
||||
table_name: 表名
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
return self.cursor.execute(f'SELECT * FROM sqlite_master WHERE type = "table" AND name = ?', (table_name,)).fetchone()
|
||||
|
||||
def first(self, model: type(LiteModel), conditions, *args, default: Any = None) -> LiteModel | None:
|
||||
"""查询第一条数据
|
||||
|
||||
Args:
|
||||
model: 模型
|
||||
conditions: 查询条件
|
||||
*args: 参数化查询条件参数
|
||||
default: 未查询到结果默认返回值
|
||||
|
||||
Returns: 数据
|
||||
"""
|
||||
table_name = model.__name__
|
||||
|
||||
if not self._detect_for_table(table_name):
|
||||
return default
|
||||
|
||||
self.cursor.execute(f'SELECT * FROM {table_name} WHERE {conditions}', args)
|
||||
if row_data := self.cursor.fetchone():
|
||||
data = dict(row_data)
|
||||
return model(**self.convert_to_dict(data))
|
||||
return default
|
||||
|
||||
def all(self, model: type(LiteModel), conditions=None, *args, default: Any = None) -> list[LiteModel] | None:
|
||||
"""查询所有数据
|
||||
|
||||
Args:
|
||||
model: 模型
|
||||
conditions: 查询条件
|
||||
*args: 参数化查询条件参数
|
||||
default: 未查询到结果默认返回值
|
||||
|
||||
Returns: 数据
|
||||
"""
|
||||
table_name = model.__name__
|
||||
|
||||
if not self._detect_for_table(table_name):
|
||||
return default
|
||||
|
||||
if conditions:
|
||||
self.cursor.execute(f'SELECT * FROM {table_name} WHERE {conditions}', args)
|
||||
else:
|
||||
self.cursor.execute(f'SELECT * FROM {table_name}')
|
||||
if row_datas := self.cursor.fetchall():
|
||||
datas = [dict(row_data) for row_data in row_datas]
|
||||
return [model(**self.convert_to_dict(d)) for d in datas] if datas else default
|
||||
return default
|
||||
|
||||
def delete(self, model: type(LiteModel), conditions, *args):
|
||||
"""删除数据
|
||||
|
||||
Args:
|
||||
model: 模型
|
||||
conditions: 查询条件
|
||||
*args: 参数化查询条件参数
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
table_name = model.__name__
|
||||
|
||||
if not self._detect_for_table(table_name):
|
||||
return
|
||||
nonebot.logger.debug(f'DELETE FROM {table_name} WHERE {conditions}')
|
||||
self.cursor.execute(f'DELETE FROM {table_name} WHERE {conditions}', args)
|
||||
self.conn.commit()
|
||||
|
||||
def convert_to_dict(self, data: dict) -> dict:
|
||||
"""将json字符串转换为字典
|
||||
|
||||
Args:
|
||||
data: json字符串
|
||||
|
||||
Returns: 字典
|
||||
"""
|
||||
|
||||
def load(d: BaseIterable) -> BaseIterable:
|
||||
"""递归加载数据，去除前缀"""
|
||||
if isinstance(d, dict):
|
||||
new_d = {}
|
||||
for k, v in d.items():
|
||||
if k.startswith(self.FOREIGNID):
|
||||
new_d[k.replace(self.FOREIGNID, '')] = load(
|
||||
dict(self.cursor.execute(f'SELECT * FROM {v.split(":", 2)[1]} WHERE id = ?', (v.split(":", 2)[2],)).fetchone()))
|
||||
|
||||
elif k.startswith(self.LIST):
|
||||
if v == '': v = '[]'
|
||||
new_d[k.replace(self.LIST, '')] = load(json.loads(v))
|
||||
elif k.startswith(self.DICT):
|
||||
if v == '': v = '{}'
|
||||
new_d[k.replace(self.DICT, '')] = load(json.loads(v))
|
||||
elif k.startswith(self.JSON):
|
||||
if v == '': v = '[]'
|
||||
new_d[k.replace(self.JSON, '')] = load(json.loads(v))
|
||||
else:
|
||||
new_d[k] = v
|
||||
elif isinstance(d, list | tuple | set):
|
||||
new_d = []
|
||||
for i, v in enumerate(d):
|
||||
if isinstance(v, str) and v.startswith(self.ID):
|
||||
new_d.append(load(dict(self.cursor.execute(f'SELECT * FROM {v.split(":", 2)[1]} WHERE id = ?', (v.split(":", 2)[2],)).fetchone())))
|
||||
elif isinstance(v, BaseIterable):
|
||||
new_d.append(load(v))
|
||||
else:
|
||||
new_d = d
|
||||
return new_d
|
||||
|
||||
return load(data)
|
||||
45
liteyuki/utils/data_manager.py
Normal file
45
liteyuki/utils/data_manager.py
Normal file
@ -0,0 +1,45 @@
|
||||
import os
|
||||
|
||||
from pydantic import Field
|
||||
|
||||
from liteyuki.utils.data import LiteModel, Database as DB
|
||||
|
||||
DATA_PATH = "data/liteyuki"
|
||||
|
||||
user_db = DB(os.path.join(DATA_PATH, 'users.ldb'))
|
||||
group_db = DB(os.path.join(DATA_PATH, 'groups.ldb'))
|
||||
plugin_db = DB(os.path.join(DATA_PATH, 'plugins.ldb'))
|
||||
common_db = DB(os.path.join(DATA_PATH, 'common.ldb'))
|
||||
|
||||
|
||||
class User(LiteModel):
|
||||
user_id: str = Field(str(), alias='user_id')
|
||||
username: str = Field(str(), alias='username')
|
||||
profile: dict[str, str] = Field(dict(), alias='profile')
|
||||
enabled_plugins: list[str] = Field(list(), alias='enabled_plugins')
|
||||
disabled_plugins: list[str] = Field(list(), alias='disabled_plugins')
|
||||
|
||||
|
||||
class GroupChat(LiteModel):
|
||||
# Group是一个关键字，所以这里用GroupChat
|
||||
group_id: str = Field(str(), alias='group_id')
|
||||
group_name: str = Field(str(), alias='group_name')
|
||||
enabled_plugins: list[str] = Field([], alias='enabled_plugins')
|
||||
disabled_plugins: list[str] = Field([], alias='disabled_plugins')
|
||||
|
||||
|
||||
class InstalledPlugin(LiteModel):
|
||||
module_name: str = Field(str(), alias='module_name')
|
||||
version: str = Field(str(), alias='version')
|
||||
|
||||
|
||||
class GlobalPlugin(LiteModel):
|
||||
module_name: str = Field(str(), alias='module_name')
|
||||
enabled: bool = Field(True, alias='enabled')
|
||||
|
||||
|
||||
def auto_migrate():
|
||||
user_db.auto_migrate(User())
|
||||
group_db.auto_migrate(GroupChat())
|
||||
plugin_db.auto_migrate(InstalledPlugin())
|
||||
common_db.auto_migrate(GlobalPlugin())
|
||||
167
liteyuki/utils/language.py
Normal file
167
liteyuki/utils/language.py
Normal file
@ -0,0 +1,167 @@
|
||||
"""
|
||||
语言模块,添加对多语言的支持
|
||||
"""
|
||||
|
||||
import json
|
||||
import locale
|
||||
import os
|
||||
from typing import Any
|
||||
|
||||
import nonebot
|
||||
|
||||
from liteyuki.utils.config import config
|
||||
from liteyuki.utils.data_manager import User, user_db
|
||||
|
||||
_default_lang_code = "en"
|
||||
_language_data = {
|
||||
"en": {
|
||||
"name": "English",
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
def load_from_lang(file_path: str, lang_code: str = None):
|
||||
"""
|
||||
从lang文件中加载语言数据，用于简单的文本键值对
|
||||
|
||||
Args:
|
||||
file_path: lang文件路径
|
||||
lang_code: 语言代码，如果为None则从文件名中获取
|
||||
"""
|
||||
try:
|
||||
if lang_code is None:
|
||||
lang_code = os.path.basename(file_path).split('.')[0]
|
||||
with open(file_path, 'r', encoding='utf-8') as file:
|
||||
data = {}
|
||||
for line in file:
|
||||
line = line.strip()
|
||||
if not line or line.startswith('#'): # 空行或注释
|
||||
continue
|
||||
key, value = line.split('=', 1)
|
||||
data[key.strip()] = value.strip()
|
||||
if lang_code not in _language_data:
|
||||
_language_data[lang_code] = {}
|
||||
_language_data[lang_code].update(data)
|
||||
except Exception as e:
|
||||
nonebot.logger.error(f"Failed to load language data from {file_path}: {e}")
|
||||
|
||||
|
||||
def load_from_json(file_path: str, lang_code: str = None):
|
||||
"""
|
||||
从json文件中加载语言数据，可以定义一些变量
|
||||
|
||||
Args:
|
||||
lang_code: 语言代码，如果为None则从文件名中获取
|
||||
file_path: json文件路径
|
||||
"""
|
||||
try:
|
||||
if lang_code is None:
|
||||
lang_code = os.path.basename(file_path).split('.')[0]
|
||||
with open(file_path, 'r', encoding='utf-8') as file:
|
||||
data = json.load(file)
|
||||
if lang_code not in _language_data:
|
||||
_language_data[lang_code] = {}
|
||||
_language_data[lang_code].update(data)
|
||||
nonebot.logger.debug(f"Loaded language data from {file_path}")
|
||||
except Exception as e:
|
||||
nonebot.logger.error(f"Failed to load language data from {file_path}: {e}")
|
||||
|
||||
|
||||
def load_from_dir(dir_path: str):
|
||||
"""
|
||||
从目录中加载语言数据
|
||||
|
||||
Args:
|
||||
dir_path: 目录路径
|
||||
"""
|
||||
for file in os.listdir(dir_path):
|
||||
try:
|
||||
file_path = os.path.join(dir_path, file)
|
||||
if os.path.isfile(file_path):
|
||||
if file.endswith('.lang'):
|
||||
load_from_lang(file_path)
|
||||
elif file.endswith('.json'):
|
||||
load_from_json(file_path)
|
||||
except Exception as e:
|
||||
nonebot.logger.error(f"Failed to load language data from {file}: {e}")
|
||||
continue
|
||||
|
||||
|
||||
def load_from_dict(data: dict, lang_code: str):
|
||||
"""
|
||||
从字典中加载语言数据
|
||||
|
||||
Args:
|
||||
lang_code: 语言代码
|
||||
data: 字典数据
|
||||
"""
|
||||
if lang_code not in _language_data:
|
||||
_language_data[lang_code] = {}
|
||||
_language_data[lang_code].update(data)
|
||||
|
||||
|
||||
class Language:
|
||||
def __init__(self, lang_code: str = None, fallback_lang_code: str = "en"):
|
||||
if lang_code is None:
|
||||
lang_code = get_system_lang_code()
|
||||
self.lang_code = lang_code
|
||||
self.fallback_lang_code = fallback_lang_code
|
||||
|
||||
def get(self, item: str, *args, **kwargs) -> str | Any:
|
||||
"""
|
||||
获取当前语言文本
|
||||
Args:
|
||||
item: 文本键
|
||||
*args: 格式化参数
|
||||
|
||||
Returns:
|
||||
str: 当前语言的文本
|
||||
|
||||
"""
|
||||
try:
|
||||
if self.lang_code in _language_data and item in _language_data[self.lang_code]:
|
||||
return _language_data[self.lang_code][item].format(*args, **kwargs)
|
||||
if self.fallback_lang_code in _language_data and item in _language_data[self.fallback_lang_code]:
|
||||
return _language_data[self.fallback_lang_code][item].format(*args, **kwargs)
|
||||
return item
|
||||
except Exception as e:
|
||||
nonebot.logger.error(f"Failed to get language text or format: {e}")
|
||||
return item
|
||||
|
||||
|
||||
def get_user_lang(user_id: str) -> Language:
|
||||
"""
|
||||
获取用户的语言代码
|
||||
"""
|
||||
user = user_db.first(User, "user_id = ?", user_id, default=User(
|
||||
user_id=user_id,
|
||||
username="Unknown"
|
||||
))
|
||||
|
||||
return Language(user.profile.get('lang', config.get("default_language", get_system_lang_code())))
|
||||
|
||||
|
||||
def get_system_lang_code() -> str:
|
||||
"""
|
||||
获取系统语言代码
|
||||
"""
|
||||
return locale.getdefaultlocale()[0].replace('_', '-')
|
||||
|
||||
|
||||
def get_default_lang() -> Language:
|
||||
"""
|
||||
获取默认/系统语言
|
||||
"""
|
||||
return Language(config.get("default_language", get_system_lang_code()))
|
||||
|
||||
|
||||
def get_all_lang() -> dict[str, str]:
|
||||
"""
|
||||
获取所有语言
|
||||
Returns
|
||||
{'en': 'English'}
|
||||
"""
|
||||
d = {}
|
||||
for key in _language_data:
|
||||
d[key] = _language_data[key].get("language.name", key)
|
||||
return d
|
||||
67
liteyuki/utils/log.py
Normal file
67
liteyuki/utils/log.py
Normal file
@ -0,0 +1,67 @@
|
||||
import sys
|
||||
import logging
|
||||
from typing import TYPE_CHECKING
|
||||
from colored import fg
|
||||
from .language import get_default_lang
|
||||
import loguru
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from loguru import Logger, Record
|
||||
|
||||
logger: "Logger" = loguru.logger
|
||||
|
||||
|
||||
class LoguruHandler(logging.Handler): # pragma: no cover
|
||||
"""logging 与 loguru 之间的桥梁，将 logging 的日志转发到 loguru。"""
|
||||
|
||||
def emit(self, record: logging.LogRecord):
|
||||
try:
|
||||
level = logger.level(record.levelname).name
|
||||
except ValueError:
|
||||
level = record.levelno
|
||||
|
||||
frame, depth = sys._getframe(6), 6
|
||||
while frame and frame.f_code.co_filename == logging.__file__:
|
||||
frame = frame.f_back
|
||||
depth += 1
|
||||
|
||||
logger.opt(depth=depth, exception=record.exc_info).log(
|
||||
level, record.getMessage()
|
||||
)
|
||||
|
||||
|
||||
def default_filter(record: "Record"):
|
||||
"""默认的日志过滤器，根据 `config.log_level` 配置改变日志等级。"""
|
||||
log_level = record["extra"].get("nonebot_log_level", "INFO")
|
||||
levelno = logger.level(log_level).no if isinstance(log_level, str) else log_level
|
||||
return record["level"].no >= levelno
|
||||
|
||||
|
||||
default_format: str = (
|
||||
"<c>{time:YYYY-MM-DD}</c> <blue>{time:HH:mm:ss}</blue> "
|
||||
"<lvl>[{level.icon}]</lvl> "
|
||||
"<c><{name}></c> "
|
||||
"{message}"
|
||||
)
|
||||
"""默认日志格式"""
|
||||
|
||||
logger.remove()
|
||||
logger_id = logger.add(
|
||||
sys.stdout,
|
||||
level=0,
|
||||
diagnose=False,
|
||||
filter=default_filter,
|
||||
format=default_format,
|
||||
)
|
||||
slang = get_default_lang()
|
||||
logger.level("DEBUG", color="<blue>", icon=f"*️⃣ DDDEBUG")
|
||||
logger.level("INFO", color="<white>", icon=f"ℹ️ IIIINFO")
|
||||
logger.level("SUCCESS", color="<green>", icon=f"✅ SUCCESS")
|
||||
logger.level("WARNING", color="<yellow>", icon=f"⚠️ WARNING")
|
||||
logger.level("ERROR", color="<red>", icon=f"⭕ EEERROR")
|
||||
|
||||
"""默认日志处理器 id"""
|
||||
|
||||
__autodoc__ = {
|
||||
"logger_id": False
|
||||
}
|
||||
7
liteyuki/utils/ly_typing.py
Normal file
7
liteyuki/utils/ly_typing.py
Normal file
@ -0,0 +1,7 @@
|
||||
from nonebot.adapters.onebot import v11, v12
|
||||
|
||||
T_Bot = v11.Bot | v12.Bot
|
||||
T_GroupMessageEvent = v11.GroupMessageEvent | v12.GroupMessageEvent
|
||||
T_PrivateMessageEvent = v11.PrivateMessageEvent | v12.PrivateMessageEvent
|
||||
T_MessageEvent = v11.MessageEvent | v12.MessageEvent
|
||||
T_Message = v11.Message | v12.Message
|
||||
120
liteyuki/utils/message.py
Normal file
120
liteyuki/utils/message.py
Normal file
@ -0,0 +1,120 @@
|
||||
import nonebot
|
||||
from nonebot.adapters.onebot import v11, v12
|
||||
from typing import Any
|
||||
|
||||
from .tools import de_escape, encode_url
|
||||
from .ly_typing import T_Bot, T_MessageEvent
|
||||
|
||||
|
||||
async def send_markdown(markdown: str, bot: T_Bot, *, message_type: str = None, session_id: str | int = None, event: T_MessageEvent = None, **kwargs) -> dict[
|
||||
str, Any]:
|
||||
formatted_md = de_escape(markdown).replace("\n", r"\n").replace("\"", r'\\\"')
|
||||
if event is not None and message_type is None:
|
||||
message_type = event.message_type
|
||||
session_id = event.user_id if event.message_type == "private" else event.group_id
|
||||
try:
|
||||
forward_id = await bot.call_api(
|
||||
api="send_forward_msg",
|
||||
messages=[
|
||||
v11.MessageSegment(
|
||||
type="node",
|
||||
data={
|
||||
"name" : "Liteyuki.OneBot",
|
||||
"uin" : bot.self_id,
|
||||
"content": [
|
||||
{
|
||||
"type": "markdown",
|
||||
"data": {
|
||||
"content": '{"content":"%s"}' % formatted_md
|
||||
}
|
||||
},
|
||||
]
|
||||
},
|
||||
)
|
||||
]
|
||||
)
|
||||
data = await bot.send_msg(
|
||||
user_id=session_id,
|
||||
group_id=session_id,
|
||||
message_type=message_type,
|
||||
message=[
|
||||
v11.MessageSegment(
|
||||
type="longmsg",
|
||||
data={
|
||||
"id": forward_id
|
||||
}
|
||||
),
|
||||
],
|
||||
**kwargs
|
||||
)
|
||||
except Exception as e:
|
||||
nonebot.logger.warning("send_markdown error, send as plain text: %s" % e.__repr__())
|
||||
if isinstance(bot, v11.Bot):
|
||||
data = await bot.send_msg(
|
||||
message_type=message_type,
|
||||
message=markdown,
|
||||
user_id=int(session_id),
|
||||
group_id=int(session_id),
|
||||
**kwargs
|
||||
)
|
||||
elif isinstance(bot, v12.Bot):
|
||||
data = await bot.send_message(
|
||||
detail_type=message_type,
|
||||
message=v12.Message(
|
||||
v12.MessageSegment.text(
|
||||
text=markdown
|
||||
)
|
||||
),
|
||||
user_id=str(session_id),
|
||||
group_id=str(session_id),
|
||||
**kwargs
|
||||
)
|
||||
else:
|
||||
nonebot.logger.error("send_markdown: bot type not supported")
|
||||
data = {}
|
||||
return data
|
||||
|
||||
|
||||
class Markdown:
|
||||
@staticmethod
|
||||
def button(name: str, cmd: str, reply: bool = False, enter: bool = True) -> str:
|
||||
"""生成点击回调按钮
|
||||
Args:
|
||||
name: 按钮显示内容
|
||||
cmd: 发送的命令，已在函数内url编码，不需要再次编码
|
||||
reply: 是否以回复的方式发送消息
|
||||
enter: 自动发送消息则为True，否则填充到输入框
|
||||
|
||||
Returns:
|
||||
markdown格式的可点击回调按钮
|
||||
|
||||
"""
|
||||
return f"[{name}](mqqapi://aio/inlinecmd?command={encode_url(cmd)}&reply={str(reply).lower()}&enter={str(enter).lower()})"
|
||||
|
||||
@staticmethod
|
||||
def link(name: str, url: str) -> str:
|
||||
"""生成点击链接按钮
|
||||
Args:
|
||||
name: 链接显示内容
|
||||
url: 链接地址
|
||||
|
||||
Returns:
|
||||
markdown格式的链接
|
||||
|
||||
"""
|
||||
return f"[🔗{name}]({url})"
|
||||
|
||||
@staticmethod
|
||||
def escape(text: str) -> str:
|
||||
"""转义特殊字符
|
||||
Args:
|
||||
text: 需要转义的文本，请勿直接把整个markdown文本传入，否则会转义掉所有字符
|
||||
|
||||
Returns:
|
||||
转义后的文本
|
||||
|
||||
"""
|
||||
chars = "*[]()~_`>#+=|{}.!"
|
||||
for char in chars:
|
||||
text = text.replace(char, f"\\\\{char}")
|
||||
return text
|
||||
207
liteyuki/utils/orm.py
Normal file
207
liteyuki/utils/orm.py
Normal file
@ -0,0 +1,207 @@
|
||||
import os
|
||||
import pickle
|
||||
import sqlite3
|
||||
from types import NoneType
|
||||
from typing import Any
|
||||
|
||||
import nonebot
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class LiteModel(BaseModel):
|
||||
"""轻量级模型基类
|
||||
类型注解统一使用Python3.9的PEP585标准，如需使用泛型请使用typing模块的泛型类型
|
||||
不允许使用id, table_name以及其他SQLite关键字作为字段名，不允许使用JSON和ID，必须指定默认值，且默认值类型必须与字段类型一致
|
||||
"""
|
||||
__ID__: int = Field(None, alias='id')
|
||||
__TABLE_NAME__: str = Field(None, alias='table_name')
|
||||
|
||||
|
||||
class Database:
|
||||
TYPE_MAPPING = {
|
||||
int : "INTEGER",
|
||||
float : "REAL",
|
||||
str : "TEXT",
|
||||
bool : "INTEGER",
|
||||
bytes : "BLOB",
|
||||
NoneType: "NULL",
|
||||
|
||||
dict : "BLOB", # LITEYUKIDICT{key_name}
|
||||
list : "BLOB", # LITEYUKILIST{key_name}
|
||||
tuple : "BLOB", # LITEYUKITUPLE{key_name}
|
||||
set : "BLOB", # LITEYUKISET{key_name}
|
||||
}
|
||||
|
||||
# 基础类型
|
||||
BASIC_TYPE = [int, float, str, bool, bytes, NoneType]
|
||||
# 可序列化类型
|
||||
ITERABLE_TYPE = [dict, list, tuple, set]
|
||||
|
||||
LITEYUKI = "LITEYUKI"
|
||||
|
||||
# 字段前缀映射，默认基础类型为""
|
||||
FIELD_PREFIX_MAPPING = {
|
||||
dict : f"{LITEYUKI}DICT",
|
||||
list : f"{LITEYUKI}LIST",
|
||||
tuple : f"{LITEYUKI}TUPLE",
|
||||
set : f"{LITEYUKI}SET",
|
||||
type(LiteModel): f"{LITEYUKI}MODEL"
|
||||
}
|
||||
|
||||
def __init__(self, db_name: str):
|
||||
if not os.path.exists(os.path.dirname(db_name)):
|
||||
os.makedirs(os.path.dirname(db_name))
|
||||
self.conn = sqlite3.connect(db_name) # 连接对象
|
||||
self.conn.row_factory = sqlite3.Row # 以字典形式返回查询结果
|
||||
self.cursor = self.conn.cursor() # 游标对象
|
||||
|
||||
def auto_migrate(self, *args: LiteModel):
|
||||
"""
|
||||
自动迁移模型
|
||||
Args:
|
||||
*args: 模型类实例化对象，支持空默认值，不支持嵌套迁移
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
for model in args:
|
||||
if not model.__TABLE_NAME__:
|
||||
raise ValueError(f"数据模型{model.__class__.__name__}未提供表名")
|
||||
|
||||
# 若无则创建表
|
||||
self.cursor.execute(
|
||||
f'CREATE TABLE IF NOT EXISTS {model.__TABLE_NAME__} (id INTEGER PRIMARY KEY AUTOINCREMENT)'
|
||||
)
|
||||
|
||||
# 获取表结构
|
||||
new_fields, new_stored_types = (
|
||||
zip(
|
||||
*[(self._get_stored_field_prefix(model.__getattribute__(field)) + field, self._get_stored_type(model.__getattribute__(field)))
|
||||
for field in model.__annotations__]
|
||||
)
|
||||
)
|
||||
|
||||
# 原有的字段列表
|
||||
existing_fields = self.cursor.execute(f'PRAGMA table_info({model.__TABLE_NAME__})').fetchall()
|
||||
existing_types = [field['name'] for field in existing_fields]
|
||||
|
||||
# 检测缺失字段，由于SQLite是动态类型，所以不需要检测类型
|
||||
for n_field, n_type in zip(new_fields, new_stored_types):
|
||||
if n_field not in existing_types:
|
||||
nonebot.logger.debug(f'ALTER TABLE {model.__TABLE_NAME__} ADD COLUMN {n_field} {n_type}')
|
||||
self.cursor.execute(
|
||||
f'ALTER TABLE {model.__TABLE_NAME__} ADD COLUMN {n_field} {n_type}'
|
||||
)
|
||||
|
||||
# 检测多余字段进行删除
|
||||
for e_field in existing_types:
|
||||
if e_field not in new_fields and e_field not in ['id']:
|
||||
nonebot.logger.debug(f'ALTER TABLE {model.__TABLE_NAME__} DROP COLUMN {e_field}')
|
||||
self.cursor.execute(
|
||||
f'ALTER TABLE {model.__TABLE_NAME__} DROP COLUMN {e_field}'
|
||||
)
|
||||
|
||||
self.conn.commit()
|
||||
|
||||
def save(self, *args: LiteModel) -> [int | tuple[int, ...]]:
|
||||
"""
|
||||
保存或更新模型
|
||||
Args:
|
||||
*args: 模型类实例化对象，支持空默认值，不支持嵌套迁移
|
||||
Returns:
|
||||
|
||||
"""
|
||||
ids = []
|
||||
for model in args:
|
||||
if not model.__TABLE_NAME__:
|
||||
raise ValueError(f"数据模型{model.__class__.__name__}未提供表名")
|
||||
if not self.cursor.execute(f'PRAGMA table_info({model.__TABLE_NAME__})').fetchall():
|
||||
raise ValueError(f"数据表{model.__TABLE_NAME__}不存在，请先迁移{model.__class__.__name__}模型")
|
||||
|
||||
stored_fields, stored_values = [], []
|
||||
for r_field in model.__annotations__:
|
||||
r_value = model.__getattribute__(r_field)
|
||||
stored_fields.append(self._get_stored_field_prefix(r_value) + r_field)
|
||||
|
||||
if type(r_value) in Database.BASIC_TYPE:
|
||||
# int str float bool bytes NoneType
|
||||
stored_values.append(r_value)
|
||||
|
||||
elif type(r_value) in Database.ITERABLE_TYPE:
|
||||
# dict list tuple set
|
||||
stored_values.append(pickle.dumps(self._flat_save(r_value)))
|
||||
|
||||
elif isinstance(r_value, LiteModel):
|
||||
# LiteModel TABLE_NAME:ID
|
||||
stored_values.append(f"{r_value.__TABLE_NAME__}:{self.save(r_value)}")
|
||||
|
||||
else:
|
||||
raise ValueError(f"不支持的数据类型{type(r_value)}")
|
||||
nonebot.logger.debug(f"INSERT OR REPLACE INTO {model.__TABLE_NAME__} ({','.join(stored_fields)}) VALUES ({','.join([_ for _ in stored_values])})")
|
||||
self.cursor.execute(
|
||||
f"INSERT OR REPLACE INTO {model.__TABLE_NAME__} ({','.join(stored_fields)}) VALUES ({','.join(['?' for _ in stored_values])})",
|
||||
stored_values
|
||||
)
|
||||
ids.append(self.cursor.lastrowid)
|
||||
self.conn.commit()
|
||||
return tuple(ids) if len(ids) > 1 else ids[0]
|
||||
|
||||
# 检测id字段是否有1，有则更新，无则插入
|
||||
|
||||
def _flat_save(self, obj) -> Any:
|
||||
"""扁平化存储
|
||||
|
||||
Args:
|
||||
obj: 需要存储的对象
|
||||
|
||||
Returns:
|
||||
存储的字节流
|
||||
"""
|
||||
# TODO 递归扁平化存储
|
||||
if type(obj) in Database.ITERABLE_TYPE:
|
||||
for i, item in enumerate(obj) if type(obj) in [list, tuple, set] else obj.items():
|
||||
if type(item) in Database.BASIC_TYPE:
|
||||
continue
|
||||
elif type(item) in Database.ITERABLE_TYPE:
|
||||
obj[i] = pickle.dumps(self._flat_save(item))
|
||||
elif isinstance(item, LiteModel):
|
||||
obj[i] = f"{item.__TABLE_NAME__}:{self.save(item)}"
|
||||
else:
|
||||
raise ValueError(f"不支持的数据类型{type(item)}")
|
||||
else:
|
||||
raise ValueError(f"不支持的数据类型{type(obj)}")
|
||||
|
||||
@staticmethod
|
||||
def _get_stored_field_prefix(value) -> str:
|
||||
"""获取存储字段前缀，一定在后加上字段名
|
||||
|
||||
LiteModel -> LITEYUKIID
|
||||
|
||||
dict -> LITEYUKIDICT
|
||||
|
||||
list -> LITEYUKILIST
|
||||
|
||||
tuple -> LITEYUKITUPLE
|
||||
|
||||
set -> LITEYUKISET
|
||||
|
||||
* -> ""
|
||||
Args:
|
||||
value: 储存的值
|
||||
|
||||
Returns:
|
||||
Sqlite3存储字段
|
||||
"""
|
||||
return Database.FIELD_PREFIX_MAPPING.get(type(value), "")
|
||||
|
||||
@staticmethod
|
||||
def _get_stored_type(value) -> str:
|
||||
"""获取存储类型
|
||||
|
||||
Args:
|
||||
value: 储存的值
|
||||
|
||||
Returns:
|
||||
Sqlite3存储类型
|
||||
"""
|
||||
return Database.TYPE_MAPPING.get(type(value), "TEXT")
|
||||
7
liteyuki/utils/permission.py
Normal file
7
liteyuki/utils/permission.py
Normal file
@ -0,0 +1,7 @@
|
||||
from nonebot.adapters.onebot import v11
|
||||
|
||||
from liteyuki.utils.ly_typing import T_GroupMessageEvent, T_MessageEvent
|
||||
|
||||
GROUP_ADMIN = v11.GROUP_ADMIN
|
||||
GROUP_OWNER = v11.GROUP_OWNER
|
||||
|
||||
54
liteyuki/utils/resource.py
Normal file
54
liteyuki/utils/resource.py
Normal file
@ -0,0 +1,54 @@
|
||||
import os
|
||||
|
||||
import nonebot
|
||||
import yaml
|
||||
from typing import Any
|
||||
|
||||
from liteyuki.utils.data import LiteModel
|
||||
|
||||
_resource_data = {}
|
||||
_loaded_resource_packs = [] # 按照加载顺序排序
|
||||
|
||||
|
||||
class ResourceMetadata(LiteModel):
|
||||
name: str = "Unknown"
|
||||
version: str = "0.0.1"
|
||||
description: str = "Unknown"
|
||||
path: str
|
||||
|
||||
|
||||
def load_resource_from_dir(path: str):
|
||||
"""
|
||||
把资源包按照文件相对路径加载到资源包中，后加载的优先级更高，顺便加载语言
|
||||
Args:
|
||||
path: 资源文件夹
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
for root, dirs, files in os.walk(path):
|
||||
for file in files:
|
||||
relative_path = os.path.relpath(os.path.join(root, file), path).replace("\\", "/")
|
||||
abs_path = os.path.join(root, file).replace("\\", "/")
|
||||
_resource_data[relative_path] = abs_path
|
||||
if os.path.exists(os.path.join(path, "metadata.yml")):
|
||||
with open(os.path.join(path, "metadata.yml"), "r", encoding="utf-8") as f:
|
||||
metadata = yaml.safe_load(f)
|
||||
else:
|
||||
metadata = ResourceMetadata()
|
||||
metadata["path"] = path
|
||||
if os.path.exists(os.path.join(path, "lang")):
|
||||
from liteyuki.utils.language import load_from_dir
|
||||
load_from_dir(os.path.join(path, "lang"))
|
||||
_loaded_resource_packs.append(ResourceMetadata(**metadata))
|
||||
|
||||
|
||||
def get(path: str, default: Any = None) -> str | Any:
|
||||
"""
|
||||
获取资源包中的文件
|
||||
Args:
|
||||
default: 默认
|
||||
path: 文件相对路径
|
||||
Returns: 文件绝对路径
|
||||
"""
|
||||
return _resource_data.get(path, default)
|
||||
83
liteyuki/utils/tools.py
Normal file
83
liteyuki/utils/tools.py
Normal file
@ -0,0 +1,83 @@
|
||||
from importlib.metadata import PackageNotFoundError, version
|
||||
from urllib.parse import quote
|
||||
|
||||
|
||||
def convert_size(size: int, precision: int = 2, add_unit: bool = True, suffix: str = "iB") -> str:
|
||||
"""把字节数转换为人类可读的字符串，计算正负
|
||||
|
||||
Args:
|
||||
|
||||
add_unit: 是否添加单位，False后则suffix无效
|
||||
suffix: iB或B
|
||||
precision: 浮点数的小数点位数
|
||||
size (int): 字节数
|
||||
|
||||
Returns:
|
||||
|
||||
str: The human-readable string, e.g. "1.23 GB".
|
||||
"""
|
||||
is_negative = False
|
||||
if size < 0:
|
||||
is_negative = True
|
||||
size = -size
|
||||
|
||||
for unit in ["", "K", "M", "G", "T", "P", "E", "Z", "Y"]:
|
||||
if size < 1024:
|
||||
if add_unit:
|
||||
result = f"{size:.{precision}f} {unit}" + suffix
|
||||
return f"-{result}" if is_negative else result
|
||||
else:
|
||||
return f"{size:.{precision}f}"
|
||||
size /= 1024
|
||||
if add_unit:
|
||||
return f"{size:.{precision}f} Y" + suffix
|
||||
else:
|
||||
return f"{size:.{precision}f}"
|
||||
|
||||
|
||||
def de_escape(text: str) -> str:
|
||||
str_map = {
|
||||
"[": "[",
|
||||
"]": "]",
|
||||
"&": "&",
|
||||
",": ",",
|
||||
}
|
||||
for k, v in str_map.items():
|
||||
text = text.replace(k, v)
|
||||
|
||||
return text
|
||||
|
||||
|
||||
def encode_url(text: str) -> str:
|
||||
return quote(text, safe="")
|
||||
|
||||
|
||||
def keywords_in_text(keywords: list[str], text: str, all_matched: bool) -> bool:
|
||||
"""
|
||||
检查关键词是否在文本中
|
||||
Args:
|
||||
keywords: 关键词列表
|
||||
text: 文本
|
||||
all_matched: 是否需要全部匹配
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
if all_matched:
|
||||
for keyword in keywords:
|
||||
if keyword not in text:
|
||||
return False
|
||||
return True
|
||||
else:
|
||||
for keyword in keywords:
|
||||
if keyword in text:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def check_for_package(package_name: str) -> bool:
|
||||
try:
|
||||
version(package_name)
|
||||
return True
|
||||
except PackageNotFoundError:
|
||||
return False
|
||||
Reference in New Issue
Block a user