Python通用日志组件使用教程-平芜编程栈

本教程封装了TaskTracker 工具类，用于追踪长耗时任务的执行过程，并将结果汇总写入SQL Server数据库

1.准备工作：数据库表设计（sql代码）

-- 适用于 SQL Server CREATE TABLE [sys_task_log] ( [id] INT IDENTITY(1,1) NOT NULL PRIMARY KEY, [task_id] VARCHAR(100) NULL, -- 任务唯一标识 [status] VARCHAR(20) NULL, -- 状态: SUCCESS / ERROR [message] NVARCHAR(MAX) NULL, -- 简短描述 -- 核心统计 [total_rows] INT DEFAULT 0, -- 处理行数/数据量 [total_cost] DECIMAL(10,3) DEFAULT 0, -- 总耗时(秒) -- 分阶段耗时 (预留3个阶段) [step1_cost] DECIMAL(10,3) DEFAULT 0, [step2_cost] DECIMAL(10,3) DEFAULT 0, [step3_cost] DECIMAL(10,3) DEFAULT 0, -- 详细内容 [details] NVARCHAR(MAX) NULL, -- 详细流水账 (文本大字段) [traceback] NVARCHAR(MAX) NULL, -- 报错堆栈 [create_time] DATETIME DEFAULT GETDATE() ); -- 创建索引 CREATE INDEX [idx_task_id] ON [sys_task_log] ([task_id]);

2.核心封装代码（db_logger.py）

2.1 新建一个文件db_logger.py，这里包含两个类：

SQLServerHandler: 底层类，负责连接数据库并执行 INSERT
askTracker: 业务类，提供 info、step、success、error 等简单方法供开发者调用

2.2 依赖安装

pip install pymssql

2.3 代码实现

import logging import pymssql import traceback import os import time from datetime import datetime # ================= 配置区域 (可提取到单独的 config 文件) ================= DB_CONFIG = { 'server': '127.0.0.1', # 数据库IP 'port': 1433, # 端口 'user': 'sa', # 用户名 'password': 'your_password',# 密码 'database': 'your_db', # 数据库名 'charset': 'utf8' } TABLE_NAME = 'sys_task_log' LOG_DIR = 'logs' # 本地日志存放目录 # ====================================================================== class SQLServerHandler(logging.Handler): """ 自定义日志处理器：将特定日志写入 SQL Server """ def emit(self, record): try: # 过滤逻辑：只有 ERROR 或 带有 is_summary=True 的日志才入库 # 普通的 info 日志只打印在本地，不入库 is_error = record.levelno >= logging.ERROR is_summary = getattr(record, 'is_summary', False) if not is_error and not is_summary: return # 提取日志记录中的字段 msg = self.format(record) task_id = getattr(record, 'task_id', 'UNKNOWN') status = 'ERROR' if is_error else 'SUCCESS' rows = getattr(record, 'rows', 0) cost_time = getattr(record, 'cost_time', 0.0) step1 = getattr(record, 'step1_cost', 0.0) step2 = getattr(record, 'step2_cost', 0.0) step3 = getattr(record, 'step3_cost', 0.0) details = getattr(record, 'details', '') # 如果是报错，获取堆栈信息 tb_str = "" if record.exc_info: tb_str = "".join(traceback.format_exception(*record.exc_info)) details += f"\n\n=== Error Traceback ===\n{tb_str}" # 执行数据库写入 self._insert_to_db(status, msg, task_id, rows, cost_time, step1, step2, step3, details, tb_str) except Exception as e: self._fallback_log(f"Handler Internal Error: {e}") def _insert_to_db(self, status, msg, task_id, rows, cost, s1, s2, s3, details, tb): conn = None try: conn = pymssql.connect(**DB_CONFIG) conn.autocommit(True) with conn.cursor() as cursor: sql = f""" INSERT INTO [{TABLE_NAME}] ([status], [message], [task_id], [total_rows], [total_cost], [step1_cost], [step2_cost], [step3_cost], [details], [traceback], [create_time]) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """ cursor.execute(sql, (status, msg, task_id, rows, cost, s1, s2, s3, details, tb, datetime.now())) except Exception as e: self._fallback_log(f"Database Insert Failed: {e}") finally: if conn: conn.close() def _fallback_log(self, msg): """当数据库写不进去时，写本地文件兜底""" if not os.path.exists(LOG_DIR): os.makedirs(LOG_DIR) with open(os.path.join(LOG_DIR, 'db_error.txt'), 'a', encoding='utf-8') as f: f.write(f"[{datetime.now()}] {msg}\n") class TaskTracker: """ 任务追踪器：封装了计时、日志缓冲和入库逻辑 """ def __init__(self, task_id): self.task_id = task_id self.start_time = time.time() self.log_buffer = [] # 内存缓冲区，存储详细流水账 self.rows = 0 self.step_costs = {'step1': 0.0, 'step2': 0.0, 'step3': 0.0} self.logger = self._setup_logger() self.log(f"任务初始化: {task_id}") def _setup_logger(self): """初始化 Logger，挂载 DBHandler 和 ConsoleHandler""" logger_name = f"tracker_{self.task_id}_{os.getpid()}" logger = logging.getLogger(logger_name) logger.setLevel(logging.INFO) if not logger.handlers: # 1. 数据库处理器 db_handler = SQLServerHandler() fmt = logging.Formatter('%(message)s') # 数据库里只存纯消息 db_handler.setFormatter(fmt) logger.addHandler(db_handler) # 2. 控制台处理器 (本地显示) console = logging.StreamHandler() console.setFormatter(logging.Formatter('[%(asctime)s] %(message)s')) logger.addHandler(console) return logger def log(self, msg): """记录普通日志：打印到控制台 + 存入内存缓冲区""" timestamp = time.strftime('%H:%M:%S') self.log_buffer.append(f"[{timestamp}] {msg}") self.logger.info(f"[{self.task_id}] {msg}") def set_rows(self, count): """设置处理的数据行数""" self.rows = count def mark_step(self, step_key, cost_seconds): """记录阶段耗时，key 必须是 step1, step2, step3""" if step_key in self.step_costs: self.step_costs[step_key] = cost_seconds def success(self, summary_msg="任务完成"): """任务成功：触发汇总日志入库""" total_cost = time.time() - self.start_time self.log(f"任务结束. 总耗时: {total_cost:.2f}s") # 发送带有 is_summary=True 的日志，触发入库 self.logger.info( summary_msg, extra={ 'is_summary': True, # 关键标记 'task_id': self.task_id, 'rows': self.rows, 'cost_time': total_cost, 'step1_cost': self.step_costs['step1'], 'step2_cost': self.step_costs['step2'], 'step3_cost': self.step_costs['step3'], 'details': "\n".join(self.log_buffer) # 将缓冲区的所有日志拼接 } ) def error(self, exception_obj, summary_msg="任务异常"): """任务失败：触发错误日志入库""" total_cost = time.time() - self.start_time self.log(f"任务崩溃: {str(exception_obj)}") # ERROR 级别会自动触发入库 self.logger.error( summary_msg, exc_info=True, # 自动抓取堆栈 extra={ 'task_id': self.task_id, 'rows': self.rows, 'cost_time': total_cost, 'step1_cost': self.step_costs['step1'], 'step2_cost': self.step_costs['step2'], 'step3_cost': self.step_costs['step3'], 'details': "\n".join(self.log_buffer) } )

