#!/usr/bin/env python3 # -*- coding: utf-8 -*- import re, time, json, asyncio import aiomysql import datetime import telnetlib3 from librouteros import connect import subprocess, platform from collections import defaultdict # ========== Device Ping Status Checker ========== def is_device_up(ip): try: if platform.system().lower() == "windows": command = ["ping", "-n", "1", "-w", "2000", ip] else: command = ["ping", "-c", "1", "-W", "2", ip] result = subprocess.run( command, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=5, shell=False, ) return result.returncode == 0 except Exception: return False # ========== DB Connection Pool ========== async def create_db_pool(): return await aiomysql.create_pool( host="db1.plexuscloud.com.bd", user="c2backend", password="XdQAo2Yvy@u9", db="c2erp_backend", charset="utf8mb4", minsize=5, maxsize=30, autocommit=False, cursorclass=aiomysql.DictCursor, ) # ========== MikroTik PPP Fetch ========== def get_mikrotik_active_ppp(host, user, password): print(f"\n--- Fetching PPP from Router: {host} ---") if not host or not user or not password: return {} try: api = connect(host=host, username=user, password=password, port=8728, timeout=10) active = api.path("/ppp/active").select("name", "caller-id") ppp_map = {} for u in active: raw_mac = u.get("caller-id") if raw_mac: mac = str(raw_mac).lower().replace(":", "").replace("-", "").replace(".", "") name = u.get("name") if mac and name: name_val = str(name).strip() ppp_map[mac] = name_val.zfill(11) if name_val.isdigit() else name_val api.close() print(f" => Fetched {len(ppp_map)} PPP users from {host}.") return ppp_map except Exception as e: print(f" => MikroTik API Error ({host}): {e}") return {} # ========== Telnet Command Helper ========== async def execute_command(writer, reader, command, timeout=60, silent=False, sleep_time=0): if sleep_time > 0: await asyncio.sleep(sleep_time) # Fast buffer flush for _ in range(15): try: await asyncio.wait_for(reader.read(4096), timeout=0.2) except: break writer.write(command + "\r\n") await writer.drain() output = "" start_time = time.time() last_data_time = time.time() read_timeout = 4.0 if timeout > 10 else timeout while True: try: chunk = await asyncio.wait_for(reader.read(4096), timeout=read_timeout) if not chunk: break output += chunk last_data_time = time.time() if not silent: print(".", end="", flush=True) if "More" in chunk: writer.write(" ") await writer.drain() await asyncio.sleep(0.1) if time.time() - start_time > timeout: if not silent: print(" [Global Timeout]", end="") break lines = output.strip().splitlines() if lines: last_line = lines[-1].strip() if (last_line.endswith("#") or last_line.endswith(">")) and " " not in last_line: if len(lines) > 1: break except asyncio.TimeoutError: if not silent: print("~", end="", flush=True) if time.time() - last_data_time > 25: if not silent: print(" [No Data Timeout]", end="") break if time.time() - start_time > timeout: break except Exception: break clean_text = "" for char in output: if char == "\x08": if clean_text: clean_text = clean_text[:-1] else: clean_text += char ansi_escape = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])") clean_text = ansi_escape.sub("", clean_text) clean_text = re.sub(r"--More--|---- More ----|More:", "", clean_text) if not silent: print("") return clean_text # ========================================================== # ========== BDCOM SECTION (Original Logic) ================ # ========================================================== async def bdcom_fetch_all_rx_powers(writer, reader, active_onus): """BDCom port-wise batch Rx power fetch.""" rx_power_map = {} unique_ports = sorted(set( re.search(r"(epon\d+/\d+|gpon\d+/\d+)", onu["interface"], re.IGNORECASE).group(1).lower() for onu in active_onus if re.search(r"(epon\d+/\d+|gpon\d+/\d+)", onu["interface"], re.IGNORECASE) )) print(f" => Fetching Rx Powers for {len(unique_ports)} port(s): {', '.join(unique_ports)}") for port in unique_ports: print(f" -> {port}...", end="", flush=True) command = f"show epon onu-ctc-optical-transceiver-diagnosis interface {port}" await asyncio.sleep(1) pwr_raw = await execute_command(writer, reader, command, timeout=30, sleep_time=2, silent=True) count = 0 for line in pwr_raw.splitlines(): intf_match = re.search(r"(epon\d+/\d+:\d+|gpon\d+/\d+:\d+)", line, re.IGNORECASE) if intf_match: parts = line.split() if len(parts) >= 2: rx_val = parts[-1] if rx_val != "--" and re.match(r"^-?\d+\.?\d*$", rx_val): rx_power_map[intf_match.group(1).lower()] = rx_val count += 1 if count == 0: print(" [Retrying...]", end="", flush=True) writer.write("\r\n") await writer.drain() await asyncio.sleep(2) pwr_raw = await execute_command(writer, reader, command, timeout=35, sleep_time=3, silent=True) for line in pwr_raw.splitlines(): intf_match = re.search(r"(epon\d+/\d+:\d+|gpon\d+/\d+:\d+)", line, re.IGNORECASE) if intf_match: parts = line.split() if len(parts) >= 2: rx_val = parts[-1] if rx_val != "--" and re.match(r"^-?\d+\.?\d*$", rx_val): rx_power_map[intf_match.group(1).lower()] = rx_val count += 1 print(f" {count} found.") # Re-fetch missed UP ONUs individually (max 20) missed_onus = [ o for o in active_onus if o["conn_state"] == "Connect" and rx_power_map.get(o["interface"].lower(), "0") == "0" ] if missed_onus: if len(missed_onus) <= 20: print(f" => Re-fetching {len(missed_onus)} missed ONU(s)...") for o in missed_onus: try: pwr_out = await execute_command( writer, reader, f"show epon onu-ctc-optical-transceiver-diagnosis interface {o['interface']}", timeout=8, sleep_time=0, silent=True, ) for line in pwr_out.splitlines(): intf_match = re.search(r"(epon\d+/\d+:\d+)", line, re.IGNORECASE) if intf_match and intf_match.group(1).lower() == o["interface"].lower(): parts = line.split() if len(parts) >= 2: rx_val = parts[-1] if rx_val != "--" and re.match(r"^-?\d+\.?\d*$", rx_val): rx_power_map[o["interface"].lower()] = rx_val break except Exception: pass else: print(f" => {len(missed_onus)} missed — relying on DB restore.") print(f" => Total Rx Powers mapped: {len(rx_power_map)}") return rx_power_map async def bdcom_scrape(writer, reader, olt_name): """BDCom: Fetch MAC table, distance map, ONU info, Rx powers.""" print(f" [{olt_name}] BDCOM mode. Fetching MAC Table...") # MAC Table mac_output = await execute_command(writer, reader, "show mac address-table", sleep_time=1) mac_map = {} mac_inserts = [] for line in mac_output.splitlines(): m = re.search(r"(\d+)\s+([0-9a-fA-F\.]+)\s+(\w+)\s+(epon\d+/\d+:\d+)", line.lower()) if m: vlan, mac, _, iface = m.groups() clean_mac = mac.replace(".", "") mac_map.setdefault(iface, []).append(clean_mac) mac_inserts.append((vlan, mac, iface)) # Buffer flush for _ in range(8): try: await asyncio.wait_for(reader.read(4096), timeout=0.4) except: break await asyncio.sleep(1) # Distance Map dist_out = await execute_command(writer, reader, "show epon active-onu", timeout=60, sleep_time=1) distance_map = { m.group(1).lower(): m.group(2) for m in re.finditer( r"(epon\d+/\d+:\d+)\s+[0-9a-f\.]+\s+[\w-]+\s+[\w-]+\s+(\d+)", dist_out.lower(), ) } # Buffer flush for _ in range(5): try: await asyncio.wait_for(reader.read(4096), timeout=0.3) except: break await asyncio.sleep(0.5) # ONU Information onu_output = await execute_command(writer, reader, "show epon onu-information", timeout=120, sleep_time=1) # Line Stitcher full_lines = [] current_line = "" for line in onu_output.splitlines(): clean_line = re.sub(r"--More--|\x08+|\x1b\[\d+[A-Za-z]", "", line).strip() if not clean_line: continue if clean_line.endswith("#") or clean_line.endswith(">"): continue if re.match(r"^epon\d+/\d+:\d+", clean_line, re.IGNORECASE): if current_line: full_lines.append(current_line) current_line = clean_line elif current_line: if ( clean_line.lower().startswith("interface") or clean_line.startswith("IntfName") or clean_line.startswith("---") or "registered" in clean_line.lower() or "ONUs:" in clean_line ): full_lines.append(current_line) current_line = "" else: current_line += " " + clean_line if current_line: full_lines.append(current_line) # Parse ONUs active_onus = [] for line in full_lines: mac_search = re.search(r"([0-9a-f]{4}\.[0-9a-f]{4}\.[0-9a-f]{4})", line, re.IGNORECASE) if not mac_search: continue mac = mac_search.group(1) mac_start, mac_end = mac_search.start(), mac_search.end() left_part = line[:mac_start].strip().split() intf = left_part[0].lower() vendor = left_part[1] if len(left_part) > 1 and left_part[1] != "----" else "" model = ( left_part[2] if len(left_part) > 2 and left_part[2] not in ["----", "0x00000000"] else "" ) right_part = line[mac_end:].strip().split() if len(right_part) >= 3: dereg = right_part[-1] status = right_part[-2].lower().replace("_", "-") bind = right_part[-3] desc = " ".join(right_part[:-3]) else: status, dereg, bind, desc = "down", "N/A", "static", "" conn_state = ( "Connect" if any(x in status for x in ["auto", "up", "working", "oper"]) else "Down" ) active_onus.append({ "interface": intf, "vendor": vendor, "model": model, "mac": mac, "description": "" if desc.upper() == "N/A" else desc, "bind_type": bind, "status": status, "conn_state": conn_state, "dereg": "" if dereg.upper() == "N/A" else dereg, "distance": distance_map.get(intf, "0"), }) # Rx Power Batch Fetch rx_power_map = await bdcom_fetch_all_rx_powers(writer, reader, active_onus) return active_onus, mac_map, mac_inserts, rx_power_map # ========================================================== # ========== C-DATA SECTION (New Logic) ==================== # ========================================================== def _cdata_normalize_mac(mac_raw): if not mac_raw: return "" clean = re.sub(r"[^0-9a-fA-F]", "", str(mac_raw)).lower() if len(clean) != 12: return "" return f"{clean[0:4]}.{clean[4:8]}.{clean[8:12]}" def _cdata_normalize_interface(intf_raw): """ Normalize C-Data interface to BDCom-style. C-Data formats vary: - 'EPON 0/1:5' -> 'epon0/1:5' - 'EPON0/1:5' -> 'epon0/1:5' - 'GPON 0/1 5' -> 'gpon0/1:5' - '0/1/5' -> cannot map (return as-is lowered) """ if not intf_raw: return "" s = str(intf_raw).strip().lower() # Combine slot/port (e.g. "epon 0/1 :5" -> "epon0/1:5") s = re.sub(r"\s+", "", s) # If form is "epon0/1/5" or "gpon0/1/5", convert last slash to colon m = re.match(r"^(epon|gpon)(\d+)/(\d+)/(\d+)$", s) if m: return f"{m.group(1)}{m.group(2)}/{m.group(3)}:{m.group(4)}" return s async def cdata_scrape(writer, reader, olt_name): active_onus = [] mac_map = {} mac_inserts = [] rx_power_map = {} # empty — not available # 1. MAC Table mac_out = await execute_command(writer, reader, "show mac-address dynamic", sleep_time=1, timeout=60) for line in mac_out.splitlines(): m = re.search( r"([0-9a-fA-F:]{17})\s+(\d+)\s+(epon\d+/\d+/\d+)\s+dynamic", line, re.IGNORECASE ) if m: # raw_mac, vlan, iface = m.group(1), m.group(2), m.group(3).lower() raw_mac, vlan, raw_iface = m.group(1), m.group(2), m.group(3).lower() iface = _cdata_normalize_interface(raw_iface) dotted_mac = _cdata_normalize_mac(raw_mac) # dotted_mac = _cdata_normalize_mac(raw_mac) clean_mac = dotted_mac.replace(".", "") mac_map.setdefault(iface, []).append(clean_mac) mac_inserts.append((vlan, dotted_mac, iface)) # 2. ONU Info onu_out = await execute_command(writer, reader, "show ont info all", sleep_time=1, timeout=120) for line in onu_out.splitlines(): # Format: 0/1 4 20 A2:8F:01:07:3E:28 Active Online success match dying-gasp 01714335998 m = re.match( r"\s*(\d+)/(\d+)\s+(\d+)\s+(\d+)\s+" r"([0-9a-fA-F:]{17})\s+" r"(\S+)\s+(\S+)\s+\S+\s+\S+\s+(\S+)\s*(.*)?$", line ) if not m: continue slot, pon, _, ont_id = m.group(1), m.group(2), m.group(3), m.group(4) raw_mac = m.group(5) run_state = m.group(7).lower() last_down = m.group(8) desc = m.group(9).strip() if m.group(9) else "" # iface = f"epon{slot}/{pon}/{ont_id}" iface = f"epon{slot}/{pon}:{ont_id}" dotted_mac = _cdata_normalize_mac(raw_mac) conn_state = "Connect" if run_state == "online" else "Down" dereg = "" if last_down in ("--", "") else last_down active_onus.append({ "interface": iface, "vendor": "CDATA", "model": "", "mac": dotted_mac, "description": desc, "bind_type": "static", "status": run_state, "conn_state": conn_state, "dereg": dereg, "distance": "0", }) return active_onus, mac_map, mac_inserts, rx_power_map # ========================================================== # ========== Main Telnet Wrapper =========================== # ========================================================== async def scrape_olt_via_telnet(olt_name, olt_ip, olt_user, olt_pass, ppp_map, olt_device_id, pool, brand): """ Unified telnet wrapper. Vendor-specific parsing is dispatched based on `brand` (BDCOM / CDATA). """ print(f"\n>>>>> Processing OLT: {olt_name} ({olt_ip}) [{brand}] <<<<<") try: # Telnet Login reader, writer = await asyncio.wait_for( telnetlib3.open_connection(olt_ip, 23), timeout=15 ) # Read until username prompt — works for both "Username:" and "User name:" login_banner = "" try: for _ in range(10): chunk = await asyncio.wait_for(reader.read(4096), timeout=3) login_banner += chunk if "name:" in login_banner.lower() or "login:" in login_banner.lower(): break except: pass writer.write(olt_user + "\r\n") await asyncio.sleep(1) # Read until password prompt pass_banner = "" try: for _ in range(10): chunk = await asyncio.wait_for(reader.read(4096), timeout=3) pass_banner += chunk if "password:" in pass_banner.lower(): break except: pass writer.write(olt_pass + "\r\n") await asyncio.sleep(2) # Terminal setup for cmd in ["enable", "terminal length 0", "terminal width 512", "terminal length 0"]: writer.write(cmd + "\r\n") await asyncio.sleep(0.8) try: resp = await asyncio.wait_for(reader.read(4096), timeout=1) if "password" in resp.lower(): writer.write(olt_pass + "\r\n") await asyncio.sleep(0.8) except: pass # Flush remaining buffer for _ in range(5): try: await asyncio.wait_for(reader.read(4096), timeout=0.3) except: break # Dispatch to vendor parser if brand.upper() == "CDATA": active_onus, mac_map, mac_inserts, rx_power_map = await cdata_scrape( writer, reader, olt_name ) else: active_onus, mac_map, mac_inserts, rx_power_map = await bdcom_scrape( writer, reader, olt_name ) writer.close() if not active_onus: print(f" [{olt_name}] No ONUs found. Skipping DB sync.") return print(f" [{olt_name}] Found {len(active_onus)} ONUs. Syncing DB...") # ===== DB SYNC via Pool (vendor-agnostic) ===== async with pool.acquire() as conn: async with conn.cursor(aiomysql.DictCursor) as cursor: # 1. Bulk MAC insert if mac_inserts: existing_macs = set() chunk_size = 200 for i in range(0, len(mac_inserts), chunk_size): chunk = [(r[1], r[2]) for r in mac_inserts[i:i+chunk_size]] placeholders = ",".join(["(%s,%s)"] * len(chunk)) flat = [v for pair in chunk for v in pair] await cursor.execute( f"SELECT mac_address, interface FROM mac_address_table WHERE (mac_address, interface) IN ({placeholders})", flat, ) for row in await cursor.fetchall(): existing_macs.add((row["mac_address"], row["interface"])) new_macs = [ (vlan, mac, "DYNAMIC", iface) for vlan, mac, iface in mac_inserts if (mac, iface) not in existing_macs ] if new_macs: await cursor.executemany( "INSERT INTO mac_address_table (vlan, mac_address, item_category, interface, created_at, updated_at) VALUES (%s,%s,%s,%s, NOW(), NOW())", new_macs, ) # 2. Bulk prefetch existing ONUs existing_onus = {} onu_keys = [(o["interface"], o["mac"]) for o in active_onus] for i in range(0, len(onu_keys), 200): chunk = onu_keys[i:i+200] placeholders = ",".join(["(%s,%s)"] * len(chunk)) flat = [v for pair in chunk for v in pair] await cursor.execute( f"SELECT id, interface_name, mac_address, connection_state, detail, last_down_time, last_up_time FROM onu_interface_state WHERE (interface_name, mac_address) IN ({placeholders})", flat, ) for row in await cursor.fetchall(): existing_onus[(row["interface_name"], row["mac_address"])] = row # 3. Build batch upsert lists invalid_powers = ["0", "0.0", "0.00", "-inf", "", "none", "n/a"] updates = [] inserts = [] for onu in active_onus: existing_row = existing_onus.get((onu["interface"], onu["mac"])) rx_power = "0" if onu["conn_state"] == "Connect": rx_power = rx_power_map.get(onu["interface"].lower(), "0") u_macs = mac_map.get(onu["interface"], []) ppp_name = next((ppp_map[m] for m in u_macs if m in ppp_map), "") db_last_down = existing_row.get("last_down_time") if existing_row else None db_last_up = existing_row.get("last_up_time") if existing_row else None if existing_row: old_state = str(existing_row.get("connection_state") or "").strip().lower() new_state = str(onu["conn_state"] or "").strip().lower() if old_state == "connect" and new_state == "down": db_last_down = datetime.datetime.now() elif old_state == "down" and new_state == "connect": db_last_up = datetime.datetime.now() if existing_row.get("detail"): try: old_detail = json.loads(existing_row["detail"]) if onu["conn_state"] == "Connect" and str(rx_power).strip().lower() in invalid_powers: saved = str(old_detail.get("rx_power", "0")).strip().lower() if saved not in invalid_powers: rx_power = old_detail.get("rx_power") if onu["distance"] == "0" and old_detail.get("distance"): onu["distance"] = old_detail.get("distance") if not ppp_name and old_detail.get("ppp_name"): ppp_name = old_detail.get("ppp_name") if not u_macs and old_detail.get("user_macs"): u_macs = old_detail.get("user_macs") except Exception: pass detail = json.dumps({ "rx_power": rx_power, "distance": onu["distance"], "user_macs": u_macs, "ppp_name": ppp_name, "olt_ip": olt_ip, "brand": brand, }) updates.append(( onu["conn_state"], onu["status"], onu["dereg"], onu["description"], onu["bind_type"], onu["vendor"], onu["model"], detail, db_last_down, db_last_up, onu["interface"], onu["mac"], )) else: detail = json.dumps({ "rx_power": rx_power, "distance": onu["distance"], "user_macs": u_macs, "ppp_name": ppp_name, "olt_ip": olt_ip, "brand": brand, }) inserts.append(( onu["interface"], onu["conn_state"], onu["vendor"], onu["model"], onu["mac"], onu["description"], onu["bind_type"], onu["status"], onu["dereg"], detail, olt_device_id, )) # 4. Batch execute if updates: await cursor.executemany( """UPDATE onu_interface_state SET connection_state=%s, status=%s, dereg_reason=%s, description_info=%s, bind_type=%s, vendor_id=%s, model_id=%s, detail=%s, last_down_time=%s, last_up_time=%s, updated_at=NOW() WHERE interface_name=%s AND mac_address=%s""", updates, ) if inserts: await cursor.executemany( """INSERT INTO onu_interface_state (interface_name, connection_state, vendor_id, model_id, mac_address, description_info, bind_type, status, dereg_reason, detail, olt_device_id, last_down_time, last_up_time, created_at, updated_at) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s, NULL, NULL, NOW(), NOW())""", inserts, ) await conn.commit() print(f" [{olt_name}] Done! Updated: {len(updates)} | Inserted: {len(inserts)}") except Exception as e: print(f"\n [!] Error processing {olt_name}: {e}") # ========== Main ========== async def main_async(): print(f"--- Global Process Started: {time.strftime('%Y-%m-%d %H:%M:%S')} ---") # Sync DB connection only for setup. import pymysql db = pymysql.connect( host="db1.plexuscloud.com.bd", user="c2backend", password="XdQAo2Yvy@u9", database="c2erp_backend", charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, autocommit=True, ) cursor = db.cursor() # NOTE: `manufacturer` column holds vendor name. # Expected values: 'bdcom' or 'cdata' (case-insensitive). cursor.execute( "SELECT id, name, ip_address, username, password, zone, manufacturer " "FROM network_devices WHERE type='OLT' AND status='Active'" ) olts = cursor.fetchall() cursor.execute( "SELECT zone, ip_address, username, password " "FROM network_devices WHERE type='Router' AND status='Active'" ) raw_routers = cursor.fetchall() # Ping all device ping_stats = {"total": 0, "up": 0, "down": 0} for r in raw_routers: r["is_up"] = is_device_up(r["ip_address"]) ping_stats["total"] += 1 status = "Up" if r["is_up"] else "Down" ping_stats["up" if r["is_up"] else "down"] += 1 cursor.execute("UPDATE network_devices SET ping_status=%s WHERE ip_address=%s", (status, r["ip_address"])) for olt in olts: olt["is_up"] = is_device_up(olt["ip_address"]) ping_stats["total"] += 1 status = "Up" if olt["is_up"] else "Down" ping_stats["up" if olt["is_up"] else "down"] += 1 cursor.execute("UPDATE network_devices SET ping_status=%s WHERE id=%s", (status, olt["id"])) db.close() print(f" => Pinged: {ping_stats['total']} | UP: {ping_stats['up']} | DOWN: {ping_stats['down']}") # Zone-wise router map zone_routers = defaultdict(list) for r in raw_routers: zone_routers[r["zone"]].append(r) # ===== PPP Data at a glance fetch, all OLT will share ===== zone_ppp = defaultdict(dict) fetched_routers = set() for olt in olts: zone = olt["zone"] if zone not in zone_ppp: for router in zone_routers.get(zone, []): router_ip = router["ip_address"] if router_ip not in fetched_routers: fetched_routers.add(router_ip) if router.get("is_up", False): zone_ppp[zone].update( get_mikrotik_active_ppp( router_ip, router["username"], router["password"] ) ) else: print(f" [!] Router {router_ip} is DOWN. Skipping PPP fetch.") print(f"\n => PPP Cache ready for {len(zone_ppp)} zone(s).") # ===== DB Connection Pool ===== pool = await create_db_pool() # ===== all OLTs in parallel ===== tasks = [] for olt in olts: if olt.get("is_up", False): brand = (olt.get("manufacturer") or "BDCOM").strip().upper() tasks.append( scrape_olt_via_telnet( olt["name"], olt["ip_address"], olt["username"], olt["password"], zone_ppp.get(olt["zone"], {}), olt["id"], pool, brand, ) ) else: print(f"\n>>>>> Skipping OLT: {olt['name']} ({olt['ip_address']}) - DOWN <<<<<") print(f"\n => Starting {len(tasks)} OLT(s) in parallel...") start = time.time() # Semaphore — 1 at a time (your original setting). Increase if needed. semaphore = asyncio.Semaphore(1) async def run_with_sem(task): async with semaphore: return await task await asyncio.gather(*[run_with_sem(t) for t in tasks], return_exceptions=True) pool.close() await pool.wait_closed() elapsed = time.time() - start print(f"\n--- Process Finished in {elapsed:.1f}s ({elapsed/60:.1f} min) ---") def main(): asyncio.run(main_async()) if __name__ == "__main__": main()