Python实用工具库srsly深度解析：从数据读写到高效开发的全场景实践

# 基础安装（支持JSON、Pickle、MsgPack）
pip install srsly

# 可选：安装YAML支持
pip install srsly[yaml]

import srsly

# 读取普通JSON文件
data = srsly.read_json("data.json")
print(f"读取到的数据类型：{type(data)}")  # 输出：<class 'dict'>或<class 'list'>

# 读取压缩JSON文件（.json.gz）
compressed_data = srsly.read_json("data.json.gz")
print(f"压缩文件数据长度：{len(compressed_data)}")

# 准备数据
user = {
    "name": "Alice",
    "age": 30,
    "skills": ["Python", "Machine Learning"]
}

# 写入普通JSON文件（默认缩进4空格）
srsly.write_json("user.json", user)

# 写入压缩JSON文件（指定压缩级别=6）
srsly.write_json("user.json.gz", user, compress=6)

with srsly.open_jsonl("large_data.jsonl", "r") as f:
    for line in f:
        process(line)  # 处理每一行数据

pip install pyyaml  # 或通过srsly[yaml]安装

# 读取配置文件
config = srsly.read_yaml("config.yaml")
print(f"学习率：{config['training']['learning_rate']}")

# 写入YAML文件
new_config = {
    "model": "BERT",
    "epoch": 10,
    "optimizer": "AdamW"
}
srsly.write_yaml("new_config.yaml", new_config)

training:
  learning_rate: 0.001
  batch_size: 32
model:
  name: "spaCy NER"
  version: "3.7.0"

# 使用safe_load模式防止恶意YAML代码执行
config = srsly.read_yaml("config.yaml", loader="safe")

import pickle
from sklearn.linear_model import LogisticRegression

# 训练模型并保存
model = LogisticRegression()
model.fit(X_train, y_train)
srsly.write_pickle(model, "model.pkl")

# 加载模型
loaded_model = srsly.read_pickle("model.pkl")
predictions = loaded_model.predict(X_test)

pip install msgpack  # srsly默认包含对msgpack的支持

# 序列化数据
data = {"id": 1, "name": "Bob", "scores": [90, 85, 92]}
srsly.write_msgpack("data.msgpack", data)

# 反序列化
loaded_data = srsly.read_msgpack("data.msgpack")
print(f"反序列化后的数据：{loaded_data}")

# 写入压缩JSON文件（自动识别扩展名）
data = [{"key": "value"}] * 100000
srsly.write_json("large_data.json.gz", data)  # 等效于gzip压缩

# 读取压缩Pickle文件
model = srsly.read_pickle("model.pkl.bz2")  # 自动解压缩bz2格式

# 设置gzip压缩级别为9（最高压缩率）
srsly.write_json(
    "data.json.gz", 
    data, 
    compress=("gzip", {"level": 9})
)

# 使用bz2压缩（默认级别为5）
srsly.write_pickle("data.pkl.bz2", data, compress="bz2")

with srsly.Tempfile() as tmp:
    # 在临时文件中写入数据
    tmp.write_json({"key": "temp_value"})
    # 获取临时文件路径
    print(f"临时文件路径：{tmp.name}")
# 上下文结束后文件自动删除

from srsly import paths

# 确保目录存在（递归创建）
paths.ensure_dir("output/models")

# 检查文件是否存在且可读
if paths.is_readable("data.csv"):
    process_data()

# 生成唯一文件名（避免覆盖）
unique_path = paths.unique_path("output/", "report", ext=".csv")

import srsly
from typing import List, Dict

# 1. 读取配置文件
config = srsly.read_yaml("pipeline_config.yaml")
input_path = config["data"]["input_path"]
min_score = config["processing"]["min_score"]

# 2. 读取并清洗数据
def filter_records(records: List[Dict]) -> List[Dict]:
    """过滤掉分数低于阈值的记录"""
    return [
        record for record in records
        if record.get("score", 0) >= min_score
    ]

with srsly.open_jsonl(input_path, "r") as f:
    raw_data = list(f)
clean_data = filter_records(raw_data)

# 3. 保存为压缩MsgPack文件
output_path = "processed_data.msgpack.gz"
srsly.write_msgpack(output_path, clean_data)
print(f"清洗后数据已保存至：{output_path}")

# 4. 保存统计信息（总记录数、有效率）
stats = {
    "total_records": len(raw_data),
    "valid_records": len(clean_data),
    "efficiency": len(clean_data) / len(raw_data)
}
srsly.write_pickle(stats, "processing_stats.pkl")

data:
  input_path: "raw_data.jsonl"
processing:
  min_score: 60  # 过滤分数≥60的记录

# 验证MsgPack文件读取
loaded_data = srsly.read_msgpack("processed_data.msgpack.gz")
print(f"有效记录数：{len(loaded_data)}")  # 应等于clean_data长度

# 验证统计信息
stats = srsly.read_pickle("processing_stats.pkl")
print(f"数据有效率：{stats['efficiency']:.2%}")