3. 使用示例

这里简单举了个关于耗时任务的例子，你也可以用到你的任务下面，修改你的主任务文件，然后根据你的功能实现你的想法

import time from task_logger import TaskTracker def run_my_business_task(): # 1. 初始化追踪器 (传入任务ID或名称) tracker = TaskTracker("EXPORT_JOB_20251217_001") try: # --- 阶段 1: 数据获取 --- t1 = time.time() tracker.log("开始从数据库拉取数据...") time.sleep(1) # 模拟耗时 # 假设获取到了 5000 行 row_count = 5000 tracker.set_rows(row_count) t2 = time.time() tracker.set_cost('s1', t2 - t1) # 记录阶段1耗时 tracker.log(f"数据获取完成，共 {row_count} 行") # --- 阶段 2: 数据处理 --- tracker.log("开始生成 Excel 文件...") time.sleep(1.5) # 模拟耗时 t3 = time.time() tracker.set_cost('s2', t3 - t2) # 记录阶段2耗时 tracker.log("Excel 生成完毕") # --- 阶段 3: 上传/分发 --- tracker.log("开始上传云存储...") time.sleep(0.5) # 模拟耗时 # 模拟一个随机报错 (取消注释可测试报错逻辑) # x = 1 / 0 t4 = time.time() tracker.set_cost('s3', t4 - t3) # 记录阶段3耗时 tracker.log("上传完成") # 2. 任务成功 (这步会将所有信息汇总写入数据库) tracker.success("导出任务执行成功") except Exception as e: # 3. 任务失败 (这步会立即记录错误堆栈和之前的日志) tracker.error(e, "导出任务执行失败") if __name__ == '__main__': run_my_business_task()

4.效果展示

4.1 控制台展示

[10:00:01] 任务初始化: EXPORT_JOB_... [10:00:01] 开始从数据库拉取数据... [10:00:02] 数据获取完成，共 5000 行 [10:00:02] 开始生成 Excel 文件... [10:00:03] Excel 生成完毕 [10:00:04] 导出任务执行成功