本文目录导读：
我来为您提供几种筛选失败定时任务的方法，包括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检查
-
定期执行：
可以将其加入定时任务，配合监控系统使用
-
告警通知：
配合邮件、短信等方式发送告警
-
日志保留：
保存历史记录，便于问题追踪
需要根据您的具体使用场景选择合适的方案。
版权声明：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直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
