本文目录导读：
我来为您提供几种筛选失败定时任务的方法，包括Windows任务计划程序和Linux cron任务。
Windows任务计划程序
使用PowerShell（推荐）
import subprocess
import json
def filter_failed_windows_tasks():
"""筛选Windows失败的定时任务"""
try:
# 获取所有任务状态
cmd = 'powershell -Command "Get-ScheduledTask | Where-Object {$_.State -ne \'Ready\'} | Select-Object TaskName, State, LastRunTime, LastTaskResult | ConvertTo-Json"'
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
if result.stdout.strip():
tasks = json.loads(result.stdout) if result.stdout.startswith('[') else [json.loads(result.stdout)]
failed_tasks = []
for task in tasks:
# 非0表示失败
if task.get('LastTaskResult') and task['LastTaskResult'] != 0:
failed_tasks.append(task)
return failed_tasks
except Exception as e:
print(f"错误: {e}")
return []
# 使用示例
failed = filter_failed_windows_tasks()
for task in failed:
print(f"任务名: {task['TaskName']}")
print(f"状态: {task['State']}")
print(f"返回码: {task['LastTaskResult']}")
print("-" * 50)
使用任务日志文件
import win32com.client
from datetime import datetime, timedelta
def check_task_failures_by_log():
"""通过事件日志检查失败任务"""
try:
# 连接到事件日志
log_type = "Microsoft-Windows-TaskScheduler/Operational"
log = win32com.client.GetObject("winmgmts:{impersonationLevel=impersonate}!\\\\.\\root\\cimv2")
# 查询24小时内的失败事件
query = f"""
SELECT * FROM Win32_NTLogEvent
WHERE LogFile='{log_type}'
AND EventCode=103
AND TimeGenerated > '{datetime.now() - timedelta(hours=24):%Y%m%d%H%M%S}'
"""
events = log.ExecQuery(query)
failed_tasks = []
for event in events:
failed_tasks.append({
'task_name': event.InsertionStrings[1] if len(event.InsertionStrings) > 1 else 'Unknown',
'time': event.TimeGenerated,
'message': event.Message
})
return failed_tasks
except Exception as e:
print(f"错误: {e}")
return []
# 使用示例
failed_tasks = check_task_failures_by_log()
for task in failed_tasks:
print(f"失败任务: {task['task_name']}")
print(f"时间: {task['time']}")
Linux Cron任务
检查系统日志
import subprocess
import re
from datetime import datetime, timedelta
def filter_failed_cron_jobs():
"""筛选Linux失败的cron任务"""
failed_jobs = []
try:
# 从系统日志获取cron失败信息
cmd = "grep -i 'FAILED\|ERROR\|CRON' /var/log/syslog | grep -v 'CRON.*INFO'"
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
if result.stdout:
lines = result.stdout.strip().split('\n')
for line in lines:
if 'failed' in line.lower() or 'error' in line.lower():
# 解析日志行
match = re.search(r'(\w+\s+\d+\s+\d+:\d+:\d+).*?CRON.*?\((.*?)\)', line)
if match:
failed_jobs.append({
'time': match.group(1),
'command': match.group(2),
'log': line
})
except Exception as e:
print(f"错误: {e}")
return failed_jobs
# 使用示例
failed = filter_failed_cron_jobs()
for job in failed:
print(f"时间: {job['time']}")
print(f"命令: {job['command']}")
检查邮件通知
import subprocess
def check_cron_mail_failures():
"""通过邮件检查cron失败"""
try:
# 检查mail命令的输出
cmd = "mail -H | head -50"
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
if result.stdout:
failed_jobs = []
for line in result.stdout.strip().split('\n'):
if 'Cron' in line or 'cron' in line:
failed_jobs.append(line)
return failed_jobs
except Exception as e:
print(f"错误: {e}")
return []
通用方案：监控脚本
#!/usr/bin/python3
import os
import sys
import subprocess
import json
from datetime import datetime
class TaskMonitor:
def __init__(self):
self.failed_tasks = []
def check_windows_tasks(self):
"""检查Windows任务"""
if os.name == 'nt':
cmd = 'powershell -Command "Get-ScheduledTask | Where-Object {$_.State -ne \'Ready\'} | Select-Object TaskName, State, LastTaskResult | ConvertTo-Json"'
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
if result.stdout.strip():
tasks = json.loads(result.stdout) if result.stdout.startswith('[') else [json.loads(result.stdout)]
for task in tasks:
if task.get('LastTaskResult') and task['LastTaskResult'] != 0:
self.failed_tasks.append(task)
def check_linux_tasks(self):
"""检查Linux任务"""
if os.name == 'posix':
cmd = "grep -i 'FAILED\|ERROR' /var/log/cron 2>/dev/null || grep -i 'FAILED\|ERROR' /var/log/syslog 2>/dev/null"
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
if result.stdout:
for line in result.stdout.strip().split('\n'):
self.failed_tasks.append({
'task': line,
'type': 'linux_cron',
'time': datetime.now().isoformat()
})
def run(self):
"""执行检查"""
self.check_windows_tasks()
self.check_linux_tasks()
# 输出结果
if self.failed_tasks:
print(f"发现 {len(self.failed_tasks)} 个失败任务:")
for task in self.failed_tasks:
print(json.dumps(task, indent=2, ensure_ascii=False))
else:
print("未发现失败任务")
return self.failed_tasks
# 使用示例
if __name__ == "__main__":
monitor = TaskMonitor()
failed_tasks = monitor.run()
# 可以添加告警功能
if failed_tasks:
print("发送告警通知...")
# 这里添加邮件、短信等通知逻辑
高级用法：Web监控界面
from flask import Flask, jsonify
import subprocess
import json
app = Flask(__name__)
@app.route('/api/failed-tasks')
def get_failed_tasks():
"""API接口获取失败任务"""
failed_tasks = []
# Windows检查
if os.name == 'nt':
cmd = 'powershell -Command "Get-ScheduledTask | Where-Object {$_.LastTaskResult -ne 0} | Select-Object TaskName, State, LastRunTime, LastTaskResult | ConvertTo-Json"'
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
if result.stdout.strip():
failed_tasks = json.loads(result.stdout) if result.stdout.startswith('[') else [json.loads(result.stdout)]
return jsonify({
'timestamp': datetime.now().isoformat(),
'failed_tasks': failed_tasks,
'total_failed': len(failed_tasks)
})
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000, debug=True)
使用建议
-
根据操作系统选择：
- Windows：使用PowerShell脚本
- Linux：使用日志分析和mail检查
-
定期执行：
可以将其加入定时任务，配合监控系统使用
-
告警通知：
配合邮件、短信等方式发送告警
-
日志保留：
保存历史记录，便于问题追踪
需要根据您的具体使用场景选择合适的方案。
