A collection of reusable python core library from AI Lingues.
Project description
PyCoreLibs
AI Lingues 基础组件库- PyCoreLibs
Message
消息库,基于redis的异步消息框架
参考:redis/README.md
Network
search 搜索引擎
互联网搜索引擎
requests 请求
-
fetch_url 强化版通用网络请求函数,增加了重试机制,并适用于复杂场景。
-
is_url 验证是否为合法url
scrawer 网页文本获取
- HTMLContent 提取网页内容
security 模块
base64s 模块
cryptographys 模块
AsymmetricCrypto: 稳健可靠的非对称加解密 + 签名/验签 工具类
说明
- 加解密:RSA-OAEP(SHA-256) + AES-GCM(256) 的混合加密(任意长度数据)
- 签名:RSA-PSS(SHA-256),返回 JSON 打包(Base64)
- 设计目标:健壮、跨语言友好(JSON+Base64)、异常统一封装
| 函数 | 说明 |
|---|---|
generate_keypair |
生成密钥对(PEM) |
encrypt |
公钥加密(混合加密) |
decrypt |
私钥解密 |
sign |
私钥签名(RSA-PSS + SHA-256) |
verify |
公钥验签 |
示例
from pycorelibs.security.cryptographys import AsymmetricCrypto
message = b"hello, asymmetric crypto! \xf0\x9f\x94\x90"
# 1) 生成密钥对(内存)
priv, pub = AsymmetricCrypto.generate_keypair(
key_size=2048, private_password=b"pass"
)
# 2) 加密(可选 AAD)
aad = b"meta:example"
blob = AsymmetricCrypto.encrypt(message, pub, aad=aad)
# 3) 解密
plain = AsymmetricCrypto.decrypt(blob, priv, private_password=b"pass")
assert plain == message
print("Decrypt OK:", plain.decode("utf-8"))
# 4) 签名
sig_blob = AsymmetricCrypto.sign(message, priv, private_password=b"pass")
# 5) 验签(成功)
ok = AsymmetricCrypto.verify(message, sig_blob, pub)
print("Verify OK:", ok)
# 6) 验签(失败示例:数据被改动)
tampered = message + b"!"
bad = AsymmetricCrypto.verify(tampered, sig_blob, pub)
print("Verify with tamper:", bad)
generate_keypair
生成 RSA 密钥对(PKCS#8 私钥 + X.509 公钥,PEM 格式)
参数说明
-
key_size (int, optional)
2048/3072/4096(推荐 2048+). Defaults to 2048.
-
private_password (Optional[bytes], optional)
给私钥加密的口令(None 表示不加密). Defaults to None.
-
save_to (Optional[str], optional)
可选保存目录(写入 private.pem / public.pem). Defaults to None.
Raises
-
ValueError: RSA key_size 仅支持 2048/3072/4096。
-
AsymmetricCryptoError: 生成密钥对失败
Returns
-
Tuple[bytes, bytes]
生成 RSA 密钥对(PKCS#8 私钥 + X.509 公钥,PEM 格式)
fingerprints 模块
跨平台硬件指纹
- 只用 Python 库,不走系统命令;信息更稳定
- 多网卡:枚举全部,带过滤选项(是否包含虚拟/回环、是否只取UP、是否要求有IP)
- 主 MAC 选择策略可配置
- 生成哈希指纹(可传入 salt 提高伪造成本)
generate_hardware_fingerprint
生成硬件指纹(不包含TPM)
参数说明
-
isGPU (bool):
是否获取GPU信息,缺省为False
-
salt (Optional[str], optional):
额外盐值(字符串)。如用于授权/反爬,建议传入,提高伪造成本。. Defaults to None.
Returns
Dict[str, Any]: { "fid": "<sha512 hex>", "hash": "sha512", "input": data, }
-
"fid": "<sha512 hex>"
最终指纹ID,长度128个字符
-
"hash": "sha512"
固定标记,加密算法名
-
input": data
加密原始数据
示例
from pycorelibs.security.fingerprints import generate_hardware_fingerprint
# 典型:只要UP、排除虚拟和回环;无IP也可(离线机器也能给出指纹)
fp = generate_hardware_fingerprint()
print(fp)
print("*" * 80)
# 更“联网语义”:要求接口必须有非回环IP
fp_net = generate_hardware_fingerprint(require_ip=True)
print(fp_net)
print("*" * 80)
# 有授权/反爬需求:加盐
fp_salted = generate_hardware_fingerprint(salt="your-secret-salt-123")
print(fp_salted)
print("*" * 80)
md5s 模块
md5
计算输入内容的 MD5 值
参数说明
-
text (Any):
任意可转为字符串的输入内容。
-
length (int):
结果长度,可选 16 或 32(默认 32)。
-
isHex (bool):
是否以十六进制格式输出(默认为 True)。
返回
-
(bool, Optional[str])
成功与否、以及 MD5 值字符串或 None。
system模块
获取当前系统信息
- OS 操作系统
- CPU
- Macs 网卡
- MB 主板
runtime 运行期信息模块
获取运行期相关信息
RunTimeInfo类
get_boot_time
获取系统启动时间和系统已运行时长
跨平台细节:
- Linux:boot_time 基本来自 btime(自启动以来的挂钟差,包含睡眠/休眠的停留时间)。
- Windows:psutil 内部通常用系统 tick 估算,一般不包含睡眠时间(更接近“活跃运行时长”)。
参数说明
-
fmt (Optional[str], optional):
时间格式化字符串,默认为 None 使用当前系统设定时区的本地格式.
Returns
- Dict[str, Any]: 包含启动时间戳、格式化时间、已运行秒数和可读格式
get_uptime
获取自 start_mono(monotonic 起点)至今的运行时长(不受系统时钟跳变影响,如被 NTP/手动调整,或夏令时切换)
参数说明
-
start_mono (float, optional):
计算开始时间. Defaults to SERVICE_START.
-
fmt (Optional[str], optional):
时间格式化字符串,默认为 None 使用当前系统设定时区的本地格式.. Defaults to None.
Returns
-
Dict[str, Any]:
包含启动时间戳、格式化时间、已运行秒数和可读格式
Utils
config配置模块
通用的本地配置存取层(与数据结构无关)
- 面向任意Python对象:通过可插拔 Serializer 适配(JSON/YAML/TOML/Pickle/自定义均可)
- 原子写入、自动创建目录、可选自动保存
- 多命名空间:各自文件、各自序列化器,互不干扰
- 默认保存到“应用同目录”
- 支持配置文件加密存储
ConfigRepo类
仓库:管理多个命名空间。
一个应用一个仓库,里面可注册多个命名空间。
每个命名空间:独立文件 + 独立序列化器 + 独立默认值
构造函数
__init__(self, root: Optional[str] = None, subdir: Optional[str] = None)
作用:创建一个配置仓库实例,确定所有命名空间文件的根目录。
参数说明
-
root:str | None
仓库根目录的绝对/相对路径。若为 None,则自动解析为应用同目录:
-
PyInstaller 打包后:os.path.dirname(sys.executable)
-
普通脚本:os.path.dirname(os.path.abspath(sys.argv[0] or file))
-
-
subdir:str | None
root 下的子目录名(例如 "config"),用于把配置集中到 root/subdir。
返回
无。
副作用
若目录不存在会自动创建。
示例
```python
# 保存到应用目录下的 ./config/
repo = ConfigRepo(subdir="config")
# 保存到指定绝对路径
repo = ConfigRepo(root="D:/myapp/config")
```
-
register注册命名空间
register(self, name: str, serializer: Serializer[Any], filename: Optional[str] = None, default_factory: Optional[Callable[[], Any]] = None, autosave: bool = False, backup_on_save: bool = False, cipher: Optional[AesGcmBox] = None, cipher_policy: str = "auto", encrypt_on_save_if_cipher: bool = True) -> Namespace[Any]
作用:
注册一个命名空间(即一个逻辑“配置空间”),绑定文件、序列化器、默认值与加密策略。
参数:
-
name:str 命名空间名称(在仓库内唯一)。
-
serializer:Serializer[Any] 序列化器实例:如 JsonSerializer()、PickleSerializer() 或自定义实现(需提供 dumps(obj)->bytes 与 loads(bytes)->obj)。
-
filename:str | None 文件名。若省略:当 serializer 为 JsonSerializer 时默认 ".json",否则为 ".bin"。
-
default_factory:() -> Any | None 当文件不存在或需重置时,用此函数惰性生成默认对象(如 dict/list/自定义对象)。若省略,默认值为 None。
-
autosave:bool 设为 True 时,调用 set() / update() / reset() 等修改后会自动保存到文件。
-
backup_on_save:bool 保存前若目标文件已存在,将其重命名为 *.bak 作为备份。
-
cipher:AesGcmBox | None 可选的加密盒;提供则进行加密存储(AES-GCM)。不提供则明文存储。
-
cipher_policy:"auto" | "require"(默认 "auto")
-
"auto":读取时若检测到加密魔数则解密,否则按明文加载(兼容旧文件);
-
"require":强制要求文件为加密格式,若无魔数直接抛错(适合敏感空间)。
-
-
encrypt_on_save_if_cipher:bool(默认 True) 当 cipher 存在且此前是明文加载的旧文件,保存时会自动迁移为加密格式。
返回:
Namespace[Any]:已注册的命名空间对象。
异常:
无(但后续加载/保存时可能抛出序列化/IO/解密类异常)。
示例:
# 明文 JSON 偏好设置 ui = repo.register( name="ui", serializer=JsonSerializer(), filename="ui.json", default_factory=lambda: {"theme": "light", "recent_files": []}, autosave=True ) # 加密的 Pickle 缓存(旧明文文件可兼容读取,保存后自动迁移为密文) cache = repo.register( name="cache", serializer=PickleSerializer(), filename="cache.dat", default_factory=list, autosave=True, cipher=AesGcmBox(password=os.environ["CACHE_PASSWORD"]), cipher_policy="auto", encrypt_on_save_if_cipher=True, )
-
-
space 取命名空间对象
space(self, name: str) -> Namespace[Any]
作用:按名称获取已注册的命名空间对象。
参数:
- name:str 注册时使用的命名空间名称。
返回:
- Namespace[Any]:命名空间对象。
异常:
- KeyError:当 name 尚未注册时抛出。
示例:
cache_ns = repo.space("cache")
-
get直接存取对象(便捷方法)
get(self, name: str) -> Any
作用:从指定命名空间读取对象(内部若未加载会自动调用 Namespace.load())。
参数:
- name:str
返回:
- Any:命名空间中的对象(可能是 dict/list/自定义对象/None 等)。
异常:
-
KeyError:命名空间未注册;
-
反序列化/解密/IO 异常:底层抛出的 ValueError、OSError 等。
示例:
ui_data = repo.get("ui") # 例如: {"theme": "light", "recent_files": []}
-
set(self, name: str, value: Any, save: Optional[bool] = None) -> None
作用:设置命名空间对象并可选择保存。
参数:
-
name:str
-
value:Any 要写入的对象(会通过绑定的 serializer 持久化)。
-
save:bool | None
True:立即保存;
False:只更新内存缓存,不落盘;
None:跟随命名空间的 autosave 设置(若为 True 则保存)。
返回:无。
异常:同 get()。
示例:
repo.set("ui", {"theme": "dark"}, save=True)
-
-
update(self, name: str, mutator: Callable[[Any], Any], save: Optional[bool] = None) -> Any
作用:在原对象基础上以函数式方式变换并回写(读 -> 变换 -> 写)。
参数:
-
name:str
-
mutator:(obj: Any) -> Any 入参是当前对象,返回变换后的新对象(函数内应返回新值)。
-
save:bool | None 语义同 set() 的 save。
返回:
- Any:变换后的新对象。
异常:同 get()/set()。
示例:
def add_recent(d): d = d or {} rec = (d.get("recent_files") or []) + ["a.sdf"] return {**d, "recent_files": rec[-10:]} # 只保留最近10条 new_ui = repo.update("ui", add_recent, save=True)
-
-
批量操作 save_all(self) -> None
作用:对所有“已加载过”的命名空间执行 save()(避免无意义的空写)。
参数:无。
返回:无。
异常:底层保存可能抛出 IO/序列化/加密相关异常。
注意:
仅对内部标记 _loaded == True 的空间生效;
若你需要确保所有空间都有初值并保存,可先 load_all() 或逐个 space(...).get() 激活。
示例:
# 某些空间 autosave=False,集中一次性保存 repo.save_all()
-
load_all(self) -> None
作用:对所有已注册的命名空间执行 load()(将文件内容加载到内存缓存)。
参数:无。
返回:无。
异常:底层加载可能抛出 IO/反序列化/解密相关异常。
示例:
# 应用启动时预热全部配置 repo.load_all()
常见问题与注意事项
-
路径选择:
默认落在应用同目录,便于“就地携带”。如需跨用户共享或写权限受限(Windows Program Files),建议显式指定 root。
-
默认值:
default_factory 推荐传入 可调用(如 dict/list/lambda),避免可变对象作为默认参数带来的共享副作用。
-
加密策略:
-
"auto":适合从旧明文平滑迁移为密文;
-
"require":适合强约束的敏感空间(若非加密立刻报错)。
-
-
备份文件:
backup_on_save=True 时会生成 *.bak;其内容可能为上一次保存时的明文或密文,读取时可用 AesGcmBox.is_encrypted() 判别。
-
异常处理:
解密失败通常是口令/密钥不匹配或文件损坏;请捕获 ValueError 并引导用户检查口令来源。
-
线程/进程并发:
当前实现是单进程内使用的简单本地层;若存在多进程/多实例同时写入同一文件,请自行加锁或采用更高级别的存储方案。
典型用法示例(整合)
from pycorelibs.utils.config import ConfigRepo, JsonSerializer, PickleSerializer
from pycorelibs.security.aesgcmbox import AesGcmBox
import os
repo = ConfigRepo(subdir="config")
# 明文 UI 配置
ui = repo.register(
name="ui",
serializer=JsonSerializer(),
filename="ui.json",
default_factory=lambda: {"theme": "light", "recent_files": []},
autosave=True,
)
# 加密的缓存(从旧明文自动迁移)
cache = repo.register(
name="cache",
serializer=PickleSerializer(),
filename="cache.dat",
default_factory=list,
autosave=True,
cipher=AesGcmBox(password=os.environ.get("CACHE_PW", "CHANGE_ME")),
cipher_policy="auto",
encrypt_on_save_if_cipher=True,
)
# 读取 / 更新 / 批量保存
ui_data = repo.get("ui")
repo.update("ui", lambda d: {**(d or {}), "theme": "dark"})
repo.update("cache", lambda lst: (lst or []) + ["session#1"])
repo.save_all()
files文件处理
-
EnumFileSizeUnit
-
FilesInfo 类
-
get_file_size 获取文件大小,并返回指定单位的大小。
支持B,KB,MB,GB,TB -
line_count
获取指定文件的总行数 Windows系统: 使用换行符作为判断标记 Linux系统: 使用系统自带的wc命令统计 注: 近适用于文本类型文件,其他类型如二进制文件统计结果不具备参考意义 -
get_directory_last_modified_time 获取指定目录最后修改时间
-
get_directory_last_modified_timestamp 获取指定目录最后修改时间戳
-
formate 数据格式化指定类型
Formates 类,静态方法如下:
- fInt 转换为整数
- fFloat 转换为浮点数
- fBool 转换为布尔值
- fDate 按指定格式解析日期
- fEnum 校验值是否在允许集合内(用于枚举字段)
- fJSON 解析为 JSON 对象
logger日志
- 支持自动产生日志
- 支持定制向控制台和日志文件双管道输出
- 支持rich组件进度条绑定控制台输出
- 支持debug、info、warning、error和critical五个级别的日志
- 支持异常信息输出
- 全局单例模式
1. 初始化与单例行为
-
第一次调用 Logger(...) 会真正完成初始化并创建日志文件。
-
后续 Logger(...) 调用将直接返回同一个实例,忽略新传入参数。
Exsample:
from pycorelibs.utils.logger import Logger
logger = Logger(
log_dir="./logs/app", # 日志目录,默认 "./logs"
filename_prefix="biotech", # 文件名前缀,可为空
level_console=Logger.INFO, # 控制台最小级别,默认 INFO
level_file=Logger.DEBUG, # 文件最小级别,默认 DEBUG
)
2. 日志输出目标(LogOutputMode)
LogOutputMode = Literal["both", "file", "console"]
-
output="both"(默认):控制台 + 文件
-
output="file":仅文件
-
output="console":仅控制台
3. 日志输出层级
DEBUG = Logger.DEBUG
INFO = Logger.INFO
WARNING = Logger.WARNING
ERROR = Logger.ERROR
CRITICAL = Logger.CRITICAL
4. 核心方法
# 多参数类似 print,内部用 sep 拼接为字符串
logger.debug(*messages, output="both", sep=" ", **kwargs)
logger.info(*messages, output="both", sep=" ", **kwargs)
logger.warning(*messages, output="both", sep=" ", **kwargs)
logger.error(*messages, output="both", sep=" ", **kwargs)
logger.critical(*messages, output="both", sep=" ", **kwargs)
# 异常日志(自动附带堆栈)
logger.exception(*messages, output="both", sep=" ", exc_info=True, **kwargs)
- *messages 会被转换为字符串并用 sep 连接。
- 默认 sep=" ",行为接近 print("a", "b")。
- exception() 等价于 error(..., exc_info=True),用于异常场景。
5. 辅助方法
# 返回底层 logging.Logger(不支持 output 参数)
std_logger = logger.get_logger()
# 返回 Rich Console,可用于手工渲染表格/进度条
console = logger.get_console()
# 返回当前日志文件路径(pathlib.Path)
log_path = logger.get_log_file_path()
6. 使用示例
6.1 基础用法(默认配置)
from pycorelibs.utils.logger import Logger
logger = Logger(filename_prefix="TRAIN")
console = Logger().get_console()
# 打印日志
logger.info("开始处理数据集", "size=", 5000)
logger.warning("这是 WARNING 信息,这条日志只写入文件,不在控制台出现",output="file")
logger.error("这是 ERROR 信息")
logger.critical("这是 CRITICAL 信息")
logger.debug("临时调试信息,仅控制台可见", output="console")
# 使用 Rich 的 Table
table = Table(title="表格")
table.add_column("姓名", justify="right", style="cyan", no_wrap=True)
table.add_column("年龄", justify="center", style="magenta")
table.add_column("城市", justify="center", style="green")
table.add_row("Alice", "24", "New York")
table.add_row("Bob", "30", "San Francisco")
table.add_row("Charlie", "29", "Chicago")
console.print(table)
# 使用 Rich 的 Progress
with Progress(console=console) as progress:
task = progress.add_task("Processing...", total=10)
for _ in range(100):
progress.update(task, advance=1)
console.print(f"日志已写入文件")
输出:
[11/17/25 11:09:19] INFO 开始处理数据集 size= 5000
ERROR 这是 ERROR 信息
CRITICAL 这是 CRITICAL 信息
DEBUG 临时调试信息,仅控制台可见
表格
┏━━━━━━━━━┳━━━━━━┳━━━━━━━━━━━━━━━┓
┃ 姓名 ┃ 年龄 ┃ 城市 ┃
┡━━━━━━━━━╇━━━━━━╇━━━━━━━━━━━━━━━┩
│ Alice │ 24 │ New York │
│ Bob │ 30 │ San Francisco │
│ Charlie │ 29 │ Chicago │
└─────────┴──────┴───────────────┘
Processing... ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 100% 0:00:00
日志已写入文件
6.2 指定日志目录和文件名前缀
from pycorelibs.utils.logger import Logger
logger = Logger(
log_dir="/var/log/",
filename_prefix="train record",
level_console=Logger.INFO,
level_file=Logger.DEBUG,
)
logger.info("训练任务启动", "job_id=", "20251117_001")
print("当前日志文件:", logger.get_log_file_path())
输出:
/var/log/train_record_20251117_101530.log
6.3 打印异常堆栈
logger = Logger()
try:
1 / 0
except Exception as e:
logger.exception("计算过程发生异常:", e)
# 默认 output="both",即控制台 + 文件都会看到完整 traceback
6.4 结合底层 logging.Logger 使用
如果第三方库需要一个 logging.Logger 实例,可以这样:
logger = Logger()
std_logger = logger.get_logger()
# 标准 logging 使用方式(不支持 output、print 风格多参数)
std_logger.info("third-party module initialized")
strings 字符串处理
-
UniCodeGenerator 模块
创建随机码,日期前缀 + 随机码
可配置、使用 secrets 生成高强度随机码的编码生成类
场景 参数示例 邀请码(短) prefix_date=False, random_length=6订单号(当天唯一) prefix_date=True, random_length=4URL 安全短码 charset=string.ascii_letters + string.digits -
json_dumps_bytes 函数
将 dict 序列化为紧凑 UTF-8 JSON bytes
-
json_loads_bytes 函数
将 UTF-8 JSON bytes 解析为 dict,并进行基本类型校验
-
split_text_by_marker函数
作用: 按照指定字符串 marker 分割文本,可选择是否保留 marker,并支持严格模式。
参数:
- text (str): 需要分割的原始文本。
- marker (str): 分割标记字符串。
- keep_marker (bool): 是否保留 marker 到结果中,默认为 True。
- strict (bool): 严格模式。当 marker 不存在时抛异常。默认为 False。
返回:
- List[str]: 分割后的文本列表。 如果 strict=False 且 marker 不存在,返回空列表。 如果 strict=True 且 marker 不存在,抛出 ValueError。
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distributions
Built Distributions
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file pycorelibs-0.3.0-cp311-cp311-win_amd64.whl.
File metadata
- Download URL: pycorelibs-0.3.0-cp311-cp311-win_amd64.whl
- Upload date:
- Size: 3.8 MB
- Tags: CPython 3.11, Windows x86-64
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.12.3
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
9b41a00b8b934371f90cdfb7ca1102f802c021ff2e4375f77970578b1f7c27b6
|
|
| MD5 |
aca6bcce2f8741c1b90baf9d21acd892
|
|
| BLAKE2b-256 |
cd5817881d45bcf2df3d4e66eb85637bbce3d94155a4d3ee38640fdba417190e
|
File details
Details for the file pycorelibs-0.3.0-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl.
File metadata
- Download URL: pycorelibs-0.3.0-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl
- Upload date:
- Size: 8.7 MB
- Tags: CPython 3.11, manylinux: glibc 2.17+ x86-64, manylinux: glibc 2.28+ x86-64
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.12.3
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
8082dde5fc04ee96fe96eaba61d2d21973e694deced3f0b9b3b7671231231720
|
|
| MD5 |
221e63fb379bb3c1fdbcc67fe6ac968f
|
|
| BLAKE2b-256 |
1d19f212328dbef2621d03075f5f7255ed49fdf26ff39ca7064fba349badc268
|