IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

Python脚本如何记录定时任务日志

wen python案例 188

本文目录导读：

Python脚本如何记录定时任务日志

  1. 使用标准logging模块
  2. 基于APScheduler的日志记录
  3. 带日志轮转的高级日志系统
  4. 基于装饰器的日志记录
  5. 数据库日志存储
  6. 选择建议

我来介绍几种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()

选择建议

  1. 简单任务：使用基础logging模块
  2. 复杂调度：使用APScheduler + logging
  3. 需要分析：使用数据库存储日志
  4. 性能监控：使用装饰器 + 性能日志

选择哪种方式取决于你的具体需求,建议从简单的开始，根据需要逐步升级。

版权声明：NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播

上一篇Python脚本如何重启失败定时任务

下一篇Python脚本如何规避定时任务重复执行

相关文章

抱歉，评论功能暂时关闭!

友情链接攻略窝 · 礼包窝 · 配置窝 · 换算窝 · 算法窝 · 查文窝 · 手册窝 · 问答窝 · 全文搜 · 手册库
相关站点礼包窝 · 查文窝