Python循环性脚本实践要点：打造稳定高效的定时任务

在Python开发中，循环性脚本（长时间运行并定期执行任务的脚本）非常常见，比如监控系统、数据采集程序、定时清理任务等。这类脚本虽然看似简单，但实际开发中容易遇到各种陷阱。本文将分享六大核心实践要点，帮助你构建稳定高效的循环性脚本。

引子：从一次"幽灵"Bug说起

我曾开发过一个简单的日志监控脚本，它每5分钟扫描一次日志文件并发送告警。但上线后发现，最初几天还能正常工作，一周后开始频繁发送重复告警。经过排查，发现问题出在状态累积——我使用了一个全局列表来存储日志条目，但没有定期重置，导致列表不断膨胀，误判了"新日志"的出现。

这个惨痛教训让我意识到，循环脚本虽然简单，但细节决定成败。下面我将分享我的实战经验。

1. 日志策略：短日志+时间戳归档

循环脚本的日志管理需要特别注意。我的建议是：
代码不要紧，主要是要实现，这样每次运行时，都产生日志方便查看情况。

import logging
import os
from datetime import datetimedef setup_logger(log_dir="logs"):if not os.path.exists(log_dir):os.makedirs(log_dir)# 使用当前时间为日志文件名log_file = os.path.join(log_dir, f"script_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log")# 设置日志格式logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(levelname)s - %(message)s',handlers=[logging.FileHandler(log_file),logging.StreamHandler()])# 每次运行任务时创建新的日志片段（简化示例）
def run_task():# 为每个任务创建临时日志temp_logger = logging.getLogger(f"temp_{datetime.now().strftime('%Y%m%d_%H%M%S')}")# ... 实际任务逻辑 ...# 最后将关键日志归档到主日志

实践技巧：

每个任务周期生成独立日志片段，避免单一日志文件过大
添加时间戳并归档，方便回溯特定运行周期的问题
同时记录到文件和标准输出，兼顾实时监控和事后分析

2. 状态重置：内存管理的隐藏陷阱

循环执行时，某些状态变量会持续累积，必须定期重置：

class DataProcessor:def __init__(self):# 每次运行前重置self.collected_data = []def reset_state(self):self.collected_data = []self.temp_files = []# 其他需要清理的资源...def run(self):self.reset_state()  # 关键点# 任务逻辑...

常见需要重置的项目：

临时数据存储结构（列表、字典等）
文件处理器/网络连接
第三方库的会话状态
自定义的logger状态（如果有）

3. 跨天问题：时间处理的黄金法则

时间相关的任务最容易在跨天时出错：

# 错误示范 - 假设每天8点运行
if datetime.now().hour == 8:# 随着时间推移，这个判断可能永远为Falsepass# 正确做法 - 每次任务时获取最新时间
def scheduled_task():now = datetime.now()if now.hour == 8 and now.minute == 0:  # 精确到分钟# 执行操作pass# 或者更健壮的定时方案next_run = datetime(now.year, now.month, now.day, 8, 0)if now > next_run:next_run += timedelta(days=1)  # 计算明天同一时间time_to_wait = (next_run - now).total_seconds()time.sleep(time_to_wait)

关键点：

避免使用"今天"、"昨天"等相对时间，每次都基于绝对时间计算
对于每天/每周任务，明确区分"今天是否已运行"和"下次运行时间"
考虑夏令时、时区等复杂情况（如果需要）

4. 功能解耦：模块化设计

将大任务拆分为独立可运行的子任务：

class TaskManager:def __init__(self):self.tasks = {  #这里定义要运行的任务"data_import": self.data_import,"report_generation": self.generate_report,"notification": self.send_notification}def run_all(self):results = {}for name, task in self.tasks.items():results[name] = self.safe_run(task)return resultsdef safe_run(self, task):   #这里可以输出每个任务的运行情况，是一套更简单的结果，方便不熟悉的人看try:success = task()  # 任务应返回True/Falsereturn {"name": task.__name__, "success": success, "error": None}except Exception as e:return {"name": task.__name__, "success": False, "error": str(e)}def data_import(self):# 导入数据逻辑return True  # 或False   任务建议输出True or False , 除了排除的情况，上一篇文章有说明这个# 其他任务方法...

优势：

便于单独测试某个功能
更好的日志记录（可知道哪个具体任务失败）
某个任务失败不会影响其他任务运行

4. 另一种实现方式
这个其实跟上一点的是差不多的，也是说明任务解耦和得到单个运行结果，这里优化任务可以返回非True or False的情况

设计统一的任务状态反馈机制：

def track_execution(task_func):"""装饰器，标准化任务结果格式"""def wrapper(*args, kwargs):task_name = task_func.__name__try:success = task_func(*args, kwargs)return {"task": task_name,"success": bool(success),   兼容返回True/False或其他结果"result": success if isinstance(success, (bool, str)) else "completed","error": None}except Exception as e:return {"task": task_name,"success": False,"result": None,"error": str(e)}return wrapper@track_execution
def data_processing():处理数据return True@track_execution
def send_email():发送邮件if mail_sent_successfully:return Trueelse:return "Failed to connect to SMTP server"

输出示例：

{"task": "data_processing","success": true,"result": true,"error": null
}

测试策略：让脚本"跑"得久一点

稳定性来自充分测试：

环境测试：不同操作系统、Python版本
时间测试：
- 模拟长时间运行（用time.sleep或测试框架的monkeypatch）
- 测试跨天、跨月边界条件
异常测试：
- 模拟任务失败
- 测试资源耗尽情况（内存、文件句柄等）
压力测试：模拟高频运行场景

测试框架示例：

import unittest
from unittest.mock import patch
from datetime import datetime, timedeltaclass TestScript(unittest.TestCase):@patch('datetime.datetime')def test_time_calculation(self, mock_dt):mock_dt.now.return_value = datetime(2023, 1, 1, 23, 55)测试你的时间逻辑def test_task_failure(self):模拟任务失败情况result = run_task(with_mock_failure=True)self.assertFalse(result"success")@patch('time.sleep', return_value=None)   避免真实等待def test_long_running(self, mock_sleep):模拟长时间运行results = run_multiple_times(1000)   测试1000次迭代self.assertTrue(all(r"success" for r in results:-1))   除了最后一个故意失败的