# -*- coding: utf-8 -*- import subprocess, re, mysql.connector, sys, os, time import concurrent.futures BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) DB_PHP_PATH = '/var/www/html/db.php' import os import os import os import os def __get_db_config_dynamic(): cfg = {'host': '127.0.0.1', 'user': 'saas_admin', 'password': 'AdminSaaS2026!', 'database': 'adminolt_db'} for db_path in ['/var/www/html/db.php', '/root/Workspace360_paquete/db.php']: try: if os.path.exists(db_path): with open(db_path, 'r', encoding='utf-8', errors='ignore') as f: for line in f: line = line.strip() if line.startswith('$host') and '=' in line: part = line.split("'")[1] if "'" in line else line.split('"')[1] cfg['host'] = '127.0.0.1' if part == 'localhost' else part elif line.startswith('$user') and '=' in line: cfg['user'] = line.split("'")[1] if "'" in line else line.split('"')[1] elif line.startswith('$pass') and '=' in line: cfg['password'] = line.split("'")[1] if "'" in line else line.split('"')[1] elif line.startswith('$db') and '=' in line: cfg['database'] = line.split("'")[1] if "'" in line else line.split('"')[1] break except: pass return cfg DB_CONFIG = __get_db_config_dynamic() def decode_huawei_index(index): index = int(index) if index > 65535: return (index >> 13) & 0x3F, (index >> 8) & 0x1F return (index >> 8) & 0xFF, index & 0xFF def parse_huawei_time(raw_val): import re if not raw_val or "No Such" in raw_val: return None val_str = raw_val.split("=", 1)[-1].strip() if "=" in raw_val else raw_val.strip() # 1. Intento de lectura estandar m_str = re.search(r'(\d{4})[-/](\d{1,2})[-/](\d{1,2})[ ,T]+(\d{1,2}):(\d{1,2}):(\d{1,2})', val_str) if m_str: return f"{m_str.group(1)}-{int(m_str.group(2)):02d}-{int(m_str.group(3)):02d} {int(m_str.group(4)):02d}:{int(m_str.group(5)):02d}:{int(m_str.group(6)):02d}" # 2. Intento de lectura Hex-STRING (Logica implementada del usuario) val_clean = val_str.replace("Hex-STRING:", "").replace("STRING:", "").replace('"', '').strip() hex_parts = val_clean.split() if len(hex_parts) >= 7: try: y = int(hex_parts[0]+hex_parts[1], 16) m = int(hex_parts[2], 16) d = int(hex_parts[3], 16) h = int(hex_parts[4], 16) min_ = int(hex_parts[5], 16) s = int(hex_parts[6], 16) return f"{y}-{m:02d}-{d:02d} {h:02d}:{min_:02d}:{s:02d}" except: pass return None def get_snmp_bulkwalk_num(ip, port, comm, oid): cmd = f"snmpbulkwalk -v2c -c {comm} -On -t 4 -r 2 -Cr15 {ip}:{port} {oid}" try: return subprocess.check_output(cmd, shell=True, stderr=subprocess.STDOUT).decode('latin-1', errors='ignore').splitlines() except subprocess.CalledProcessError as e: return e.output.decode('latin-1', errors='ignore').splitlines() except: return [] def get_snmp_bulkwalk_str(ip, port, comm, oid): cmd = f"snmpbulkwalk -v2c -c {comm} -Ona -t 4 -r 2 -Cr15 {ip}:{port} {oid}" try: return subprocess.check_output(cmd, shell=True, stderr=subprocess.STDOUT).decode('latin-1', errors='ignore').splitlines() except subprocess.CalledProcessError as e: return e.output.decode('latin-1', errors='ignore').splitlines() except: return [] def get_snmp_bulkwalk_hex(ip, port, comm, oid): cmd = f"snmpbulkwalk -v2c -c {comm} -Onx -t 4 -r 2 -Cr30 {ip}:{port} {oid}" try: return subprocess.check_output(cmd, shell=True, stderr=subprocess.STDOUT).decode('latin-1', errors='ignore').splitlines() except subprocess.CalledProcessError as e: return e.output.decode('latin-1', errors='ignore').splitlines() except: return [] def procesar_olt_rapido(olt_data): olt_id, ip, port, comm, nombre = olt_data start_time = time.time() print(f"[{nombre}] Iniciando descarga SNMP...") # Ejecucion secuencial contra la OLT (protege a Huawei), pero en nucleos paralelos del servidor raw_states = get_snmp_bulkwalk_num(ip, port, comm, "1.3.6.1.4.1.2011.6.128.1.1.2.46.1.15") raw_causes = get_snmp_bulkwalk_num(ip, port, comm, "1.3.6.1.4.1.2011.6.128.1.1.2.46.1.24") raw_power = get_snmp_bulkwalk_num(ip, port, comm, "1.3.6.1.4.1.2011.6.128.1.1.2.51.1.4") raw_up = get_snmp_bulkwalk_hex(ip, port, comm, "1.3.6.1.4.1.2011.6.128.1.1.2.46.1.22") raw_down = get_snmp_bulkwalk_hex(ip, port, comm, "1.3.6.1.4.1.2011.6.128.1.1.2.46.1.23") reg_st = re.compile(r'\.46\.1\.15\.(\d+)\.(\d+)\s*=\s*INTEGER:\s*[^\d-]*(-?\d+)') reg_cs = re.compile(r'\.46\.1\.24\.(\d+)\.(\d+)\s*=\s*INTEGER:\s*[^\d-]*(-?\d+)') reg_pw = re.compile(r'\.51\.1\.4\.(\d+)\.(\d+)\s*=\s*INTEGER:\s*[^\d-]*(-?\d+)') states = {f"{m.group(1)}.{m.group(2)}": int(m.group(3)) for l in raw_states if (m := reg_st.search(l))} causes = {f"{m.group(1)}.{m.group(2)}": int(m.group(3)) for l in raw_causes if (m := reg_cs.search(l))} powers = {f"{m.group(1)}.{m.group(2)}": int(m.group(3)) for l in raw_power if (m := reg_pw.search(l))} uptimes = {f"{m.group(1)}.{m.group(2)}": parse_huawei_time(m.group(3)) for l in raw_up if (m := re.search(r'\.46\.1\.22\.(\d+)\.(\d+)\s*=\s*(.*)', l))} downtimes = {f"{m.group(1)}.{m.group(2)}": parse_huawei_time(m.group(3)) for l in raw_down if (m := re.search(r'\.46\.1\.23\.(\d+)\.(\d+)\s*=\s*(.*)', l))} if not states: print(f"[{nombre}] 0 ONUs procesadas o Bloqueo Huawei. Tiempo: {time.time() - start_time:.2f}s") return 0 l_updates_realtime = [] l_updates_fallback = [] for key, run_state in states.items(): if_idx, onu_id = key.split('.') slot, port_num = decode_huawei_index(if_idx) final_status, rx_power, real_time = "OFFLINE", None, None if run_state == 1: final_status, real_time = "ONLINE", uptimes.get(key) val_pwr = powers.get(key) if val_pwr is not None and -10000 < val_pwr < 10000: rx_power = float(val_pwr) / 100.0 else: real_time, cause_code = downtimes.get(key), causes.get(key, 0) if cause_code in [3, 13]: final_status = "SIN ENERGIA" elif cause_code in [1, 2]: final_status = "LOS (FIBRA)" if real_time: l_updates_realtime.append((final_status, rx_power, real_time, olt_id, slot, port_num, onu_id)) else: l_updates_fallback.append((final_status, final_status, rx_power, olt_id, slot, port_num, onu_id)) conn = mysql.connector.connect(**DB_CONFIG) conn.autocommit = False cursor = conn.cursor() total_updates = len(l_updates_realtime) + len(l_updates_fallback) try: if l_updates_realtime: cursor.executemany("UPDATE onus SET estado_actual=%s, potencia_rx=COALESCE(%s, potencia_rx), fecha_estado=%s WHERE olt_id=%s AND slot=%s AND port=%s AND onu_id=%s", l_updates_realtime) if l_updates_fallback: cursor.executemany("UPDATE onus SET fecha_estado = IF(estado_actual != %s, NOW(), fecha_estado), estado_actual=%s, potencia_rx=COALESCE(%s, potencia_rx) WHERE olt_id=%s AND slot=%s AND port=%s AND onu_id=%s", l_updates_fallback) conn.commit() print(f"[{nombre}] Exito: {total_updates} clientes. Tiempo CPU: {time.time() - start_time:.2f}s") except Exception as e: conn.rollback() print(f"[{nombre}] Error DB: {e}") finally: cursor.close() conn.close() return total_updates def monitor_inteligente(): start_master = time.time() print(f"=== INICIANDO MONITOR RAPIDO MULTI-CORE ===") conn = mysql.connector.connect(**DB_CONFIG) cursor = conn.cursor() cursor.execute("SELECT id, ip_address, snmp_port, snmp_community, nombre FROM olts") olts = cursor.fetchall() conn.close() # ProcessPoolExecutor distribuye el trabajo en multiples nucleos fisicos del procesador # max_workers=4 significa que procesara 4 OLTs en paralelo exprimiendo al maximo tu servidor Ubuntu with concurrent.futures.ProcessPoolExecutor(max_workers=4) as ex: resultados = list(ex.map(procesar_olt_rapido, olts)) total_tiempo = time.time() - start_master print(f"=== FIN: {sum(resultados)} ONUs procesadas. Tiempo absoluto: {total_tiempo:.2f} segundos ===") if __name__ == "__main__": monitor_inteligente()