#!/bin/python3 from datetime import datetime import subprocess import tarfile import threading import hashlib import fcntl import time import json import os import re SYSTEM_CFG = "/app/config/system_cfg.json" # 配置自动打包监控目录 CONFIG_DIR = "/app/config" CONFIG_DIR_ROOT = "/app/" # 配置打包路径 TAR_FILE_DIR = "/app/temp/config" LIST_MD5_FILE = "/app/temp/config/config_md5" CONFIG_MD5_FILE = "/app/temp/config/md5" LOG_FILE = "/tmp/ems/configlog.txt" CONFIG_FILE_CONTINUE = ["shm", "swp", "wal"] LC_LIST_CONFIG_FILE = "/app/config/ems_lc_list.json" # 配置下载设置 MQTT_DP_CONFIG_SET_DIR = "/app/temp/config/set" MQTT_DP_CONFIG_SET_INFO = "/app/temp/config/set/info" SSH_PASSWORD = "lnxall123" SSH_ARGV = "-oConnectTimeout=5 -oStrictHostKeyChecking=no -oUserKnownHostsFile=/dev/null -q" OPCUA_CMD = "/opt/lnxall_app/bin/opcua_cli" EMS_MODE_TAG = "EMS.ControlMode" EMS_MODE_MANUAL = 0 EMS_POWER_TAG = "EMS.SetPower" CONFIG_FILE_CONTINUE = (".shm", "-shm", ".swp", ".wal", ".tmp", ".bak", "~") original_print = __builtins__.print def print(*args, **kwargs): kwargs['flush'] = True return original_print(*args, **kwargs) class TarConfigMonitor: last_check_time = 0 check_interval = 3600 sn = "" md5 = "" def __init__(self, sn): self.last_check_time = 0 self.check_interval = 3600 self.sn = sn self.md5 = "" self.check_path() def check_path(self): os.makedirs(TAR_FILE_DIR, exist_ok=True) os.makedirs(os.path.dirname(LOG_FILE), exist_ok=True) if not os.path.exists(CONFIG_MD5_FILE): with open(CONFIG_MD5_FILE, "w") as f: pass def calculate_md5(self, file_path): """计算文件的MD5值""" hash_md5 = hashlib.md5() if os.path.exists(file_path): with open(file_path, "rb") as f: for chunk in iter(lambda: f.read(4096), b""): hash_md5.update(chunk) return hash_md5.hexdigest() return "" def get_config_md5(self): """获取当前目录下所有文件的MD5值""" md5_list = [] for root, _, files in os.walk(CONFIG_DIR): for file in files: if file.endswith(tuple(CONFIG_FILE_CONTINUE)): continue; file_path = os.path.join(root, file) s = self.calculate_md5(file_path) md5_list.append(s) return hashlib.md5("".join(md5_list).encode()).hexdigest() def skip_temp_files(self, tarinfo): """检查文件是否为临时文件，如果是则返回None以跳过""" # 检查文件扩展名是否在跳过列表中 if any(tarinfo.name.endswith(ext) for ext in CONFIG_FILE_CONTINUE): return None # 检查文件名是否以~结尾（通常是编辑器临时文件） if tarinfo.name.endswith('~'): return None return tarinfo def tar_config_dir(self): tgzfile = f"{TAR_FILE_DIR}/{self.sn}.tar.gz" """删除旧文件""" if os.path.exists(tgzfile): os.remove(tgzfile) """备份配置文件""" with tarfile.open(tgzfile, "w:gz") as tar: tar.add(CONFIG_DIR, arcname=os.path.basename(CONFIG_DIR), filter=self.skip_temp_files) self.md5 = self.calculate_md5(tgzfile) """记录备份日志""" timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") with open(LOG_FILE, "a") as f: f.write(f"[{timestamp}] 配置文件 {os.path.basename(tgzfile)} 已备份，MD5: {self.md5}\n") print(f"[{timestamp}] 配置文件 {os.path.basename(tgzfile)} 已备份，MD5: {self.md5}") self.update_md5_file() def update_md5_file(self): file = f"{self.sn}.tar.gz" info = {} try: with open(CONFIG_MD5_FILE, "r") as f: info = json.load(f) except json.JSONDecodeError as e: info = {} with open(CONFIG_MD5_FILE, "w") as f: info[file] = self.md5 json.dump(info, f) return True def check_config(self): """检查配置文件是否有变化""" try: current_md5 = self.get_config_md5() saved_md5 = "" if os.path.exists(LIST_MD5_FILE): with open(LIST_MD5_FILE, "r") as f: saved_md5 = f.read().strip() if current_md5 != saved_md5: self.tar_config_dir() with open(LIST_MD5_FILE, "w") as f: f.write(current_md5) except Exception as e: print(f"Error: {e}") class LC: def __init__(self, info): self.sn = info['sn'] self.ip = info['ipaddr'] self.password = SSH_PASSWORD self.md5 = "" def check_lc_tar_file_md5(self): file = f"{TAR_FILE_DIR}/{self.sn}.tar.gz" cmd = f"sshpass -p {SSH_PASSWORD} ssh {SSH_ARGV} root@{self.ip} \"md5sum {file}\"" result = subprocess.run(cmd, shell=True, text=True, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) if result.returncode == 0: line = result.stdout if line.endswith(".tar.gz"): md5 = line.split()[0] print("md5", md5) if md5 == self.md5: return True else: self.md5 = md5 return False else: return True def copy_lc_tar_file(self): file = f"{TAR_FILE_DIR}/{self.sn}.tar.gz" cmd = f"sshpass -p {SSH_PASSWORD} scp {SSH_ARGV} root@{self.ip}:{file} {file}" result = subprocess.run(cmd, shell=True) return result.returncode == 0 def update_md5_file(self): file = f"{self.sn}.tar.gz" info = {} try: with open(CONFIG_MD5_FILE, "r") as f: info = json.load(f) except json.JSONDecodeError as e: info = {} with open(CONFIG_MD5_FILE, "w") as f: info[file] = self.md5 json.dump(info, f) return True def get_md5(self, info): file = f"{self.sn}.tar.gz" if file in info: self.md5 = info[file] class LCConfigMonitor: lc_list = [] sn = "" last_check_time = 0 check_interval = 600 def __init__(self, sn): self.sn = sn with open(LC_LIST_CONFIG_FILE, "r") as f: root = json.load(f) if "devices" not in root: return for lc in root['devices']: if lc['sn'] == self.sn: continue self.lc_list.append(LC(lc)) try: with open(CONFIG_MD5_FILE, "r") as f: md5_info = json.load(f) for lc in self.lc_list: lc.get_md5(md5_info) except json.JSONDecodeError as e: pass def check_lc_config(self): try: for lc in self.lc_list: if lc.check_lc_tar_file_md5(): continue if lc.copy_lc_tar_file(): lc.update_md5_file() else: print("拷贝从机配置文件失败", lc.ip, lc.sn) except Exception as e: print(f"Error: {e}") class SetConfigMonitor: sn = "" last_check_time = 0 check_interval = 2 def __init__(self, sn): self.sn = sn def set_control_mode(self, mode): try: result = subprocess.run([OPCUA_CMD, "-s", EMS_MODE_TAG, str(mode)], check=True) if result.returncode != 0: print("设置控制模式失败") except subprocess.CalledProcessError as e: print(f"设置控制模式失败: {e}") def set_power_to_zero(self): try: result = subprocess.run([OPCUA_CMD, "-s", EMS_POWER_TAG, "0"], check=True) if result.returncode != 0: print("功率置0失败") except subprocess.CalledProcessError as e: print(f"功率置0失败: {e}") def calculate_md5(self, file_path): """计算文件的MD5值""" hash_md5 = hashlib.md5() with open(file_path, "rb") as f: for chunk in iter(lambda: f.read(4096), b""): hash_md5.update(chunk) return hash_md5.hexdigest() def change_local_ip_config(self,file): print("覆盖本机配置") try: self.set_control_mode(EMS_MODE_MANUAL) self.set_power_to_zero() # 解压至配置目录 result = subprocess.run(["tar", "-xf", file, "-C", CONFIG_DIR_ROOT]) print(result) if result.returncode != 0: print("tar 失败, 覆盖本地配置包失败") return # 更新本地配置 result = subprocess.run(["systemctl", "restart", "emsd"]) print(result) result = subprocess.run(["systemctl", "restart", "emud"]) if result.returncode != 0: print("systemctl 重启失败") return except Exception as e: print(f"更新本地配置时发生错误: {e}") def _run_ssh_command(self, ip, command): cmd = f"sshpass -p {SSH_PASSWORD} ssh {SSH_ARGV} root@{ip} {command}" result = subprocess.run(cmd, shell=True) return result.returncode def change_remote_ip_config(self, file, filename, ip): print("覆盖从机配置") try: # SCP file to remote host scp_cmd = f"sshpass -p {SSH_PASSWORD} scp {SSH_ARGV} {file} root@{ip}:/root/" scp_result = subprocess.run(scp_cmd, shell=True) if scp_result.returncode != 0: print("scp 失败") return # Extract the file on the remote host tar_cmd = f"tar -xf /root/{filename} -C {CONFIG_DIR_ROOT}" tar_result = self._run_ssh_command(ip, tar_cmd) if tar_result != 0: print("tar 失败") return # Remove the compressed file from the remote host rm_cmd = f"rm -rf /root/{filename}" rm_code = self._run_ssh_command(ip, rm_cmd) # Restart emsd and emud on the remote host restart_cmd = "systemctl restart emsd; systemctl restart emud" restart_ret = self._run_ssh_command(ip, restart_cmd) if restart_ret.returncode != 0: print("重启失败") if rm_code != 0: print("删除压缩包失败") except Exception as e: print(f"更改远程IP配置时发生错误: {e}") def check_config_set(self): if not os.path.exists(MQTT_DP_CONFIG_SET_INFO): return try: data = "" with open(MQTT_DP_CONFIG_SET_INFO, "r+") as f: fcntl.flock(f, fcntl.LOCK_EX) data = f.read() fcntl.flock(f, fcntl.LOCK_UN) if data != "": try: info = json.loads(data) except json.JSONDecodeError as e: print("配置文件格式错误", e) return print("覆盖配置文件信息, sn:", info.get('sn'), "ip:", info.get('ip'), "file:", info.get('file')) file_path = os.path.join(MQTT_DP_CONFIG_SET_DIR, info.get('file', '')) if not os.path.exists(file_path): print("配置包不存在") return if self.calculate_md5(file_path) != info.get('md5'): print("配置包MD5不匹配") return if info.get('sn') == self.sn: self.change_local_ip_config(file_path) elif info.get('sn'): ip = info.get('ip') if ip: self.change_remote_ip_config(file_path, info['file'], ip) else: print("覆盖远程设备配置失败, ip为空") else: print("覆盖远程设备配置失败, sn为空") os.remove(file_path) except Exception as e: print(f"处理配置文件时发生错误: {e}") finally: with open(MQTT_DP_CONFIG_SET_INFO, "w") as f: fcntl.flock(f, fcntl.LOCK_EX) f.seek(0) f.truncate() fcntl.flock(f, fcntl.LOCK_UN) # elf包升级 UPGRADE_DOWNLOAD_DIR = "/app/temp/upgrade" UPGRADE_INFO_FILE = "/app/temp/upgrade/upgrade_info" PACKAPP_BIN_FILE = "/opt/elect_resources/resources/app/bin/packapp" class EMSUpgrade: sn = '' def __init__(self, sn): self.sn = sn self.last_check_time = 0 self.check_interval = 5 self.check_path() self.threads = [] def check_path(self): os.makedirs(UPGRADE_DOWNLOAD_DIR, exist_ok=True) if not os.path.exists(UPGRADE_INFO_FILE): with open(UPGRADE_INFO_FILE, "w") as f: f.write("{}") def calculate_md5(self, file_path): """计算文件的MD5值""" hash_md5 = hashlib.md5() with open(file_path, "rb") as f: for chunk in iter(lambda: f.read(4096), b""): hash_md5.update(chunk) return hash_md5.hexdigest() def do_upgrade(self,sn_list, ip_list, file_path): try: self.set_control_mode(0) self.set_power_to_zero() except Exception as e: print(f"设置控制模式或功率失败失败: {e}") lc_threads = [] for i in range(len(sn_list)): if sn_list[i] == self.sn: continue thread = threading.Thread(target=self.upgrade_device, args= (ip_list[i], sn_list[i], file_path)) thread.start() lc_threads.append(thread) for thread in lc_threads: thread.join() self.handle_local_upgrade(sn_list, ip_list, file_path) if os.path.exists(file_path): os.remove(file_path) def get_upgrade_info(self, data): lines = data.splitlines() for line in lines: if not line: continue info = {} try: info = json.loads(line) except Exception as e: print(e) continue if not all(key in info for key in ("file", "md5", "sn", "ip")): continue file_path = os.path.join(UPGRADE_DOWNLOAD_DIR, info['file']) if not os.path.exists(file_path) or self.calculate_md5(file_path) != info['md5']: raise Exception("配置包不存在或MD5不匹配") sn_list = info['sn'] ip_list = info['ip'] if not sn_list or not ip_list or len(sn_list) != len(ip_list): raise Exception("配置包SN/IP配置错误") self.do_upgrade(sn_list, ip_list, file_path) def check_upgrade_set(self): if not os.path.exists(UPGRADE_INFO_FILE): return try: data = "" try: with open(UPGRADE_INFO_FILE, "r+") as f: # 使用 'r+' 模式 fcntl.flock(f, fcntl.LOCK_EX) data = f.read() fcntl.flock(f, fcntl.LOCK_UN) f.close() except Exception as e: print(f"读取升级信息失败: {e}") finally: with open(UPGRADE_INFO_FILE, "w") as f: fcntl.flock(f, fcntl.LOCK_EX) f.seek(0) f.truncate() fcntl.flock(f, fcntl.LOCK_UN) f.close() if data != '': thread = threading.Thread(target=self.get_upgrade_info, args=(data,)); thread.start() self.threads.append(thread) except Exception as e: print(f"check_upgrade_set error: {e}") def upgrade_device(self, ip, sn, file_path): try: result = subprocess.run([PACKAPP_BIN_FILE, ip, sn, "upgrade", file_path], check=True) if result.returncode != 0: print(f"{ip}, {sn}, {file_path}, 升级失败, {result.returncode}") else : print(f"{ip}, {sn}, {file_path}, 升级成功") except subprocess.CalledProcessError as e: print(f"升级设备 {ip}, {sn} 失败: {e}") def handle_local_upgrade(self, sn_list, ip_list, file_path): for i in range(len(sn_list)): if sn_list[i] == self.sn: for thread in self.threads: if(thread.ident != threading.current_thread().ident): thread.join() self.upgrade_device(ip_list[i], sn_list[i], file_path) break def get_control_mode(self): try: result = subprocess.run([OPCUA_CMD, "-g", "EMS.ControlMode"], check=True, stdout=subprocess.PIPE) mode = int(result.stdout.decode().split(":")[1].strip()) print(f"获取控制模式 {mode}") return mode except subprocess.CalledProcessError as e: print(f"获取控制模式失败: {e}") return 0 def set_control_mode(self, mode): try: result = subprocess.run([OPCUA_CMD, "-s", "EMS.ControlMode", str(mode)], check=True) if result.returncode != 0: print("设置控制模式失败") except subprocess.CalledProcessError as e: print(f"设置控制模式失败: {e}") def set_power_to_zero(self): try: result = subprocess.run([OPCUA_CMD, "-s", "EMS.SetPower", "0"], check=True) if result.returncode != 0: print("功率置0失败") except subprocess.CalledProcessError as e: print(f"功率置0失败: {e}") def restore_control_mode(self, mode): if mode > 0: try: result = subprocess.run([OPCUA_CMD, "-s", "EMS.ControlMode", str(mode)], check=True) if result.returncode != 0: print("设置控制模式失败") except subprocess.CalledProcessError as e: print(f"设置控制模式失败: {e}") FRPC_CONFIG_FIFO = "/tmp/ems/frpc_queue" FRPC_PATH = '/usr/bin/frpc' FRPC_CONFIG_DIR = '/tmp/ems/frpc' FRPC_LOG_DIR = '/app/temp/frpc_log' FRPC_INFO_FILE = "info" FRPC_LOG_FILE = "log" FRPC_PID_FILE = "pid" FRPC_INI_FILE = "frpc.ini" FRPC_LOG_RETENTION = 3 * 86400 # 日志保留3天 class FRPinfo: def __init__(self, name): self.name = name self.status = "" self.remoteIp = "" self.remotePort = 0 self.localIp = "" self.localPort = 0 class FRPConfig: def __init__(self, name,stop): self.name = name self.stop = stop self.status = "started" self.info = FRPinfo(self.name) self.processes = None self.not_running_time = 0 self.config_file = FRPC_CONFIG_DIR self.info_file = FRPC_CONFIG_DIR + '/' + self.name + '/' +FRPC_INFO_FILE self.log_file = FRPC_CONFIG_DIR + '/' + self.name + '/' +FRPC_LOG_FILE self.pid_file = FRPC_CONFIG_DIR + '/' + self.name + '/' +FRPC_PID_FILE self.ini_file = FRPC_CONFIG_DIR + '/' + self.name + '/' +FRPC_INI_FILE def update_stop(self, stop): self.stop = stop def update_info(self): if self.status == "running": return None try: result = subprocess.run([FRPC_PATH, 'status', '-c', self.ini_file], capture_output=True, text=True) self._process_status_output(result.stdout) except subprocess.CalledProcessError as e: print(f"获取FRP状态失败: {e}") except Exception as e: print(f"更新信息时发生错误: {e}") def _process_status_output(self, output): pattern = re.compile(r'(\S+)\s+(\S+)\s+(\S+):(\d+)\s+(\S+):(\d+)') for line in output.splitlines(): if self.name not in line: continue match = pattern.match(line) if match: self._update_info_from_match(match) self._write_info_to_file() break def _update_info_from_match(self, match): self.info.status = match.group(2) self.info.remoteIp = match.group(5) self.info.remotePort = int(match.group(6)) self.info.localIp = match.group(3) self.info.localPort = int(match.group(4)) self.status = "running" if self.info.status == 'running' else "stopped" self.not_running_time = 0 if self.status == "running" else time.time() def _write_info_to_file(self): info = { "status": self.info.status, "remoteIp": self.info.remoteIp, "remotePort": self.info.remotePort, "localIp": self.info.localIp, "localPort": self.info.localPort } with open(self.info_file, 'w+') as f: f.write(json.dumps(info, indent=4, ensure_ascii=False)) def save_pid(self): try: print("FRPC ",self.name,"save pid",self.pid_file, self.processes.pid) with open(self.pid_file, 'w+') as f: f.write(str(self.processes.pid)) return True except Exception as e: print(f"保存PID失败: {e}") return False def start_frp(self): if not os.path.exists(self.ini_file): print ("配置文件不存在 ",self.ini_file) return False try: # 启动frpc进程 print ("启动FRPC ",self.name," ",self.ini_file) process = subprocess.Popen(['frpc', '-c', self.ini_file]) self.processes = process time.sleep(2) # 等待服务启动 self.update_info() print(f"启动FRP成功: {self.name}, PID: {process.pid}") return self.save_pid() except Exception as e: print(f"启动FRP失败: {e}") return False def stop_frp(self): print(f"停止FRP: {self.name}") try: self.status = "stoped" if self.processes != {}: self.processes.terminate() self.processes.wait() return True except Exception as e: print(f"停止FRP失败: {e}") return False def check_schedule(self): is_done = False now = time.time() if self.status == "running": if now > self.stop + 5 : self.stop_frp() is_done = True elif self.not_running_time > 0 and now - self.not_running_time > 65 : self.stop_frp() elif self.status == "started": self.update_info() return is_done class FRPManager: def __init__(self): self.frpc = {} self.log = {} self.last_check_time = 0 self.check_interval = 2 self.check_path() self.load_log_dir() def check_path(self): os.makedirs(os.path.dirname(FRPC_CONFIG_FIFO), exist_ok=True) os.makedirs(FRPC_LOG_DIR, exist_ok=True) # 检查配置文件目录是否存在,如果存在,则删除下面的文件 if os.path.exists(FRPC_CONFIG_DIR): for file in os.listdir(FRPC_CONFIG_DIR): file_path = os.path.join(FRPC_CONFIG_DIR, file) try: print(f"打包日志: {file}") self.check_dir(FRPC_LOG_DIR) file_name = file +'_'+ time.strftime('%Y%m%d-%H%M%S') + '.tar.gz' # 经命令行将这个文件夹打包为tar.gz subprocess.run(['tar', '-czf', FRPC_LOG_DIR + '/' + file_name, '-C', FRPC_CONFIG_DIR, file]) # 删除这个文件夹 dir = FRPC_CONFIG_DIR + '/' + file subprocess.run(['rm', '-rf', dir]) self.log[file_name] = { 'file': file_name, 'time': time.time() } print(f"打包日志成功: {self.log[file_name] }") except Exception as e: print(f"打包日志失败: {e}") else: os.makedirs(FRPC_CONFIG_DIR, exist_ok=True) def load_log_dir(self): for file in os.listdir(FRPC_LOG_DIR): if file.endswith(".tar.gz"): self.log[file] = { 'file': file, 'time': os.path.getmtime(FRPC_LOG_DIR + '/' + file), 'name': file.split('_')[0] } def check_schedule(self): for name in list(self.frpc.keys()): try: if self.frpc[name].check_schedule(): self.packaging_log(self.frpc[name]) del self.frpc[name] except Exception as e: print(f"处理{name}时出错: {str(e)}") for name in list(self.log.keys()): try: log = self.log[name] if time.time() - log['time'] > FRPC_LOG_RETENTION: print(f"删除日志: {log['file']}") os.remove(FRPC_LOG_DIR + '/' + log['file']) del self.log[name] except Exception as e: print(f"删除日志失败: {e}") def check_dir(self, dir): os.makedirs(dir, exist_ok=True) def packaging_log(self,frpc): print(f"打包日志: {frpc.name}") try: self.check_dir(FRPC_LOG_DIR) file_name = frpc.name +'_'+ time.strftime('%Y%m%d-%H%M%S') + '.tar.gz' # 经命令行将这个文件夹打包为tar.gz subprocess.run(['tar', '-czf', FRPC_LOG_DIR + '/' + file_name, '-C', FRPC_CONFIG_DIR, frpc.name]) # 删除这个文件夹 dir = FRPC_CONFIG_DIR + '/' + frpc.name subprocess.run(['rm', '-rf', dir]) self.log[file_name] = { 'file': file_name, 'time': time.time() } print(f"打包日志成功: {self.log[file_name] }") except Exception as e: print(f"打包日志失败: {e}") def read_config(self): if not os.path.exists(FRPC_CONFIG_FIFO): return try: with open(FRPC_CONFIG_FIFO, 'r+') as f: fcntl.flock(f, fcntl.LOCK_EX) self._process_lines(f) # 清空文件内容 f.seek(0) f.truncate() fcntl.flock(f, fcntl.LOCK_UN) except (IOError, OSError) as e: print(f"文件操作错误: {e}") except json.JSONDecodeError as e: print(f"JSON解析错误: {e}") def _process_lines(self, file): for line in file: if line.strip(): # 去除空行 try: data = json.loads(line) print(data) if 'name' not in data or 'stop' not in data: continue name = data['name'] stop = data['stop'] if name not in self.frpc: self.frpc[name] = FRPConfig(name, stop) if self.frpc[name].start_frp() == False: self.frpc[name].stop = time.time() else: self.frpc[name].stop = stop except json.JSONDecodeError as e: print(f"JSON解析错误: {e}") def check_frpc_run(self): try: self.read_config() self.check_schedule() except Exception as e: print(f"检查FRP运行状态失败: {e}") class EmsMonitor: enable_lc = False sn = "" def __init__(self): self.sn = subprocess.check_output("factory get | awk -F'=' '/^SN=/{print $2}'", shell=True).decode().strip() print("当前设备SN: ", self.sn) with open(SYSTEM_CFG, "r") as f: root = json.load(f) if "en_lc_ctrl" in root and "mode" in root: self.enable_lc = (root['en_lc_ctrl'] == 1 and root['mode'] == 1) self.tar_checker = TarConfigMonitor(self.sn) self.set_checker = SetConfigMonitor(self.sn) self.frpc_manager = FRPManager() if self.enable_lc: self.lc_checker = LCConfigMonitor(self.sn) else: self.lc_checker = None self.upgrade_checker = EMSUpgrade(self.sn) print("配置监控初始化完成, 当前为 {} 模式".format("LC" if self.enable_lc else "非LC")) def _monitor_tar_config(self): while True: now = time.time() if now - self.tar_checker.last_check_time > self.tar_checker.check_interval: self.tar_checker.check_config() self.tar_checker.last_check_time = now time.sleep(1) def _monitor_set_config(self): while True: now = time.time() if now - self.set_checker.last_check_time > self.set_checker.check_interval: self.set_checker.check_config_set() self.set_checker.last_check_time = now time.sleep(1) def _monitor_upgrade(self): while True: now = time.time() if now - self.upgrade_checker.last_check_time > self.upgrade_checker.check_interval: self.upgrade_checker.check_upgrade_set() self.upgrade_checker.last_check_time = now time.sleep(1) def _monitor_frpc(self): while True: now = time.time() if now - self.frpc_manager.last_check_time > self.frpc_manager.check_interval: self.frpc_manager.check_frpc_run() self.frpc_manager.last_check_time = now time.sleep(1) def _monitor_lc_config(self): if not self.enable_lc: return while True: now = time.time() if now - self.lc_checker.last_check_time > self.lc_checker.check_interval: self.lc_checker.check_lc_config() self.frpc_manager.last_check_time = now time.sleep(1) def run(self): # 创建并启动线程 threads = [ threading.Thread(target=self._monitor_tar_config, daemon=True), threading.Thread(target=self._monitor_set_config, daemon=True), threading.Thread(target=self._monitor_upgrade, daemon=True), threading.Thread(target=self._monitor_frpc, daemon=True), ] if self.enable_lc: threads.append(threading.Thread(target=self._monitor_lc_config, daemon=True)) for thread in threads: thread.start() # 主线程保持运行 try: while True: time.sleep(1) except KeyboardInterrupt: print("监控程序停止") # 使用示例 if __name__ == "__main__": os.makedirs(TAR_FILE_DIR, exist_ok=True) os.makedirs(MQTT_DP_CONFIG_SET_DIR, exist_ok=True) monitor = EmsMonitor() print("配置监控启动") monitor.run()