import os import requests import time import json import threading import logging from datetime import datetime from collections import defaultdict # ============================================================ # CONFIGURACIÓN ← edita solo esto # ============================================================ def _cargar_dotenv(): ruta = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env") try: with open(ruta) as f: for linea in f: linea = linea.strip() if not linea or linea.startswith("#") or "=" not in linea: continue k, v = linea.split("=", 1) os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'")) except IOError: pass _cargar_dotenv() SMARTOLT_BASE = os.environ["SMARTOLT_BASE"] API_KEY = os.environ["SMARTOLT_API_KEY"] BOT_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"] # int() importa: Telegram manda los chat_id como numero, y autorizado() # los compara contra ALLOWED_CHATS. Si quedaran como texto, no casarian nunca. CHAT_ID = int(os.environ["TELEGRAM_CHAT_ID"]) GROUP_ID = int(os.environ["TELEGRAM_GROUP_ID"]) # ID del grupo (empieza con -100) ALERT_TARGETS = [CHAT_ID, GROUP_ID] # destinatarios de alertas automáticas STATE_FILE = "onu_state.json" ONU_STATE_FILE = "onu_individual_state.json" # estado por ONU (alta/LOS/recuperada) # Umbral: si el % de ONUs en LOS sobre las ONLINE del PON supera esto → alerta LOS_PCT_UMBRAL = 20 # 20% sobre las que estaban online # Si una caja ya está en alerta (cascada), no repetir aviso individual por cada # ONT en LOS de esa misma caja (evita spam de decenas de mensajes en una caída masiva). SUPRIMIR_LOS_INDIVIDUAL_SI_CAJA = True # ONTs que llevan en LOS ininterrumpido este número de días o más se consideran # crónicas (equipo muerto/cliente de baja) y dejan de contar para alertas y # para el % de caída de caja — aunque sigan en LOS no se vuelve a avisar de ellas. UMBRAL_LOS_ANTIGUA_DIAS = 14 MONITOR_INTERVAL = 4 * 60 # 240s = 15 llamadas/hora, respeta límite SmartOLT # ============================================================ # LOGGING # ============================================================ logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[logging.FileHandler("bot.log"), logging.StreamHandler()] ) log = logging.getLogger("smartolt") TELEGRAM = "https://api.telegram.org/bot{}".format(BOT_TOKEN) # ============================================================ # TELEGRAM # ============================================================ def tg(method, data): try: return requests.post("{}/{}".format(TELEGRAM, method), json=data, timeout=10).json() except Exception as e: log.error("telegram %s: %s", method, e) return {} def send(chat, text, reply_markup=None): p = {"chat_id": chat, "text": text[:4096], "parse_mode": "Markdown", "protect_content": True} if reply_markup: p["reply_markup"] = reply_markup tg("sendMessage", p) def edit(chat, msg_id, text, reply_markup=None): p = {"chat_id": chat, "message_id": msg_id, "text": text[:4096], "parse_mode": "Markdown"} if reply_markup: p["reply_markup"] = reply_markup tg("editMessageText", p) def answer(cb_id): tg("answerCallbackQuery", {"callback_query_id": cb_id}) ALLOWED_CHATS = {CHAT_ID, GROUP_ID} def autorizado(chat): """Solo el chat privado del admin y el grupo NOC pueden usar el bot.""" if chat in ALLOWED_CHATS: return True log.warning("Acceso denegado desde chat %s", chat) return False # ============================================================ # API SMARTOLT — tracker de límites # ============================================================ _api_calls = defaultdict(list) _api_lock = threading.Lock() API_LIMITS = {"onu/get_all_onus_details": (15, 3600)} def _puede_llamar(endpoint): if endpoint not in API_LIMITS: return True max_calls, ventana = API_LIMITS[endpoint] ahora = time.time() with _api_lock: _api_calls[endpoint] = [t for t in _api_calls[endpoint] if ahora - t < ventana] if len(_api_calls[endpoint]) >= max_calls: log.warning("⛔ Límite API %s (%d/%d)", endpoint, len(_api_calls[endpoint]), max_calls) return False return True def _registrar(endpoint): with _api_lock: _api_calls[endpoint].append(time.time()) def api(endpoint, params=None): if not _puede_llamar(endpoint): return None try: r = requests.get( "{}/{}".format(SMARTOLT_BASE, endpoint), headers={"X-Token": API_KEY}, params=params, timeout=20 ) _registrar(endpoint) return r.json() except Exception as e: log.error("api %s: %s", endpoint, e) return None def calls_usadas(): ahora = time.time() with _api_lock: return len([t for t in _api_calls["onu/get_all_onus_details"] if ahora - t < 3600]) # ============================================================ # CACHÉ ONUs # ============================================================ _cache_onus = [] _cache_time = 0 _potencia_debug_logged = False def get_onus_cached(): return _cache_onus def get_onus_fresh(): global _cache_onus, _cache_time, _potencia_debug_logged data = api("onu/get_all_onus_details") if data is not None and "onus" not in data: log.warning("API respondio sin 'onus' (posible limite remoto): %s", str(data)[:200]) if data and "onus" in data: _cache_onus = data["onus"] _cache_time = time.time() log.info("Caché actualizado: %d ONUs", len(_cache_onus)) if not _potencia_debug_logged and _cache_onus: log.info("DEBUG campos disponibles por ONU: %s", sorted(_cache_onus[0].keys())) _potencia_debug_logged = True return _cache_onus def edad_cache(): return int(time.time() - _cache_time) if _cache_time else 9999 # ============================================================ # POTENCIA — el nombre exacto del campo varía según versión de API, # se prueban los alias más habituales. Si sale N/A siempre, revisa # en bot.log la línea "DEBUG campos disponibles por ONU" y añade # aquí el nombre real del campo de potencia (rx power). # ============================================================ CAMPOS_POTENCIA = ("signal_1310", "signal_1490", "signal") def get_potencia(o): for campo in CAMPOS_POTENCIA: v = o.get(campo) if v not in (None, "", "N/A", "-"): return v return "N/A" def potencia_txt(v): """Potencia lista para mostrar, con la unidad puesta una sola vez. La API ya devuelve el valor con unidad dentro ("-20.05 dBm"), asi que se quita y se vuelve a poner aqui; antes salia "-20.05 dBm dBm". Ojo con el campo 'signal', que es cualitativo ("Very good") y no una medida: en ese caso se muestra tal cual, sin dBm.""" txt = str(v).strip() if v not in (None, "") else "N/A" if txt.lower().endswith("dbm"): txt = txt[:-3].strip() if txt in ("", "-", "N/A"): return "N/A" try: float(txt) except ValueError: return txt # texto cualitativo, sin unidad return txt + " dBm" def get_potencia_mostrar(o, estado_onu): """Potencia a mostrar: en vivo si está online; última registrada (guardada la última vez que estuvo online) si está en LOS/offline, ya que en LOS la lectura en vivo no es fiable.""" status = (o.get("status") or "").strip() if status.lower() == "online": return get_potencia(o) prev = _normalizar_prev(estado_onu.get(clave_onu(o))) return prev.get("ultima_potencia") or "N/A" # ============================================================ # ESTADO PERSISTENTE — por caja (PON) # ============================================================ def cargar_estado(): try: with open(STATE_FILE) as f: return json.load(f) except Exception: return {} def guardar_estado(e): try: with open(STATE_FILE, "w") as f: json.dump(e, f) except Exception as ex: log.error("guardar_estado: %s", ex) # ============================================================ # ESTADO PERSISTENTE — por ONU individual (alta / LOS / recuperada) # ============================================================ def cargar_estado_onus(): try: with open(ONU_STATE_FILE) as f: return json.load(f) except Exception: return {} def guardar_estado_onus(e): try: with open(ONU_STATE_FILE, "w") as f: json.dump(e, f) except Exception as ex: log.error("guardar_estado_onus: %s", ex) def clave_onu(o): """Identificador estable de la ONU: SN si existe, si no id, si no nombre.""" return o.get("sn") or o.get("id") or o.get("name") def _normalizar_prev(prev): """Compatibilidad con el formato antiguo del state file (string plano).""" if isinstance(prev, str): return {"status": prev, "los_desde": None} return prev or {} def actualizar_estado_los(onus, estado_anterior): """ Recorre todas las ONUs y devuelve: estado_actual: dict key -> {status, los_desde, ultima_potencia} cronicos: set de keys en LOS ininterrumpido >= UMBRAL_LOS_ANTIGUA_DIAS días """ estado_actual = {} cronicos = set() ahora = datetime.now() for o in onus: key = clave_onu(o) if not key: continue status = (o.get("status") or "").strip() if status.upper() == "LOS": status_norm = "los" elif status.lower() == "online": status_norm = "online" else: status_norm = "offline" prev = _normalizar_prev(estado_anterior.get(key)) if status_norm == "los": if prev.get("status") == "los" and prev.get("los_desde"): los_desde = prev["los_desde"] else: los_desde = ahora.strftime('%Y-%m-%d %H:%M:%S') else: los_desde = None # potencia: si está online se refresca; si no, se conserva la última conocida if status_norm == "online": ultima_potencia = get_potencia(o) else: ultima_potencia = prev.get("ultima_potencia") estado_actual[key] = { "status": status_norm, "los_desde": los_desde, "ultima_potencia": ultima_potencia, "recuperacion_pendiente": prev.get("recuperacion_pendiente", False), } if status_norm == "los" and los_desde: dias = (ahora - datetime.strptime(los_desde, '%Y-%m-%d %H:%M:%S')).days if dias >= UMBRAL_LOS_ANTIGUA_DIAS: cronicos.add(key) return estado_actual, cronicos # ============================================================ # ANALIZAR PONs — agrupa por OLT+board+port # ============================================================ def analizar_pons(onus): pons = defaultdict(lambda: { "olt": "", "board": "", "port": "", "zona": "", "total": 0, "online": 0, "los": 0, "offline": 0, "onus_los": [], "onus_offline": [], "onus_online": [] }) for o in onus: olt = o.get("olt_name", "") board = str(o.get("board", "")) port = str(o.get("port", "")) if not olt or board == "" or port == "": continue status = o.get("status", "").strip() key = "{}/{}/{}".format(olt, board, port) pons[key]["olt"] = olt pons[key]["board"] = board pons[key]["port"] = port pons[key]["zona"] = o.get("zone_name", "") pons[key]["total"] += 1 if status.upper() == "LOS": pons[key]["los"] += 1 pons[key]["onus_los"].append(o) elif status.lower() == "online": pons[key]["online"] += 1 pons[key]["onus_online"].append(o) else: pons[key]["offline"] += 1 pons[key]["onus_offline"].append(o) return pons # ============================================================ # CALCULAR % LOS — base = online + los (los que DEBERÍAN estar online) # ============================================================ def calcular_pct_los(stats): """% de LOS sobre (online + LOS), es decir, sobre las que estaban activas.""" base = stats["online"] + stats["los"] return round(stats["los"] * 100 / base) if base else 0 # ============================================================ # FORMATO MENSAJES — caja (PON) # ============================================================ def formato_alerta(stats, tipo="CAÍDA DE CAJA DETECTADA"): olt = stats["olt"] board = stats["board"] port = stats["port"] zona = stats["zona"] total = stats["total"] n_los = stats["los"] n_off = stats["offline"] n_on = stats["online"] pct_los = calcular_pct_los(stats) msg = "🚨 {}\n\n".format(tipo) msg += "📡 OLT: `{}`\n".format(olt) msg += "📦 Board `{}` / PON `{}`\n".format(board, port) msg += "📍 Zona: {}\n\n".format(zona) msg += "🟢 Online: {}\n".format(n_on) msg += "⚠️ LOS: {} ({}% de las activas)\n".format(n_los, pct_los) msg += "🔴 Offline: {}\n".format(n_off) msg += "Total PON: {}\n\n".format(total) estado_onu = cargar_estado_onus() if stats["onus_los"]: msg += "⚠️ ONUs en LOS:\n" for o in stats["onus_los"]: nombre = o.get("name", "Sin nombre") addr = o.get("address", "") potencia = get_potencia_mostrar(o, estado_onu) msg += "⚠️ {}".format(nombre) if addr: msg += " — {}".format(addr) msg += " — 🔋 `{}`\n".format(potencia_txt(potencia)) if stats["onus_offline"]: msg += "\n🔴 ONUs Offline:\n" for o in stats["onus_offline"]: nombre = o.get("name", "Sin nombre") addr = o.get("address", "") msg += "🔴 {}".format(nombre) if addr: msg += " — {}".format(addr) msg += "\n" if stats["onus_online"]: msg += "\n🟢 ONUs Online:\n" for o in stats["onus_online"]: nombre = o.get("name", "Sin nombre") potencia = get_potencia_mostrar(o, estado_onu) msg += "🟢 {} — 🔋 `{}`\n".format(nombre, potencia_txt(potencia)) msg += "\n⏱ `{}`".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S')) return msg def formato_recuperacion(stats): msg = "✅ CAJA RECUPERADA\n\n" msg += "📡 OLT: `{}`\n".format(stats['olt']) msg += "📦 Board `{}` / PON `{}`\n".format(stats['board'], stats['port']) msg += "📍 Zona: {}\n\n".format(stats['zona']) msg += "🟢 Online: {}/{}\n".format(stats['online'], stats['total']) msg += "⚠️ LOS: {}\n".format(stats['los']) msg += "🔴 Offline: {}\n\n".format(stats['offline']) msg += "⏱ `{}`".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S')) return msg # ============================================================ # FORMATO MENSAJES — ONU individual (alta / LOS / recuperada) # ============================================================ def formato_alta_onu(o): nombre = o.get("name", "Sin nombre") olt = o.get("olt_name", "") board = o.get("board", "") port = o.get("port", "") addr = o.get("address", "") potencia = get_potencia(o) msg = "🆕 ALTA REALIZADA\n\n" msg += "👤 Cliente: {}\n".format(nombre) msg += "📡 OLT: `{}` — Board `{}` / PON `{}`\n".format(olt, board, port) if addr: msg += "📍 {}\n".format(addr) msg += "🔋 Potencia: `{}`\n".format(potencia_txt(potencia)) msg += "⏱ `{}`".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S')) return msg def formato_los_onu(o, ultima_potencia=None): nombre = o.get("name", "Sin nombre") olt = o.get("olt_name", "") board = o.get("board", "") port = o.get("port", "") addr = o.get("address", "") msg = "⚠️ ONT EN LOS\n\n" msg += "👤 Cliente: {}\n".format(nombre) msg += "📡 OLT: `{}` — Board `{}` / PON `{}`\n".format(olt, board, port) if addr: msg += "📍 {}\n".format(addr) msg += "🔋 Ultima potencia registrada: `{}`\n".format(potencia_txt(ultima_potencia)) msg += "⏱ `{}`".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S')) return msg def formato_recuperada_onu(o): nombre = o.get("name", "Sin nombre") olt = o.get("olt_name", "") board = o.get("board", "") port = o.get("port", "") potencia = get_potencia(o) msg = "✅ RECUPERADA\n\n" msg += "👤 Cliente: {}\n".format(nombre) msg += "📡 OLT: `{}` — Board `{}` / PON `{}`\n".format(olt, board, port) msg += "🔋 Potencia de recuperación: `{}`\n".format(potencia_txt(potencia)) msg += "⏱ Hora de recuperación: `{}`".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S')) return msg # ============================================================ # MONITOREO AUTOMÁTICO — por ONU individual # ============================================================ def check_onus_individual(onus, pons_en_caida, estado_anterior, estado_actual): """ Compara el estado de cada ONU frente al chequeo anterior: - Aparece por primera vez y está online → ALTA REALIZADA (tras el arranque inicial) - Pasa de online/offline a LOS → ONT EN LOS - Pasa de LOS a online → NO avisa aún, marca recuperacion_pendiente (la potencia recién recuperada no es fiable hasta el siguiente chequeo de la API) - Sigue online con recuperacion_pendiente → RECUPERADA (potencia ya fresca, confirma) `onus` ya viene filtrada sin las ONTs crónicas en LOS (ver UMBRAL_LOS_ANTIGUA_DIAS). `pons_en_caida` es el set de claves olt/board/port en alerta de caja, para no duplicar aviso individual por cada ONT de una caída masiva. """ bootstrap = len(estado_anterior) == 0 # primera ejecución: solo crea la foto base for o in onus: key = clave_onu(o) if not key: continue prev = _normalizar_prev(estado_anterior.get(key)) prev_status = prev.get("status") prev_pendiente = prev.get("recuperacion_pendiente", False) status_norm = estado_actual[key]["status"] nombre = o.get("name", "Sin nombre") pon_key = "{}/{}/{}".format(o.get('olt_name',''), o.get('board',''), o.get('port','')) if key not in estado_anterior: # ONU vista por primera vez desde que existe el fichero de estado if not bootstrap and status_norm == "online": msg = formato_alta_onu(o) for target in ALERT_TARGETS: send(target, msg) log.info("🆕 Alta: %s (%s)", nombre, key) continue if status_norm == "los" and prev_status != "los": # nueva caída (o recaída durante confirmación de recuperación) estado_actual[key]["recuperacion_pendiente"] = False if pon_key in pons_en_caida and SUPRIMIR_LOS_INDIVIDUAL_SI_CAJA: log.info("⚠️ LOS individual omitido (caja %s ya en alerta): %s", pon_key, key) else: msg = formato_los_onu(o, estado_actual[key].get("ultima_potencia")) for target in ALERT_TARGETS: send(target, msg) log.info("⚠️ LOS individual: %s (%s)", nombre, key) elif status_norm == "online" and prev_pendiente: # segundo ciclo online tras LOS: potencia ya fiable, confirma recuperación estado_actual[key]["recuperacion_pendiente"] = False msg = formato_recuperada_onu(o) for target in ALERT_TARGETS: send(target, msg) log.info("✅ Recuperada (confirmada): %s (%s) potencia=%s", nombre, key, get_potencia(o)) elif status_norm == "online" and prev_status == "los": # primer ciclo online tras LOS: NO avisar todavía, potencia recién leída no es fiable estado_actual[key]["recuperacion_pendiente"] = True log.info("⏳ Recuperación pendiente de confirmar (esperando potencia fresca): %s (%s)", nombre, key) else: # online normal / offline sin cambios relevantes estado_actual[key]["recuperacion_pendiente"] = False # ============================================================ # MONITOREO AUTOMÁTICO — por caja (PON) # ============================================================ def check_onus(): log.info("🔍 Chequeo (llamadas usadas: %d/15)", calls_usadas()) onus = get_onus_fresh() if not onus: return # Estado por ONU (alta/LOS/recuperada) + detección de crónicas en LOS estado_onu_anterior = cargar_estado_onus() estado_onu_actual, cronicos = actualizar_estado_los(onus, estado_onu_anterior) if cronicos: log.info("💀 ONTs crónicas en LOS (≥%d días, sin avisar): %d", UMBRAL_LOS_ANTIGUA_DIAS, len(cronicos)) # Las crónicas no cuentan para cascada de caja ni para avisos individuales onus_para_alertar = [o for o in onus if clave_onu(o) not in cronicos] estado_anterior = cargar_estado() estado_actual = {} pons = analizar_pons(onus_para_alertar) pons_en_caida = set() for key, stats in pons.items(): pct_los = calcular_pct_los(stats) # Condición: LOS supera el umbral sobre las activas (online + LOS) es_caida = stats["los"] > 0 and pct_los >= LOS_PCT_UMBRAL ya_alertado = estado_anterior.get(key, False) estado_actual[key] = es_caida if es_caida: pons_en_caida.add(key) if es_caida and not ya_alertado: msg = formato_alerta(stats) for target in ALERT_TARGETS: # ← envía a todos los destinos send(target, msg) log.info("🚨 Alerta %s B%s/P%s — %d LOS (%d%% de activas)", stats["olt"], stats["board"], stats["port"], stats["los"], pct_los) elif not es_caida and ya_alertado: msg = formato_recuperacion(stats) for target in ALERT_TARGETS: # ← envía a todos los destinos send(target, msg) log.info("✅ Recuperada %s B%s/P%s", stats["olt"], stats["board"], stats["port"]) guardar_estado(estado_actual) # Chequeo por ONU individual (alta / LOS / recuperada con potencia) check_onus_individual(onus_para_alertar, pons_en_caida, estado_onu_anterior, estado_onu_actual) guardar_estado_onus(estado_onu_actual) log.info("✅ Chequeo OK. PONs analizados: %d", len(pons)) def loop_monitor(): time.sleep(5) while True: try: check_onus() except Exception as e: log.error("loop_monitor: %s", e) time.sleep(MONITOR_INTERVAL) # ============================================================ # COMANDOS TELEGRAM # ============================================================ def cmd_start(chat): send(chat, "🤖 SmartOLT NOC Bot\n\n" "Alerta cuando ≥{}% de las activas de un PON está en LOS.\n" "También avisa por ONT individual: alta realizada (con potencia), " "LOS y recuperada (con potencia y hora, confirmada un ciclo después).\n" "ONTs con ≥{} días seguidos en LOS se consideran " "crónicas y dejan de generar avisos.\n\n" "Comandos:\n" "• /onts — Ver ONUs por OLT y puerto\n" "• /cajas — Ver PONs con LOS ahora\n" "• /los — Ver todas las ONTs en LOS (sin crónicas)\n" "• /status — Resumen global\n" "• /api — Uso de la API\n\n" "_Caché: hace {} min_".format(LOS_PCT_UMBRAL, UMBRAL_LOS_ANTIGUA_DIAS, edad_cache()//60) ) def cmd_status(chat): onus = get_onus_cached() if not onus: send(chat, "⏳ Caché no disponible aún.") return total = len(onus) online = sum(1 for o in onus if o.get("status","").lower() == "online") los = sum(1 for o in onus if o.get("status","").upper() == "LOS") offline = total - online - los send(chat, "📊 ESTADO GLOBAL\n\n" "Total ONUs: `{}`\n" "🟢 Online: `{}`\n" "⚠️ LOS: `{}`\n" "🔴 Offline: `{}`\n\n" "_Datos de hace {} min_".format(total, online, los, offline, edad_cache()//60) ) def cmd_cajas(chat): onus = get_onus_cached() if not onus: send(chat, "⏳ Caché no disponible aún.") return pons = analizar_pons(onus) cajas = [] for stats in pons.values(): pct_los = calcular_pct_los(stats) if stats["los"] > 0 and pct_los >= LOS_PCT_UMBRAL: cajas.append((pct_los, stats)) cajas.sort(key=lambda x: x[0], reverse=True) if not cajas: send(chat, "✅ Sin caídas de caja ahora mismo\n\n_Datos de hace {} min_".format(edad_cache()//60)) return msg = "🚨 PONs con caída de caja ({})\n\n".format(len(cajas)) for pct, s in cajas: msg += ("• `{}` B`{}`P`{}` " "— ⚠️ {} LOS ({}% activas) / 🔴 {} Offline / 🟢 {} Online" " — {}\n".format(s['olt'], s['board'], s['port'], s['los'], pct, s['offline'], s['online'], s['zona'])) msg += "\n_Datos de hace {} min_".format(edad_cache()//60) send(chat, msg) def cmd_los(chat): onus = get_onus_cached() if not onus: send(chat, "⏳ Caché no disponible aún.") return estado_onu = cargar_estado_onus() ahora = datetime.now() en_los = [] for o in onus: if (o.get("status") or "").strip().upper() != "LOS": continue key = clave_onu(o) prev = _normalizar_prev(estado_onu.get(key)) los_desde = prev.get("los_desde") ultima_potencia = prev.get("ultima_potencia") or "N/A" dias = None if los_desde: dias = (ahora - datetime.strptime(los_desde, '%Y-%m-%d %H:%M:%S')).days # excluye crónicas, mismo umbral que las alertas if dias is not None and dias >= UMBRAL_LOS_ANTIGUA_DIAS: continue en_los.append({ "o": o, "los_desde": los_desde or "desconocida", "potencia": ultima_potencia, "dias": dias, }) if not en_los: send(chat, "✅ Sin ONTs en LOS ahora mismo\n\n_Datos de hace {} min_\n_(excluidas crónicas ≥{}d)_".format(edad_cache()//60, UMBRAL_LOS_ANTIGUA_DIAS)) return en_los.sort(key=lambda x: (x["o"].get("olt_name",""), str(x["o"].get("board","")), str(x["o"].get("port","")))) msg = "⚠️ ONTs en LOS — {} total\n".format(len(en_los)) olt_actual = None for item in en_los: o = item["o"] olt = o.get("olt_name", "") board = o.get("board", "") port = o.get("port", "") nombre = o.get("name", "Sin nombre") addr = o.get("address", "") if olt != olt_actual: msg += "\n━━━━━━━━━━━━━━━━━━━━\n" msg += "📡 {}\n".format(olt) msg += "━━━━━━━━━━━━━━━━━━━━\n" olt_actual = olt msg += "\n👤 {}\n".format(nombre) msg += "📦 Board `{}` / PON `{}`\n".format(board, port) if addr: msg += "📍 {}\n".format(addr) msg += "⏱ Cayó: `{}`".format(item['los_desde']) if item["dias"] is not None: msg += " _({}d)_".format(item['dias']) msg += "\n" msg += "🔋 Potencia previa: `{}`\n".format(potencia_txt(item['potencia'])) if len(msg) > 4000: msg = msg[:3950] + "\n\n⚠️ _Lista truncada_" msg += "\n_Datos de hace {} min_ — _excluidas crónicas ≥{}d_".format(edad_cache()//60, UMBRAL_LOS_ANTIGUA_DIAS) send(chat, msg) def cmd_onts(chat): onus = get_onus_cached() if not onus: send(chat, "⏳ Caché no disponible aún.") return olts = sorted(set(o.get("olt_name","") for o in onus if o.get("olt_name"))) kb = [[{"text": "📡 {}".format(olt), "callback_data": "OLT:{}".format(olt)}] for olt in olts] send(chat, "Selecciona OLT:\n_Datos de hace {} min_".format(edad_cache()//60), {"inline_keyboard": kb}) def cmd_api(chat): usadas = calls_usadas() send(chat, "📈 Uso API\n\n" "get_all_onus_details: `{}/15` esta hora\n" "Intervalo monitoreo: `{} min`\n" "Caché actualizado: `hace {} min`\n\n" "_Umbral alerta: ≥{}% LOS sobre activas por PON_".format(usadas, MONITOR_INTERVAL//60, edad_cache()//60, LOS_PCT_UMBRAL) ) # ============================================================ # CALLBACKS — vista por puertos # ============================================================ def handle_callback(cq): answer(cq["id"]) chat = cq["message"]["chat"]["id"] if not autorizado(chat): return msg_id = cq["message"]["message_id"] parts = cq["data"].split(":", 3) onus = get_onus_cached() # Paso 1: eligió OLT → listar puertos PON if parts[0] == "OLT" and len(parts) == 2: olt = parts[1] pons = analizar_pons([o for o in onus if o.get("olt_name") == olt]) if not pons: edit(chat, msg_id, "Sin ONUs en `{}`.".format(olt)) return ordenados = sorted(pons.values(), key=lambda x: (x["board"], int(x["port"]) if x["port"].isdigit() else 0)) kb = [] for s in ordenados: pct_los = calcular_pct_los(s) icono = "🚨" if pct_los >= LOS_PCT_UMBRAL else ("⚠️" if s["los"] > 0 else "🟢") label = "{} B{}/PON{} — {}✅ {}⚠️ {}🔴 ({} total)".format(icono, s['board'], s['port'], s['online'], s['los'], s['offline'], s['total']) kb.append([{"text": label, "callback_data": "PON:{}:{}:{}".format(olt, s['board'], s['port'])}]) edit(chat, msg_id, "📡 {}\nSelecciona un puerto PON:\n_Datos de hace {} min_".format(olt, edad_cache()//60), reply_markup={"inline_keyboard": kb}) # Paso 2: eligió PON → detalle completo de ese puerto elif parts[0] == "PON" and len(parts) == 4: _, olt, board, port = parts onus_pon = [ o for o in onus if o.get("olt_name") == olt and str(o.get("board","")) == board and str(o.get("port","")) == port ] if not onus_pon: edit(chat, msg_id, "Sin ONUs en este puerto.") return stats = { "olt": olt, "board": board, "port": port, "zona": onus_pon[0].get("zone_name",""), "total": len(onus_pon), "online": 0, "los": 0, "offline": 0, "onus_los": [], "onus_offline": [], "onus_online": [] } for o in onus_pon: s = o.get("status","").strip() if s.upper() == "LOS": stats["los"] += 1 stats["onus_los"].append(o) elif s.lower() == "online": stats["online"] += 1 stats["onus_online"].append(o) else: stats["offline"] += 1 stats["onus_offline"].append(o) pct_los = calcular_pct_los(stats) tipo = "🚨 CAÍDA DE CAJA DETECTADA" if pct_los >= LOS_PCT_UMBRAL else "📊 PON {} — detalle".format(port) msg = "{}\n\n".format(tipo) msg += "📡 OLT: `{}`\n".format(olt) msg += "📦 Board `{}` / PON `{}`\n".format(board, port) msg += "📍 Zona: {}\n\n".format(stats['zona']) msg += "🟢 Online: {}\n".format(stats['online']) msg += "⚠️ LOS: {} ({}% de las activas)\n".format(stats['los'], pct_los) msg += "🔴 Offline: {}\n".format(stats['offline']) msg += "Total PON: {}\n".format(stats['total']) estado_onu = cargar_estado_onus() if stats["onus_los"]: msg += "\n⚠️ ONUs en LOS:\n" for o in stats["onus_los"]: nombre = o.get("name","Sin nombre") addr = o.get("address","") potencia = get_potencia_mostrar(o, estado_onu) msg += "⚠️ {}".format(nombre) if addr: msg += " — {}".format(addr) msg += " — 🔋 `{}`\n".format(potencia_txt(potencia)) if stats["onus_offline"]: msg += "\n🔴 ONUs Offline:\n" for o in stats["onus_offline"]: nombre = o.get("name","Sin nombre") addr = o.get("address","") msg += "🔴 {}".format(nombre) if addr: msg += " — {}".format(addr) msg += "\n" if stats["onus_online"]: msg += "\n🟢 ONUs Online:\n" for o in stats["onus_online"]: nombre = o.get("name","Sin nombre") potencia = get_potencia_mostrar(o, estado_onu) msg += "🟢 {} — 🔋 `{}`\n".format(nombre, potencia_txt(potencia)) if len(msg) > 4000: msg = msg[:3950] + "\n\n⚠️ _Lista truncada_" msg += "\n⏱ `{}`".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S')) edit(chat, msg_id, msg) # ============================================================ # POLLING # ============================================================ def poll(): tg("deleteWebhook", {}) log.info("🤖 Polling iniciado") offset = 0 while True: try: r = requests.get( "{}/getUpdates".format(TELEGRAM), params={"offset": offset, "timeout": 30}, timeout=35 ) for u in r.json().get("result", []): offset = u["update_id"] + 1 if "message" in u: chat = u["message"]["chat"]["id"] text = u["message"].get("text","").split("@")[0].strip() log.info("Cmd: %s %s", chat, text) if not autorizado(chat): continue if text == "/start": cmd_start(chat) elif text == "/onts": cmd_onts(chat) elif text == "/status": cmd_status(chat) elif text == "/cajas": cmd_cajas(chat) elif text == "/los": cmd_los(chat) elif text == "/api": cmd_api(chat) elif "callback_query" in u: handle_callback(u["callback_query"]) except Exception as e: log.error("poll: %s", e) time.sleep(5) # ============================================================ # MAIN # ============================================================ if __name__ == "__main__": log.info("🚀 Bot iniciado — umbral LOS %d%% sobre activas — intervalo %ds", LOS_PCT_UMBRAL, MONITOR_INTERVAL) get_onus_fresh() threading.Thread(target=loop_monitor, daemon=True).start() poll()