amh-utils

AMH 系统通用工具库，封装 MySQL 客户端、配置加载、日志记录、Shell 执行、进程锁、钉钉告警、守护进程等常用功能。

版本： 1.1.4 | 作者： nbyue | Python： ≥ 3.6.8

---

安装

pip install amh-utils

或从源码安装：

git clone <repo-url>
cd amh-utils
pip install .

依赖：

包	版本
PyMySQL	≥ 1.0.2
PyYAML	≥ 5.4
requests	钉钉告警功能需要

---

模块一览

模块	导入路径	说明
MySQL 客户端	MySQLClient	PyMySQL 封装，支持连接池、断线重连、游标类型切换
配置加载	ConfigLoader, load_config_from_path, get_db_config_from_env	多位置自动搜索 YAML/JSON 配置，支持环境区分与类型校验
日志记录	EnhancedLogger, create_logger, setup_logging	多处理器/滚动/JSON 格式/异常捕获增强日志
Shell 执行	shell, Shell, ShellException, ShellTimeoutExpired	subprocess 封装，支持超时、编码、交互式输入
进程锁	ProcessLock, LockBusyError, ProcessLockError	flock 单实例锁，支持 stale PID 检测与自动接管
钉钉告警	DingRobot, DingRobotError	钉钉自定义机器人，支持 text/markdown/actionCard + 签名 + 重试
守护进程	Daemon, DaemonError, run_cli	双 fork 守护进程框架，集成 ProcessLock + 信号处理

所有模块均可从顶层直接导入：

from amh.utils import MySQLClient, ConfigLoader, create_logger, shell, ProcessLock, DingRobot, Daemon

---

使用方法

1. MySQL 客户端 — MySQLClient

from amh.utils import MySQLClient

# 方式一：直接传参
client = MySQLClient(host='10.0.0.1', port=3306, user='root', password='xxx', database='test')
with client:
    rows = client.query("SELECT * FROM users WHERE age > %s", (18,))
    for row in rows:
        print(row)

# 方式二：从配置文件加载
from amh.utils import get_db_config_from_env
db_cfg = get_db_config_from_env('ro')     # 读取脚本同级目录 config.yaml 的 database.ro 配置
client = MySQLClient(**db_cfg)

# 游标类型：dict 返回字典，tuple 返回元组
client_dict = MySQLClient(..., cursor_type='dict')
rows = client_dict.query("SELECT 1")      # [{'1': 1}]

# 常用方法
client.execute("INSERT INTO t (name) VALUES (%s)", ('foo',))  # 执行写操作
client.like("SELECT * FROM t WHERE name LIKE %s", '%foo%')    # LIKE 查询（自动转义 %）
client.is_connected    # 检查连接状态

特性： 断线自动重连 + 重建游标、close() 安全调用、__exit__ 不吞异常、commit() 保护、% 转义只认 %s、LIKE 参数值不双写 %。

---

2. 配置加载 — ConfigLoader

from amh.utils import ConfigLoader, load_config_from_path, get_db_config_from_env

# 自动搜索（脚本同级目录 → cwd 降级）
loader = ConfigLoader()                # 找脚本目录下的 config.yaml
config = loader.load()

# 显式路径
loader = ConfigLoader.from_path("/etc/app/config.yaml")

# 必须存在（找不到抛 FileNotFoundError）
config = loader.load_required()

# 嵌套键取值
host = loader.get_value("database.ro.host", default="localhost")

# 获取数据库配置（含必填校验 + 类型转换）
db_cfg = loader.get_database_config(env="ro")

# 快捷函数
db_cfg = get_db_config_from_env("ro")
config = load_config_from_path("/path/to/config.yaml")

配置文件搜索优先级： 显式路径 → 脚本所在目录 → 当前工作目录(cwd)。

支持格式： .yaml / .yml / .json / .conf。

特性： 环境区分（ro/rw/ops）、必填字段校验、类型转换（smart_bool 防止 bool("false") → True）。

---

3. 日志记录 — EnhancedLogger

from amh.utils import EnhancedLogger, create_logger, setup_logging

# 快捷创建
logger = create_logger("myapp", root_dir="./")

# 全局日志配置（所有 getLogger 继承）
setup_logging(verbose=True)

# 使用
logger.info("服务启动")
logger.error("连接失败", exc_info=True)

# JSON 格式输出
logger = EnhancedLogger("myapp", root_dir="./")
# 配置中可指定 json_format=True

特性： 单例模式、多处理器（TimedRotatingFileHandler / RotatingFileHandler）、JSON 格式、异常捕获装饰器、clear_cache()、重复初始化警告。

---

4. Shell 执行 — Shell

