Merge pull request #1050 from LibreQoE/visp_2

Improve VISP subscriber coverage and IRM topology import

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. |
 

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 `<lqos_directory>/.visp_token_cache_*.json`.

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`.
 

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 `<lqos_directory>/.visp_token_cache_*.json`.

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 `<lqos_directory>/.visp_token_cache_*.json`.
 
 Run a manual import with:

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. |
 

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,

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()