第11章 运维监控与文档交付:让脚本具备”自我报告”能力#
11.1 崩溃报警:10行代码接入企业 IM 机器人#
一个凌晨三点的电话#
凌晨三点,你被电话吵醒。产线负责人语气焦急:“测试工装的自动校准脚本从昨晚十点就挂了,五个小时的产能全废了。”
你打开电脑一看——日志里写着一行 ConnectionResetError,脚本在十点零三分就崩溃了,没有任何人知道。
你心想:如果脚本崩溃的那一秒,钉钉群里就弹出一条报警消息,这一切就不会发生。
这就是本节要解决的问题——让你的 Python 脚本具备”自我报警”能力。
11.1.1 Webhook 机器人:三分钟配置#
主流企业 IM 平台(钉钉、企业微信、飞书)都提供了”自定义机器人”功能,通过一个 Webhook URL,你的程序就能往群里发消息。
钉钉机器人的创建步骤:
- 打开钉钉群 → 群设置 → 智能群助手 → 添加机器人
- 选择”自定义”类型
- 设置机器人名称(如”测试工装报警”)
- 安全设置选择”自定义关键词”,填入
报警(这样只有包含”报警”关键词的消息才会被发送,防止误发) - 复制 Webhook URL,格式类似:
plaintexthttps://oapi.dingtalk.com/robot/send?access_token=xxxxxxxxxxxxxxxx
企业微信机器人的创建步骤:
- 打开企业微信群 → 群设置 → 群机器人 → 添加
- 设置名称和头像
- 复制 Webhook URL,格式类似:
plaintexthttps://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxxxxxx
飞书机器人的创建步骤:
- 打开飞书群 → 设置 → 群机器人 → 添加机器人 → 自定义机器人
- 设置名称和描述
- 复制 Webhook URL,格式类似:
plaintexthttps://open.feishu.cn/open-apis/bot/v2/hook/xxxxxxxx
安全提示:Webhook URL 等同于”群发消息的钥匙”,不要硬编码在代码里。建议放在配置文件(第1章讲的 JSON/INI 配置文件)或环境变量中。
11.1.2 核心实现:10行代码发一条报警#
钉钉、企业微信、飞书的 Webhook 接口格式略有不同,但本质上都是向一个 URL POST 一个 JSON。我们先从最简单的开始:
钉钉发送文本消息:
import requests
import json
DINGTALK_WEBHOOK = "https://oapi.dingtalk.com/robot/send?access_token=你的token"
def send_dingtalk_alert(message: str) -> bool:
"""向钉钉群发送报警消息"""
payload = {
"msgtype": "text",
"text": {
"content": f"【设备报警】{message}" # 包含"报警"关键词才能通过安全校验
}
}
headers = {"Content-Type": "application/json"}
try:
resp = requests.post(DINGTALK_WEBHOOK, data=json.dumps(payload), headers=headers, timeout=10)
result = resp.json()
return result.get("errcode") == 0
except Exception as e:
print(f"发送钉钉报警失败: {e}")
return Falsepython企业微信发送文本消息:
WECOM_WEBHOOK = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=你的key"
def send_wecom_alert(message: str) -> bool:
"""向企业微信群发送报警消息"""
payload = {
"msgtype": "text",
"text": {
"content": f"【设备报警】{message}"
}
}
headers = {"Content-Type": "application/json"}
try:
resp = requests.post(WECOM_WEBHOOK, data=json.dumps(payload), headers=headers, timeout=10)
result = resp.json()
return result.get("errcode") == 0
except Exception as e:
print(f"发送企业微信报警失败: {e}")
return Falsepython飞书发送文本消息:
FEISHU_WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/你的hook_id"
def send_feishu_alert(message: str) -> bool:
"""向飞书群发送报警消息"""
payload = {
"msg_type": "text",
"content": {
"text": f"【设备报警】{message}"
}
}
headers = {"Content-Type": "application/json"}
try:
resp = requests.post(FEISHU_WEBHOOK, data=json.dumps(payload), headers=headers, timeout=10)
result = resp.json()
return result.get("code") == 0
except Exception as e:
print(f"发送飞书报警失败: {e}")
return Falsepython三种平台的核心逻辑完全一样:构造 JSON → POST 到 Webhook URL → 检查返回值。唯一区别是 JSON 字段名称和返回码字段名略有不同。
11.1.3 统一封装:一个 AlertManager 搞定三种平台#
实际项目中,你不会只用一个平台,而且你不想在业务代码里到处写 requests.post。我们需要一个统一的报警管理器:
# alert_manager.py
import requests
import json
import time
from enum import IntEnum
from datetime import datetime
from typing import Optional
class AlertLevel(IntEnum):
"""报警等级定义"""
INFO = 0 # 信息通知:正常事件记录
WARNING = 1 # 警告:需要关注但不紧急
ERROR = 2 # 错误:需要人工介入
CRITICAL = 3 # 严重:需要立即处理,可能需要电话告警
class AlertManager:
"""统一报警管理器 - 支持钉钉、企业微信、飞书"""
def __init__(self, config: dict):
"""
初始化报警管理器
config 示例:
{
"platform": "dingtalk", # 平台:dingtalk / wecom / feishu
"webhook_url": "https://...",
"alert_level": "WARNING", # 最低触发等级
"rate_limit_seconds": 60, # 同一消息的最小发送间隔(秒)
"contact_phone": "13800138000" # CRITICAL 级别的电话告警联系人
}
"""
self.platform = config.get("platform", "dingtalk")
self.webhook_url = config["webhook_url"]
self.min_level = AlertLevel[config.get("alert_level", "WARNING")]
self.rate_limit = config.get("rate_limit_seconds", 60)
self.contact_phone = config.get("contact_phone")
# 发送历史,用于限频(防止报警风暴)
self._last_sent: dict[str, float] = {}
def _should_send(self, message: str) -> bool:
"""限频检查:同一消息在 rate_limit 秒内不重复发送"""
now = time.time()
last_time = self._last_sent.get(message, 0)
if now - last_time < self.rate_limit:
return False
self._last_sent[message] = now
return True
def _build_message(self, level: AlertLevel, title: str, content: str) -> str:
"""构建带上下文信息的完整消息"""
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
level_names = {
AlertLevel.INFO: "信息",
AlertLevel.WARNING: "警告",
AlertLevel.ERROR: "错误",
AlertLevel.CRITICAL: "严重"
}
level_icon = {
AlertLevel.INFO: "ℹ️",
AlertLevel.WARNING: "⚠️",
AlertLevel.ERROR: "❌",
AlertLevel.CRITICAL: "🔴"
}
return (
f"{level_icon[level]}【{level_names[level]}】{title}\n"
f"时间:{timestamp}\n"
f"详情:{content}"
)
def _send_to_platform(self, message: str) -> bool:
"""根据平台类型发送消息"""
if self.platform == "dingtalk":
payload = {
"msgtype": "text",
"text": {"content": f"报警\n{message}"} # "报警"是安全关键词
}
elif self.platform == "wecom":
payload = {
"msgtype": "text",
"text": {"content": f"报警\n{message}"}
}
elif self.platform == "feishu":
payload = {
"msgtype": "text",
"content": {"text": f"报警\n{message}"}
}
else:
raise ValueError(f"不支持的平台: {self.platform}")
headers = {"Content-Type": "application/json"}
try:
resp = requests.post(
self.webhook_url,
data=json.dumps(payload),
headers=headers,
timeout=10
)
return resp.status_code == 200
except Exception as e:
# 报警发送失败时,打印到控制台(不能静默吞掉)
print(f"[AlertManager] 发送失败: {e}")
return False
def send(self, level: AlertLevel, title: str, content: str) -> bool:
"""
发送报警消息
Args:
level: 报警等级
title: 报警标题(简短描述)
content: 报警详情(具体信息)
Returns:
是否发送成功
"""
# 等级过滤:低于最低触发等级的消息不发送
if level < self.min_level:
return False
# 限频过滤
full_message = self._build_message(level, title, content)
if not self._should_send(full_message):
return False
# 发送消息
success = self._send_to_platform(full_message)
# CRITICAL 级别额外处理:未来可扩展为电话告警
if level == AlertLevel.CRITICAL and self.contact_phone:
self._trigger_phone_alert(title, content)
return success
def _trigger_phone_alert(self, title: str, content: str):
"""
触发电话告警(预留接口)
实际实现需要对接电话告警服务,如:
- 阿里云语音通知
- 腾讯云语音通知
- 自建电话通知服务
这里仅打印提示,实际项目中按需对接。
"""
print(f"[电话告警] 联系人: {self.contact_phone}, 内容: {title} - {content}")
# TODO: 对接实际的电话告警 API
# 示例:阿里云语音通知
# from aliyunsdkdyvmsapi.request.v20170525.SingleCallByTtsRequest import SingleCallByTtsRequest
# ...python11.1.4 分级报警策略#
报警不能”什么都往群里发”——否则报警消息太多,真正重要的报警反而被淹没。我们需要分级策略:
┌──────────┬──────────────┬─────────────────────────────────────┐
│ 等级 │ 触发场景 │ 报警动作 │
├──────────┼──────────────┼─────────────────────────────────────┤
│ INFO │ 正常事件记录 │ 仅写入日志文件,不发送消息 │
│ WARNING │ 一般异常 │ 发送群消息通知 │
│ ERROR │ 严重异常 │ 发送群消息并 @负责人 │
│ CRITICAL│ 系统崩溃 │ 发送群消息 + @负责人 + 电话告警 │
└──────────┴──────────────┴─────────────────────────────────────┘plaintext在业务代码中使用:
# main.py
import logging
from alert_manager import AlertManager, AlertLevel
from config import ALERT_CONFIG # 从配置文件加载
# 初始化报警管理器
alerter = AlertManager(ALERT_CONFIG)
# INFO 级别:静默记录到日志,不发消息(因为 min_level 设为 WARNING)
alerter.send(AlertLevel.INFO, "设备上线", "工装A已连接,端口 COM3")
# WARNING 级别:发一条群消息
alerter.send(AlertLevel.WARNING, "温度偏高",
"设备A温度达到 65°C,阈值为 60°C,请关注散热情况")
# ERROR 级别:发群消息 + @负责人
alerter.send(AlertLevel.ERROR, "通讯中断",
"设备A串口连接丢失,最后心跳时间 14:30:05,已自动重试3次失败")
# CRITICAL 级别:发群消息 + 电话告警
alerter.send(AlertLevel.CRITICAL, "系统崩溃",
"测试工装主程序异常退出,异常类型:OSError: [Errno 28] No space left on device")python11.1.5 与 logging 集成:报警自动触发#
你可能不想在业务代码里到处写 alerter.send()。更好的方式是把报警管理器集成到 logging 框架中,让日志达到特定级别时自动触发报警:
# alert_handler.py
import logging
from alert_manager import AlertManager, AlertLevel
class AlertLogHandler(logging.Handler):
"""自定义日志处理器 - 当日志级别达到 WARNING 及以上时自动发送报警"""
# Python logging 级别与 AlertLevel 的映射
LEVEL_MAP = {
logging.WARNING: AlertLevel.WARNING,
logging.ERROR: AlertLevel.ERROR,
logging.CRITICAL: AlertLevel.CRITICAL,
}
def __init__(self, alert_manager: AlertManager):
super().__init__()
self.alert_manager = alert_manager
self.setLevel(logging.WARNING) # 只处理 WARNING 及以上级别
def emit(self, record: logging.LogRecord):
"""当日志记录被触发时调用"""
alert_level = self.LEVEL_MAP.get(record.levelno)
if alert_level is None:
return
# 获取异常信息(如果有)
title = f"{record.funcName}:{record.lineno}"
content = self.format(record)
self.alert_manager.send(alert_level, title, content)python注册到 logging:
import logging
from alert_manager import AlertManager
from alert_handler import AlertLogHandler
from config import ALERT_CONFIG
# 配置 logging
logger = logging.getLogger("my_app")
logger.setLevel(logging.DEBUG)
# 控制台输出
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.DEBUG)
logger.addHandler(console_handler)
# 报警输出(WARNING 及以上自动触发)
alerter = AlertManager(ALERT_CONFIG)
alert_handler = AlertLogHandler(alerter)
alert_handler.setLevel(logging.WARNING)
logger.addHandler(alert_handler)python之后,业务代码只需要正常写日志,报警会自动触发:
logger.debug("正在解析第 1024 帧数据") # 不触发报警
logger.info("设备A已上线") # 不触发报警
logger.warning("温度超过阈值: 65°C") # → 自动发送 WARNING 级别报警
logger.error("串口连接断开") # → 自动发送 ERROR 级别报警
logger.critical("硬盘空间不足,系统即将停止") # → 自动发送 CRITICAL 级别报警python架构示意:
业务代码
│
├── logger.info() ──→ 日志文件 + 控制台
├── logger.debug() ──→ 日志文件 + 控制台
│
├── logger.warning() ──→ 日志文件 + 控制台 + 钉钉群消息
├── logger.error() ──→ 日志文件 + 控制台 + 钉钉群消息(@负责人)
└── logger.critical()──→ 日志文件 + 控制台 + 钉钉群消息 + 电话告警plaintext11.1.6 发送富文本消息:Markdown 卡片#
纯文本消息信息密度较低。钉钉和飞书支持 Markdown 格式的消息,可以让报警信息更加结构化:
def send_dingtalk_markdown(title: str, content: str, level: str = "WARNING") -> bool:
"""发送钉钉 Markdown 格式消息"""
color_map = {
"WARNING": "orange",
"ERROR": "red",
"CRITICAL": "red"
}
payload = {
"msgtype": "markdown",
"markdown": {
"title": f"报警 - {title}",
"text": (
f"### {title}\n\n"
f"- **等级**: {level}\n"
f"- **时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"
f"- **详情**: {content}\n\n"
f"---\n"
f"> 请及时处理,[查看详情](http://your-monitor-dashboard.com)"
)
}
}
headers = {"Content-Type": "application/json"}
try:
resp = requests.post(DINGTALK_WEBHOOK, data=json.dumps(payload), headers=headers, timeout=10)
return resp.json().get("errcode") == 0
except Exception:
return Falsepython企业微信支持 Markdown 格式的消息:
def send_wecom_markdown(title: str, content: str, level: str = "WARNING") -> bool:
"""发送企业微信 Markdown 格式消息"""
payload = {
"msgtype": "markdown",
"markdown": {
"content": (
f"## 报警: {title}\n"
f"> 等级: <font color=\"warning\">{level}</font>\n"
f"> 时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"
f"> 详情: {content}\n"
)
}
}
headers = {"Content-Type": "application/json"}
try:
resp = requests.post(WECOM_WEBHOOK, data=json.dumps(payload), headers=headers, timeout=10)
return resp.json().get("errcode") == 0
except Exception:
return Falsepython飞书 Markdown 注意事项:飞书的自定义机器人对 Markdown 的支持有特殊语法要求(如需要使用
<font>标签而非标准 Markdown 语法),建议参考飞书开放平台的官方文档进行适配。
11.2 日志归档:用 Loguru 实现日志自动切割与清理#
11.2.1 为什么 logging 不够用?#
在第1章,我们介绍了 Python 标准库的 logging 模块,它确实是”够用”的日志方案。但在嵌入式工控机上长期运行时,logging 有一个让人头疼的问题:
它不会自动切割日志文件。
想象一下:你的测试工装脚本每天 24 小时运行,每天产生大约 200MB 的日志。一个月后,日志文件涨到了 6GB。工控机的硬盘通常只有 32GB~64GB,很快就被撑满了。然后系统报 No space left on device,脚本崩溃,产线停摆。
你当然可以用 logging.handlers.RotatingFileHandler 来做日志切割,但配置起来相当繁琐:
# 标准库 logging 的日志切割配置(繁琐但能用)
import logging
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler
logger = logging.getLogger("my_app")
logger.setLevel(logging.DEBUG)
# 按大小切割:单文件最大 50MB,最多保留 10 个文件
size_handler = RotatingFileHandler(
"app.log",
maxBytes=50*1024*1024, # 50MB
backupCount=10,
encoding="utf-8"
)
# 按时间切割:每天午夜切割,保留 30 天
time_handler = TimedRotatingFileHandler(
"app.log",
when="midnight",
interval=1,
backupCount=30,
encoding="utf-8"
)
# 还需要设置 formatter...
formatter = logging.Formatter("%(asctime)s [%(levelname)s] %(name)s: %(message)s")
size_handler.setFormatter(formatter)
logger.addHandler(size_handler)python这段代码能用,但有几个问题:
- 代码量大:光配置日志就要写十几行
- 不支持压缩:切割后的旧日志还是
.log文件,白白占用空间 - 不支持异步写入:高频日志写入可能阻塞业务线程
- 配置复杂:
RotatingFileHandler和TimedRotatingFileHandler是两个独立的处理器,无法同时按大小和按时间切割
11.2.2 Loguru:开箱即用的日志方案#
Loguru 是一个第三方日志库,设计哲学是”日志应该开箱即用,不需要繁琐配置”。它解决了标准库 logging 的所有痛点。
安装:
pip install logurubash最简单的用法——一行代码替代整个 logging 体系:
from loguru import logger
logger.debug("调试信息")
logger.info("普通信息")
logger.warning("警告信息")
logger.error("错误信息")
logger.critical("严重错误")python不需要创建 logger 实例、不需要配置 handler、不需要设置 formatter。from loguru import logger 这一行就够了。
11.2.3 配置日志切割:保留30天,单文件50MB,自动压缩#
这是嵌入式场景最核心的需求——在工控机有限的硬盘上,安全地保留足够的日志历史:
from loguru import logger
import sys
# 移除默认的控制台输出(可选)
logger.remove()
# 添加控制台输出(带颜色,方便本地调试)
logger.add(
sys.stdout,
format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | "
"<level>{level: <8}</level> | "
"<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> | "
"<level>{message}</level>",
level="DEBUG",
colorize=True
)
# 添加文件输出 - 按天切割,保留30天,自动压缩
logger.add(
"logs/app_{time:YYYY-MM-DD}.log", # 日志文件路径(按日期自动命名)
rotation="00:00", # 每天午夜切割(也可以写 "50 MB" 按大小切割)
retention="30 days", # 只保留最近30天的日志
compression="zip", # 旧日志自动压缩为 zip
encoding="utf-8", # 中文支持
level="INFO", # 文件中只记录 INFO 及以上
format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | {name}:{function}:{line} | {message}",
enqueue=True # 异步写入(不阻塞业务线程)
)
# 添加错误日志单独文件(只记录 ERROR 及以上,方便快速排查)
logger.add(
"logs/error_{time:YYYY-MM-DD}.log",
rotation="00:00",
retention="90 days", # 错误日志保留更久:90天
compression="zip",
encoding="utf-8",
level="ERROR",
format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | {name}:{function}:{line} | {message}",
enqueue=True
)python关键参数说明:
| 参数 | 说明 | 常用值 |
|---|---|---|
rotation | 切割条件 | "00:00"(每天午夜)、"50 MB"(按大小)、"1 week"(按周) |
retention | 保留策略 | "30 days"、"10 files"、"2 GB" |
compression | 压缩格式 | "zip"、"gz"、"tar.gz"、"bz2" |
enqueue | 异步写入 | True(推荐,避免日志写入阻塞业务) |
encoding | 字符编码 | "utf-8"(必选,否则中文乱码) |
backtrace | 异常回溯增强 | True(显示完整调用栈) |
diagnose | 变量值诊断 | True(在异常信息中显示变量值,调试利器) |
rotation 还支持组合条件:
# 同时满足"每天"或"50MB"时切割
logger.add("logs/app.log", rotation="00:00", ...)
# 或者纯按大小
logger.add("logs/app.log", rotation="50 MB", ...)python11.2.4 实际效果演示#
配置好之后,你的 logs/ 目录结构会自动变成这样:
logs/
├── app_2025-07-28.log # 今天的日志(正在写入)
├── app_2025-07-27.log.zip # 昨天的(已压缩)
├── app_2025-07-26.log.zip # 前天的(已压缩)
├── ...
├── error_2025-07-28.log # 今天的错误日志
├── error_2025-07-27.log.zip # 昨天的错误日志
└── ...(超过30天的自动删除)plaintext你不需要写任何”清理旧日志”的脚本——Loguru 帮你处理了一切。
11.2.5 Loguru 高级技巧#
技巧1:捕获未处理的异常#
标准库 logging 不会自动捕获未处理的异常——程序崩溃时,最后的错误信息不会被写入日志文件。Loguru 可以:
# 自动捕获所有未处理的异常并记录到日志
logger.add(
"logs/crash.log",
rotation="10 MB",
retention="60 days",
compression="zip",
encoding="utf-8",
backtrace=True,
diagnose=True, # 关键:在异常信息中显示变量值
level="ERROR"
)python当程序崩溃时,crash.log 会记录完整的调用栈和每个变量的值:
2025-07-28 14:30:05.123 | ERROR | test_station:run_calibration:156 | 未捕获的异常
Traceback (most recent call last):
File "test_station.py", line 154, in run_calibration
result = self.device.read_sensor(channel=3)
│ │ └ 3
│ └ <Device object at 0x7f8b8c0d4a90>
└ <TestStation object at 0x7f8b8c0d4b50>
ConnectionError: 串口读取超时(等待 1000ms 无响应)plaintext注意它自动打印了 channel=3、self.device 的类型和地址——这比标准 traceback 信息量大得多,排查问题时省去大量加 print 调试的时间。
技巧2:使用 @logger.catch 装饰器#
@logger.catch
def connect_device(port: str, baudrate: int):
"""连接设备 - 如果出错,自动记录详细异常信息"""
import serial
ser = serial.Serial(port, baudrate, timeout=1)
return ser
# 即使这个函数抛出异常,Loguru 也会完整记录,而不是让程序无声崩溃python技巧3:结构化日志(JSON 格式)#
当你需要把日志导入到 ELK(Elasticsearch + Logstash + Kibana)等日志分析平台时,JSON 格式的日志更方便解析:
import json
def json_formatter(record):
"""自定义 JSON 格式化器"""
log_entry = {
"timestamp": record["time"].strftime("%Y-%m-%d %H:%M:%S.%f"),
"level": record["level"].name,
"module": record["name"],
"function": record["function"],
"line": record["line"],
"message": record["message"],
}
if record["exception"]:
log_entry["exception"] = {
"type": record["exception"].type.__name__,
"value": str(record["exception"].value),
}
record["extra"]["serialized"] = json.dumps(log_entry, ensure_ascii=False)
return "{extra[serialized]}\n"
logger.add(
"logs/structured.jsonl",
format=json_formatter,
rotation="100 MB",
retention="30 days",
compression="gz",
encoding="utf-8",
)python11.2.6 从 logging 迁移到 Loguru#
如果你的项目已经在用 logging,不需要全部重写。Loguru 提供了拦截器,可以把所有 logging 日志转发到 Loguru:
import logging
from loguru import logger
class InterceptHandler(logging.Handler):
"""将标准库 logging 的所有日志转发到 Loguru"""
def emit(self, record):
try:
level = logger.level(record.levelname).name
except ValueError:
level = record.levelno
frame, depth = logging.currentframe(), 2
while frame and frame.f_code.co_filename == logging.__file__:
frame = frame.f_back
depth += 1
logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())
# 拦截所有第三方库的 logging 输出
logging.basicConfig(handlers=[InterceptHandler()], level=0, force=True)
# 特别拦截某些第三方库
for name in logging.root.manager.loggerDict:
logging.getLogger(name).handlers = []
logging.getLogger(name).propagate = Truepython这样,pymodbus、pyserial 等第三方库的日志也会自动通过 Loguru 输出,统一格式、统一管理。
11.2.7 与报警模块联动#
把 11.1 的报警模块和 Loguru 结合起来,实现”日志自动归档 + 重要日志自动报警”的完整方案:
from loguru import logger
from alert_manager import AlertManager, AlertLevel
from alert_handler import AlertLogHandler # 11.1 中定义的
# 日志配置
logger.remove()
logger.add(sys.stdout, level="DEBUG")
# 文件日志:自动切割、压缩、清理
logger.add(
"logs/app_{time:YYYY-MM-DD}.log",
rotation="00:00",
retention="30 days",
compression="zip",
encoding="utf-8",
level="INFO",
enqueue=True
)
# 报警处理器:WARNING 及以上自动发送钉钉报警
alerter = AlertManager({
"platform": "dingtalk",
"webhook_url": "https://oapi.dingtalk.com/robot/send?access_token=xxx",
"alert_level": "WARNING",
"rate_limit_seconds": 60
})
# 将 AlertLogHandler 注册到 logging
# (因为 Loguru 的拦截器会把 logging 日志转发过来,所以也能被 AlertLogHandler 捕获)
alert_handler = AlertLogHandler(alerter)
alert_handler.setLevel(logging.WARNING)
logging.getLogger().addHandler(alert_handler)python最终效果:
- 所有日志:自动写入文件,按天切割,30天自动清理,压缩归档
- WARNING 及以上:除了写日志,还会自动发送钉钉群消息
- ERROR 及以上:发送钉钉群消息 + @负责人
- CRITICAL:发送钉钉群消息 + @负责人 + 电话告警
- 硬盘不会撑满:Loguru 自动管理日志生命周期
- 业务代码零侵入:只需要
logger.info()/logger.warning(),报警和归档全自动
11.3 文档交付:基于代码结构扫描的 AI 文档生成策略#
11.3.1 场景与痛点:文档永远过期#
“文档和代码不一致”是软件工程中最古老的问题之一。在嵌入式项目中,这个问题尤其严重:
- 代码在迭代,文档停在三个月前:你改了
parse_frame()的参数签名,但 README 里还是旧的调用示例 - 新人看不懂代码,老员工没时间写文档:项目交付时,文档往往是最先被牺牲的
- 手动写文档成本高、容易出错:一个有 50 个函数的模块,手动写文档要花一整天
核心矛盾:文档需要与代码保持同步,但人没有精力持续同步。
解法:让 AI 扫描你的代码结构,自动生成文档。当代码变了,重新跑一遍扫描就行——文档永远跟代码同步。
11.3.2 方法论:AST 扫描 + AI 生成#
整个流程分两步:
第一步:扫描代码结构(机器做)
Python 代码 ──→ AST 解析 / inspect 模块 ──→ 结构化数据(函数名、参数、类型、docstring)
第二步:生成文档(AI 做)
结构化数据 ──→ AI ──→ README.md / API 文档 / 用户操作手册plaintext第一步 是确定性的——机器精确地读取代码结构,不会遗漏、不会出错。
第二步 是创造性的——AI 基于结构化信息,用自然语言组织成人类可读的文档。
11.3.3 实战:用 inspect 模块扫描代码结构#
Python 标准库的 inspect 模块是代码扫描的瑞士军刀。它能在运行时获取任何 Python 对象的详细信息:
# code_scanner.py
import inspect
import importlib
import pkgutil
from typing import Any
from dataclasses import dataclass, field
@dataclass
class FunctionInfo:
"""函数信息的数据结构"""
name: str
signature: str
docstring: str
parameters: list[dict]
return_annotation: str
source_file: str
line_number: int
@dataclass
class ClassInfo:
"""类信息的数据结构"""
name: str
docstring: str
methods: list[FunctionInfo]
bases: list[str]
source_file: str
@dataclass
class ModuleInfo:
"""模块信息的数据结构"""
name: str
docstring: str
functions: list[FunctionInfo]
classes: list[ClassInfo]
source_file: str
def scan_function(func) -> FunctionInfo:
"""扫描单个函数的详细信息"""
sig = inspect.signature(func)
params = []
for name, param in sig.parameters.items():
params.append({
"name": name,
"type": str(param.annotation) if param.annotation != inspect.Parameter.empty else "未指定",
"default": str(param.default) if param.default != inspect.Parameter.empty else "无默认值",
})
return FunctionInfo(
name=func.__name__,
signature=str(sig),
docstring=inspect.getdoc(func) or "无文档字符串",
parameters=params,
return_annotation=str(sig.return_annotation) if sig.return_annotation != inspect.Signature.empty else "未指定",
source_file=inspect.getfile(func) if not inspect.isbuiltin(func) else "内置函数",
line_number=inspect.getsourcelines(func)[1] if not inspect.isbuiltin(func) else 0,
)
def scan_class(cls) -> ClassInfo:
"""扫描单个类的详细信息"""
methods = []
for name, method in inspect.getmembers(cls, predicate=inspect.isfunction):
if not name.startswith('_'): # 跳过私有方法
methods.append(scan_function(method))
return ClassInfo(
name=cls.__name__,
docstring=inspect.getdoc(cls) or "无文档字符串",
methods=methods,
bases=[base.__name__ for base in cls.__bases__ if base.__name__ != 'object'],
source_file=inspect.getfile(cls),
)
def scan_module(module) -> ModuleInfo:
"""扫描整个模块的结构"""
functions = []
classes = []
for name, obj in inspect.getmembers(module):
if name.startswith('_'):
continue
if inspect.isfunction(obj) and inspect.getmodule(obj) == module:
functions.append(scan_function(obj))
elif inspect.isclass(obj) and inspect.getmodule(obj) == module:
classes.append(scan_class(obj))
return ModuleInfo(
name=module.__name__,
docstring=inspect.getdoc(module) or "无模块文档",
functions=functions,
classes=classes,
source_file=inspect.getfile(module),
)
def scan_project(package_path: str) -> list[ModuleInfo]:
"""扫描整个项目包的所有模块"""
import sys
sys.path.insert(0, package_path)
modules_info = []
package_name = package_path.split('/')[-1].split('\\')[-1]
try:
package = importlib.import_module(package_name)
except ImportError as e:
print(f"无法导入包 {package_name}: {e}")
return []
# 遍历包中的所有模块
for importer, modname, ispkg in pkgutil.walk_packages(
path=package.__path__,
prefix=package.__name__ + ".",
):
try:
module = importlib.import_module(modname)
modules_info.append(scan_module(module))
except Exception as e:
print(f"扫描模块 {modname} 时出错: {e}")
return modules_infopython使用示例:
# 扫描项目并输出结构化信息
modules = scan_project("./src")
for mod in modules:
print(f"\n{'='*60}")
print(f"模块: {mod.name}")
print(f"文件: {mod.source_file}")
print(f"文档: {mod.docstring[:100]}...")
print(f"函数数量: {len(mod.functions)}")
print(f"类数量: {len(mod.classes)}")
for func in mod.functions:
print(f" - {func.name}{func.signature}")
for cls in mod.classes:
print(f" + class {cls.name}({', '.join(cls.bases)})")
for method in cls.methods:
print(f" - {method.name}{method.signature}")python输出示例:
============================================================
模块: src.protocol_parser
文件: /home/user/project/src/protocol_parser.py
文档: 协议解析模块 - 支持 V1.0/V2.0/V3.0 协议帧解析...
函数数量: 5
类数量: 2
- parse_frame(frame: bytes) -> dict
- build_frame(cmd: int, data: bytes) -> bytes
+ class ProtocolParser()
- parse(data: bytes) -> dict
- build(cmd: int, payload: bytes) -> bytes
- validate(frame: bytes) -> bool
+ class DeviceProtocol(serial_port: str)
- connect() -> None
- send_command(cmd: int, data: bytes) -> dict
- close() -> Noneplaintext11.3.4 用 AST 模块做更深度的扫描#
inspect 模块适合运行时扫描,但有时候你需要更细粒度的信息(比如函数体内的逻辑结构)。Python 的 ast(Abstract Syntax Tree,抽象语法树)模块可以直接解析 .py 源文件:
# ast_scanner.py
import ast
from pathlib import Path
def scan_py_file(filepath: str) -> dict:
"""用 AST 扫描单个 Python 文件的结构"""
source = Path(filepath).read_text(encoding="utf-8")
tree = ast.parse(source)
result = {
"file": filepath,
"imports": [],
"functions": [],
"classes": [],
}
for node in ast.walk(tree):
if isinstance(node, ast.Import):
for alias in node.names:
result["imports"].append(alias.name)
elif isinstance(node, ast.ImportFrom):
module = node.module or ""
for alias in node.names:
result["imports"].append(f"{module}.{alias.name}")
elif isinstance(node, ast.FunctionDef):
func_info = {
"name": node.name,
"args": [arg.arg for arg in node.args.args],
"decorators": [
ast.dump(d) for d in node.decorator_list
],
"line": node.lineno,
"has_docstring": ast.get_docstring(node) is not None,
"docstring": ast.get_docstring(node) or "",
}
result["functions"].append(func_info)
elif isinstance(node, ast.ClassDef):
class_info = {
"name": node.name,
"bases": [
ast.dump(base) for base in node.bases
],
"methods": [],
"line": node.lineno,
"docstring": ast.get_docstring(node) or "",
}
for item in node.body:
if isinstance(item, ast.FunctionDef):
class_info["methods"].append({
"name": item.name,
"args": [arg.arg for arg in item.args.args],
"line": item.lineno,
"has_docstring": ast.get_docstring(item) is not None,
})
result["classes"].append(class_info)
return result
def scan_project_ast(project_dir: str) -> list[dict]:
"""扫描项目目录下所有 Python 文件"""
results = []
for py_file in Path(project_dir).rglob("*.py"):
if py_file.name.startswith("_"):
continue
try:
results.append(scan_py_file(str(py_file)))
except SyntaxError as e:
print(f"语法错误,跳过 {py_file}: {e}")
return resultspythonAST 扫描 vs inspect 扫描的区别:
| 维度 | inspect 模块 | ast 模块 |
|---|---|---|
| 工作方式 | 运行时扫描(需要 import 模块) | 静态分析(只读源文件) |
| 速度 | 较慢(需要执行导入) | 极快(纯文本解析) |
| 信息丰富度 | 能获取运行时类型、默认值 | 能获取代码结构、装饰器、注释 |
| 适用场景 | 需要精确类型信息 | 快速扫描大量文件、检查文档覆盖率 |
| 依赖要求 | 需要所有 import 可用 | 零依赖,纯 Python |
实用建议:快速生成项目概览用 AST;生成精确的 API 文档用 inspect。
11.3.5 让 AI 基于扫描结果生成文档#
有了结构化数据,接下来就是 AI 发挥的时刻了。你需要把扫描结果喂给 AI,让它生成各种文档。
Prompt 模板:生成 README.md
以下是项目的代码结构扫描结果:
[粘贴 scan_project() 或 scan_project_ast() 的输出]
请基于以上代码结构,生成一份完整的 README.md 文档,包含以下部分:
1. **项目简介**:一句话说明项目功能(基于模块名和docstring推断)
2. **功能列表**:列出所有主要模块及其功能
3. **快速开始**:
- 安装依赖(从 import 语句推断)
- 最简使用示例(基于核心类和函数的签名)
4. **项目结构**:目录树 + 每个文件的功能说明
5. **配置说明**:如果有配置文件或环境变量相关的代码,生成配置说明
6. **注意事项**:基于代码中的异常处理逻辑,列出常见问题
要求:
- 语言简洁,面向嵌入式工程师
- 代码示例必须基于实际函数签名,不要编造不存在的参数
- 如果某个模块有 docstring,优先使用 docstring 中的描述plaintextPrompt 模板:生成 API 文档
以下是项目中 protocol_parser 模块的代码结构:
[粘贴该模块的 scan_module() 输出]
请生成该模块的 API 参考文档,格式如下:
## 模块名
### 概述
(基于模块docstring描述)
### 类
#### ClassName
- **说明**:(基于类docstring)
- **继承**:(基于bases列表)
- **构造参数**:(基于__init__签名)
##### 方法列表
对每个公开方法:
- **方法签名**
- **说明**:(基于方法docstring)
- **参数说明**:每个参数的类型和含义
- **返回值**:返回类型和含义
- **异常**:如果代码中有 raise 语句,列出可能的异常
### 函数
(同上格式)
要求:
- 所有类型信息直接使用扫描结果中的 type annotation
- 如果 docstring 写的是"无文档字符串",根据函数名和参数名合理推断功能
- 每个方法附带一个简短的使用示例代码plaintextPrompt 模板:生成用户操作手册
以下是项目的完整代码结构:
[粘贴扫描结果]
请生成一份面向非技术人员的用户操作手册,包含:
1. **准备工作**:
- 硬件连接说明(基于代码中的串口号、设备名等推断)
- 软件安装步骤
2. **操作步骤**:
- 启动程序
- 基本操作流程
- 参数配置方法
3. **常见问题排查**:
- 基于代码中的异常处理和日志输出,列出用户可能遇到的错误信息及解决方法
4. **注意事项**:
- 基于代码中的限制条件(如最大数据长度、超时时间等)
要求:
- 用通俗易懂的语言,避免技术术语
- 如果必须使用技术术语,加括号解释
- 配以步骤编号,方便用户跟着操作plaintext11.3.6 自动化流水线:扫描 + 生成 + 提交#
把整个过程自动化,加入到你的开发流程中:
# generate_docs.py
"""
文档自动生成脚本
用法:python generate_docs.py --project ./src --output ./docs
"""
import argparse
import json
from code_scanner import scan_project
from pathlib import Path
def format_scan_results_for_ai(modules: list) -> str:
"""将扫描结果格式化为 AI 可读的文本"""
output = []
for mod in modules:
output.append(f"\n### 模块: {mod.name}")
output.append(f"文件: {mod.source_file}")
output.append(f"说明: {mod.docstring[:200]}")
if mod.functions:
output.append(f"\n#### 函数 ({len(mod.functions)} 个):")
for func in mod.functions:
output.append(f"\n- `{func.name}{func.signature}`")
output.append(f" 说明: {func.docstring[:100]}")
output.append(f" 返回值: {func.return_annotation}")
for p in func.parameters:
output.append(f" 参数 `{p['name']}`: 类型={p['type']}, 默认值={p['default']}")
if mod.classes:
output.append(f"\n#### 类 ({len(mod.classes)} 个):")
for cls in mod.classes:
output.append(f"\n- **class {cls.name}**({', '.join(cls.bases)})")
output.append(f" 说明: {cls.docstring[:100]}")
for method in cls.methods:
output.append(f" - `{method.name}{method.signature}`: {method.docstring[:80]}")
return "\n".join(output)
def main():
parser = argparse.ArgumentParser(description="扫描项目代码结构,输出格式化文档素材")
parser.add_argument("--project", required=True, help="项目源码目录路径")
parser.add_argument("--output", default="./docs", help="输出目录")
parser.add_argument("--format", choices=["text", "json"], default="text", help="输出格式")
args = parser.parse_args()
# 扫描项目
print(f"正在扫描项目: {args.project}")
modules = scan_project(args.project)
print(f"扫描完成,共 {len(modules)} 个模块")
# 输出结果
output_dir = Path(args.output)
output_dir.mkdir(parents=True, exist_ok=True)
if args.format == "json":
# JSON 格式(方便程序处理)
data = []
for mod in modules:
data.append({
"name": mod.name,
"docstring": mod.docstring,
"source_file": mod.source_file,
"functions": [
{"name": f.name, "signature": f.signature, "docstring": f.docstring}
for f in mod.functions
],
"classes": [
{
"name": c.name, "docstring": c.docstring, "bases": c.bases,
"methods": [
{"name": m.name, "signature": m.signature, "docstring": m.docstring}
for m in c.methods
]
}
for c in mod.classes
],
})
output_file = output_dir / "code_structure.json"
output_file.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
else:
# 文本格式(直接粘贴给 AI)
text = format_scan_results_for_ai(modules)
output_file = output_dir / "code_structure.txt"
output_file.write_text(text, encoding="utf-8")
print(f"扫描结果已保存到: {output_file}")
print(f"请将内容复制粘贴给 AI,使用配套的 Prompt 模板生成文档。")
if __name__ == "__main__":
main()python使用流程:
# 第一步:扫描代码
python generate_docs.py --project ./src --output ./docs --format text
# 第二步:复制 docs/code_structure.txt 内容,粘贴给 AI
# 第三步:使用上面的 Prompt 模板,让 AI 生成 README.md / API 文档 / 操作手册
# 第四步:将 AI 生成的文档保存到 docs/ 目录
# 第五步:git commit + push(文档与代码一起版本管理)bash11.3.7 关键价值:文档与代码始终同步#
传统的文档流程:
写代码 → (手动) 写文档 → 代码迭代 → 文档过期 → (没人有时间更新) → 文档废弃plaintextAI 驱动的文档流程:
写代码 → 运行扫描脚本 → AI 生成文档 → 代码迭代 → 重新运行扫描 → AI 重新生成plaintext核心差异:文档生成的成本从”人工数小时”降低到”机器几秒钟”。当成本足够低时,你就能做到”每次代码变更都同步更新文档”——这是手动文档永远做不到的事情。
11.4 AI 协作指南#
11.4.1 让 AI 根据代码结构自动生成标准 API 文档#
前面的 11.3 节已经给出了详细的 Prompt 模板。这里补充几个实用技巧:
技巧1:分模块喂给 AI
当项目较大时,不要一次性把整个项目的扫描结果都丢给 AI——太长了,AI 可能遗漏细节。建议一个模块一个模块地喂:
# 不推荐:一次喂整个项目(可能几万字)
[粘贴 50 个模块的扫描结果]
# 推荐:一个模块一个模块来
[粘贴 protocol_parser 模块的扫描结果]
→ 生成该模块的 API 文档
→ 审查、修正
→ 继续下一个模块plaintext技巧2:让 AI 根据函数签名推断功能
当函数没有 docstring 时,AI 可以根据函数名、参数名、类型注解推断功能:
以下函数没有 docstring:
def parse_frame(frame: bytes) -> dict: ...
def build_frame(cmd: int, data: bytes, version: int = 1) -> bytes: ...
def validate_checksum(data: bytes, expected: int) -> bool: ...
请根据函数名和参数名,为每个函数推断功能描述,生成中文 docstring。
要求:
- 描述函数的输入、处理逻辑、输出
- 包含 Args、Returns、Raises 三部分
- 保持简洁(每个函数描述不超过 5 行)plaintext技巧3:Docstring 格式选择
嵌入式项目推荐使用 Google 风格的 docstring,因为它对 AI 和人类都最友好:
def parse_frame(frame: bytes) -> dict:
"""解析协议帧,返回结构化数据。
支持 V1.0/V2.0/V3.0 三种协议版本,自动识别版本号并使用对应的解析逻辑。
Args:
frame: 原始字节流,必须包含完整的帧头和帧尾。
Returns:
包含以下键的字典:
- version (int): 协议版本号
- cmd (int): 命令字
- data (bytes): 数据域
- valid (bool): 帧是否有效
Raises:
ValueError: 当帧长度小于最小帧长度时。
ProtocolError: 当帧格式不符合任何已知协议版本时。
"""python11.4.2 让 AI 编写 Webhook 报警模块的 Prompt#
请帮我编写一个 Python 报警管理类 AlertManager,要求:
功能需求:
1. 支持钉钉、企业微信、飞书三种 Webhook 平台
2. 支持四个报警等级:INFO / WARNING / ERROR / CRITICAL
3. WARNING 及以上级别自动发送群消息
4. CRITICAL 级别支持电话告警(预留接口)
5. 同一消息在 60 秒内不重复发送(防报警风暴)
6. 支持限频配置、最低报警等级配置
技术要求:
1. 使用 requests 库发送 HTTP 请求
2. 使用 dataclass 或标准类实现
3. 配置从 dict 传入,不硬编码
4. 所有网络请求设置 10 秒超时
5. 发送失败时不抛异常,返回 False 并打印错误信息
6. 包含完整的类型注解和 docstring
请生成完整的、可直接运行的代码,包含:
- AlertLevel 枚举类
- AlertManager 类
- 使用示例plaintext11.4.3 让 AI 帮你写 Loguru 配置的 Prompt#
我有一个嵌入式工控机上运行的 Python 脚本,需求如下:
1. 日志按天切割,保留最近 30 天
2. 单个日志文件最大 50MB(超过也切割)
3. 旧日志自动压缩为 zip
4. 错误日志单独保存一份,保留 90 天
5. 使用异步写入,避免阻塞业务线程
6. 需要支持中文路径
7. 工控机硬盘总共 32GB,需要防止日志撑满硬盘
请用 Loguru 库生成完整的日志配置代码,包含:
- 控制台输出配置(带颜色)
- 普通日志文件配置
- 错误日志文件配置
- 磁盘空间检查的额外保障措施
代码需要可直接复制使用,包含详细注释。plaintext11.4.4 审查清单#
当 AI 生成代码后,你需要重点审查:
| 审查项 | 关注点 |
|---|---|
| Webhook URL 安全性 | 不应硬编码在代码中,应从配置文件或环境变量读取 |
| 超时设置 | 所有 HTTP 请求必须有超时,否则网络故障时程序会卡死 |
| 异常处理 | 报警发送失败不应导致业务程序崩溃 |
| 限频机制 | 必须有防报警风暴的限频策略,否则一个 bug 可能在 1 秒内发出 1000 条消息 |
| 日志切割路径 | 确认工控机上有足够的磁盘空间存放日志 |
| 编码设置 | 日志中的中文必须使用 UTF-8 编码 |
| 异步写入 | enqueue=True 确保日志写入不阻塞业务线程 |
| 日志敏感信息 | 日志中不应出现密码、密钥等敏感信息 |
本章小结#
| 知识点 | 解决的工程痛点 |
|---|---|
| Webhook 机器人接入 | 崩溃时无人知晓,错失最佳修复时间 |
| AlertManager 统一封装 | 一套代码支持钉钉/企微/飞书三种平台 |
| 分级报警策略 | 避免报警风暴,重要信息不被淹没 |
| logging + AlertLogHandler | 报警自动触发,业务代码零侵入 |
| Loguru 日志切割 | 防止日志撑爆工控机硬盘 |
| Loguru 异步写入 | 高频日志不阻塞业务线程 |
| Loguru 异常捕获 | 程序崩溃时完整记录现场 |
| inspect 模块扫描 | 精确获取函数签名和类型信息 |
| AST 模块扫描 | 快速扫描大量文件的代码结构 |
| AI 生成文档 | 文档成本从”人工数小时”降到”机器几秒” |
| 扫描→生成→提交 | 文档与代码始终同步,告别”文档过期” |
一句话总结:报警让脚本”会喊救命”,日志归档让脚本”有记忆”,AI 文档生成让脚本”会自述”——三者合在一起,你的脚本就从一个”沉默的程序”变成了一个”有自我报告能力的工程化产品”。
下一章,我们将讨论最后一公里的工程化问题——如何把你的 Python 脚本打包成可交付的产品:Windows 下用 PyInstaller 打包为 .exe,Linux 工控机上用 Docker 容器化部署,以及如何优雅地处理打包过程中遇到的各种”幽灵错误”。
动手练习:
- 在钉钉/企业微信/飞书中创建一个自定义机器人,用本章的代码成功发送一条测试消息
- 为你的项目配置 Loguru 日志,设置按天切割、保留 7 天、自动压缩
- 运行
code_scanner.py扫描你当前的项目,把输出结果粘贴给 AI,让它生成一份 README.md- 挑战题:把 AlertLogHandler 注册到你的项目中,让
logger.error()自动触发钉钉报警