from amh.utils import shell, Shell, ShellTimeoutExpired

# 快捷函数
result = shell('ls -alh *.py')
for line in result.output():
    print(line)

# 带超时
result = shell('ping 10.0.0.1', timeout=5)
if result.timed_out:
    print("命令超时")

# 交互式输入
sh = Shell(has_input=True)
sh.run('cat -u')
sh.write('Hello, world!')
print(sh.output())

# 获取错误输出
errors = result.errors()

特性： timeout 超时控制、Windows GBK 编码自动处理、交互式输入、strip_empty 行过滤。

---

5. 进程锁 — ProcessLock（Linux/Unix only）

from amh.utils import ProcessLock, LockBusyError

# 上下文管理器（推荐）
with ProcessLock('/tmp/my_script.lock'):
    do_work()                     # 保证只有一个进程在执行

# 手动 acquire / release
lock = ProcessLock('/tmp/my_script.lock')
lock.acquire()                    # 失败抛 LockBusyError
try:
    do_work()
finally:
    lock.release()

# 非阻塞 + 返回值（不抛异常）
lock = ProcessLock('/tmp/my_script.lock', raise_on_fail=False)
if not lock.acquire():
    print("Another instance is running")
    sys.exit(0)
try:
    do_work()
finally:
    lock.release()

特性： 非阻塞排他锁（LOCK_EX | LOCK_NB）、stale PID 检测与自动接管、atexit 自动释放、__lock_acquired 防误删标记。

---

6. 钉钉告警 — DingRobot

from amh.utils import DingRobot, DingRobotError

robot = DingRobot(secret="SECxxx", token="xxx")

# 文本消息
robot.send_text("监控告警：MySQL slave 延迟超过 60s")

# Markdown 消息
robot.send_markdown("告警通知", "## MySQL 延迟\n延迟: **60s**")

# @指定人
robot.send_text("紧急告警", atMobiles=["138xxxx", "139xxxx"])

特性： HMAC-SHA256 签名验证、超时控制 + 自动重试、响应状态码检查、支持 text / markdown / actionCard 三种消息类型。

---

7. 守护进程 — Daemon（Linux/Unix only）

from amh.utils import Daemon, run_cli

class MyDaemon(Daemon):
    def run(self):
        while not self._shutdown_event.is_set():
            do_work()
            self._shutdown_event.wait(timeout=60)

if __name__ == '__main__':
    daemon = MyDaemon(pidfile='/tmp/my_daemon.pid')
    run_cli(daemon)

CLI 命令：

python my_daemon.py start       # 后台启动
python my_daemon.py stop        # 停止
python my_daemon.py restart     # 重启
python my_daemon.py status      # 查看状态
python my_daemon.py foreground  # 前台运行（调试用）

特性： 双 fork 守护进程化（经典 Unix daemon 模式）、ProcessLock flock 互斥锁 + PID 文件双保险、内置 shutdown_event（threading.Event）、SIGTERM/SIGINT 信号自动触发停机、foreground 模式适合调试、run_cli() 一行搞定 argparse。

---

项目结构

amh-utils/
├── amh/
│   ├── __init__.py                  # namespace package (pkgutil 风格)
│   ├── utils/
│   │   ├── __init__.py              # 顶层导出 + 版本号
│   │   ├── mysql_client/            # MySQL 客户端
│   │   ├── enhanced_logging/        # 增强日志
│   │   ├── load_config.py           # 配置加载
│   │   ├── shell.py                 # Shell 执行
│   │   ├── process_lock.py          # 进程锁
│   │   ├── ding_robot/              # 钉钉告警
│   │   └── daemon/                  # 守护进程
│   └── tests/                       # 测试脚本
├── pyproject.toml
├── requirements.txt
└── README.md

---

更新日志

日期	版本/模块	变更
2026-07-02	v1.1.4 / Daemon	新增通用守护进程类
2026-06-26	DingRobot	新增钉钉告警功能（text/markdown/actionCard + 签名 + 重试）
2026-06-24	ProcessLock	新增进程锁功能（flock + stale PID 检测）
2026-06-20	ConfigLoader	默认搜索脚本所在目录而非 cwd
2026-06-19	MySQLClient	v1.2：close 安全、断线重连 + 重建 cursor、% 转义优化
2026-06-18	Shell	v1.1：新增 timeout 参数、修复 Windows 编码问题
2026-06-18	EnhancedLogger	v1.4：catch_exceptions bug 修复、新增 clear_cache
2026-03-17	EnhancedLogger	新增日志记录器
2026-03-10	v1.0.0	初始发布，utils 新增 config 文件夹
2026-02-25	MySQLClient	新增 is_connected 方法