本文目录导读：
我可以为您提供几种统计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：适合分布式系统或多进程场景
选择哪种方法取决于您的具体需求,比如是否需要持久化、是否需要历史数据查询、是否在分布式环境下运行等。
版权声明：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直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
