基于Python实现：通过TaskManager监控脚本状态并生成记录DataFrame

核心实现思路 #

• 每个定时脚本启动时，立即向中央监控文件写入执行启动记录；脚本结束/异常/被终止时，更新对应状态
• 新增一个独立的监控校验脚本，由Task Manager按相同周期（每分钟）运行，负责检查各任务的预期执行时间点，对比监控记录后，自动标记未执行任务并追加null状态记录

改造后的测试脚本 #

修改原测试脚本，确保进程启动、终止时都能写入监控记录，避免无执行痕迹：

import multiprocessing
import time
import pandas as pd
import datetime
import os

# 中央监控文件路径（可按需修改）
MONITOR_FILE = "script_monitor.csv"

def init_monitor_file():
    # 初始化监控文件（首次运行时创建表头）
    if not os.path.exists(MONITOR_FILE):
        pd.DataFrame(columns=['Script', 'DateTime', 'Processing']).to_csv(MONITOR_FILE, index=False)

def log_execution(script_name, status):
    # 写入/追加监控记录
    init_monitor_file()
    new_record = pd.DataFrame({
        'Script': [script_name],
        'DateTime': [datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")],
        'Processing': [status]
    })
    new_record.to_csv(MONITOR_FILE, mode='a', header=False, index=False)

def foo(n):
    # 脚本启动时立即记录运行状态
    log_execution("Foo", "running")
    try:
        for _ in range(10000 * n):
            time.sleep(1)
        # 正常结束时记录完成状态
        log_execution("Foo", "completed")
    except Exception as e:
        # 异常终止时记录错误信息
        log_execution("Foo", f"error: {str(e)}")

if __name__ == '__main__':
    p = multiprocessing.Process(target=foo, name="Foo", args=(10,))
    p.start()
    time.sleep(300)  # 运行5分钟后手动终止
    p.terminate()
    p.join()
    # 终止后记录终止状态
    log_execution("Foo", "terminated")

监控校验脚本（核心：检测未执行任务） #

编写该脚本并设置Task Manager每分钟运行一次，用来校验任务是否按预期执行：

import pandas as pd
import datetime
import os

MONITOR_FILE = "script_monitor.csv"
# 配置需监控的任务：{脚本名称: 执行周期（分钟）}
MONITORED_TASKS = {
    "Foo": 1
}

def get_expected_minute():
    # 获取上一分钟的时间节点（对应Task Manager的执行周期）
    now = datetime.datetime.now()
    expected_time = now - datetime.timedelta(minutes=1)
    return expected_time.strftime("%Y-%m-%d %H:%M")

def check_missing_executions():
    init_monitor_file()
    monitor_df = pd.read_csv(MONITOR_FILE)
    # 转换时间格式，提取分钟级时间用于匹配
    monitor_df['DateTime'] = pd.to_datetime(monitor_df['DateTime'])
    monitor_df['Minute'] = monitor_df['DateTime'].dt.strftime("%Y-%m-%d %H:%M")
    
    expected_minute = get_expected_minute()
    for task_name in MONITORED_TASKS.keys():
        # 检查该任务在上一分钟是否有执行记录
        has_record = not monitor_df[(monitor_df['Script'] == task_name) & (monitor_df['Minute'] == expected_minute)].empty
        if not has_record:
            # 未执行，写入null状态记录
            missing_record = pd.DataFrame({
                'Script': [task_name],
                'DateTime': [datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")],
                'Processing': [None]
            })
            missing_record.to_csv(MONITOR_FILE, mode='a', header=False, index=False)

def init_monitor_file():
    if not os.path.exists(MONITOR_FILE):
        pd.DataFrame(columns=['Script', 'DateTime', 'Processing']).to_csv(MONITOR_FILE, index=False)

if __name__ == '__main__':
    check_missing_executions()

配置与验证步骤 #

1. 将改造后的测试脚本设置为Task Manager每分钟运行一次
2. 将监控校验脚本同样设置为Task Manager每分钟运行一次
3. 测试脚本运行5分钟被终止后，第6分钟Task Manager尝试启动时，若因进程占用等原因无法执行，监控校验脚本会检测到该分钟无执行记录，自动写入Processing为null的记录

监控文件说明 #

script_monitor.csv会包含以下状态的记录：

• running：脚本启动执行
• completed：脚本正常结束
• terminated：脚本被外部终止
• null：脚本未在预期时间点执行

内容的提问来源于stack exchange，提问作者tonydanza123