diff --git a/netx_api/biz_state/collect_runner.py b/netx_api/biz_state/collect_runner.py index c0d9a16..cca562b 100644 --- a/netx_api/biz_state/collect_runner.py +++ b/netx_api/biz_state/collect_runner.py @@ -29,7 +29,7 @@ from ..models import ( from ..ne_netmiko import disable_target_paging, send_show_command from ..ne_session_factory import close_netmiko_connection, open_netmiko_connection from .command_match import expand_from_bindings, match_command, normalize_command -from .parsers import get_parser +from .parsers import get_parser, run_parser from .profiles import get_profile _log = logging.getLogger("netx.biz_state.runner") @@ -470,8 +470,7 @@ def _run_collect_session( merged = {**params, **hit.params} cmd_row.params_json = merged - parser = get_parser(hit.profile.parser_id) - if not parser: + if not get_parser(hit.profile.parser_id): any_fail = True cmd_row.parse_status = "failed" cmd_row.message = f"unknown parser {hit.profile.parser_id}" @@ -480,13 +479,22 @@ def _run_collect_session( continue try: - records = parser( - raw_text=cmd_row.raw_text, + records, fsm_tables, rule_keys = run_parser( + hit.profile.parser_id, + raw_text=cmd_row.raw_text or "", vendor=vendor_eff, device_type=device_type_eff, command=hit.profile.textfsm_command or concrete, params=merged, + textfsm_command=hit.profile.textfsm_command or "", ) + # Debug hint only — avoid persisting large FSM JSON. + if rule_keys: + nonempty = [k for k in rule_keys if fsm_tables.get(k)] + cmd_row.message = ( + f"fsm_keys={','.join(rule_keys)}" + + (f";hit={','.join(nonempty)}" if nonempty else ";hit=") + )[:1020] except Exception as exc: any_fail = True cmd_row.parse_status = "failed" diff --git a/netx_api/biz_state/discover.py b/netx_api/biz_state/discover.py index 80ae9df..7c7b8bc 100644 --- a/netx_api/biz_state/discover.py +++ b/netx_api/biz_state/discover.py @@ -14,7 +14,7 @@ from ..lldp_shared import resolve_vendor_key from ..models import BizStateTask from ..ne_netmiko import disable_target_paging, send_show_command from ..ne_session_factory import close_netmiko_connection, open_netmiko_connection -from .parsers import get_parser +from .parsers import get_parser, run_parser from .profiles import get_profile @@ -124,16 +124,17 @@ def discover_params( "raw_preview": str(raw or "")[:4000], } - parser = get_parser(disc.parser_id) records: list[dict[str, Any]] = [] - if parser: + if get_parser(disc.parser_id): try: - records = parser( + records, _fsm_tables, _rule_keys = run_parser( + disc.parser_id, raw_text=raw, vendor=vendor, device_type=device_type, command=disc.textfsm_command or command, params={}, + textfsm_command=disc.textfsm_command or "", ) except Exception as exc: return { diff --git a/netx_api/biz_state/parsers/__init__.py b/netx_api/biz_state/parsers/__init__.py index 9011cba..8f05b2a 100644 --- a/netx_api/biz_state/parsers/__init__.py +++ b/netx_api/biz_state/parsers/__init__.py @@ -3,7 +3,7 @@ Layout:: parsers/ - common/ # cross-vendor (lldp, vrf_list, vrf_route_summary) + common/ # cross-vendor (lldp, vrf_list, vrf_route_summary, pipeline) zte/ # implemented status tables cisco/ # skeleton — add command modules here huawei/ @@ -13,12 +13,27 @@ Layout:: ericsson/ Each vendor package exports ``PARSERS: dict[parser_id, normalize_fn]``. + +Adding a status metric +---------------------- +1. (Recommended) ``cli_templates//.textfsm`` + ``index`` line +2. ``parsers//.py``: ``RULE_KEYS = ("",)`` + + ``normalize(..., fsm_tables=...)`` (use ``common.pipeline.prefer_fsm``) +3. Register in the vendor ``PARSERS`` dict +4. ``profiles.py``: add ``ParseProfile`` (command match + ``FieldDef`` schema) + +Collect runs TextFSM rules first → ``fsm_tables``, then calls ``normalize`` with +both ``raw_text`` and ``fsm_tables``. The parser decides: return mapped FSM rows, +post-process them, or ignore FSM and hand-parse ``raw_text``. """ from __future__ import annotations -from typing import Any, Callable +import inspect +from typing import Any, Callable, Sequence +from ...lldp_shared import resolve_vendor_key +from ...ntc_parse import apply_rules, resolve_cli_platform, rules_for_command from .cisco import PARSERS as _CISCO_PARSERS from .common.lldp_neighbors import normalize_lldp_neighbors from .common.vrf_list import normalize_vrf_list @@ -59,3 +74,107 @@ def get_parser(parser_id: str) -> NormalizeFn | None: def registered_parser_ids() -> list[str]: return sorted(_REGISTRY.keys()) + + +def _rule_keys_for_fn(fn: NormalizeFn) -> tuple[tuple[str, ...], bool]: + """Return ``(keys, declared)``. ``declared`` is True when RULE_KEYS is set on fn/module.""" + keys_raw = None + declared = False + if hasattr(fn, "RULE_KEYS"): + keys_raw = getattr(fn, "RULE_KEYS") + declared = True + else: + mod = inspect.getmodule(fn) + if mod is not None and hasattr(mod, "RULE_KEYS"): + keys_raw = getattr(mod, "RULE_KEYS") + declared = True + if not keys_raw: + return (), declared + out: list[str] = [] + for k in keys_raw: + s = str(k or "").strip() + if s.lower().endswith(".textfsm"): + s = s[: -len(".textfsm")] + if s and s not in out: + out.append(s) + return tuple(out), declared + + +def get_parser_meta(parser_id: str) -> dict[str, Any] | None: + """Return ``{fn, rule_keys, rule_keys_declared}`` for a registered parser, or None.""" + fn = get_parser(parser_id) + if not fn: + return None + keys, declared = _rule_keys_for_fn(fn) + return {"fn": fn, "rule_keys": keys, "rule_keys_declared": declared} + + +def resolve_rule_keys( + *, + parser_id: str, + platform: str = "", + command: str = "", + textfsm_command: str = "", +) -> list[str]: + """RULE_KEYS from parser when declared; else index stems for textfsm_command/command.""" + meta = get_parser_meta(parser_id) + if not meta: + return [] + if meta.get("rule_keys_declared"): + return list(meta["rule_keys"]) + plat = str(platform or "").strip() + cmd = str(textfsm_command or command or "").strip() + if plat and cmd: + return rules_for_command(plat, cmd) + return [] + + +def run_parser( + parser_id: str, + *, + raw_text: str, + vendor: str = "", + device_type: str = "", + command: str = "", + params: dict[str, str] | None = None, + textfsm_command: str = "", +) -> tuple[list[dict[str, Any]], dict[str, list[dict[str, Any]]], list[str]]: + """Apply TextFSM rules then normalize. + + Returns ``(records, fsm_tables, rule_keys_used)``. + """ + meta = get_parser_meta(parser_id) + if not meta: + raise KeyError(f"unknown parser {parser_id}") + fn: NormalizeFn = meta["fn"] + platform = resolve_cli_platform( + vendor=vendor, + device_type=device_type, + vendor_key=resolve_vendor_key(vendor, device_type), + ) + cmd_hint = str(textfsm_command or command or "").strip() + rule_keys = resolve_rule_keys( + parser_id=parser_id, + platform=platform, + command=command, + textfsm_command=textfsm_command, + ) + fsm_tables: dict[str, list[dict[str, Any]]] = {} + if rule_keys and str(raw_text or "").strip(): + fsm_tables = apply_rules( + platform=platform, + text=raw_text, + rule_keys=rule_keys, + command=cmd_hint, + ) + records = fn( + raw_text=raw_text, + fsm_tables=fsm_tables, + vendor=vendor, + device_type=device_type, + command=cmd_hint or command, + params=params or {}, + ) + if not isinstance(records, list): + records = [] + return records, fsm_tables, rule_keys diff --git a/netx_api/biz_state/parsers/cisco/arp.py b/netx_api/biz_state/parsers/cisco/arp.py index f5c3688..de025e6 100644 --- a/netx_api/biz_state/parsers/cisco/arp.py +++ b/netx_api/biz_state/parsers/cisco/arp.py @@ -4,14 +4,19 @@ from __future__ import annotations from typing import Any +RULE_KEYS: tuple[str, ...] = () + def normalize_arp( *, raw_text: str, + fsm_tables=None, vendor: str = "", device_type: str = "", command: str = "", params: dict[str, str] | None = None, ) -> list[dict[str, Any]]: - _ = (raw_text, vendor, device_type, command, params) + _ = (raw_text, vendor, device_type, command, params, fsm_tables) raise NotImplementedError("cisco arp parser not implemented yet") + +normalize_arp.RULE_KEYS = RULE_KEYS diff --git a/netx_api/biz_state/parsers/cisco/bgp_peer.py b/netx_api/biz_state/parsers/cisco/bgp_peer.py index 8c74d85..036b1c2 100644 --- a/netx_api/biz_state/parsers/cisco/bgp_peer.py +++ b/netx_api/biz_state/parsers/cisco/bgp_peer.py @@ -4,14 +4,19 @@ from __future__ import annotations from typing import Any +RULE_KEYS: tuple[str, ...] = () + def normalize_bgp_peer( *, raw_text: str, + fsm_tables=None, vendor: str = "", device_type: str = "", command: str = "", params: dict[str, str] | None = None, ) -> list[dict[str, Any]]: - _ = (raw_text, vendor, device_type, command, params) + _ = (raw_text, vendor, device_type, command, params, fsm_tables) raise NotImplementedError("cisco bgp_peer parser not implemented yet") + +normalize_bgp_peer.RULE_KEYS = RULE_KEYS diff --git a/netx_api/biz_state/parsers/cisco/interface_brief.py b/netx_api/biz_state/parsers/cisco/interface_brief.py index 003f05e..7b6bead 100644 --- a/netx_api/biz_state/parsers/cisco/interface_brief.py +++ b/netx_api/biz_state/parsers/cisco/interface_brief.py @@ -4,14 +4,19 @@ from __future__ import annotations from typing import Any +RULE_KEYS: tuple[str, ...] = () + def normalize_interface_brief( *, raw_text: str, + fsm_tables=None, vendor: str = "", device_type: str = "", command: str = "", params: dict[str, str] | None = None, ) -> list[dict[str, Any]]: - _ = (raw_text, vendor, device_type, command, params) + _ = (raw_text, vendor, device_type, command, params, fsm_tables) raise NotImplementedError("cisco interface_brief parser not implemented yet") + +normalize_interface_brief.RULE_KEYS = RULE_KEYS diff --git a/netx_api/biz_state/parsers/cisco/isis_adjacency.py b/netx_api/biz_state/parsers/cisco/isis_adjacency.py index fd6194c..40b3309 100644 --- a/netx_api/biz_state/parsers/cisco/isis_adjacency.py +++ b/netx_api/biz_state/parsers/cisco/isis_adjacency.py @@ -4,14 +4,19 @@ from __future__ import annotations from typing import Any +RULE_KEYS: tuple[str, ...] = () + def normalize_isis_adjacency( *, raw_text: str, + fsm_tables=None, vendor: str = "", device_type: str = "", command: str = "", params: dict[str, str] | None = None, ) -> list[dict[str, Any]]: - _ = (raw_text, vendor, device_type, command, params) + _ = (raw_text, vendor, device_type, command, params, fsm_tables) raise NotImplementedError("cisco isis_adjacency parser not implemented yet") + +normalize_isis_adjacency.RULE_KEYS = RULE_KEYS diff --git a/netx_api/biz_state/parsers/cisco/nd6_cache.py b/netx_api/biz_state/parsers/cisco/nd6_cache.py index c044a1b..3989bff 100644 --- a/netx_api/biz_state/parsers/cisco/nd6_cache.py +++ b/netx_api/biz_state/parsers/cisco/nd6_cache.py @@ -4,14 +4,19 @@ from __future__ import annotations from typing import Any +RULE_KEYS: tuple[str, ...] = () + def normalize_nd6_cache( *, raw_text: str, + fsm_tables=None, vendor: str = "", device_type: str = "", command: str = "", params: dict[str, str] | None = None, ) -> list[dict[str, Any]]: - _ = (raw_text, vendor, device_type, command, params) + _ = (raw_text, vendor, device_type, command, params, fsm_tables) raise NotImplementedError("cisco nd6_cache parser not implemented yet") + +normalize_nd6_cache.RULE_KEYS = RULE_KEYS diff --git a/netx_api/biz_state/parsers/common/lldp_neighbors.py b/netx_api/biz_state/parsers/common/lldp_neighbors.py index a1d56d2..7960b70 100644 --- a/netx_api/biz_state/parsers/common/lldp_neighbors.py +++ b/netx_api/biz_state/parsers/common/lldp_neighbors.py @@ -6,10 +6,13 @@ from typing import Any from ....lldp_shared import NeighborHit, parse_neighbor_output +RULE_KEYS: tuple[str, ...] = () + def normalize_lldp_neighbors( *, raw_text: str, + fsm_tables=None, vendor: str = "", device_type: str = "", command: str = "", @@ -21,7 +24,7 @@ def normalize_lldp_neighbors( device_type=device_type, command=command, ) - _ = params + _ = (params, fsm_tables) rows: list[dict[str, Any]] = [] for h in hits: rows.append( @@ -34,3 +37,5 @@ def normalize_lldp_neighbors( } ) return rows + +normalize_lldp_neighbors.RULE_KEYS = RULE_KEYS diff --git a/netx_api/biz_state/parsers/common/pipeline.py b/netx_api/biz_state/parsers/common/pipeline.py new file mode 100644 index 0000000..cd001fe --- /dev/null +++ b/netx_api/biz_state/parsers/common/pipeline.py @@ -0,0 +1,31 @@ +"""Shared FSM-vs-hand helpers for biz_state parsers.""" + +from __future__ import annotations + +from typing import Any, Callable, Mapping, Sequence + +MapRowsFn = Callable[[list[dict[str, Any]]], list[dict[str, Any]]] +HandFn = Callable[..., list[dict[str, Any]]] + + +def prefer_fsm( + fsm_tables: Mapping[str, list[dict[str, Any]]] | None, + keys: Sequence[str], + map_rows: MapRowsFn, + hand_fn: HandFn, + *, + raw_text: str, + **kw: Any, +) -> list[dict[str, Any]]: + """Use the first non-empty FSM table (mapped); otherwise hand-parse raw_text.""" + tables = fsm_tables or {} + for key in keys or (): + k = str(key or "").strip() + if not k: + continue + rows = tables.get(k) or [] + if rows: + mapped = map_rows(rows) + if mapped: + return mapped + return hand_fn(raw_text=raw_text, **kw) diff --git a/netx_api/biz_state/parsers/common/vrf_list.py b/netx_api/biz_state/parsers/common/vrf_list.py index 48f943c..c6449ff 100644 --- a/netx_api/biz_state/parsers/common/vrf_list.py +++ b/netx_api/biz_state/parsers/common/vrf_list.py @@ -3,28 +3,16 @@ from __future__ import annotations import re -from typing import Any +from typing import Any, Mapping from ....lldp_shared import resolve_vendor_key -from ....ntc_parse import parse_cli, resolve_cli_platform, row_get +from ....ntc_parse import apply_rules, resolve_cli_platform, row_get +from .pipeline import prefer_fsm + +RULE_KEYS = ("zte_zxros_show_ip_vrf",) -def normalize_vrf_list( - *, - raw_text: str, - vendor: str = "", - device_type: str = "", - command: str = "", - params: dict[str, str] | None = None, -) -> list[dict[str, Any]]: - _ = params - platform = resolve_cli_platform( - vendor=vendor, - device_type=device_type, - vendor_key=resolve_vendor_key(vendor, device_type), - ) - cmd = str(command or "").strip() - rows = parse_cli(platform=platform, command=cmd, text=raw_text) if platform and cmd else [] +def _map_fsm_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: out: list[dict[str, Any]] = [] seen: set[str] = set() for r in rows: @@ -38,11 +26,15 @@ def normalize_vrf_list( { "vrf_name": name[:128], "rd": row_get(r, "DEFAULT_RD", "RD", "rd")[:64], - "protocols": row_get(r, "PROTOCOLS", "ADDRESS_FAMILY", "protocols")[:64], + "protocols": row_get(r, "PROTOCOLS", "ADDRESS_FAMILY", "INTERFACES", "protocols")[:64], } ) - if out: - return out + return out + + +def _hand_parse(*, raw_text: str, **_kw: Any) -> list[dict[str, Any]]: + out: list[dict[str, Any]] = [] + seen: set[str] = set() for line in str(raw_text or "").splitlines(): m = re.match(r"^\s*([A-Za-z0-9_./:-]+)\s+(\d+:\d+||\S+:\S+)\s*", line) if not m: @@ -55,3 +47,37 @@ def normalize_vrf_list( seen.add(name) out.append({"vrf_name": name[:128], "rd": m.group(2)[:64], "protocols": ""}) return out + + +def normalize_vrf_list( + *, + raw_text: str, + fsm_tables: Mapping[str, list[dict[str, Any]]] | None = None, + vendor: str = "", + device_type: str = "", + command: str = "", + params: dict[str, str] | None = None, +) -> list[dict[str, Any]]: + _ = params + tables = dict(fsm_tables or {}) + if not any(tables.get(k) for k in RULE_KEYS): + platform = resolve_cli_platform( + vendor=vendor, + device_type=device_type, + vendor_key=resolve_vendor_key(vendor, device_type), + ) + cmd = str(command or "show ip vrf").strip() + if platform and cmd: + tables = apply_rules( + platform=platform, text=raw_text, rule_keys=RULE_KEYS, command=cmd + ) + return prefer_fsm( + tables, + RULE_KEYS, + _map_fsm_rows, + _hand_parse, + raw_text=raw_text, + ) + + +normalize_vrf_list.RULE_KEYS = RULE_KEYS diff --git a/netx_api/biz_state/parsers/common/vrf_route_summary.py b/netx_api/biz_state/parsers/common/vrf_route_summary.py index 5fe7261..e885d75 100644 --- a/netx_api/biz_state/parsers/common/vrf_route_summary.py +++ b/netx_api/biz_state/parsers/common/vrf_route_summary.py @@ -16,16 +16,19 @@ _SOURCE_PATTERNS: list[tuple[str, re.Pattern[str]]] = [ ("total", re.compile(r"(?i)^\s*(?:total|totals?)\s+(?:routes?\s+)?(\d+)\s*$")), ] +RULE_KEYS: tuple[str, ...] = () + def normalize_vrf_route_summary( *, raw_text: str, + fsm_tables=None, vendor: str = "", device_type: str = "", command: str = "", params: dict[str, str] | None = None, ) -> list[dict[str, Any]]: - _ = (vendor, device_type, command) + _ = (vendor, device_type, command, fsm_tables) vrf = str((params or {}).get("vrf") or (params or {}).get("vrf_name") or "").strip() text = str(raw_text or "") found: dict[str, int] = {} @@ -53,3 +56,5 @@ def normalize_vrf_route_summary( if vrf and not rows: rows.append({"vrf": vrf[:128], "source": "empty", "networks": 0}) return rows + +normalize_vrf_route_summary.RULE_KEYS = RULE_KEYS diff --git a/netx_api/biz_state/parsers/huawei/arp.py b/netx_api/biz_state/parsers/huawei/arp.py index cf28553..56f1f9f 100644 --- a/netx_api/biz_state/parsers/huawei/arp.py +++ b/netx_api/biz_state/parsers/huawei/arp.py @@ -4,14 +4,19 @@ from __future__ import annotations from typing import Any +RULE_KEYS: tuple[str, ...] = () + def normalize_arp( *, raw_text: str, + fsm_tables=None, vendor: str = "", device_type: str = "", command: str = "", params: dict[str, str] | None = None, ) -> list[dict[str, Any]]: - _ = (raw_text, vendor, device_type, command, params) + _ = (raw_text, vendor, device_type, command, params, fsm_tables) raise NotImplementedError("huawei arp parser not implemented yet") + +normalize_arp.RULE_KEYS = RULE_KEYS diff --git a/netx_api/biz_state/parsers/huawei/bgp_peer.py b/netx_api/biz_state/parsers/huawei/bgp_peer.py index 27082e7..516cafc 100644 --- a/netx_api/biz_state/parsers/huawei/bgp_peer.py +++ b/netx_api/biz_state/parsers/huawei/bgp_peer.py @@ -4,14 +4,19 @@ from __future__ import annotations from typing import Any +RULE_KEYS: tuple[str, ...] = () + def normalize_bgp_peer( *, raw_text: str, + fsm_tables=None, vendor: str = "", device_type: str = "", command: str = "", params: dict[str, str] | None = None, ) -> list[dict[str, Any]]: - _ = (raw_text, vendor, device_type, command, params) + _ = (raw_text, vendor, device_type, command, params, fsm_tables) raise NotImplementedError("huawei bgp_peer parser not implemented yet") + +normalize_bgp_peer.RULE_KEYS = RULE_KEYS diff --git a/netx_api/biz_state/parsers/huawei/interface_brief.py b/netx_api/biz_state/parsers/huawei/interface_brief.py index 930fcea..fa67c79 100644 --- a/netx_api/biz_state/parsers/huawei/interface_brief.py +++ b/netx_api/biz_state/parsers/huawei/interface_brief.py @@ -4,14 +4,19 @@ from __future__ import annotations from typing import Any +RULE_KEYS: tuple[str, ...] = () + def normalize_interface_brief( *, raw_text: str, + fsm_tables=None, vendor: str = "", device_type: str = "", command: str = "", params: dict[str, str] | None = None, ) -> list[dict[str, Any]]: - _ = (raw_text, vendor, device_type, command, params) + _ = (raw_text, vendor, device_type, command, params, fsm_tables) raise NotImplementedError("huawei interface_brief parser not implemented yet") + +normalize_interface_brief.RULE_KEYS = RULE_KEYS diff --git a/netx_api/biz_state/parsers/huawei/isis_adjacency.py b/netx_api/biz_state/parsers/huawei/isis_adjacency.py index 4035caa..f67c7ce 100644 --- a/netx_api/biz_state/parsers/huawei/isis_adjacency.py +++ b/netx_api/biz_state/parsers/huawei/isis_adjacency.py @@ -4,14 +4,19 @@ from __future__ import annotations from typing import Any +RULE_KEYS: tuple[str, ...] = () + def normalize_isis_adjacency( *, raw_text: str, + fsm_tables=None, vendor: str = "", device_type: str = "", command: str = "", params: dict[str, str] | None = None, ) -> list[dict[str, Any]]: - _ = (raw_text, vendor, device_type, command, params) + _ = (raw_text, vendor, device_type, command, params, fsm_tables) raise NotImplementedError("huawei isis_adjacency parser not implemented yet") + +normalize_isis_adjacency.RULE_KEYS = RULE_KEYS diff --git a/netx_api/biz_state/parsers/huawei/nd6_cache.py b/netx_api/biz_state/parsers/huawei/nd6_cache.py index 9ecc7ec..6ff9b2f 100644 --- a/netx_api/biz_state/parsers/huawei/nd6_cache.py +++ b/netx_api/biz_state/parsers/huawei/nd6_cache.py @@ -4,14 +4,19 @@ from __future__ import annotations from typing import Any +RULE_KEYS: tuple[str, ...] = () + def normalize_nd6_cache( *, raw_text: str, + fsm_tables=None, vendor: str = "", device_type: str = "", command: str = "", params: dict[str, str] | None = None, ) -> list[dict[str, Any]]: - _ = (raw_text, vendor, device_type, command, params) + _ = (raw_text, vendor, device_type, command, params, fsm_tables) raise NotImplementedError("huawei nd6_cache parser not implemented yet") + +normalize_nd6_cache.RULE_KEYS = RULE_KEYS diff --git a/netx_api/biz_state/parsers/vrf.py b/netx_api/biz_state/parsers/vrf.py new file mode 100644 index 0000000..b0add91 --- /dev/null +++ b/netx_api/biz_state/parsers/vrf.py @@ -0,0 +1,8 @@ +"""Compat shim — prefer ``parsers.common.vrf_list`` / ``vrf_route_summary``.""" + +from __future__ import annotations + +from .common.vrf_list import normalize_vrf_list +from .common.vrf_route_summary import normalize_vrf_route_summary + +__all__ = ["normalize_vrf_list", "normalize_vrf_route_summary"] diff --git a/netx_api/biz_state/parsers/zte/arp.py b/netx_api/biz_state/parsers/zte/arp.py index 37a195c..25a37e2 100644 --- a/netx_api/biz_state/parsers/zte/arp.py +++ b/netx_api/biz_state/parsers/zte/arp.py @@ -7,6 +7,8 @@ from typing import Any _ARP_AGE_TIME_RE = re.compile(r"^\d{1,2}:\d{2}:\d{2}$") +RULE_KEYS: tuple[str, ...] = () + def is_valid_arp_age(age: str) -> bool: """True when Age looks like a dynamic timer (HH:MM:SS), not static flags like H.""" @@ -16,12 +18,13 @@ def is_valid_arp_age(age: str) -> bool: def normalize_arp( *, raw_text: str, + fsm_tables=None, vendor: str = "", device_type: str = "", command: str = "", params: dict[str, str] | None = None, ) -> list[dict[str, Any]]: - _ = (vendor, device_type, command, params) + _ = (vendor, device_type, command, params, fsm_tables) out: list[dict[str, Any]] = [] seen: set[tuple[str, str]] = set() ip_re = re.compile(r"^\d{1,3}(?:\.\d{1,3}){3}$") @@ -61,3 +64,5 @@ def normalize_arp( } ) return out + +normalize_arp.RULE_KEYS = RULE_KEYS diff --git a/netx_api/biz_state/parsers/zte/bgp_peer.py b/netx_api/biz_state/parsers/zte/bgp_peer.py index 1a29eee..5552bdd 100644 --- a/netx_api/biz_state/parsers/zte/bgp_peer.py +++ b/netx_api/biz_state/parsers/zte/bgp_peer.py @@ -11,6 +11,8 @@ _BGP_PEER_RE = re.compile( re.I, ) +RULE_KEYS: tuple[str, ...] = () + def _detect_bgp_afi(command: str, params: dict[str, str] | None) -> str: if params and params.get("afi"): @@ -30,12 +32,13 @@ def _detect_bgp_afi(command: str, params: dict[str, str] | None) -> str: def normalize_bgp_peer( *, raw_text: str, + fsm_tables=None, vendor: str = "", device_type: str = "", command: str = "", params: dict[str, str] | None = None, ) -> list[dict[str, Any]]: - _ = (vendor, device_type) + _ = (vendor, device_type, fsm_tables) afi = _detect_bgp_afi(command, params) out: list[dict[str, Any]] = [] seen: set[str] = set() @@ -77,3 +80,5 @@ def normalize_bgp_peer( } ) return out + +normalize_bgp_peer.RULE_KEYS = RULE_KEYS diff --git a/netx_api/biz_state/parsers/zte/interface_brief.py b/netx_api/biz_state/parsers/zte/interface_brief.py index b9c9989..956611d 100644 --- a/netx_api/biz_state/parsers/zte/interface_brief.py +++ b/netx_api/biz_state/parsers/zte/interface_brief.py @@ -3,10 +3,13 @@ from __future__ import annotations import re -from typing import Any +from typing import Any, Mapping from ....lldp_shared import resolve_vendor_key -from ....ntc_parse import parse_cli, resolve_cli_platform, row_get +from ....ntc_parse import apply_rules, resolve_cli_platform, row_get +from ..common.pipeline import prefer_fsm + +RULE_KEYS = ("zte_zxros_show_interface_brief",) _IFACE_RE = re.compile( r"^(?P\S+)\s+(?P\S+)\s+(?P\S+)" @@ -16,22 +19,7 @@ _IFACE_RE = re.compile( ) -def normalize_interface_brief( - *, - raw_text: str, - vendor: str = "", - device_type: str = "", - command: str = "", - params: dict[str, str] | None = None, -) -> list[dict[str, Any]]: - _ = params - platform = resolve_cli_platform( - vendor=vendor, - device_type=device_type, - vendor_key=resolve_vendor_key(vendor, device_type), - ) - cmd = str(command or "show interface brief").strip() or "show interface brief" - rows = parse_cli(platform=platform, command=cmd, text=raw_text) if platform else [] +def _map_fsm_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: out: list[dict[str, Any]] = [] seen: set[str] = set() for r in rows: @@ -53,8 +41,20 @@ def normalize_interface_brief( "description": row_get(r, "DESCRIPTION", "description")[:256], } ) - if out: - return out + return out + + +def _hand_parse( + *, + raw_text: str, + vendor: str = "", + device_type: str = "", + command: str = "", + **_kw: Any, +) -> list[dict[str, Any]]: + _ = (vendor, device_type, command) + out: list[dict[str, Any]] = [] + seen: set[str] = set() for raw in str(raw_text or "").splitlines(): line = raw.strip() if not line or line.lower().startswith("interface"): @@ -79,3 +79,40 @@ def normalize_interface_brief( } ) return out + + +def normalize_interface_brief( + *, + raw_text: str, + fsm_tables: Mapping[str, list[dict[str, Any]]] | None = None, + vendor: str = "", + device_type: str = "", + command: str = "", + params: dict[str, str] | None = None, +) -> list[dict[str, Any]]: + _ = params + tables = dict(fsm_tables or {}) + if not any(tables.get(k) for k in RULE_KEYS): + platform = resolve_cli_platform( + vendor=vendor, + device_type=device_type, + vendor_key=resolve_vendor_key(vendor, device_type), + ) + cmd = str(command or "show interface brief").strip() or "show interface brief" + if platform: + tables = apply_rules( + platform=platform, text=raw_text, rule_keys=RULE_KEYS, command=cmd + ) + return prefer_fsm( + tables, + RULE_KEYS, + _map_fsm_rows, + _hand_parse, + raw_text=raw_text, + vendor=vendor, + device_type=device_type, + command=command, + ) + + +normalize_interface_brief.RULE_KEYS = RULE_KEYS diff --git a/netx_api/biz_state/parsers/zte/isis_adjacency.py b/netx_api/biz_state/parsers/zte/isis_adjacency.py index fd0822c..30cb5ef 100644 --- a/netx_api/biz_state/parsers/zte/isis_adjacency.py +++ b/netx_api/biz_state/parsers/zte/isis_adjacency.py @@ -13,16 +13,19 @@ _ISIS_ROW_RE = re.compile( ) _PROCESS_RE = re.compile(r"(?i)^\s*Process\s+ID\s*:\s*(\d+)\s*$") +RULE_KEYS: tuple[str, ...] = () + def normalize_isis_adjacency( *, raw_text: str, + fsm_tables=None, vendor: str = "", device_type: str = "", command: str = "", params: dict[str, str] | None = None, ) -> list[dict[str, Any]]: - _ = (vendor, device_type, command, params) + _ = (vendor, device_type, command, params, fsm_tables) out: list[dict[str, Any]] = [] process_id = "" for raw in str(raw_text or "").splitlines(): @@ -54,3 +57,5 @@ def normalize_isis_adjacency( } ) return out + +normalize_isis_adjacency.RULE_KEYS = RULE_KEYS diff --git a/netx_api/biz_state/parsers/zte/nd6_cache.py b/netx_api/biz_state/parsers/zte/nd6_cache.py index 3871cc8..91c3fcc 100644 --- a/netx_api/biz_state/parsers/zte/nd6_cache.py +++ b/netx_api/biz_state/parsers/zte/nd6_cache.py @@ -11,16 +11,19 @@ _ND6_ROW_RE = re.compile( re.I, ) +RULE_KEYS: tuple[str, ...] = () + def normalize_nd6_cache( *, raw_text: str, + fsm_tables=None, vendor: str = "", device_type: str = "", command: str = "", params: dict[str, str] | None = None, ) -> list[dict[str, Any]]: - _ = (vendor, device_type, command, params) + _ = (vendor, device_type, command, params, fsm_tables) out: list[dict[str, Any]] = [] seen: set[tuple[str, str]] = set() for raw in str(raw_text or "").splitlines(): @@ -52,3 +55,5 @@ def normalize_nd6_cache( } ) return out + +normalize_nd6_cache.RULE_KEYS = RULE_KEYS diff --git a/netx_api/ntc_parse.py b/netx_api/ntc_parse.py index 6fa35d3..6aa8f55 100644 --- a/netx_api/ntc_parse.py +++ b/netx_api/ntc_parse.py @@ -3,15 +3,21 @@ CliTable/ParseCmd is not safe to share across threads. Each call builds a fresh CliTable under a process-wide lock. Template files on disk are read-only and safe for multi-process workers. + +Rule pipeline (biz_state): + apply_rule / apply_rules — run TextFSM by template stem (filename without .textfsm) + rules_for_command — list stems matching platform + CLI command via index """ from __future__ import annotations import logging +import re import threading +from dataclasses import dataclass from functools import lru_cache from pathlib import Path -from typing import Any, Callable, TypeVar +from typing import Any, Callable, Sequence, TypeVar from .cli_wrap import apply_cli_wrap @@ -103,6 +109,238 @@ def row_get(row: dict[str, Any], *names: str) -> str: return "" +@dataclass(frozen=True) +class _IndexEntry: + stem: str + rel_path: str + platform: str + command_pat: str + template_dir: Path + + +def _stem_of_template_path(rel: str) -> str: + name = Path(str(rel or "").replace("\\", "/")).name + if name.lower().endswith(".textfsm"): + name = name[: -len(".textfsm")] + return name + + +def _parse_index_file(template_dir: Path) -> list[_IndexEntry]: + index_path = template_dir / "index" + if not index_path.is_file(): + return [] + out: list[_IndexEntry] = [] + try: + lines = index_path.read_text(encoding="utf-8", errors="ignore").splitlines() + except Exception: + _log.debug("read index failed dir=%s", template_dir, exc_info=True) + return [] + for line in lines: + s = line.strip() + if not s or s.startswith("#"): + continue + # Template, Hostname, Platform, Command + parts = [p.strip() for p in s.split(",")] + if len(parts) < 4: + continue + if parts[0].lower() == "template": + continue + rel, _host, platform, cmd_pat = parts[0], parts[1], parts[2], ",".join(parts[3:]).strip() + stem = _stem_of_template_path(rel) + if not stem or not platform: + continue + out.append( + _IndexEntry( + stem=stem, + rel_path=rel.replace("\\", "/"), + platform=str(platform).strip(), + command_pat=cmd_pat, + template_dir=template_dir, + ) + ) + return out + + +@lru_cache(maxsize=4) +def _all_index_entries(include_community: bool = True) -> tuple[_IndexEntry, ...]: + entries = list(_parse_index_file(CUSTOM_TEMPLATE_DIR)) + if include_community: + try: + entries.extend(_parse_index_file(Path(_community_template_dir()))) + except Exception: + _log.debug("community index unavailable", exc_info=True) + return tuple(entries) + + +def _index_cmd_to_regex(pat: str) -> re.Pattern[str] | None: + """Convert ntc index Command column (sh[[ow]] foo) to an anchored regex.""" + raw = str(pat or "").strip() + if not raw: + return None + out: list[str] = [r"^\s*"] + i = 0 + while i < len(raw): + if raw.startswith("[[", i): + i += 2 + continue + m = re.match(r"([A-Za-z0-9_.|/\-]+)(?:\[\[([A-Za-z0-9_.|/\-]+)\]\])?", raw[i:]) + if m: + base, opt = m.group(1), m.group(2) + if opt: + out.append(re.escape(base) + r"(?:" + re.escape(opt) + r")?") + else: + out.append(re.escape(base)) + i += m.end() + ws = re.match(r"\s+", raw[i:]) + if ws: + out.append(r"\s+") + i += ws.end() + continue + ch = raw[i] + if ch in ".$^?*+()[]{}|\\": + out.append(re.escape(ch)) + else: + out.append(re.escape(ch)) + i += 1 + out.append(r"\s*$") + try: + return re.compile("".join(out), re.I) + except re.error: + return None + + +def _command_matches_index(pat: str, command: str) -> bool: + cmd = str(command or "").strip() + if not cmd: + return False + rx = _index_cmd_to_regex(pat) + if rx and rx.match(cmd): + return True + # Fallback: loose token containment after expanding [[opt]] + flat = re.sub(r"\[\[([^\]]+)\]\]", r"\1", pat) + flat = re.sub(r"\s+", " ", flat).strip().lower() + norm = re.sub(r"\s+", " ", cmd).strip().lower() + return bool(flat) and flat in norm + + +def rules_for_command(platform: str, command: str) -> list[str]: + """Return TextFSM template stems matching platform + CLI command (index order).""" + plat = str(platform or "").strip() + cmd = str(command or "").strip() + if not plat or not cmd: + return [] + seen: set[str] = set() + out: list[str] = [] + for e in _all_index_entries(True): + if e.platform != plat: + continue + if not _command_matches_index(e.command_pat, cmd): + continue + if e.stem in seen: + continue + seen.add(e.stem) + out.append(e.stem) + return out + + +def _find_entries_for_stem(rule_key: str) -> list[_IndexEntry]: + stem = str(rule_key or "").strip() + if stem.lower().endswith(".textfsm"): + stem = stem[: -len(".textfsm")] + if not stem: + return [] + return [e for e in _all_index_entries(True) if e.stem == stem] + + +def _parse_textfsm_file(template_path: Path, text: str) -> list[dict[str, Any]]: + try: + from textfsm import TextFSM + except Exception: + _log.debug("textfsm unavailable", exc_info=True) + return [] + try: + with template_path.open("r", encoding="utf-8", errors="ignore") as fh: + fsm = TextFSM(fh) + rows = fsm.ParseTextToDicts(str(text or "")) + except Exception: + _log.debug("textfsm parse failed path=%s", template_path, exc_info=True) + return [] + if not isinstance(rows, list): + return [] + # Match ntc-templates parse_output: lowercase dict keys. + out: list[dict[str, Any]] = [] + for r in rows: + if not isinstance(r, dict): + continue + out.append({str(k).lower(): v for k, v in r.items()}) + return out + + +def apply_rule(*, platform: str, rule_key: str, text: str, command: str = "") -> list[dict[str, Any]]: + """Run one TextFSM rule by template stem; empty list if missing/failed.""" + plat = str(platform or "").strip() + key = str(rule_key or "").strip() + raw = str(text or "") + if not key or not raw.strip(): + return [] + wrap_cmd = str(command or "").strip() or key.replace("_", " ") + if plat: + raw = apply_cli_wrap(raw, platform=plat, command=wrap_cmd) + + entries = _find_entries_for_stem(key) + if plat: + plat_entries = [e for e in entries if e.platform == plat] + if plat_entries: + entries = plat_entries + with _PARSE_LOCK: + for e in entries: + path = e.template_dir / e.rel_path + if not path.is_file(): + alt = e.template_dir / Path(e.rel_path).name + path = alt if alt.is_file() else path + if not path.is_file(): + continue + rows = _parse_textfsm_file(path, raw) + if rows: + return rows + # Fallback: search directories by filename even if not in index + fname = f"{key}.textfsm" if not key.endswith(".textfsm") else key + try: + community = Path(_community_template_dir()) + except Exception: + community = None + roots = [CUSTOM_TEMPLATE_DIR] + ([community] if community else []) + for base in roots: + if not base or not base.is_dir(): + continue + for path in base.rglob(fname): + rows = _parse_textfsm_file(path, raw) + if rows: + return rows + return [] + + +def apply_rules( + *, + platform: str, + text: str, + rule_keys: Sequence[str], + command: str = "", +) -> dict[str, list[dict[str, Any]]]: + """Run multiple TextFSM rules → ``{stem: rows}``. Missing/failed keys map to ``[]``.""" + out: dict[str, list[dict[str, Any]]] = {} + for key in rule_keys or (): + k = str(key or "").strip() + if not k: + continue + if k.lower().endswith(".textfsm"): + k = k[: -len(".textfsm")] + if k in out: + continue + out[k] = apply_rule(platform=platform, rule_key=k, text=text, command=command) + return out + + def parse_cli( *, platform: str, diff --git a/tests/test_fsm_parser_pipeline.py b/tests/test_fsm_parser_pipeline.py new file mode 100644 index 0000000..2b057b6 --- /dev/null +++ b/tests/test_fsm_parser_pipeline.py @@ -0,0 +1,97 @@ +"""Tests for TextFSM rule table + parser dual-input pipeline.""" + +from __future__ import annotations + +import unittest + +from netx_api.biz_state.parsers import get_parser_meta, run_parser +from netx_api.ntc_parse import apply_rule, apply_rules, rules_for_command + + +ZTE_BRIEF = """\ +Interface Attribute Mode BW Admin Phy Prot Description +xgei-1/1/0/1 optical Duplex/full 1G up up up C2930L100-EQ2 +xgei-1/1/0/2 optical Duplex/full 1G up down down +""" + +ZTE_ARP = """\ +IP Age Hardware address Interface Exter Interface VPN name +192.168.1.1 03:22:07 0011.2233.4455 gei-0/1/0/1 N/A --- +192.168.1.2 H 00aa.bbcc.ddee gei-0/1/0/1 N/A --- +""" + + +class ApplyRulesTests(unittest.TestCase): + def test_apply_rules_interface_brief(self) -> None: + stems = rules_for_command("zte_zxros", "show interface brief") + self.assertIn("zte_zxros_show_interface_brief", stems) + tables = apply_rules( + platform="zte_zxros", + text=ZTE_BRIEF, + rule_keys=("zte_zxros_show_interface_brief",), + command="show interface brief", + ) + rows = tables.get("zte_zxros_show_interface_brief") or [] + self.assertGreaterEqual(len(rows), 2) + self.assertEqual(rows[0].get("INTERFACE") or rows[0].get("interface"), "xgei-1/1/0/1") + + def test_apply_rule_missing_stem(self) -> None: + rows = apply_rule( + platform="zte_zxros", + rule_key="zte_zxros_show_arp_does_not_exist", + text=ZTE_ARP, + ) + self.assertEqual(rows, []) + + def test_apply_rules_empty_key_on_miss(self) -> None: + tables = apply_rules( + platform="zte_zxros", + text=ZTE_ARP, + rule_keys=("no_such_template_xyz",), + ) + self.assertEqual(tables.get("no_such_template_xyz"), []) + + +class ParserMetaTests(unittest.TestCase): + def test_interface_brief_meta(self) -> None: + meta = get_parser_meta("interface_brief") + self.assertIsNotNone(meta) + assert meta is not None + self.assertIn("zte_zxros_show_interface_brief", meta["rule_keys"]) + + def test_arp_empty_rule_keys(self) -> None: + meta = get_parser_meta("arp") + self.assertIsNotNone(meta) + assert meta is not None + self.assertEqual(meta["rule_keys"], ()) + + def test_run_parser_interface_brief(self) -> None: + records, tables, keys = run_parser( + "interface_brief", + raw_text=ZTE_BRIEF, + vendor="zte", + device_type="zte_zxros", + command="show interface brief", + ) + self.assertIn("zte_zxros_show_interface_brief", keys) + self.assertGreaterEqual(len(tables.get("zte_zxros_show_interface_brief") or []), 2) + self.assertGreaterEqual(len(records), 2) + self.assertEqual(records[0]["interface"], "xgei-1/1/0/1") + + def test_run_parser_arp_hand_only(self) -> None: + # Declared empty RULE_KEYS → no index auto-bind; hand parse only. + records, tables, keys = run_parser( + "arp", + raw_text=ZTE_ARP, + vendor="zte", + device_type="zte_zxros", + command="show arp", + ) + self.assertEqual(keys, []) + self.assertEqual(tables, {}) + self.assertGreaterEqual(len(records), 1) + self.assertTrue(any(r["ip"] == "192.168.1.1" for r in records)) + + +if __name__ == "__main__": + unittest.main()