diff --git a/docs-es/v2.0/integrations-es.md b/docs-es/v2.0/integrations-es.md index bca7020a..936a2b71 100644 --- a/docs-es/v2.0/integrations-es.md +++ b/docs-es/v2.0/integrations-es.md @@ -51,7 +51,7 @@ LibreQoS soporta un campo genérico opcional `"id"` en los nodos de `network.jso | Splynx | Sí | Los nodos de topología de network sites y AP/site exportan `id` genérico. | | Sonar | Sí | Los nodos de topología de sitios y AP exportan `id` genérico. | | Netzur | Parcial | Se exporta solo cuando los datos upstream de zonas incluyen un ID de zona estable. | -| VISP | No | El importador actual shapea servicios/dispositivos pero no construye nodos de topología en `network.json`. | +| VISP | Parcial | Importa topología de sitio/upstream e IDs genéricos estables cuando VISP IRM tiene pobladas esas relaciones; los suscriptores sin mapeo usable siguen cayendo en adjunto plano. | | Powercode | No | El importador actual no construye nodos de topología en `network.json`. | | WISPGate | No | El importador actual no construye nodos de topología a partir de identificadores estables upstream. | diff --git a/docs-es/v2.0/integrations-reference-es.md b/docs-es/v2.0/integrations-reference-es.md index 3761142a..7ffa173f 100644 --- a/docs-es/v2.0/integrations-reference-es.md +++ b/docs-es/v2.0/integrations-reference-es.md @@ -181,7 +181,9 @@ timeout_secs = 20 Notas: -- La importación VISP usa GraphQL y actualmente trabaja en estrategia `flat`. +- La importación VISP usa GraphQL. +- El importador prioriza consultas bulk rápidas y completa de forma selectiva los suscriptores/servicios que no estén cubiertos por esa ruta. +- Cuando las relaciones IRM de dispositivo upstream están pobladas en VISP, el importador también construye topología de sitio/upstream para los suscriptores importados. - La integración reescribe `ShapedDevices.csv` en cada ejecución. - `network.json` sigue siendo una entrada DIY/manual operada por el usuario. - Los tokens de VISP se cachean en `/.visp_token_cache_*.json`. diff --git a/docs-es/v2.0/integrations-visp-es.md b/docs-es/v2.0/integrations-visp-es.md index bc7b2b95..a856b64f 100644 --- a/docs-es/v2.0/integrations-visp-es.md +++ b/docs-es/v2.0/integrations-visp-es.md @@ -13,6 +13,8 @@ Use esta integración cuando VISP sea su fuente de verdad CRM/NMS. ## Notas operativas - `ShapedDevices.csv` se reescribe en cada ejecución. +- La importación VISP usa GraphQL y prioriza las consultas bulk rápidas; si faltan suscriptores o servicios, intenta completar la cobertura con consultas adicionales de GraphQL. +- Cuando VISP IRM tiene pobladas las relaciones de dispositivo upstream, la importación también construye topología de sitio/upstream para esos suscriptores en lugar de quedarse siempre en modo plano. - `network.json` sigue siendo una entrada DIY/manual operada por el usuario. - Para despliegues guiados por integración, use `topology_import.json` y `network.effective.json`. diff --git a/docs/v2.0/integrations-reference.md b/docs/v2.0/integrations-reference.md index 68b86217..1a245f83 100644 --- a/docs/v2.0/integrations-reference.md +++ b/docs/v2.0/integrations-reference.md @@ -182,7 +182,9 @@ timeout_secs = 20 ``` Notes: -- VISP import is GraphQL-based and currently defaults to a flat topology strategy. +- VISP import is GraphQL-based. +- The importer prefers fast bulk service pulls and selectively backfills uncovered subscriber/service data from other VISP GraphQL queries. +- When VISP IRM upstream-device relationships are populated, the importer also builds site/upstream topology for imported subscribers. - The integration writes `ShapedDevices.csv` every run. - `network.json` remains a DIY/manual ingress file; built-in integrations do not overwrite it. - VISP auth tokens are cached in `/.visp_token_cache_*.json`. diff --git a/docs/v2.0/integrations-visp.md b/docs/v2.0/integrations-visp.md index 0320bcc0..67e0e0a1 100644 --- a/docs/v2.0/integrations-visp.md +++ b/docs/v2.0/integrations-visp.md @@ -17,9 +17,10 @@ timeout_secs = 20 ``` Notes: -- VISP import is GraphQL-based and currently defaults to a flat topology strategy. -- The integration refreshes its shaping data every run. -- `network.json` is for DIY/manual deployments; built-in integrations do not write it. +- VISP import is GraphQL-based. +- The importer prefers fast bulk service pulls and automatically backfills missing subscriber/service data from other VISP GraphQL queries when needed. +- When VISP IRM upstream-device data is populated, the importer also builds site/upstream topology for imported subscribers instead of staying flat-only. +- `network.json` is for DIY/manual deployments; built-in integrations do not overwrite it. - VISP auth tokens are cached in `/.visp_token_cache_*.json`. Run a manual import with: diff --git a/docs/v2.0/integrations.md b/docs/v2.0/integrations.md index efc15a0f..c0462c81 100644 --- a/docs/v2.0/integrations.md +++ b/docs/v2.0/integrations.md @@ -80,7 +80,7 @@ Current behavior: | Splynx | Yes | Network sites and AP/site topology nodes export generic `id`. | | Sonar | Yes | Site and AP topology nodes export generic `id`. | | Netzur | Partial | Exported only when the upstream zone data includes a stable zone ID. | -| VISP | No | Current importer shapes services/devices but does not build topology nodes in `network.json`. | +| VISP | Partial | Imports site/upstream topology and stable generic IDs when VISP IRM relationships are populated; subscribers without usable upstream mapping still fall back to flat attachment. | | Powercode | No | Current importer does not build topology nodes in `network.json`. | | WISPGate | No | Current importer does not build topology nodes from stable upstream topology identifiers. | diff --git a/src/integrationVISP.py b/src/integrationVISP.py index 5d94f627..4eb3ccb4 100644 --- a/src/integrationVISP.py +++ b/src/integrationVISP.py @@ -5,7 +5,8 @@ import json import os import time import hashlib -from typing import Any, Dict, List, Optional +import ipaddress +from typing import Any, Dict, List, Optional, Set, Tuple import requests from collections import Counter @@ -31,7 +32,7 @@ from integrationCommon import ( NetworkNode, NodeType, apply_client_bandwidth_multiplier, - isIpv4Permitted, + isIntegrationOutputIpAllowed, ) @@ -256,6 +257,158 @@ def _wifi_bulk(visp: VispClient) -> List[Dict[str, Any]]: res = data.get("customerCPEwithWirelessServiceSpeeds") return res if isinstance(res, list) else [] + +def _subscribers(visp: VispClient) -> List[Dict[str, Any]]: + q = """ + query Subs($isp_id:Int!, $limit:Int!, $offset:Int!){ + rows: subscribers(isp_id:$isp_id, limit:$limit, offset:$offset){ + customer_id + username + first_name + last_name + status + package + package_ip + equipment_ip + equipment_status + tower + isp_site + access_point + subscriber_type + } + count: subscribersCount(isp_id:$isp_id) + } + """ + rows: List[Dict[str, Any]] = [] + limit = 500 + offset = 0 + total = None + while True: + data = visp.gql(q, {"isp_id": visp.isp_id, "limit": limit, "offset": offset}, "Subs") + batch = data.get("rows") or [] + if not isinstance(batch, list) or not batch: + break + rows.extend([row for row in batch if isinstance(row, dict)]) + if total is None: + try: + total = int(data.get("count") or 0) + except Exception: + total = 0 + offset += len(batch) + if len(batch) < limit: + break + if total and offset >= total: + break + return rows + + +def _customer_services(visp: VispClient, customer_id: int) -> List[Dict[str, Any]]: + q = """ + query CustomerServices($customer_id:Int!){ + rows: customerServices(customer_id:$customer_id, active:true){ + id + service_id + service_name + service_type + service_label + status + package_number + bill_separately + service_details { + __typename + ... on ServiceTypeWifi { + service_number + service_name + network_id + network_name + down_speed + down_speed_unit + up_speed + up_speed_unit + ip_address + mac_address + username + } + ... on ServiceTypeVoip { + service_number + } + } + } + } + """ + data = visp.gql(q, {"customer_id": int(customer_id)}, "CustomerServices") + rows = data.get("rows") or [] + return rows if isinstance(rows, list) else [] + + +def _customer_ip_list(visp: VispClient, customer_id: int) -> Dict[str, Any]: + q = """ + query CustomerIpList($customer_id:Int!){ + row: customerIPList(customer_id:$customer_id){ + package_ip + equipment_ip + } + } + """ + data = visp.gql(q, {"customer_id": int(customer_id)}, "CustomerIpList") + row = data.get("row") or {} + return row if isinstance(row, dict) else {} + + +def _equipment_assemblies_cpe(visp: VispClient) -> List[Dict[str, Any]]: + q = """ + query EquipmentAssemblies($isp_id:Int!){ + rows: equipmentAssemblies(isp_id:$isp_id, ap_upstream:true, type:"CPE"){ + id + parent_id + description + location_name + item_id + parent_site_id + equipment_data + parent_data + } + } + """ + data = visp.gql(q, {"isp_id": visp.isp_id}, "EquipmentAssemblies") + rows = data.get("rows") or [] + return rows if isinstance(rows, list) else [] + + +def _equipment_pg_bulk(visp: VispClient) -> List[Dict[str, Any]]: + q = """ + query EquipmentPg($isp_id:Int!, $limit:Int!, $offset:Int!){ + rows: equipmentPg(isp_id:$isp_id, limit:$limit, offset:$offset){ + id + parent_id + name + description + type + default_type + location_id + location_name + parent_site_id + item_id + child_count + equipment_data + parent_data + } + } + """ + rows: List[Dict[str, Any]] = [] + limit = 500 + offset = 0 + while True: + data = visp.gql(q, {"isp_id": visp.isp_id, "limit": limit, "offset": offset}, "EquipmentPg") + batch = data.get("rows") or [] + if not isinstance(batch, list) or not batch: + break + rows.extend([row for row in batch if isinstance(row, dict)]) + offset += len(batch) + if len(batch) < limit: + break + return rows + def _extract_ipv4_list(ip_val: Any) -> List[str]: out: List[str] = [] if not ip_val: @@ -263,11 +416,245 @@ def _extract_ipv4_list(ip_val: Any) -> List[str]: ip = str(ip_val).strip() if not ip: return out - if isIpv4Permitted(ip): - out.append(ip) + base_ip = ip.split("/", 1)[0].strip() + try: + parsed = ipaddress.ip_address(base_ip) + except ValueError: + return out + if parsed.version != 4: + return out + # VISP sometimes surfaces placeholder addresses such as 0.0.0.0 for + # unassigned subscriber/service IPs. These should never become shapeable + # device addresses. + if parsed.is_unspecified or base_ip == "255.255.255.255": + return out + try: + if isIntegrationOutputIpAllowed(base_ip): + out.append(base_ip) + except Exception: + return out return out +def _split_ipv4_candidates(*values: Any) -> List[str]: + seen: Set[str] = set() + out: List[str] = [] + for value in values: + if value is None: + continue + if isinstance(value, list): + items = value + else: + items = str(value).split(",") + for item in items: + ip = str(item or "").strip() + if not ip: + continue + for parsed in _extract_ipv4_list(ip): + if parsed not in seen: + seen.add(parsed) + out.append(parsed) + return out + + +def _safe_int(value: Any) -> Optional[int]: + try: + return int(value) + except (TypeError, ValueError): + return None + + +def _bulk_ipv4_candidates( + subscriber_row: Optional[Dict[str, Any]], + service_row: Dict[str, Any], + equipment_rows: List[Dict[str, Any]], +) -> List[str]: + candidate_values: List[Any] = [service_row.get("ip_address")] + if subscriber_row: + candidate_values.extend( + [ + subscriber_row.get("package_ip"), + subscriber_row.get("equipment_ip"), + ] + ) + for equipment_row in equipment_rows: + if not isinstance(equipment_row, dict): + continue + candidate_values.append(equipment_row.get("ip_address")) + return _split_ipv4_candidates(*candidate_values) + + +def _full_name(rec: Dict[str, Any]) -> str: + first = str(rec.get("first_name") or "").strip() + last = str(rec.get("last_name") or "").strip() + full = f"{first} {last}".strip() + return full + + +def _service_is_shapable(service_type: Any, service_details: Any) -> bool: + type_name = str(service_type or "").strip().lower() + details = service_details if isinstance(service_details, dict) else {} + typename = str(details.get("__typename") or "").strip() + return type_name == "wifi" or typename == "ServiceTypeWifi" + + +def _extract_mac_candidates(row: Dict[str, Any]) -> List[str]: + candidates: List[str] = [] + seen: Set[str] = set() + equipment_data = row.get("equipment_data") or {} + if not isinstance(equipment_data, dict): + equipment_data = {} + for key in ("Fiber MAC", "mac_address", "Ethernet MAC", "WAN MAC", "MAC Addr", "radio_mac"): + mac = _normalize_mac(equipment_data.get(key)) + if mac and mac not in seen: + seen.add(mac) + candidates.append(mac) + return candidates + + +def _score_attachment_row( + row: Dict[str, Any], + customer_names: Set[str], + preferred_ids: Set[int], + preferred_macs: Set[str], +) -> int: + score = 0 + parent_id = row.get("parent_id") + if parent_id: + score += 100 + location_name = str(row.get("location_name") or "").strip() + if location_name and location_name in customer_names: + score += 40 + try: + if int(row.get("id") or 0) in preferred_ids: + score += 20 + except Exception: + pass + macs = set(_extract_mac_candidates(row)) + if preferred_macs and macs.intersection(preferred_macs): + score += 15 + equipment_data = row.get("equipment_data") or {} + if isinstance(equipment_data, dict): + if equipment_data.get("Fiber MAC"): + score += 10 + if equipment_data.get("VLAN") or equipment_data.get("VLAN-1"): + score += 5 + if str(equipment_data.get("Router Mode") or "").strip().lower() == "true": + score -= 20 + description = str(row.get("description") or "").strip().lower() + if "router" in description: + score -= 10 + return score + + +def _select_attachment_equipment( + candidate_rows: List[Dict[str, Any]], + customer_names: Set[str], + preferred_ids: Set[int], + preferred_macs: Set[str], +) -> Optional[Dict[str, Any]]: + if not candidate_rows: + return None + ranked = sorted( + candidate_rows, + key=lambda row: ( + _score_attachment_row(row, customer_names, preferred_ids, preferred_macs), + int(row.get("id") or 0), + ), + reverse=True, + ) + return ranked[0] + + +def _visp_site_node_id(location_id: Any, location_name: Any) -> str: + if location_id: + return f"visp_site_{location_id}" + name = str(location_name or "unknown").strip().lower().replace(" ", "_") + return f"visp_site_name_{name}" + + +def _visp_ap_node_id(equipment_id: Any) -> str: + return f"visp_eq_{equipment_id}" + + +def _ensure_site_node( + net: NetworkGraph, + site_cache: Dict[str, str], + location_id: Any, + location_name: Any, +) -> str: + site_id = _visp_site_node_id(location_id, location_name) + existing = site_cache.get(site_id) + if existing: + return existing + if net.findNodeIndexById(site_id) == -1: + display_name = str(location_name or "VISP Site").strip() or "VISP Site" + net.addRawNode( + NetworkNode( + id=site_id, + displayName=display_name, + parentId="", + type=NodeType.site, + networkJsonId=f"visp:site:{location_id or display_name}", + ) + ) + site_cache[site_id] = site_id + return site_id + + +def _ensure_infrastructure_node( + net: NetworkGraph, + pg_by_id: Dict[int, Dict[str, Any]], + site_cache: Dict[str, str], + ap_cache: Dict[int, str], + equipment_id: Any, +) -> str: + try: + eq_id = int(equipment_id or 0) + except Exception: + return "" + if eq_id <= 0: + return "" + existing = ap_cache.get(eq_id) + if existing: + return existing + row = pg_by_id.get(eq_id) + if not row: + return "" + + parent_node_id = "" + parent_id = row.get("parent_id") + try: + parent_eq_id = int(parent_id or 0) + except Exception: + parent_eq_id = 0 + + if parent_eq_id > 0 and parent_eq_id != eq_id and parent_eq_id in pg_by_id: + parent_node_id = _ensure_infrastructure_node(net, pg_by_id, site_cache, ap_cache, parent_eq_id) + else: + parent_node_id = _ensure_site_node( + net, + site_cache, + row.get("location_id"), + row.get("location_name"), + ) + + node_id = _visp_ap_node_id(eq_id) + if net.findNodeIndexById(node_id) == -1: + display_name = str(row.get("description") or row.get("name") or f"VISP {eq_id}").strip() or f"VISP {eq_id}" + net.addRawNode( + NetworkNode( + id=node_id, + displayName=display_name, + parentId=parent_node_id, + type=NodeType.ap, + networkJsonId=f"visp:eq:{eq_id}", + ) + ) + ap_cache[eq_id] = node_id + return node_id + + def _service_is_active(details: Dict[str, Any]) -> bool: # VISP can surface multiple status-like fields. Be conservative about skipping: # only skip if it is clearly terminated/cancelled/deleted. Many VISP tenants appear @@ -309,6 +696,7 @@ def _add_circuit_and_device( ipv4s: List[str], mac: str, device_id_suffix: str = "dev", + parent_id: str = "", ) -> bool: if not ipv4s: return False @@ -322,7 +710,7 @@ def _add_circuit_and_device( id=circuit_id, displayName=customer_name, type=NodeType.client, - parentId="", + parentId=parent_id, address=customer_name, customerName=customer_name, download=apply_client_bandwidth_multiplier(down_mbps), @@ -347,9 +735,22 @@ def createShaper() -> None: t0 = time.time() visp = VispClient() net = NetworkGraph() + # Some deployed configs surface exception_cpes() as a list instead of a map. + # Normalize locally so the VISP importer doesn't rely on integrationCommon + # accepting either shape. + if not isinstance(net.exceptionCPEs, dict): + net.exceptionCPEs = {} counters = { + "subscribers_rows": 0, "wifi_bulk_services": 0, + "bulk_customers": 0, + "backfill_customers": 0, + "backfill_service_candidates": 0, + "backfill_shaped": 0, + "topology_sites": 0, + "topology_equipment": 0, + "topology_attached": 0, "shaped": 0, "skipped_inactive": 0, "skipped_no_ip": 0, @@ -357,14 +758,138 @@ def createShaper() -> None: } disable_wifi_bulk = os.environ.get("VISP_DISABLE_WIFI_BULK", "").strip() in ("1", "true", "yes", "on") - # IMPORTANT: VISP imports must be fast. By default, we only use the known bulk resolver. - # Set VISP_ENABLE_CUSTOMER_PAGING=1 to opt into the slow, full-tenant scan mode. - enable_customer_paging = os.environ.get("VISP_ENABLE_CUSTOMER_PAGING", "").strip() in ("1", "true", "yes", "on") + debug_topology = os.environ.get("VISP_DEBUG_TOPOLOGY", "").strip() in ("1", "true", "yes", "on") # 1) WiFi bulk shaping (fast path with IP + speed + MAC) debug_status = os.environ.get("VISP_DEBUG_STATUS", "").strip() in ("1", "true", "yes", "on") pkg_status_counts: Counter = Counter() status_counts: Counter = Counter() + bulk_customers: Set[int] = set() + subscriber_by_customer_id: Dict[int, Dict[str, Any]] = {} + subscribers = _subscribers(visp) + counters["subscribers_rows"] = len(subscribers) + for sub in subscribers: + customer_id = _safe_int(sub.get("customer_id")) + if customer_id is None: + continue + if customer_id > 0: + subscriber_by_customer_id[customer_id] = sub + + assembly_rows = _equipment_assemblies_cpe(visp) + assembly_by_id: Dict[int, Dict[str, Any]] = {} + assemblies_by_location_name: Dict[str, List[Dict[str, Any]]] = {} + for row in assembly_rows: + try: + row_id = int(row.get("id") or 0) + except Exception: + row_id = 0 + if row_id > 0: + assembly_by_id[row_id] = row + location_name = str(row.get("location_name") or "").strip() + if location_name: + assemblies_by_location_name.setdefault(location_name, []).append(row) + + pg_rows = _equipment_pg_bulk(visp) + pg_by_id: Dict[int, Dict[str, Any]] = {} + for row in pg_rows: + try: + row_id = int(row.get("id") or 0) + except Exception: + row_id = 0 + if row_id > 0: + pg_by_id[row_id] = row + site_cache: Dict[str, str] = {} + ap_cache: Dict[int, str] = {} + + def topology_parent_for_customer( + customer_id: Optional[int], + bulk_customer_name: str, + equipment_ids: Set[int], + preferred_macs: Set[str], + ) -> str: + customer_names: Set[str] = set() + if bulk_customer_name: + customer_names.add(bulk_customer_name) + sub = subscriber_by_customer_id.get(int(customer_id or 0)) + if sub: + full = _full_name(sub) + if full: + customer_names.add(full) + username = str(sub.get("username") or "").strip() + if username: + customer_names.add(username) + + candidates: List[Dict[str, Any]] = [] + for equipment_id in equipment_ids: + row = assembly_by_id.get(equipment_id) + if row: + candidates.append(row) + if not candidates: + for customer_name in customer_names: + candidates.extend(assemblies_by_location_name.get(customer_name, [])) + + # Deduplicate candidate rows while preserving first occurrence. + deduped: List[Dict[str, Any]] = [] + seen_candidate_ids: Set[int] = set() + for row in candidates: + try: + candidate_id = int(row.get("id") or 0) + except Exception: + candidate_id = 0 + if candidate_id in seen_candidate_ids: + continue + seen_candidate_ids.add(candidate_id) + deduped.append(row) + + selected = _select_attachment_equipment(deduped, customer_names, equipment_ids, preferred_macs) + if debug_topology: + selected_id = selected.get("id") if selected else None + selected_parent = selected.get("parent_id") if selected else None + print( + f"[VISP] Topology select customer_id={customer_id} names={sorted(customer_names)} " + f"equipment_ids={sorted(equipment_ids)} selected_id={selected_id} selected_parent={selected_parent}" + ) + if not selected: + return "" + parent_id = selected.get("parent_id") + node_id = _ensure_infrastructure_node(net, pg_by_id, site_cache, ap_cache, parent_id) + if node_id: + counters["topology_attached"] += 1 + return node_id + + def add_service_record( + circuit_id: str, + customer_id: Optional[int], + customer_name: str, + down: Optional[float], + up: Optional[float], + ipv4s: List[str], + mac: str, + device_suffix: str, + equipment_ids: Set[int], + preferred_macs: Set[str], + ) -> bool: + if not down or not up: + counters["skipped_no_speed"] += 1 + return False + if not ipv4s: + counters["skipped_no_ip"] += 1 + return False + parent_id = topology_parent_for_customer(customer_id, customer_name, equipment_ids, preferred_macs) + if _add_circuit_and_device( + net, + circuit_id, + customer_name, + down, + up, + ipv4s, + mac, + device_suffix, + parent_id=parent_id, + ): + counters["shaped"] += 1 + return True + return False if disable_wifi_bulk: print("[VISP] VISP_DISABLE_WIFI_BULK enabled; skipping wireless bulk import") @@ -373,10 +898,32 @@ def createShaper() -> None: for rec in bulk: if not isinstance(rec, dict): continue + try: + customer_id = int(rec.get("customer_id") or 0) + except Exception: + customer_id = 0 + if customer_id > 0: + bulk_customers.add(customer_id) + subscriber_row = subscriber_by_customer_id.get(customer_id) customer_name = str(rec.get("username") or "").strip() ws_list = rec.get("wireless_services") or [] if not customer_name or not isinstance(ws_list, list): continue + preferred_macs: Set[str] = set() + equipment_ids: Set[int] = set() + for eq in rec.get("equipment") or []: + if not isinstance(eq, dict): + continue + try: + eq_id = int(eq.get("equipment_id") or 0) + except Exception: + eq_id = 0 + if eq_id > 0: + equipment_ids.add(eq_id) + for key in ("mac_address", "radio_mac"): + mac = _normalize_mac(eq.get(key)) + if mac: + preferred_macs.add(mac) for ws in ws_list: if not isinstance(ws, dict): continue @@ -396,19 +943,115 @@ def createShaper() -> None: if not down or not up: counters["skipped_no_speed"] += 1 continue - ipv4s = _extract_ipv4_list(ws.get("ip_address")) + ipv4s = _bulk_ipv4_candidates(subscriber_row, ws, rec.get("equipment") or []) if not ipv4s: counters["skipped_no_ip"] += 1 continue mac = _normalize_mac(ws.get("mac_address")) - if _add_circuit_and_device(net, circuit_id, customer_name, down, up, ipv4s, mac, "wifi"): - counters["shaped"] += 1 + if mac: + preferred_macs.add(mac) + add_service_record( + circuit_id=circuit_id, + customer_id=customer_id, + customer_name=customer_name, + down=down, + up=up, + ipv4s=ipv4s, + mac=mac, + device_suffix="wifi", + equipment_ids=equipment_ids, + preferred_macs=preferred_macs, + ) - # 2) Optional slow mode: full tenant scan for additional service types - # We intentionally keep this off by default to avoid long waits during scheduled imports. - if enable_customer_paging: - print("[VISP] VISP_ENABLE_CUSTOMER_PAGING enabled, but full-tenant scan is currently disabled in this build.") - print("[VISP] If you need non-wireless bulk support, we should first confirm VISP provides bulk resolvers for them.") + counters["bulk_customers"] = len(bulk_customers) + + # 2) Coverage-first backfill: discover active customers not present in the fast bulk + # path, then selectively hydrate only the missing internet services. + for sub in subscribers: + customer_id = _safe_int(sub.get("customer_id")) + if customer_id is None: + continue + if customer_id <= 0 or customer_id in bulk_customers: + continue + counters["backfill_customers"] += 1 + customer_services: List[Dict[str, Any]] = [] + try: + customer_services = _customer_services(visp, customer_id) + except Exception as exc: + print(f"[VISP] Skipping customer_id={customer_id}: customerServices lookup failed: {exc}") + if not customer_services: + continue + + package_ips = _split_ipv4_candidates(sub.get("package_ip"), sub.get("equipment_ip")) + ip_cache = package_ips + customer_name = str(sub.get("username") or "").strip() or _full_name(sub) or str(customer_id) + customer_full_name = _full_name(sub) + equipment_ids: Set[int] = set() + preferred_macs: Set[str] = set() + if customer_full_name: + for row in assemblies_by_location_name.get(customer_full_name, []): + try: + equipment_ids.add(int(row.get("id") or 0)) + except Exception: + pass + preferred_macs.update(_extract_mac_candidates(row)) + + for svc in customer_services: + if not isinstance(svc, dict): + continue + service_details = svc.get("service_details") or {} + counters["backfill_service_candidates"] += 1 + if not _service_is_shapable(svc.get("service_type"), service_details): + continue + if not _service_is_active( + { + "status": svc.get("status"), + } + ): + counters["skipped_inactive"] += 1 + continue + + service_number = service_details.get("service_number") or svc.get("id") + if not service_number: + continue + circuit_id = str(service_number) + if net.findNodeIndexById(circuit_id) != -1: + continue + + down = _unit_to_mbps(service_details.get("down_speed"), service_details.get("down_speed_unit")) + up = _unit_to_mbps(service_details.get("up_speed"), service_details.get("up_speed_unit")) + ipv4s = _split_ipv4_candidates(service_details.get("ip_address"), ip_cache) + if not ipv4s: + try: + ip_row = _customer_ip_list(visp, customer_id) + ip_cache = _split_ipv4_candidates( + ip_cache, + ip_row.get("package_ip"), + ip_row.get("equipment_ip"), + ) + ipv4s = list(ip_cache) + except Exception: + ipv4s = list(ip_cache) + mac = _normalize_mac(service_details.get("mac_address")) + if not mac and preferred_macs: + mac = sorted(preferred_macs)[0] + + if add_service_record( + circuit_id=circuit_id, + customer_id=customer_id, + customer_name=str(service_details.get("username") or customer_name).strip() or customer_name, + down=down, + up=up, + ipv4s=ipv4s, + mac=mac, + device_suffix="backfill", + equipment_ids={eid for eid in equipment_ids if eid > 0}, + preferred_macs=preferred_macs, + ): + counters["backfill_shaped"] += 1 + + counters["topology_sites"] = len(site_cache) + counters["topology_equipment"] = len(ap_cache) # 3) Optional onlineUsers enrichment (only supplements; doesn't create new circuits) domain = str(visp_online_users_domain() or "").strip() @@ -423,7 +1066,7 @@ def createShaper() -> None: if isinstance(r, dict): u = str(r.get("username") or "").strip() ip = str(r.get("ip") or "").strip() - if u and ip and isIpv4Permitted(ip): + if u and ip and isIntegrationOutputIpAllowed(ip): ip_by_user[u] = ip # Best-effort: if device has no IPs, populate from username match for node in net.nodes: @@ -439,6 +1082,10 @@ def createShaper() -> None: print( "[VISP] shaped={shaped} wifi_bulk_services={wifi_bulk_services} " + "subscribers_rows={subscribers_rows} bulk_customers={bulk_customers} " + "backfill_customers={backfill_customers} backfill_service_candidates={backfill_service_candidates} " + "backfill_shaped={backfill_shaped} topology_sites={topology_sites} " + "topology_equipment={topology_equipment} topology_attached={topology_attached} " "skipped_inactive={skipped_inactive} " "skipped_no_ip={skipped_no_ip} skipped_no_speed={skipped_no_speed} elapsed_s={elapsed_s}".format( **counters, diff --git a/src/test_integration_visp.py b/src/test_integration_visp.py new file mode 100644 index 00000000..57a2825b --- /dev/null +++ b/src/test_integration_visp.py @@ -0,0 +1,104 @@ +import unittest +from unittest.mock import patch + +from integrationVISP import ( + _bulk_ipv4_candidates, + _select_attachment_equipment, + _service_is_shapable, + _split_ipv4_candidates, +) + + +class IntegrationVispTests(unittest.TestCase): + def test_split_ipv4_candidates_filters_and_deduplicates(self): + with patch( + "integrationVISP.isIntegrationOutputIpAllowed", + side_effect=lambda ip: ip in {"192.0.2.10", "198.51.100.20"}, + ): + result = _split_ipv4_candidates( + "192.0.2.10, 192.0.2.10, not-an-ip", + ["198.51.100.20", "", None], + ) + self.assertEqual(result, ["192.0.2.10", "198.51.100.20"]) + + def test_bulk_ipv4_candidates_uses_subscriber_and_equipment_fallbacks(self): + with patch( + "integrationVISP.isIntegrationOutputIpAllowed", + side_effect=lambda ip: ip in {"198.51.100.20", "203.0.113.7"}, + ): + result = _bulk_ipv4_candidates( + subscriber_row={ + "package_ip": "198.51.100.20", + "equipment_ip": "", + }, + service_row={ + "ip_address": "", + }, + equipment_rows=[ + {"ip_address": "203.0.113.7"}, + {"ip_address": ""}, + ], + ) + self.assertEqual(result, ["198.51.100.20", "203.0.113.7"]) + + def test_split_ipv4_candidates_keeps_public_ip_when_not_ignored(self): + with patch("integrationVISP.isIntegrationOutputIpAllowed", return_value=True): + result = _split_ipv4_candidates("203.0.113.191") + self.assertEqual(result, ["203.0.113.191"]) + + def test_split_ipv4_candidates_rejects_unspecified_placeholder_ips(self): + with patch("integrationVISP.isIntegrationOutputIpAllowed", return_value=True): + result = _split_ipv4_candidates("0.0.0.0, 255.255.255.255, 203.0.113.191") + self.assertEqual(result, ["203.0.113.191"]) + + def test_service_is_shapable_accepts_wifi_typed_service(self): + self.assertTrue( + _service_is_shapable( + "wifi", + {"__typename": "ServiceTypeWifi"}, + ) + ) + + def test_service_is_shapable_rejects_non_internet_service(self): + self.assertFalse( + _service_is_shapable( + "voip", + {"__typename": "ServiceTypeVoip"}, + ) + ) + + def test_select_attachment_prefers_parented_fiber_cpe(self): + rows = [ + { + "id": 1450, + "parent_id": None, + "location_name": "Example Subscriber", + "description": "Router-side device", + "equipment_data": { + "mac_address": "02:00:00:00:00:10", + "Router Mode": "false", + }, + }, + { + "id": 1296, + "parent_id": 1409, + "location_name": "Example Subscriber", + "description": "Fiber ONU", + "equipment_data": { + "Fiber MAC": "02:00:00:00:00:20", + "VLAN-1": "Example-VLAN", + }, + }, + ] + selected = _select_attachment_equipment( + candidate_rows=rows, + customer_names={"Example Subscriber"}, + preferred_ids={1450, 1296}, + preferred_macs={"020000000020"}, + ) + self.assertIsNotNone(selected) + self.assertEqual(selected["id"], 1296) + + +if __name__ == "__main__": + unittest.main()