import json import configparser import os import sys import subprocess import logging import threading import argparse import time import datetime import signal import functools from kafka import KafkaConsumer from datetime import datetime, timedelta import schedule # 创建日志记录器 logger = logging.getLogger(__name__) # 设置日志级别 logger.setLevel(logging.INFO) # 创建文件处理器，日志写入到文件 file_handler = logging.FileHandler('./parkwatch.log') file_handler.setLevel(logging.INFO) # 文件输出的日志级别 # 创建控制台处理器，日志输出到控制台 console_handler = logging.StreamHandler() console_handler.setLevel(logging.INFO) # 控制台输出的日志级别 # 定义日志格式 formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s') file_handler.setFormatter(formatter) console_handler.setFormatter(formatter) # 将处理器添加到日志记录器 logger.addHandler(file_handler) logger.addHandler(console_handler) # 记录日志 logger.info("parkwatch Program started") process_data = {} def read_config(config_path): """读取配置文件""" config = configparser.ConfigParser() config.read(config_path) return config def subscribe_kafka(config): """订阅Kafka topic""" kafka_server = config['KAFKA']['server'] topic = config['TRANSFER']['access_topic'] consumer = KafkaConsumer( topic, bootstrap_servers=kafka_server, value_deserializer=lambda m: json.loads(m.decode('utf-8')) #auto_offset_reset='earliest' ) logger.info(f"subscribe Kafka topic : {topic} success!") return consumer def process_kafka_message(message, config, file_index): """解析Kafka消息并按channel_size拆分写入JSON""" output_dir = os.path.dirname(config['TRANSFER']['outfile']) os.makedirs(output_dir, exist_ok=True) file_name, file_extension = os.path.splitext(os.path.basename(config['TRANSFER']['outfile'])) channel_size = int(config['TRANSFER']['channel_size']) default_skip_interval = int(config['TRANSFER']['default_skip_interval']) default_gate_interval = int(config['TRANSFER']['default_gate_interval']) source_type = config['TRANSFER']['source_type'] video_resize_ratio = round(float(config['TRANSFER']['video_resize_ratio']), 1) kfk_server = config['KAFKA']['server'] public_topic = config['KAFKA']['public_topic'] need_independent = config.getboolean('TRANSFER', 'need_independent') independent_action = [item.strip() for item in config['TRANSFER']['independent_action'].split(',')] independent_channel_size = int(config['TRANSFER']['independent_channel_size']) independent_channels = [] # 用于存储需要单独处理的通道 # 读取允许的 actions allowed_actions = set(config['TRANSFER']['actions'].split(',')) # 解析通道信息 channels = message.get("channels", []) logger.info(f"get Kafka message, contains {len(channels)} channels") # 读取配置模板 config_template = { "version": "1.0", "show_screen": False, "show_board": False, "kafka_server": kfk_server, "kafka_topic": public_topic, "max_input_per_detection": channel_size, "ba_jam_config": message.get("ba_jam_config", {}), "ba_stop_config": message.get("ba_stop_config", {}), "ba_person_gathering_config": message.get("ba_person_gathering_config", {}), "ba_person_fall_down_config": message.get("ba_person_fall_down_config", {}), "ba_wrong_direction_config": message.get("ba_wrong_direction_config", {}), "ba_barrier_gate_config": message.get("ba_barrier_gate_config", {}), "ba_fire_smoke_config": message.get("ba_fire_smoke_config", {}), "ba_junk_stack_config": message.get("ba_junk_stack_config", {}), "channels": [] } # 拆分通道，每channel_size个写入一个新文件 for i in range(0, len(channels), channel_size): config_template["channels"] = [] for channel in channels[i:i + channel_size]: processed_channel = { "id": str(channel["id"]), "url": channel["url"], "source_type": channel.get("source_type", source_type), "skip_interval": channel.get("skip_interval", default_skip_interval), "resize_ratio": channel.get("resize_ratio", video_resize_ratio), "pix_width": channel.get("pix_width", 1280), "pix_height": channel.get("pix_height", 720), "actions": [] } # 处理 actions，仅保留 STOP 和 JAM for act in channel.get("actions", []): act["action"] = act.get("action", "").upper() if act["action"] not in allowed_actions: continue # 过滤掉不在配置文件中的 action if act["action"] == independent_action[0] and need_independent: processed_channel["skip_interval"] = default_gate_interval # 将该通道添加到 independent_channels 中，而不是 config_template["channels"] independent_channels.append(processed_channel) processed_channel["actions"].append(act) # 不将该通道加入到 config_template["channels"] continue processed_channel["actions"].append(act) # 仅当 actions 不为空时才加入 channels if processed_channel["actions"]: config_template["channels"].append(processed_channel) # 写入 JSON 文件 file_path = os.path.join(output_dir, f"{file_name}_{file_index}{file_extension}") write_config_file(config_template, file_path) file_index += 1 # 如果 independent_channels 不为空，将其写入一个新的文件 if independent_channels: # 过滤掉 independent_channels 中的 actions for channel in independent_channels: channel["actions"] = [act for act in channel["actions"] if act["action"] in independent_action] # 更新 config_template，设置最大通道数为 independent_channel_size config_template["channels"] = independent_channels config_template["max_input_per_detection"] = independent_channel_size # 写入文件 file_path = os.path.join(output_dir, f"{file_name}_gate{file_extension}") write_config_file(config_template, file_path) return file_index def write_config_file(config_template, file_path): try: with open(file_path, 'w') as f: json.dump(config_template, f, indent=4) logger.info(f"created file: {file_path}") except IOError as e: logger.error(f"write file: {file_path} faild: {e}") def get_process_pids(process_name): """获取进程名称匹配的所有PID""" try: output = subprocess.check_output(["pgrep", "-x", process_name], text=True) return [int(pid) for pid in output.strip().split("\n") if pid] except subprocess.CalledProcessError: return [] def get_process_info(config): """ 获取进程信息，从配置文件中的process_file加载数据。 :param config: 配置字典，包含PROCESS配置 :return: 进程数据字典 """ process_data = {} # 从配置中获取process_file路径 process_file = config['PROCESSES']['process_file'] if not process_file: logger.error("Configuration missing 'process_file' in PROCESSES section.") return process_data try: # 打开文件并加载JSON数据 with open(process_file, 'r') as file: process_data = json.load(file) logger.info(f"Successfully loaded process data from {process_file}") except FileNotFoundError: logger.error(f"Process file '{process_file}' not found.") except json.JSONDecodeError: logger.error(f"Error decoding JSON from the process file '{process_file}'.") except Exception as e: logger.error(f"Unexpected error when reading process file '{process_file}': {e}") return process_data def manage_processes(process_data): """定期检查进程状态并在进程停止时重新启动""" if not process_data: logging.error("process_data is empty, not starting the monitor thread") return # 停止进程 process_list = process_data.get("process_list", []) for process in process_list: pids = get_process_pids(process.get("process_name")) for pid in pids: try: if os.path.exists(f"/proc/{pid}"): os.kill(pid, 9) logger.info(f"stop success {process} (PID: {pid})") else: logger.warning(f"Process {process} (PID: {pid}) does not exist") except ProcessLookupError: logger.warning(f"process {process} (PID: {pid}) is not exists") except Exception as e: logger.error(f"stop process {process} faild: {e}") # 启动进程 for process in process_data.get("process_list", []): start_process(process) def start_process(proc): """启动单个进程""" program_path = proc.get('program_path') command_ = proc.get('commands') try: if program_path == "gnome-terminal": subprocess.Popen([program_path, '--', 'bash', '-c', command_]) elif program_path == "xterm": subprocess.Popen([program_path, '-e', 'bash -c "' + command_ + '"']) elif program_path == "konsole": subprocess.Popen([program_path, '--hold', '-e', 'bash -c "' + command_ + '"']) else: subprocess.Popen([program_path, '-c', command_]) logger.info(f"start process : {command_}") except FileNotFoundError as e: logger.error(f"terminel programe not find: {e}") except Exception as e: logger.error(f"start process faild: {e}") def stop_process(pids): """停止指定PID的进程""" for pid in pids: try: os.kill(pid, signal.SIGTERM) # Try graceful stop logging.info(f"Sent SIGTERM to process with PID: {pid}") time.sleep(1) # Wait for process to terminate if os.path.exists(f"/proc/{pid}"): os.kill(pid, signal.SIGKILL) # Force kill if still running logging.info(f"Process {pid} killed with SIGKILL") except ProcessLookupError: logging.warning(f"Process with PID {pid} not found, might have exited already.") except PermissionError: logging.error(f"Permission denied while stopping process {pid}. Try running as root.") except Exception as e: logging.error(f"Error stopping process with PID {pid}: {e}") def monitor_processes(process_data): """定期检查进程状态并在进程停止时重新启动""" if not process_data: logging.error("process_data is null, not starting the monitor thread") return process_list = process_data.get("process_list", []) interval_seconds = process_data.get("interval_monitor_seconds", 10) stop_start_hour = process_data.get("stop_start_hour", 0) stop_end_hour = process_data.get("stop_end_hour", 5) if not isinstance(interval_seconds, (int, float)) or interval_seconds <= 0: logging.warning(f"the interval_seconds is not effective {interval_seconds}, defaulting to 10 seconds") interval_seconds = 10 # 兜底策略 while True: try: current_time = datetime.now() current_hour = current_time.hour if stop_start_hour <= current_hour < stop_end_hour: logging.info("Within 00:00 - 05:00, stopping all monitored processes.") for process in process_list: process_name = process.get("process_name") if not process_name or process_name=="Gate": continue pids = get_process_pids(process_name) if pids: stop_process(pids) else: for process in process_list: process_name = process.get("process_name") if not process_name: continue pids = get_process_pids(process_name) if not pids: logging.warning(f"Process {process_name} is not running, starting...") start_process(process) except Exception as e: logging.error(f"Monitoring processes encountered an exception: {e}") time.sleep(interval_seconds) # 按配置时间间隔检查 def stop_all_processes(process_data): """Stop all processes listed in the process_data.""" process_list = process_data.get("process_list", []) for process in process_list: process_name = process.get("process_name") if process_name: pids = get_process_pids(process_name) for pid in pids: try: os.kill(pid, signal.SIGTERM) # Try graceful stop logger.info(f"Sent SIGTERM to process: {process_name} (PID: {pid})") time.sleep(1) # Wait for process to terminate if os.path.exists(f"/proc/{pid}"): os.kill(pid, signal.SIGKILL) # Force kill if still running logger.info(f"Sent SIGKILL to process: {process_name} (PID: {pid})") except ProcessLookupError: logger.warning(f"Process {process_name} (PID: {pid}) not found.") except Exception as e: logger.error(f"Error stopping process {process_name} (PID: {pid}): {e}") def restart_processes(process_data): """Restart all processes listed in the process_data.""" process_list = process_data.get("process_list", []) for process in process_list: start_process(process) # Restart the process using existing start_process function def exit_handler(signum, frame, process_data): """Exit handler to stop all processes on script exit.""" logger.info("Exiting program, stopping all processes...") stop_all_processes(process_data) # Stop all processes sys.exit(0) def clean_files(log_dir,log_expiry_time,file_dir,file_expiry_time): """清理过期的日志文件和录制文件""" logger.info(f"[{datetime.now()}] Starting cleanup process...") # 清理日志文件 for filename in os.listdir(log_dir): file_path = os.path.join(log_dir, filename) if os.path.isfile(file_path): file_mtime = os.path.getmtime(file_path) # 获取文件的修改时间 if file_mtime < log_expiry_time: # 如果文件超过指定的清理时间 logger.info(f"Deleting log file: {file_path}") os.remove(file_path) # 清理录制文件 for filename in os.listdir(file_dir): file_path = os.path.join(file_dir, filename) if os.path.isfile(file_path): file_mtime = os.path.getmtime(file_path) # 获取文件的修改时间 if file_mtime < file_expiry_time: # 如果文件超过指定的清理时间 logger.info(f"Deleting file: {file_path}") os.remove(file_path) logger.info(f"[{datetime.now()}] Cleanup process completed.") def schedule_cleanup(config): """计划清理任务""" try: # 读取配置 cleanup_time = config['clean'].get('cleanup_time') log_dir = config['clean'].get('log_dir') log_expiry_days = int(config['clean'].get('log_expiry_days', 7)) # Default to 7 days if not set file_dir = config['clean'].get('file_dir') file_expiry_days = int(config['clean'].get('file_expiry_days', 7)) # Default to 7 days if not set if not cleanup_time or not log_dir or not file_dir: raise ValueError("Missing required configuration for cleanup: cleanup_time, log_dir, or file_dir.") # 计算过期时间 log_expiry_time = time.time() - (log_expiry_days * 24 * 60 * 60) file_expiry_time = time.time() - (file_expiry_days * 24 * 60 * 60) # 设置清理任务 schedule_time = datetime.strptime(cleanup_time, '%H:%M:%S').time() # 安排每天在指定时间清理文件 schedule.every().day.at(cleanup_time).do(clean_files, log_dir=log_dir, log_expiry_time=log_expiry_time, file_dir=file_dir, file_expiry_time=file_expiry_time) # 获取当前时间 now = datetime.now() target_time = now.replace(hour=schedule_time.hour, minute=schedule_time.minute, second=schedule_time.second, microsecond=0) # 如果目标时间已经过去，计划任务将推迟到明天 if now > target_time: target_time += timedelta(days=1) wait_time = (target_time - now).total_seconds() logger.info(f"Waiting {wait_time} seconds until the next cleanup task.") time.sleep(wait_time) # 进入调度循环 while True: schedule.run_pending() time.sleep(60*60) # 等待下一次任务 except ValueError as ve: logger.error(f"Configuration error: {ve}") except KeyError as ke: logger.error(f"Missing configuration key: {ke}") except Exception as e: logger.error(f"Unexpected error occurred in schedule_cleanup: {e}") def main(): """主函数""" # 设置命令行参数解析 parser = argparse.ArgumentParser(description="Process some integers.") parser.add_argument('--config', type=str, help="Path to the config file", default='msg_config.ini') args = parser.parse_args() config_path = args.config # 从命令行获取配置文件路径 config = read_config(config_path) process_data = get_process_info(config) signal.signal(signal.SIGINT, functools.partial(exit_handler, process_data=process_data)) signal.signal(signal.SIGTERM, functools.partial(exit_handler, process_data=process_data)) # 启动进程监控线程 monitor_thread = threading.Thread(target=monitor_processes, args=(process_data,), daemon=True) monitor_thread.start() # 启动清理任务调度线程 cleanup_thread = threading.Thread(target=schedule_cleanup, args=(config,), daemon=True) cleanup_thread.start() consumer = subscribe_kafka(config) if not consumer: return for message in consumer: try: process_kafka_message(message.value, config, 1) manage_processes(process_data) except Exception as e: logger.error(f"deal the Kafka messages exception: {e}") if __name__ == "__main__": main()