sololib 0.3.9.8

Python toolkit for corpus generation, utility functions, and Nacos config center

sololib

一个 Python 工具包，提供 语料生成、通用工具集、Nacos 配置中心 与 HTTP 中间件 四大功能模块。

当前版本号以 pyproject.toml 的 version 字段为单一来源；运行时可通过 sololib.__version__ 读取。

安装

pip install sololib
# 或
uv add sololib

# 可选依赖（图像处理）
pip install sololib[image]
# 可选依赖（Windows 窗口管理）
pip install sololib[win32]

快速开始

1. 语料生成 (sololib.corpus)

基于 30 个变量池（每池 30 值）和 500+ 中英文问题模板，随机生成对话数据。

from sololib import get_random_conversation

# n=1（默认）返回单组对话: [(question, topic), ...]
conv = get_random_conversation()
for q, topic in conv:
    print(f"[{topic}] {q}")

# n>1 返回多组对话: [[(question, topic), ...], ...]
convs = get_random_conversation(n=10)

# 批量生成随机问题（无追问）
from sololib import generate_questions
questions = generate_questions(n=5)

# 生成单个问题
from sololib import generate_single_question
single = generate_single_question()

# 查看语料统计
from sololib import get_corpus_stats
stats = get_corpus_stats()

# 估算模板组合数
from sololib import estimate_combinations
total = estimate_combinations()

概率模型：

• 中文概率：10%，追问概率：50%，三轮追问条件概率：40%
• 自动中英文分离（通过 Unicode 范围检测）

2. 通用工具集 (sololib.utils)

# ---- 异步命令执行 ----
from sololib.utils import run_command

result = await run_command("ls -la", timeout=30)
# {"stdout": "...", "stderr": "...", "returncode": 0}

# ---- 重试装饰器（同步/异步自适应；可指定异常白名单）----
from sololib.utils import retry

@retry(max_retries=3, delay=3.0)
def flaky_function():
    ...

@retry(max_retries=5, delay=1.0, exceptions=(ConnectionError, TimeoutError))
async def async_network_call():
    ...

# ---- 递归字典合并 ----
from sololib.utils import merge_dicts
merged = merge_dicts({"a": {"x": 1}}, {"a": {"y": 2}})
# {"a": {"x": 1, "y": 2}}

# ---- 异步 HTTP POST ----
from sololib.utils import post_data, UnauthorizedError, HttpRequestError

try:
    result = await post_data("https://api.example.com/data", {"key": "value"})
except UnauthorizedError:
    ...  # 401
except HttpRequestError as e:
    ...  # 其他网络/状态码/解析错误

# ---- Pydantic 统一响应模型 ----
from sololib.utils import success, error, result, result_page, ResponseModel

resp = success()          # ResponseModel(code=200, message="success")
resp = result({"id": 1})  # ResponseModel(code=200, message="success", data={"id": 1})
resp = error(400, "Bad request")
resp = result_page([{"id": 1}], total=100)

# ---- PyPI 版本检查 ----
from sololib.utils import check_package_update, update_package, get_current_version

current = get_current_version("sololib")
needs_update = check_package_update("sololib", current)
update_package("sololib")

# ---- YAML 配置加载 ----
from sololib.utils import load_config
from dataclasses import dataclass

@dataclass
class AppConfig:
    host: str
    port: int

config = load_config("app.yaml", AppConfig)

# ---- Loguru 日志（核心依赖）----
from sololib.utils import setup_logger, shutdown_logger, logger

setup_logger(env="dev", log_dir="logs", level="INFO")
logger.info("Hello, world!")
logger.bind(request_id="req-1").info("带上下文的日志")
shutdown_logger()  # 服务优雅退出时调用

3. Nacos 配置中心 (sololib.configs)

基于 nacos-sdk-python v3 全异步 gRPC API，对外暴露同步接口，支持配置热更新。

from sololib.configs import NacosConfig, NacosStore

# 初始化配置（连接 Nacos 并自动拉取所有配置）
nacos_config = NacosConfig(
    server_addresses="127.0.0.1:9848",
    namespace="public",
    username="nacos",
    password="nacos",
    configs=[
        NacosConfig.ConfigItem(data_id="app.yaml", group="DEFAULT_GROUP"),
        NacosConfig.ConfigItem(data_id="db.yaml", group="DEFAULT_GROUP"),
    ],
)

store = NacosStore(nacos_config, is_watcher=True)

# 获取合并后的完整配置
config = store.get_config()

# 手动刷新
store.refresh_config()

# 关闭连接
store.close()

单例约定：NacosStore 是进程级单例。第二次以不同 nacos_config 构造时，新参数会被忽略并打印 warning；要切换配置必须先 close()。

4. HTTP 中间件 (sololib.middleware)

基于 Starlette 的纯 ASGI 实现，无 BaseHTTPMiddleware 双重缓冲开销。

from fastapi import FastAPI
from sololib.middleware import RequestLogMiddleware

app = FastAPI()
app.add_middleware(
    RequestLogMiddleware,
    exclude_paths={"/health", "/metrics"},
    exclude_path_prefixes={"/api/internal"},
    log_request_body=True,
    log_response_body=True,
    request_body_max_len=200,
    response_body_max_len=200,
)

# 日志格式：
# [->] POST | path=/api/echo | rid=abc123 | ip=1.2.3.4 | body={"key":"val"}
# [<-] POST | path=/api/echo | rid=abc123 | ip=1.2.3.4 | status=200 | elapsed=12.34ms | body={"result":"ok"}

• 2xx → INFO；4xx → WARNING；5xx → ERROR
• 自动收集请求体（POST/PUT/PATCH/DELETE，RFC 7231 允许 DELETE 带 body）
• 自动跳过 text/event-stream / multipart/* / application/octet-stream 响应体

版本要求

• Python >= 3.11

License

MIT