qbtools · 青博科技常用工具集

面向青博科技内部项目的 Python 工具库，封装了时间处理、任务调度、Django 辅助、数据类转换、日志、工业协议（MQTT / OPC）等常用能力，开箱即用，减少重复造轮子。

环境要求

• Python 3.12
• 主要依赖：Django 5.2、APScheduler、Loguru、Matplotlib、psutil、paho-mqtt、python-opcua 等（随安装自动获取）

安装

在本地源码目录安装（推荐使用虚拟环境）：

pip install .
# 或
pip install -e .   # 开发模式

使用 Poetry：

poetry install

快速开始

from dataclasses import dataclass
from qbtools import app_now_time, random_string
from qbtools.datacls import from_dict

print(app_now_time())                 # 统一时区/可模拟的当前时间
print(random_string(16, alphabet='hex'))

@dataclass
class User:
    name: str
    tags: list[str] | None = None

user = from_dict(User, {"name": "qb", "tags": ["iot", "ai"]})
print(user)

模块速览与示例

• 时间工具 qbtools.datetime / qbtools.time
  • get_f_time、get_milli_timestamp、timestamp2_formate_time、app_now_time()（支持设置模拟起点和时区）。
  • 对齐计时器：AlignedTimer（基于 monotonic，支持 wake/stop/动态周期）、NaturalAlignedTimer（按自然分钟对齐）。
• 数据类转换 qbtools.datacls
  • from_dict/from_list 将嵌套 dict 转 dataclass，支持 Optional、list[dataclass]、缺失字段策略；update_dataclass 递归更新；to_dict/to_list 递归序列化。
• 通用工具
  • random_string（安全随机串），cvt_to_bool/pyjson_to_js_string/jsstr_to_pyjson，get_big_file_md5、get_file_size，round_half/generate_possible_blocks 等数学辅助。
• 日志 qbtools.loghandler
  • InterceptHandler 将标准库 logging 透传到 Loguru；ReFormater 提供终端/文件格式化（含模拟时间显示）；dump_logging_tree() 输出当前 logging 结构。
• 任务调度 qbtools.task（基于 APScheduler）
  • 定义任务：

    from qbtools.task import BaseTask, TaskAdmin
    
    @TaskAdmin.register(trigger="interval", seconds=10)
    class HelloTask(BaseTask):
        desc = "周期问候"
        def run(self):
            print("hello qbtools")
    
    if __name__ == "__main__":
        TaskAdmin.start()   # 初始化调度器
    

  • 支持暂停/恢复/移除任务、心跳写入 Redis、统一的 run_task 入口。
• Django 辅助 qbtools.django
  • init_django_env() 便捷初始化 ORM 环境；get_json_model 序列化模型；get_where_map 根据表单拼接查询；自定义 FloatField/IntegerField/BooleanField/DateTimeField 等安全转换字段；get_local_app_urls 自动收集本地 app 的路由。
• 工业协议
  • MQTT：MqttPublisher（自动重连、重试发送）。
  • OPC UA：OpcClient 封装读写，具备节点存在性检查、连接异常自动重试、类型转换（含 bool/数值字符串处理）。
• 可视化 qbtools.plt
  • load_font_support(font_path) 为 Matplotlib 注册自定义字体并修复负号显示。
• 版本守护 qbtools.version_guard
  • ensure_python_312() / ensure_python_version() 在启动阶段校验 Python 版本并给出警告。

工具详解

• qbtools.datetime
  • 统一的当前时间：app_now_time（可设置起点/时区模拟时间流逝）、real_time（真实时间）。
  • 时间格式/解析：get_f_time/get_f_time_milis、get_p_time、timestamp2_formate_time、get_timestamp/get_milli_timestamp。
  • 运行计时：CostRecord 简单耗时记录器。
• qbtools.time
  • AlignedTimer：基于 monotonic 的周期对齐计时器，支持 wake/stop、动态周期、跳过/补齐延误。
  • NaturalAlignedTimer：按自然分钟的整数倍触发（如 :00、:20、:40），可 wake/stop，支持时区。
• qbtools.datacls
  • from_dict/from_list：嵌套 dict -> dataclass，自动处理 Optional、list[dataclass]、缺省字段策略（严格/宽松）。
  • update_dataclass：递归更新已有实例；to_dict/to_list：递归序列化（可忽略 None）。
  • 辅助 _unwrap_optional 处理 PEP-604 union，支持字符串化 list 自动 json 解析。
• qbtools.common / qbtools.json
  • random_string 安全随机串（预设字符集：alnum/ascii/digits/hex/urlsafe/punct）。
  • cvt_to_bool、pyjson_to_js_string、jsstr_to_pyjson、cvt_value_to_str。
• qbtools.mmath
  • 数学/数组片段生成：round_half（0.5 对齐，三种策略）、gen_block_info/gen_block_info_by_changes（约束步长序列生成）、generate_possible_blocks（分段组合）。
  • get_uid 生成 UUID 字符串。
• qbtools.file
  • get_file_size、get_big_file_md5 支持大文件分块计算。
• qbtools.loghandler
  • InterceptHandler 将标准库 logging 转 Loguru；ReFormater 生成控制台/文件格式（支持模拟时间标记、异常堆栈）。
  • dump_logging_tree 打印当前 logger 树与 handler。
• qbtools.task
  • BaseTask 生命周期：start/run/shutdown；记录任务状态、起停时间。
  • TaskAdmin：APScheduler 封装，装饰器 @TaskAdmin.register 注册任务，支持暂停/恢复/移除、心跳写入 Redis、任务实例注册表、等待队列、全局 ready 控制。
  • run_task 统一入口，支持类路径调用、参数传递。
• qbtools.django
  • 环境与 URL：init_django_env 便捷初始化 ORM，get_local_app_urls 自动收集本地 app 路由。
  • 模型工具：get_model_field_info、get_json_model（模型序列化，日期/文件处理）、get_where_map/get_update_map 构造查询/更新字段。
  • 自定义字段：FloatField/IntegerField/BooleanField/DateTimeField/DateTimeMiliField，统一转换与校验。
  • 网络：get_ip（X-Forwarded-For 支持）、get_local_ips（本机网卡 IP 列表）。
• qbtools.mqtt
  • MqttPublisher：连接、发送 JSON，失败自动重试，可显式关闭。
• qbtools.opc
  • OpcClient：OPC UA 读写封装，节点存在性检查，连接异常自动重连/重试，字符串数值/布尔安全转换，支持只读模式。
  • 自定义异常：OpcConnectionError、OpcNodeNotFoundError。
• qbtools.plt
  • load_font_support 为 Matplotlib 注册自定义字体，修复中文/负号显示。
• qbtools.version_guard
  • ensure_python_version / ensure_python_312 启动时校验并提示兼容性风险。

运行/调试建议

• python -m qbtools.version_guard（或在入口调用 ensure_python_312）确保运行时版本符合要求。
• 日志体系建议在应用入口添加：

  import logging
  from qbtools.loghandler import InterceptHandler
  logging.basicConfig(handlers=[InterceptHandler()], level=logging.INFO)
  

• 使用 TaskAdmin 时，TaskAdmin.start() 建议放在 Django AppConfig.ready 或独立 worker 入口，确保 set_ready() 被调用后再调度任务。

许可

本项目为青博科技内部工具，版权归青博科技所有。