本文目录导读：
我来介绍几种Python记录定时任务日志的方法：
使用标准logging模块
import logging
import time
from datetime import datetime
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('scheduler.log', encoding='utf-8'),
logging.StreamHandler() # 同时输出到控制台
]
)
logger = logging.getLogger(__name__)
def scheduled_task():
"""定时任务函数"""
try:
logger.info(f"定时任务开始执行 - {datetime.now()}")
# 你的任务逻辑
time.sleep(2)
logger.info("定时任务执行完成")
except Exception as e:
logger.error(f"定时任务执行失败: {str(e)}")
# 测试
if __name__ == "__main__":
scheduled_task()
基于APScheduler的日志记录
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR
import logging
from datetime import datetime
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
filename='apscheduler.log'
)
class TaskLogger:
"""任务日志记录类"""
def __init__(self):
self.logger = logging.getLogger('scheduler')
def job_listener(self, event):
"""监听任务事件"""
if event.exception:
self.logger.error(f'任务 {event.job_id} 执行失败: {event.exception}')
else:
self.logger.info(f'任务 {event.job_id} 执行成功，返回值: {event.retval}')
def my_task():
"""示例任务"""
try:
# 记录任务开始
print(f"任务开始执行: {datetime.now()}")
# 任务逻辑
result = "任务执行成功"
# 记录任务结束
print(f"任务完成: {datetime.now()}")
return result
except Exception as e:
print(f"任务失败: {e}")
raise
# 调度器配置
scheduler = BlockingScheduler()
task_logger = TaskLogger()
# 添加任务监听
scheduler.add_listener(task_logger.job_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
# 添加定时任务
scheduler.add_job(
my_task,
'interval',
seconds=30,
id='my_task',
name='定时任务示例'
)
try:
scheduler.start()
except KeyboardInterrupt:
scheduler.shutdown()
带日志轮转的高级日志系统
import logging
import logging.handlers
import json
from datetime import datetime
import os
class AdvancedTaskLogger:
"""高级任务日志系统"""
def __init__(self, log_dir='logs'):
self.log_dir = log_dir
if not os.path.exists(log_dir):
os.makedirs(log_dir)
self.setup_loggers()
def setup_loggers(self):
"""配置多个日志处理器"""
# 1. 常规日志（带轮转）
regular_handler = logging.handlers.RotatingFileHandler(
filename=f'{self.log_dir}/task_scheduler.log',
maxBytes=10*1024*1024, # 10MB
backupCount=5,
encoding='utf-8'
)
regular_handler.setLevel(logging.INFO)
# 2. 错误日志
error_handler = logging.handlers.RotatingFileHandler(
filename=f'{self.log_dir}/error.log',
maxBytes=10*1024*1024,
backupCount=3,
encoding='utf-8'
)
error_handler.setLevel(logging.ERROR)
# 3. 性能日志
performance_handler = logging.handlers.RotatingFileHandler(
filename=f'{self.log_dir}/performance.log',
maxBytes=5*1024*1024,
backupCount=2,
encoding='utf-8'
)
performance_handler.setLevel(logging.INFO)
# 格式化器
regular_format = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
performance_format = logging.Formatter(
'%(asctime)s|%(task_name)s|%(duration)s|%(status)s'
)
regular_handler.setFormatter(regular_format)
error_handler.setFormatter(regular_format)
performance_handler.setFormatter(performance_format)
# 常规日志器
self.regular_logger = logging.getLogger('scheduler')
self.regular_logger.addHandler(regular_handler)
self.regular_logger.addHandler(error_handler)
# 性能日志器
self.performance_logger = logging.getLogger('performance')
self.performance_logger.addHandler(performance_handler)
def log_task_execution(self, task_name, duration, status, extra_info=None):
"""记录任务执行日志"""
# 记录常规日志
if status == 'success':
self.regular_logger.info(f"任务 {task_name} 执行成功，耗时: {duration:.2f}秒")
else:
self.regular_logger.error(f"任务 {task_name} 执行失败，耗时: {duration:.2f}秒")
# 记录性能日志
extra_data = {
'task_name': task_name,
'duration': f"{duration:.2f}",
'status': status,
'timestamp': datetime.now().isoformat()
}
if extra_info:
extra_data.update(extra_info)
self.performance_logger.info(json.dumps(extra_data))
# 使用示例
task_logger = AdvancedTaskLogger()
# 模拟任务执行
import time
def run_task():
start_time = time.time()
task_name = "数据备份"
try:
# 执行任务...
time.sleep(2)
duration = time.time() - start_time
task_logger.log_task_execution(
task_name=task_name,
duration=duration,
status='success',
extra_info={'records_processed': 1000}
)
except Exception as e:
duration = time.time() - start_time
task_logger.log_task_execution(
task_name=task_name,
duration=duration,
status='failed',
extra_info={'error': str(e)}
)
run_task()
基于装饰器的日志记录
import logging
from functools import wraps
import time
from datetime import datetime
# 设置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - [%(task_name)s] - %(message)s',
handlers=[
logging.FileHandler('decorator_tasks.log'),
logging.StreamHandler()
]
)
def log_task(name=None):
"""任务日志装饰器"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
task_name = name or func.__name__
logger = logging.getLogger(task_name)
try:
# 开始时间
start_time = time.time()
logger.info(f"任务开始执行")
# 执行任务
result = func(*args, **kwargs)
# 完成时间
duration = time.time() - start_time
logger.info(f"任务执行完成，耗时: {duration:.2f}秒")
return result
except Exception as e:
logger.error(f"任务执行失败: {str(e)}")
raise
return wrapper
return decorator
# 使用装饰器
@log_task(name="数据同步任务")
def sync_data():
"""数据同步"""
time.sleep(2)
return "数据同步完成"
@log_task(name="邮件发送任务")
def send_emails():
"""发送邮件"""
time.sleep(1)
# raise Exception("发送失败")
return "邮件发送成功"
# 执行任务
sync_data()
send_emails()
数据库日志存储
import sqlite3
import logging
from datetime import datetime
import json
class DatabaseTaskLogger:
"""数据库日志存储"""
def __init__(self, db_path='task_logs.db'):
self.conn = sqlite3.connect(db_path)
self.cursor = self.conn.cursor()
self.create_tables()
# 设置文件日志作为补充
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
filename='task_scheduler.log'
)
self.logger = logging.getLogger('task_db')
def create_tables(self):
"""创建日志表"""
self.cursor.execute('''
CREATE TABLE IF NOT EXISTS task_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
task_name TEXT NOT NULL,
status TEXT NOT NULL,
start_time TIMESTAMP,
end_time TIMESTAMP,
duration REAL,
error_message TEXT,
extra_data TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
self.conn.commit()
def log_task(self, task_name, status, start_time=None, end_time=None,
duration=None, error_message=None, extra_data=None):
"""记录任务日志到数据库"""
try:
self.cursor.execute('''
INSERT INTO task_logs
(task_name, status, start_time, end_time, duration, error_message, extra_data)
VALUES (?, ?, ?, ?, ?, ?, ?)
''', (
task_name,
status,
start_time or datetime.now(),
end_time,
duration,
error_message,
json.dumps(extra_data) if extra_data else None
))
self.conn.commit()
# 同时记录到文件
self.logger.info(f"任务 {task_name} - {status}")
except Exception as e:
self.logger.error(f"数据库日志记录失败: {e}")
def get_task_history(self, task_name=None, limit=100):
"""查询任务历史"""
if task_name:
self.cursor.execute(
"SELECT * FROM task_logs WHERE task_name=? ORDER BY created_at DESC LIMIT ?",
(task_name, limit)
)
else:
self.cursor.execute(
"SELECT * FROM task_logs ORDER BY created_at DESC LIMIT ?",
(limit,)
)
return self.cursor.fetchall()
def get_statistics(self, since_days=7):
"""获取任务统计信息"""
self.cursor.execute('''
SELECT
task_name,
COUNT(*) as total_executions,
SUM(CASE WHEN status='success' THEN 1 ELSE 0 END) as success_count,
SUM(CASE WHEN status='failed' THEN 1 ELSE 0 END) as failed_count,
AVG(duration) as avg_duration
FROM task_logs
WHERE created_at >= datetime('now', '-' || ? || ' days')
GROUP BY task_name
''', (since_days,))
return self.cursor.fetchall()
def close(self):
"""关闭数据库连接"""
self.conn.close()
# 使用示例
db_logger = DatabaseTaskLogger()
# 模拟任务
def simulated_task():
start = datetime.now()
task_name = "文件清理"
try:
# 模拟任务执行
import random
time.sleep(random.uniform(1, 3))
if random.random() > 0.8: # 20%概率失败
raise Exception("模拟错误")
# 记录成功
db_logger.log_task(
task_name=task_name,
status='success',
start_time=start,
end_time=datetime.now(),
duration=time.time() - time.mktime(start.timetuple()),
extra_data={'files_cleaned': 100}
)
except Exception as e:
db_logger.log_task(
task_name=task_name,
status='failed',
start_time=start,
end_time=datetime.now(),
duration=time.time() - time.mktime(start.timetuple()),
error_message=str(e)
)
# 执行多次任务
for _ in range(5):
simulated_task()
# 查看统计信息
stats = db_logger.get_statistics()
print("任务统计:", stats)
# 关闭连接
db_logger.close()
选择建议
- 简单任务：使用基础logging模块
- 复杂调度：使用APScheduler + logging
- 需要分析：使用数据库存储日志
- 性能监控：使用装饰器 + 性能日志
选择哪种方式取决于你的具体需求,建议从简单的开始，根据需要逐步升级。
