本文目录导读：
我可以为您提供几种统计Python脚本定时任务执行次数的方法：
方法1：使用文件存储计数
import json
import os
from datetime import datetime
class TaskCounter:
def __init__(self, counter_file='task_counter.json'):
self.counter_file = counter_file
self.data = self._load_counter()
def _load_counter(self):
if os.path.exists(self.counter_file):
with open(self.counter_file, 'r', encoding='utf-8') as f:
return json.load(f)
return {
'total_executions': 0,
'last_execution': None,
'daily_count': {}
}
def _save_counter(self):
with open(self.counter_file, 'w', encoding='utf-8') as f:
json.dump(self.data, f, indent=2, ensure_ascii=False)
def increment(self):
"""增加执行计数"""
self.data['total_executions'] += 1
self.data['last_execution'] = datetime.now().isoformat()
# 每日统计
today = datetime.now().strftime('%Y-%m-%d')
self.data['daily_count'][today] = self.data['daily_count'].get(today, 0) + 1
self._save_counter()
return self.data['total_executions']
def get_stats(self):
"""获取统计信息"""
return {
'total_executions': self.data['total_executions'],
'last_execution': self.data['last_execution'],
'today_count': self.data['daily_count'].get(
datetime.now().strftime('%Y-%m-%d'), 0
),
'average_daily': self._calculate_average()
}
def _calculate_average(self):
"""计算日均执行次数"""
days = len(self.data['daily_count'])
if days == 0:
return 0
total = sum(self.data['daily_count'].values())
return round(total / days, 2)
# 使用示例
if __name__ == "__main__":
counter = TaskCounter()
# 模拟定时任务执行
count = counter.increment()
print(f"第 {count} 次执行")
stats = counter.get_stats()
print(f"统计信息: {stats}")
方法2：使用数据库存储
import sqlite3
from datetime import datetime
from contextlib import contextmanager
class DatabaseCounter:
def __init__(self, db_path='task_stats.db'):
self.db_path = db_path
self._init_database()
@contextmanager
def _get_connection(self):
conn = sqlite3.connect(self.db_path)
try:
yield conn
conn.commit()
finally:
conn.close()
def _init_database(self):
with self._get_connection() as conn:
cursor = conn.cursor()
# 创建总统计表
cursor.execute('''
CREATE TABLE IF NOT EXISTS task_stats (
id INTEGER PRIMARY KEY AUTOINCREMENT,
task_name TEXT NOT NULL,
execution_time TIMESTAMP NOT NULL,
status TEXT DEFAULT 'success',
duration FLOAT,
UNIQUE(task_name, execution_time)
)
''')
# 创建每日统计表
cursor.execute('''
CREATE TABLE IF NOT EXISTS daily_stats (
id INTEGER PRIMARY KEY AUTOINCREMENT,
task_name TEXT NOT NULL,
date TEXT NOT NULL,
count INTEGER DEFAULT 0,
min_duration FLOAT,
max_duration FLOAT,
avg_duration FLOAT,
UNIQUE(task_name, date)
)
''')
def record_execution(self, task_name, duration=None, status='success'):
"""记录一次执行"""
with self._get_connection() as conn:
cursor = conn.cursor()
now = datetime.now()
# 插入执行记录
cursor.execute('''
INSERT INTO task_stats (task_name, execution_time, status, duration)
VALUES (?, ?, ?, ?)
''', (task_name, now.isoformat(), status, duration))
# 更新每日统计
date_str = now.strftime('%Y-%m-%d')
cursor.execute('''
INSERT INTO daily_stats (task_name, date, count, min_duration, max_duration, avg_duration)
VALUES (?, ?, 1, ?, ?, ?)
ON CONFLICT(task_name, date) DO UPDATE SET
count = count + 1,
min_duration = MIN(COALESCE(min_duration, ?), ?),
max_duration = MAX(COALESCE(max_duration, 0), ?),
avg_duration = (COALESCE(avg_duration, 0) * count + ?) / (count + 1)
''', (task_name, date_str, duration, duration, duration, duration, duration, duration, duration))
def get_total_count(self, task_name=None):
"""获取总执行次数"""
with self._get_connection() as conn:
cursor = conn.cursor()
if task_name:
cursor.execute('SELECT COUNT(*) FROM task_stats WHERE task_name = ?', (task_name,))
else:
cursor.execute('SELECT COUNT(*) FROM task_stats')
return cursor.fetchone()[0]
def get_daily_count(self, task_name=None, date=None):
"""获取每日执行次数"""
with self._get_connection() as conn:
cursor = conn.cursor()
if date is None:
date = datetime.now().strftime('%Y-%m-%d')
if task_name:
cursor.execute('''
SELECT count FROM daily_stats
WHERE task_name = ? AND date = ?
''', (task_name, date))
else:
cursor.execute('''
SELECT SUM(count) FROM daily_stats WHERE date = ?
''', (date,))
result = cursor.fetchone()
return result[0] if result[0] else 0
def get_statistics(self, task_name=None):
"""获取完整统计信息"""
with self._get_connection() as conn:
cursor = conn.cursor()
if task_name:
cursor.execute('''
SELECT
COUNT(*) as total,
MIN(execution_time) as first_execution,
MAX(execution_time) as last_execution,
AVG(duration) as avg_duration
FROM task_stats WHERE task_name = ?
''', (task_name,))
else:
cursor.execute('''
SELECT
COUNT(*) as total,
MIN(execution_time) as first_execution,
MAX(execution_time) as last_execution,
AVG(duration) as avg_duration
FROM task_stats
''')
return cursor.fetchone()
# 使用示例
if __name__ == "__main__":
counter = DatabaseCounter()
# 模拟定时任务执行
counter.record_execution('data_sync', duration=1.5)
counter.record_execution('data_sync', duration=2.1, status='failed')
print(f"总执行次数: {counter.get_total_count('data_sync')}")
print(f"今日执行次数: {counter.get_daily_count('data_sync')}")
stats = counter.get_statistics('data_sync')
print(f"统计数据: {stats}")
方法3：集成到定时任务中
import schedule
import time
from datetime import datetime
# 导入计数器
from task_counter import TaskCounter
def my_scheduled_task():
"""实际的定时任务"""
counter = TaskCounter()
try:
# 任务开始
start_time = time.time()
print(f"开始执行任务，时间: {datetime.now()}")
# 在这里执行你的实际任务
# 数据处理、API调用等
print("执行任务中...")
time.sleep(2) # 模拟任务执行
# 任务完成，记录执行
count = counter.increment()
print(f"任务执行完成，总计执行 {count} 次")
except Exception as e:
print(f"任务执行失败: {e}")
# 设置定时任务
def setup_scheduled_tasks():
# 每10分钟执行一次
schedule.every(10).minutes.do(my_scheduled_task)
# 每小时执行一次
# schedule.every().hour.do(my_scheduled_task)
# 每天固定时间执行
# schedule.every().day.at("10:30").do(my_scheduled_task)
# 工作日执行
# schedule.every().monday.do(my_scheduled_task)
print("定时任务已启动...")
while True:
schedule.run_pending()
time.sleep(1)
if __name__ == "__main__":
# 立即执行一次测试
my_scheduled_task()
# 启动定时任务
setup_scheduled_tasks()
方法4：使用Redis（适合分布式环境）
import redis
from datetime import datetime
class RedisTaskCounter:
def __init__(self, redis_host='localhost', redis_port=6379, redis_db=0):
self.redis_client = redis.Redis(
host=redis_host,
port=redis_port,
db=redis_db,
decode_responses=True
)
self.task_prefix = 'task_counter:'
def increment(self, task_name):
"""增加执行计数"""
pipeline = self.redis_client.pipeline()
# 总计数
total_key = f"{self.task_prefix}{task_name}:total"
pipeline.incr(total_key)
# 今日计数
today = datetime.now().strftime('%Y-%m-%d')
daily_key = f"{self.task_prefix}{task_name}:daily:{today}"
pipeline.incr(daily_key)
pipeline.expire(daily_key, 86400 * 30) # 30天后过期
# 记录最后执行时间
last_time_key = f"{self.task_prefix}{task_name}:last_time"
pipeline.set(last_time_key, datetime.now().isoformat())
results = pipeline.execute()
return results[0] # 返回总计数
def get_total_count(self, task_name):
"""获取总计数"""
key = f"{self.task_prefix}{task_name}:total"
count = self.redis_client.get(key)
return int(count) if count else 0
def get_daily_count(self, task_name, date=None):
"""获取每日计数"""
if date is None:
date = datetime.now().strftime('%Y-%m-%d')
key = f"{self.task_prefix}{task_name}:daily:{date}"
count = self.redis_client.get(key)
return int(count) if count else 0
def get_last_execution_time(self, task_name):
"""获取最后执行时间"""
key = f"{self.task_prefix}{task_name}:last_time"
return self.redis_client.get(key)
# 使用示例
if __name__ == "__main__":
counter = RedisTaskCounter()
# 模拟任务执行
count = counter.increment('data_sync')
print(f"第 {count} 次执行")
print(f"总执行次数: {counter.get_total_count('data_sync')}")
print(f"今日执行次数: {counter.get_daily_count('data_sync')}")
推荐使用场景
- 方法1：适合简单场景，不需要额外依赖
- 方法2：适合需要长期统计和历史数据查询的场景
- 方法3：与schedule库集成的方法
- 方法4：适合分布式系统或多进程场景
选择哪种方法取决于您的具体需求,比如是否需要持久化、是否需要历史数据查询、是否在分布式环境下运行等。
