import multiprocessing import subprocess import time import threading import os import signal import json from datetime import datetime from typing import Any, Dict, List, Optional import sys import hashlib import platform PROGRAM_DIR = "/opt/lnxall_app/bin" CONFIG_BASE_DIR = "/app/config" # ★ 改为基础目录，不再指向单个文件 BIN_FILE = ["bmser_trans", "bmser_alarm", "bmser_upgrade"] class ProcessManager: def __init__(self): self.processes: Dict[str, subprocess.Popen] = {} self.process_info: Dict[str, dict] = {} self.monitoring = False self.last_hash = "" self.condition_met = False self.lock = threading.Lock() self.config = { "monitor_interval": 5, "max_restart_attempts": 1000, "health_check_timeout": 10, "programs": { "bmser_trans": { "path": "/opt/lnxall_app/bin/bmser_trans", "args": [], "condition_check_interval": 5, "restart_delay": 2 }, "bmser_alarm": { "path": "/opt/lnxall_app/bin/bmser_alarm", "args": [], "condition_check_interval": 5, "restart_delay": 2 }, "bmser_upgrade": { "path": "/opt/lnxall_app/bin/bmser_upgrade", "args": [], "condition_check_interval": 5, "restart_delay": 2 } } } # ────────────── ★ 新增：动态发现所有 device_config.json ────────────── def get_config_paths(self) -> List[str]: """递归扫描 /app/config 下所有的 device_config.json""" config_paths = [] try: for root, dirs, files in os.walk(CONFIG_BASE_DIR): if "device_config.json" in files: config_paths.append(os.path.join(root, "device_config.json")) except Exception as e: print(f"扫描配置文件目录失败: {e}", file=sys.stderr) return config_paths # ────────────── ★ 修改：遍历所有文件，任一命中即返回 True ────────────── def get_protocol(self): """遍历所有 device_config.json，只要任意一个包含 bmser_lib 通道即视为条件满足""" config_paths = self.get_config_paths() self.condition_met = False for config_path in config_paths: try: with open(config_path, 'r', encoding='utf-8') as f: config: Dict[str, Any] = json.load(f) for item in config.get("dev_class", []): for dev in item.get("devs", []): channel = dev.get("channel", "") if "bmser_lib" in channel: self.condition_met = True return True except Exception as e: print(f"读取配置文件 {config_path} 失败: {e}", file=sys.stderr) continue # 单个文件读取失败不影响其他文件的检测 return False # ────────────── ★ 修改：计算所有文件的组合哈希 ────────────── def get_combined_hash(self) -> Optional[str]: """按路径排序后，拼接每个文件的 MD5，再计算总哈希。 任一文件内容变更都会导致最终哈希值变化。""" config_paths = self.get_config_paths() if not config_paths: return None combined = "" for path in sorted(config_paths): # 排序保证顺序一致 try: with open(path, 'rb') as f: file_hash = hashlib.md5(f.read()).hexdigest() combined += f"{path}:{file_hash};" except Exception: pass return hashlib.md5(combined.encode()).hexdigest() if combined else None # ────────────── ★ 修改：使用组合哈希判断变更 ────────────── def check_config_change(self): current_hash = self.get_combined_hash() if current_hash is None or current_hash == self.last_hash: return False self.last_hash = current_hash return True # ────────────── 以下方法保持不变 ────────────── def log(self, message: str): timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") print(f"[{timestamp}] {message}") def check_condition(self) -> bool: try: if self.check_config_change(): return self.get_protocol() return False except Exception as e: self.log(f"条件检查失败: {e}") return False def start_program(self, program_name: str) -> bool: with self.lock: if program_name in self.processes and self.is_process_running(program_name): self.log(f"程序 {program_name} 已在运行") return True program_config = self.config["programs"].get(program_name) if not program_config: self.log(f"未找到程序 {program_name} 的配置") return False try: cmd = [program_config["path"]] + program_config["args"] process = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) self.processes[program_name] = process self.process_info[program_name] = { "start_time": datetime.now(), "restart_count": 0, "last_restart": None } self.log(f"程序 {program_name} 已启动 (PID: {process.pid})") return True except Exception as e: self.log(f"启动程序 {program_name} 失败: {e}") return False def stop_program(self, program_name: str) -> bool: with self.lock: if program_name not in self.processes: return True process = self.processes[program_name] try: process.terminate() try: process.wait(timeout=5) except subprocess.TimeoutExpired: process.kill() process.wait() del self.processes[program_name] self.log(f"程序 {program_name} 已停止") return True except Exception as e: self.log(f"停止程序 {program_name} 失败: {e}") return False def is_process_running(self, program_name: str) -> bool: if program_name not in self.processes: return False process = self.processes[program_name] return process.poll() is None def is_process_zombie(self, program_name: str) -> bool: zombie_count = 0 try: result = subprocess.run( "ps -A -o pid,state | grep 'Z'", capture_output=True, text=True, shell=True ) if result.returncode == 0: lines = result.stdout.strip().split('\n') for line in lines: if line.strip(): parts = line.split() if len(parts) >= 2 and parts[1] == 'Z': pid = int(parts[0]) try: os.waitpid(pid, 0) zombie_count += 1 print(f"已回收僵尸进程 PID: {pid}") except: pass except Exception as e: print(f"检查僵尸进程时发生错误: {e}") def restart_program(self, program_name: str) -> bool: self.log(f"正在重启程序 {program_name}") self.stop_program(program_name) program_config = self.config["programs"].get(program_name, {}) restart_delay = program_config.get("restart_delay", 2) time.sleep(restart_delay) return self.start_program(program_name) def monitor_processes(self): while self.monitoring: time.sleep(self.config["monitor_interval"]) try: self.check_condition() condition_met = self.condition_met for program_name in BIN_FILE: if not condition_met: self.stop_program(program_name) continue if not self.is_process_running(program_name): self.log(f"程序 {program_name} 已退出，尝试重启") self.restart_program(program_name) elif self.is_process_zombie(program_name): self.log(f"程序 {program_name} 进入僵尸状态，强制重启") self.restart_program(program_name) if program_name in self.process_info: info = self.process_info[program_name] max_attempts = self.config.get("max_restart_attempts", 3) if info["restart_count"] >= max_attempts: self.log(f"程序 {program_name} 已达到最大重启次数，停止重启") continue except Exception as e: self.log(f"监控过程中发生错误: {e}") def start_monitoring(self): if self.monitoring: self.log("监控已在运行") return self.monitoring = True monitor_thread = threading.Thread(target=self.monitor_processes, daemon=True) monitor_thread.start() self.log("进程监控已启动") def stop_monitoring(self): self.monitoring = False self.log("进程监控已停止") for program_name in list(self.processes.keys()): self.stop_program(program_name) def main(): manager = ProcessManager() manager.monitoring = True try: manager.log("进程监控启动") manager.monitor_processes() except Exception as e: manager.log(f"监控过程中发生错误: {e}") manager.stop_monitoring() if __name__ == "__main__": main()