如何可靠实时监控电脑上所有运行的Python文件（含subprocess启动的）？

核心思路：直接追踪进程而非依赖文件 #

放弃通过创建/删除文件的间接标记方式，直接操作系统进程列表，结合进程的启动参数、PID精准识别目标Python进程，同时通过心跳检测判断进程是否僵死，彻底避免文件系统异常带来的状态误判。

具体实现方案 #

1. 精准识别目标Python进程 #

使用psutil库（跨平台支持Windows/Linux/macOS）遍历系统进程，通过进程命令行参数匹配你要监控的4个主脚本，精准定位目标进程，排除无关Python进程干扰。

示例代码片段：

import psutil

def get_target_python_processes(target_script_names):
    target_processes = []
    for proc in psutil.process_iter(['pid', 'name', 'cmdline']):
        try:
            # 过滤Python进程
            if 'python' in proc.info['name'].lower() or 'python.exe' in proc.info['name'].lower():
                cmdline = proc.info['cmdline']
                if cmdline:
                    # 匹配目标脚本文件名
                    for script_name in target_script_names:
                        if script_name in cmdline[-1]:
                            target_processes.append(proc)
                            break
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue
    return target_processes

2. 监控进程存活与僵死状态 #

基础存活监控

通过proc.is_running()直接判断进程是否存在，但该方法仅能检测进程存活，无法识别"进程存在但无响应"的僵死状态。

僵死进程检测（心跳机制）

让每个被监控的主脚本定期写入心跳信息（比如包含PID、时间戳的JSON文件），监控程序定期读取心跳文件，若某个进程的心跳时间戳超过设定阈值（如30秒），则判定为僵死。

被监控脚本的心跳逻辑（需添加到每个主脚本的后台线程）：

import json
import time
import os
import threading

HEARTBEAT_DIR = r"C:\Users\Computador\Desktop\utilities_code\heartbeats"
os.makedirs(HEARTBEAT_DIR, exist_ok=True)

def heartbeat_loop(pid):
    while True:
        heartbeat_path = os.path.join(HEARTBEAT_DIR, f"{pid}.json")
        with open(heartbeat_path, 'w') as f:
            json.dump({"pid": pid, "timestamp": time.time()}, f)
        time.sleep(10)

# 启动心跳线程，不阻塞主业务逻辑
threading.Thread(target=heartbeat_loop, args=(os.getpid(),), daemon=True).start()

监控程序的僵死检测代码：

import json
import time
import os

HEARTBEAT_DIR = r"C:\Users\Computador\Desktop\utilities_code\heartbeats"
STALE_THRESHOLD = 30  # 超过30秒无心跳判定为僵死

def check_stale_processes(target_pids):
    stale_pids = []
    current_time = time.time()
    for pid in target_pids:
        heartbeat_path = os.path.join(HEARTBEAT_DIR, f"{pid}.json")
        if os.path.exists(heartbeat_path):
            with open(heartbeat_path, 'r') as f:
                try:
                    data = json.load(f)
                    if current_time - data["timestamp"] > STALE_THRESHOLD:
                        stale_pids.append(pid)
                except json.JSONDecodeError:
                    stale_pids.append(pid)
        else:
            stale_pids.append(pid)
    return stale_pids

3. 整合到GUI监控界面（替换原有watchdog代码） #

将进程识别、状态检测逻辑整合到Tkinter界面，定期刷新并标记异常进程：

import tkinter as tk
from tkinter import ttk
import psutil
import time
import os
import json

TARGET_SCRIPTS = ["script1.py", "script2.py", "script3.py", "script4.py"]  # 替换为你的4个脚本名
HEARTBEAT_DIR = r"C:\Users\Computador\Desktop\utilities_code\heartbeats"
STALE_THRESHOLD = 30

class ProcessMonitorApp:
    def __init__(self, root):
        self.root = root
        self.root.title("Python进程监控")
        self.root.minsize(600, 400)
        
        # 创建进程列表树
        self.tree = ttk.Treeview(root, columns=("PID", "状态", "脚本名"), show="headings")
        self.tree.heading("PID", text="进程ID")
        self.tree.heading("状态", text="运行状态")
        self.tree.heading("脚本名", text="脚本名称")
        self.tree.column("PID", width=100)
        self.tree.column("状态", width=150)
        self.tree.column("脚本名", width=300)
        self.tree.pack(fill=tk.BOTH, expand=True, padx=10, pady=10)
        
        # 定期刷新进程状态
        self.refresh_processes()
        
    def get_target_processes(self):
        target_processes = []
        for proc in psutil.process_iter(['pid', 'cmdline']):
            try:
                if 'python' in proc.name().lower():
                    cmdline = proc.info['cmdline']
                    if cmdline:
                        for script in TARGET_SCRIPTS:
                            if script in cmdline[-1]:
                                target_processes.append((proc.info['pid'], script))
                                break
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                continue
        return target_processes
    
    def check_process_status(self, pid):
        try:
            proc = psutil.Process(pid)
            if not proc.is_running():
                return "已退出"
            # 检查心跳状态
            heartbeat_path = os.path.join(HEARTBEAT_DIR, f"{pid}.json")
            if os.path.exists(heartbeat_path):
                with open(heartbeat_path, 'r') as f:
                    data = json.load(f)
                    if time.time() - data["timestamp"] > STALE_THRESHOLD:
                        return "疑似僵死"
                return "正常运行"
            else:
                return "无心跳"
        except psutil.NoSuchProcess:
            return "已退出"
    
    def refresh_processes(self):
        # 清空现有列表
        for item in self.tree.get_children():
            self.tree.delete(item)
        
        # 获取目标进程并更新状态
        target_procs = self.get_target_processes()
        for pid, script in target_procs:
            status = self.check_process_status(pid)
            # 为僵死进程标记红色背景
            if status == "疑似僵死":
                self.tree.insert("", tk.END, values=(pid, status, script), tags=("stale",))
            else:
                self.tree.insert("", tk.END, values=(pid, status, script))
        
        # 配置标签样式
        self.tree.tag_configure("stale", background="#ffcccc")
        
        # 1秒后再次刷新
        self.root.after(1000, self.refresh_processes)

if __name__ == "__main__":
    os.makedirs(HEARTBEAT_DIR, exist_ok=True)
    root = tk.Tk()
    app = ProcessMonitorApp(root)
    root.mainloop()

方案优势 #

• 精准性：直接通过进程属性识别目标脚本，不会因文件删除失败、权限问题导致状态误判
• 可靠性：结合进程存活检测+心跳机制，能准确区分正常运行、已退出、僵死三种状态
• 可扩展性：后续可快速添加进程自动重启、日志收集、告警推送等功能

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