# -*- coding: utf-8 -*- import subprocess, re, mysql.connector, sys, concurrent.futures, os import datetime original_print = print def print(*args, **kwargs): timestamp = datetime.datetime.now().strftime("[%Y-%m-%d %H:%M:%S]") original_print(timestamp, *args, **kwargs) sys.stdout.reconfigure(line_buffering=True) 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() # REGEX PARA SNMP REG_SN = re.compile(r'\.43\.1\.3\.(\d+)\.(\d+)\s*=\s*(.*)') REG_NM = re.compile(r'\.43\.1\.9\.(\d+)\.(\d+)\s*=\s*(.*)') REG_ST = re.compile(r'\.46\.1\.15\.(\d+)\.(\d+)\s*=\s*INTEGER:\s*(\d+)') REG_CS = re.compile(r'\.46\.1\.24\.(\d+)\.(\d+)\s*=\s*INTEGER:\s*(\d+)') REG_PW = re.compile(r'\.51\.1\.4\.(\d+)\.(\d+)\s*=\s*INTEGER:\s*(-?\d+)') REG_PN = re.compile(r'\.31\.1\.1\.1\.1\.(\d+)\s*=\s*STRING:\s*(.*)') REG_PD = re.compile(r'\.21\.1\.6\.(\d+)\s*=\s*STRING:\s*(.*)') REG_MD = re.compile(r'\.45\.1\.4\.(\d+)\.(\d+)\s*=\s*STRING:\s*(.*)') 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_walk(ip, port, community, oid): cmd = f"snmpbulkwalk -v2c -c {community} -t 10 -r 2 -Cr10 -Onx {ip}:{port} {oid}" try: return subprocess.check_output(cmd, shell=True).decode('latin-1', errors='ignore').splitlines() except: return [] def get_snmp_walk_str(ip, port, community, oid): cmd = f"snmpbulkwalk -v2c -c {community} -t 10 -r 2 -Cr10 -Ona {ip}:{port} {oid}" try: return subprocess.check_output(cmd, shell=True).decode('latin-1', errors='ignore').splitlines() except: return [] def decode_huawei_index(index): index = int(index) if index > 65535: return (index >> 13) & 0x3F, (index >> 8) & 0x1F else: return (index >> 8) & 0xFF, index & 0xFF def clean_string(val): return val.replace("STRING: ", "").replace('"', '').strip() def clean_name_cliente(raw_name): val = raw_name.strip() if "Hex-STRING:" in val: hex_part = val.split("Hex-STRING:")[1] hex_clean = re.sub(r'[^0-9A-Fa-f]', '', hex_part) if len(hex_clean) % 2 != 0: hex_clean = hex_clean[:-1] try: val = bytes.fromhex(hex_clean).decode('latin-1', errors='ignore') except: pass val = val.replace("STRING:", "").replace('"', '').replace('_', ' ').strip() for corte in [" zone ", " Zone ", " extid ", " authd ", " ont ", "\x00"]: if corte in val: val = val.split(corte)[0] return val.strip() def clean_serial(raw_sn): val = raw_sn.replace("Hex-STRING: ", "").replace("STRING: ", "").replace('"', '').strip() if " " in val and len(val) > 10: try: ascii_val = bytes.fromhex(val.replace(" ", "")).decode('utf-8') if re.match(r'^[A-Za-z0-9]+$', ascii_val): return ascii_val except: pass return val.replace(" ", "") def procesar_olt(olt_data): olt_id, olt_name, olt_ip, olt_port, olt_comm = olt_data print(f"--- Procesando Sincronizacion: {olt_name} ---") seriales, nombres, estados, causas, potencias, info_puertos, modelos = {}, {}, {}, {}, {}, {}, {} tech_names, desc_names = {}, {} try: with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor: f_sn = executor.submit(get_snmp_walk, olt_ip, olt_port, olt_comm, "1.3.6.1.4.1.2011.6.128.1.1.2.43.1.3") f_nm = executor.submit(get_snmp_walk_str, olt_ip, olt_port, olt_comm, "1.3.6.1.4.1.2011.6.128.1.1.2.43.1.9") f_st = executor.submit(get_snmp_walk, olt_ip, olt_port, olt_comm, "1.3.6.1.4.1.2011.6.128.1.1.2.46.1.15") f_cs = executor.submit(get_snmp_walk, olt_ip, olt_port, olt_comm, "1.3.6.1.4.1.2011.6.128.1.1.2.46.1.24") f_pw = executor.submit(get_snmp_walk, olt_ip, olt_port, olt_comm, "1.3.6.1.4.1.2011.6.128.1.1.2.51.1.4") f_up = executor.submit(get_snmp_walk, olt_ip, olt_port, olt_comm, "1.3.6.1.4.1.2011.6.128.1.1.2.46.1.22") f_dw = executor.submit(get_snmp_walk, olt_ip, olt_port, olt_comm, "1.3.6.1.4.1.2011.6.128.1.1.2.46.1.23") f_pn = executor.submit(get_snmp_walk_str, olt_ip, olt_port, olt_comm, "1.3.6.1.2.1.31.1.1.1.1") f_pd = executor.submit(get_snmp_walk_str, olt_ip, olt_port, olt_comm, "1.3.6.1.4.1.2011.6.128.1.1.2.21.1.6") f_md = executor.submit(get_snmp_walk_str, olt_ip, olt_port, olt_comm, "1.3.6.1.4.1.2011.6.128.1.1.2.45.1.4") l_sn, l_nm, l_st, l_cs, l_pw = f_sn.result(), f_nm.result(), f_st.result(), f_cs.result(), f_pw.result() l_up, l_dw, l_pn, l_pd = f_up.result(), f_dw.result(), f_pn.result(), f_pd.result() l_md = f_md.result() if not l_pn: print(f" [ERROR CRITICO] {olt_name}: Timeout SNMP detectado. Abortando sincronizacion.") return for l in l_pn: m = REG_PN.search(l) if m: val = clean_string(m.group(2)) m_fsp = re.search(r'(GPON|EPON|XG-PON|XGS-PON)[^\d]*(\d+)/(\d+)/(\d+)', val) if m_fsp: f = int(m_fsp.group(2)) s = int(m_fsp.group(3)) p = int(m_fsp.group(4)) tech_names[(f, s, p)] = val for l in l_pd: m = REG_PD.search(l) if m: val = clean_string(m.group(2)) if val: s, p = decode_huawei_index(int(m.group(1))) desc_names[(0, s, p)] = val for k, v_tech in tech_names.items(): info_puertos[k] = desc_names.get(k, v_tech) conn = mysql.connector.connect(**DB_CONFIG) conn.autocommit = False cursor = conn.cursor() # ========================================================================= # AUTO-REPARACIN DE MYSQL: Destruye el candado defectuoso # ========================================================================= try: cursor.execute("ALTER TABLE naps DROP INDEX unique_port") conn.commit() except: pass try: cursor.execute("ALTER TABLE naps ADD UNIQUE INDEX unique_olt_fsp (olt_id, frame, slot, port)") conn.commit() except: pass try: cursor.execute("SELECT id, frame, slot, port FROM naps WHERE olt_id = %s", (olt_id,)) naps_existentes = {} for r in cursor.fetchall(): try: naps_existentes[(int(r[1]), int(r[2]), int(r[3]))] = r[0] except: pass for (f, s, p), nombre_nap in info_puertos.items(): if (f, s, p) in naps_existentes: nap_id = naps_existentes[(f, s, p)] try: cursor.execute("UPDATE naps SET alias_nombre = %s WHERE id = %s AND (alias_nombre LIKE 'GPON%%' OR alias_nombre LIKE 'Puerto%%')", (nombre_nap, nap_id)) except: pass else: try: cursor.execute("INSERT INTO naps (olt_id, frame, slot, port, alias_nombre) VALUES (%s, %s, %s, %s, %s)", (olt_id, f, s, p, nombre_nap)) except: pass conn.commit() except Exception as e: print(f" [ERROR BD NAPS] {olt_name}: {e}") # PROCESAR ONUS for l in l_sn: m = REG_SN.search(l) if m: seriales[(int(m.group(1)), int(m.group(2)))] = clean_serial(m.group(3)) if not seriales: cursor.close() conn.close() print(f" [OK] {olt_name}: Sincronizacion finalizada. No se detectaron clientes activos en la lectura.") return for l in l_nm: m = REG_NM.search(l) if m: nombres[(int(m.group(1)), int(m.group(2)))] = clean_name_cliente(m.group(3)) for l in l_st: m = REG_ST.search(l) if m: estados[(int(m.group(1)), int(m.group(2)))] = int(m.group(3)) for l in l_cs: m = REG_CS.search(l) if m: causas[(int(m.group(1)), int(m.group(2)))] = int(m.group(3)) for l in l_pw: m = REG_PW.search(l) if m: raw = int(m.group(3)) if -6000 < raw < 1000: potencias[(int(m.group(1)), int(m.group(2)))] = float(raw)/100.0 for l in l_md: m = REG_MD.search(l) if m: val = clean_string(m.group(3)) if val: modelos[(int(m.group(1)), int(m.group(2)))] = val uptimes, downtimes = {}, {} for l in l_up: m = re.search(r'\.46\.1\.22\.(\d+)\.(\d+) = (.*)', l) if m: uptimes[f"{m.group(1)}.{m.group(2)}"] = parse_huawei_time(m.group(3)) for l in l_dw: m = re.search(r'\.46\.1\.23\.(\d+)\.(\d+) = (.*)', l) if m: downtimes[f"{m.group(1)}.{m.group(2)}"] = parse_huawei_time(m.group(3)) try: cursor.execute("SELECT id, serial_number, slot, port, onu_id, nombre_cliente, modelo FROM onus WHERE olt_id = %s", (olt_id,)) db_onus = cursor.fetchall() sn_to_id = {r[1]: r[0] for r in db_onus if r[1]} pos_to_id = {(r[2], r[3], r[4]): r[0] for r in db_onus} db_nombres = {(r[2], r[3], r[4]): r[5] for r in db_onus} db_modelos = {(r[2], r[3], r[4]): r[6] for r in db_onus} l_inserts, l_updates, l_history, ids_to_keep = [], [], [], set() for k, sn in seriales.items(): slot, port = decode_huawei_index(k[0]) onu_id = k[1] nom_snmp = nombres.get(k, "").strip() nom_db = db_nombres.get((slot, port, onu_id), "") if (not nom_snmp or nom_snmp == "SIN NOMBRE") and nom_db and nom_db != "SIN NOMBRE": nom = nom_db else: nom = nom_snmp if nom_snmp else "SIN NOMBRE" mod_snmp = modelos.get(k, "").upper().strip() mod_db = db_modelos.get((slot, port, onu_id), "") if (not mod_snmp or mod_snmp == "ONT GPON") and mod_db and mod_db != "ONT GPON": mod_str = mod_db else: mod_str = mod_snmp if mod_snmp else "ONT GPON" st_val = estados.get(k, 2) if st_val == 1: st_txt, pwr, real_time = "ONLINE", potencias.get(k, -99.99), uptimes.get(f"{k[0]}.{k[1]}") else: pwr, c, real_time = -99.99, causas.get(k, 0), downtimes.get(f"{k[0]}.{k[1]}") if c in [3, 13]: st_txt = "SIN ENERGIA" elif c in [1, 2]: st_txt = "LOS (FIBRA)" else: st_txt = "OFFLINE" if sn and sn in sn_to_id: db_id = sn_to_id[sn] ids_to_keep.add(db_id) l_updates.append((slot, port, onu_id, sn, nom, pwr, real_time, st_txt, st_txt, mod_str, db_id)) if st_val == 1 and pwr > -90: l_history.append((db_id, pwr)) elif (slot, port, int(onu_id)) in pos_to_id: db_id = pos_to_id[(slot, port, int(onu_id))] ids_to_keep.add(db_id) l_updates.append((slot, port, onu_id, sn, nom, pwr, real_time, st_txt, st_txt, mod_str, db_id)) if st_val == 1 and pwr > -90: l_history.append((db_id, pwr)) else: l_inserts.append((olt_id, slot, port, onu_id, sn, nom, st_txt, pwr, real_time, mod_str)) l_deletes = [(r[0],) for r in db_onus if r[0] not in ids_to_keep] if l_inserts: cursor.executemany("INSERT INTO onus (olt_id, frame, slot, port, onu_id, serial_number, nombre_cliente, estado_actual, potencia_rx, fecha_estado, modelo) VALUES (%s, 0, %s, %s, %s, %s, %s, %s, %s, COALESCE(%s, NOW()), %s)", l_inserts) if l_updates: cursor.executemany("UPDATE onus SET frame=0, slot=%s, port=%s, onu_id=%s, serial_number=%s, nombre_cliente=%s, potencia_rx=%s, fecha_estado = COALESCE(%s, IF(estado_actual != %s, NOW(), fecha_estado)), estado_actual=%s, modelo=%s WHERE id=%s", l_updates) if l_history: cursor.executemany("INSERT INTO onu_signal_history (onu_db_id, rx_power, timestamp) VALUES (%s, %s, NOW())", l_history) cursor.execute("DELETE FROM onu_signal_history WHERE timestamp < NOW() - INTERVAL 7 DAY") if l_deletes: cursor.executemany("DELETE FROM onus WHERE id=%s", l_deletes) conn.commit() print(f" [OK] {olt_name}: Sincronizacion de Clientes Exitosa.") except Exception as e: conn.rollback() print(f" [ROLLBACK] {olt_name}: {e}") finally: cursor.close() conn.close() except Exception as e: print(f" [ERROR GENERAL] {olt_name}: {e}") if __name__ == "__main__": try: db = mysql.connector.connect(**DB_CONFIG) cur = db.cursor() if len(sys.argv) > 1: target_olt = int(sys.argv[1]) cur.execute("SELECT id, nombre, ip_address, snmp_port, snmp_community FROM olts WHERE id = %s", (target_olt,)) else: cur.execute("SELECT id, nombre, ip_address, snmp_port, snmp_community FROM olts") olts = cur.fetchall() db.close() if not olts: print("No se encontr la OLT especificada.") sys.exit(0) with concurrent.futures.ThreadPoolExecutor(max_workers=5) as ex: ex.map(procesar_olt, olts) except Exception as e: print(f"Error critico de inicio: {e}")