mirror of
https://github.com/hansjone/netx.git
synced 2026-10-09 05:30:46 +08:00
oliver
214bc56774
Split the ops wall to its own route with fullscreen charts, persist collect errors as device events, and polish wall legend layout. Co-authored-by: Cursor <cursoragent@cursor.com>
581 lines
17 KiB
Python
581 lines
17 KiB
Python
"""Parsers for ZTE / Huawei / Cisco interface brief & detail (rate bit/s)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
|
|
_BW_UNIT = {
|
|
"k": 1_000,
|
|
"m": 1_000_000,
|
|
"g": 1_000_000_000,
|
|
"t": 1_000_000_000_000,
|
|
}
|
|
|
|
_RE_BW_COMPACT = re.compile(r"^(\d+(?:\.\d+)?)\s*([kKmMgGtT])(?:bit)?s?$", re.I)
|
|
# ZTE "BW 1 Gbit/s" and Cisco "BW 1000000 Kbit/sec"
|
|
_RE_BW_DETAIL = re.compile(
|
|
r"\bBW\s+(\d+(?:\.\d+)?)\s*([kKmMgGtT])?\s*(?:G?bit|bit)/s(?:ec)?\b",
|
|
re.I,
|
|
)
|
|
_RE_RATE_PERIOD = re.compile(r"Rate\s+period\s*:\s*(\d+)\s*s", re.I)
|
|
_RE_INPUT_BPS = re.compile(r"^\s*Input\s*:\s*([\d.]+)\s*bit/s", re.I | re.M)
|
|
_RE_OUTPUT_BPS = re.compile(r"^\s*Output\s*:\s*([\d.]+)\s*bit/s", re.I | re.M)
|
|
_RE_UTIL = re.compile(
|
|
r"Intf\s+utilization\s*:\s*input\s*([\d.]+)%\s*output\s*([\d.]+)%",
|
|
re.I,
|
|
)
|
|
_RE_IF_UP = re.compile(r"^(\S+)\s+is\s+(up|down)\b", re.I | re.M)
|
|
_RE_DESC = re.compile(r"^\s*Description:\s*(.+?)\s*$", re.I | re.M)
|
|
_RE_PROMPT_LINE = re.compile(r"[#>]\s*$")
|
|
_RE_UPDOWN = re.compile(r"^(up|down)$", re.I)
|
|
_RE_UPDOWN_TOKEN = re.compile(r"^(up|down)\b", re.I)
|
|
# ZTE / Huawei / Cisco / common logical iface names.
|
|
_RE_IFNAME = re.compile(
|
|
r"^(?:"
|
|
r"xxvgei|xgei|cgei|gei|fei|qli|smartgroup|bvi|vlan|loopback|mgmt|"
|
|
r"null|pos|atm|tunnel|irb|pw|eth|ethernet|port-channel|bundle|"
|
|
r"gigabitethernet|xgigabitethernet|fastethernet|tengigabitethernet|"
|
|
r"hundredgige|fivegige|fortygige|serial|dialer|cellular|multilink|"
|
|
r"10ge|25ge|40ge|100ge|eth-trunk|vlanif|meth|loopback"
|
|
r")[\w./:-]*$",
|
|
re.I,
|
|
)
|
|
_RE_CISCO_BRIEF_ROW = re.compile(
|
|
r"^(\S+)\s+(\S+)\s+(YES|NO)\s+(\S+)\s+(.+?)\s+(up|down)\s*$",
|
|
re.I,
|
|
)
|
|
_RE_CISCO_IF_STATE = re.compile(
|
|
r"^(\S+)\s+is\s+(administratively\s+)?(up|down),\s*line\s+protocol\s+is\s+(up|down)\b",
|
|
re.I | re.M,
|
|
)
|
|
_RE_CISCO_RATE = re.compile(
|
|
r"(\d+)\s+(second|minute|hour)s?\s+(input|output)\s+rate\s+([\d.]+)\s*bits/sec",
|
|
re.I,
|
|
)
|
|
_RE_HW_BRIEF_ROW = re.compile(
|
|
r"^(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\d+)\s+(\d+)\s*$"
|
|
)
|
|
_RE_HW_STATE = re.compile(
|
|
r"^(\S+)\s+current\s+state\s*:\s*(UP|DOWN|Administratively\s+DOWN)\b",
|
|
re.I | re.M,
|
|
)
|
|
_RE_HW_IN_RATE = re.compile(
|
|
r"Last\s+(\d+)\s+seconds\s+input\s+rate\s*:\s*([\d.]+)\s*bits/sec",
|
|
re.I,
|
|
)
|
|
_RE_HW_OUT_RATE = re.compile(
|
|
r"Last\s+(\d+)\s+seconds\s+output\s+rate\s*:\s*([\d.]+)\s*bits/sec",
|
|
re.I,
|
|
)
|
|
_RE_HW_IN_UTIL = re.compile(
|
|
r"Last\s+(\d+)\s+seconds\s+input\s+utility\s+rate\s*:\s*([\d.]+)\s*%",
|
|
re.I,
|
|
)
|
|
_RE_HW_OUT_UTIL = re.compile(
|
|
r"Last\s+(\d+)\s+seconds\s+output\s+utility\s+rate\s*:\s*([\d.]+)\s*%",
|
|
re.I,
|
|
)
|
|
|
|
# Fixed-width columns from ZTE `show interface brief` header.
|
|
_COL_IF = (0, 24)
|
|
_COL_ATTR = (24, 35)
|
|
_COL_MODE = (35, 48)
|
|
_COL_BW = (48, 54)
|
|
_COL_ADMIN = (54, 60)
|
|
_COL_PHY = (60, 66)
|
|
_COL_PROT = (66, 72)
|
|
_COL_DESC = (72, None)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BriefPort:
|
|
ifname: str
|
|
attribute: str = ""
|
|
mode: str = ""
|
|
bw_raw: str = ""
|
|
bw_bps: int = 0
|
|
admin: str = ""
|
|
phy: str = ""
|
|
prot: str = ""
|
|
description: str = ""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DetailRates:
|
|
ifname: str = ""
|
|
admin_oper: str = ""
|
|
description: str = ""
|
|
bw_bps: int = 0
|
|
rate_period_sec: int = 0
|
|
in_bps: float = 0.0
|
|
out_bps: float = 0.0
|
|
in_util_pct: float = 0.0
|
|
out_util_pct: float = 0.0
|
|
|
|
|
|
def resolve_util_pct(vendor_util: float, bps: float, bw_bps: int) -> float:
|
|
"""Prefer vendor-reported util; derive from rate/BW when util is missing but traffic exists."""
|
|
u = float(vendor_util or 0.0)
|
|
if u > 0:
|
|
return u
|
|
bw = int(bw_bps or 0)
|
|
rate = float(bps or 0.0)
|
|
if bw > 0 and rate > 0:
|
|
return rate / float(bw) * 100.0
|
|
return u
|
|
|
|
|
|
def parse_bw_to_bps(raw: str) -> int:
|
|
text = (raw or "").strip()
|
|
if not text or text.upper() == "N/A":
|
|
return 0
|
|
m = _RE_BW_COMPACT.match(text.replace(" ", ""))
|
|
if m:
|
|
value = float(m.group(1))
|
|
mult = _BW_UNIT[m.group(2).lower()]
|
|
return int(value * mult)
|
|
m2 = _RE_BW_DETAIL.search(text)
|
|
if m2:
|
|
value = float(m2.group(1))
|
|
unit = (m2.group(2) or "g").lower()
|
|
# "BW 1 Gbit/s" — unit letter may be in "Gbit" when group2 empty after "1 "
|
|
if m2.group(2) is None and "gbit" in text.lower():
|
|
unit = "g"
|
|
elif m2.group(2) is None and "mbit" in text.lower():
|
|
unit = "m"
|
|
elif m2.group(2) is None and "kbit" in text.lower():
|
|
unit = "k"
|
|
mult = _BW_UNIT.get(unit, 1_000_000_000)
|
|
return int(value * mult)
|
|
# Plain "BW 1000000000" unlikely; try digits only
|
|
digits = re.sub(r"[^\d.]", "", text)
|
|
if digits:
|
|
try:
|
|
return int(float(digits))
|
|
except ValueError:
|
|
return 0
|
|
return 0
|
|
|
|
|
|
def _slice(line: str, start: int, end: int | None) -> str:
|
|
if end is None:
|
|
return line[start:].rstrip() if len(line) > start else ""
|
|
if len(line) <= start:
|
|
return ""
|
|
return line[start:end].strip()
|
|
|
|
|
|
def _looks_like_brief_port(ifname: str, admin: str, phy: str, prot: str) -> bool:
|
|
"""Reject prompts / hostname lines that leak after the brief table."""
|
|
name = (ifname or "").strip()
|
|
if not name or name.lower() == "interface":
|
|
return False
|
|
if "#" in name or ">" in name:
|
|
return False
|
|
if _RE_PROMPT_LINE.search(name):
|
|
return False
|
|
# Real brief rows always have Admin/Phy/Prot as up|down.
|
|
if not (
|
|
_RE_UPDOWN.match(admin or "")
|
|
and _RE_UPDOWN.match(phy or "")
|
|
and _RE_UPDOWN.match(prot or "")
|
|
):
|
|
return False
|
|
# Prefer known iface prefixes; still allow smartgroup/bvi style names.
|
|
if not _RE_IFNAME.match(name):
|
|
return False
|
|
return True
|
|
|
|
|
|
def parse_zte_interface_brief(text: str) -> list[BriefPort]:
|
|
"""Parse ZTE `show interface brief` into port rows."""
|
|
lines = (text or "").replace("\r\n", "\n").replace("\r", "\n").split("\n")
|
|
out: list[BriefPort] = []
|
|
started = False
|
|
for raw in lines:
|
|
line = raw.rstrip()
|
|
if not line.strip():
|
|
continue
|
|
if line.lstrip().startswith("Interface") and "Admin" in line and "Description" in line:
|
|
started = True
|
|
continue
|
|
if not started:
|
|
continue
|
|
stripped = line.strip()
|
|
# Trailing device prompt after the table (Netmiko often leaves hostname#).
|
|
if "#" in stripped or (stripped.endswith(">") and " " not in stripped):
|
|
admin_probe = _slice(line, *_COL_ADMIN).lower()
|
|
if not _RE_UPDOWN.match(admin_probe):
|
|
break
|
|
if " is " in line and "ifindex" in line.lower():
|
|
break
|
|
if_field = _slice(line, *_COL_IF)
|
|
ifname = if_field.split()[0] if if_field else ""
|
|
admin = _slice(line, *_COL_ADMIN).lower()
|
|
phy = _slice(line, *_COL_PHY).lower()
|
|
prot = _slice(line, *_COL_PROT).lower()
|
|
if not _looks_like_brief_port(ifname, admin, phy, prot):
|
|
continue
|
|
bw_raw = _slice(line, *_COL_BW)
|
|
out.append(
|
|
BriefPort(
|
|
ifname=ifname,
|
|
attribute=_slice(line, *_COL_ATTR),
|
|
mode=_slice(line, *_COL_MODE),
|
|
bw_raw=bw_raw,
|
|
bw_bps=parse_bw_to_bps(bw_raw),
|
|
admin=admin,
|
|
phy=phy,
|
|
prot=prot,
|
|
description=_slice(line, *_COL_DESC).strip(),
|
|
)
|
|
)
|
|
return out
|
|
|
|
|
|
def parse_zte_interface_detail(text: str) -> DetailRates:
|
|
"""Parse ZTE `show interface {ifname}` rate / util / BW."""
|
|
blob = text or ""
|
|
ifname = ""
|
|
admin_oper = ""
|
|
m_if = _RE_IF_UP.search(blob)
|
|
if m_if:
|
|
ifname = m_if.group(1)
|
|
admin_oper = m_if.group(2).lower()
|
|
desc = ""
|
|
m_desc = _RE_DESC.search(blob)
|
|
if m_desc:
|
|
desc = m_desc.group(1).strip()
|
|
|
|
bw_bps = 0
|
|
m_bw = _RE_BW_DETAIL.search(blob)
|
|
if m_bw:
|
|
bw_bps = parse_bw_to_bps(m_bw.group(0))
|
|
else:
|
|
# Fallback line scan
|
|
for line in blob.splitlines():
|
|
if re.search(r"\bBW\b", line, re.I):
|
|
bw_bps = parse_bw_to_bps(line)
|
|
if bw_bps:
|
|
break
|
|
|
|
rate_period = 0
|
|
m_rp = _RE_RATE_PERIOD.search(blob)
|
|
if m_rp:
|
|
rate_period = int(m_rp.group(1))
|
|
|
|
# Prefer Rate period block: first Input/Output after "Rate period"
|
|
in_bps = 0.0
|
|
out_bps = 0.0
|
|
rp_idx = blob.lower().find("rate period")
|
|
rate_blob = blob[rp_idx:] if rp_idx >= 0 else blob
|
|
# Stop before Peak rate to avoid peak Input/Output
|
|
peak_idx = rate_blob.lower().find("peak rate")
|
|
if peak_idx >= 0:
|
|
rate_blob = rate_blob[:peak_idx]
|
|
m_in = _RE_INPUT_BPS.search(rate_blob)
|
|
m_out = _RE_OUTPUT_BPS.search(rate_blob)
|
|
if m_in:
|
|
in_bps = float(m_in.group(1))
|
|
if m_out:
|
|
out_bps = float(m_out.group(1))
|
|
|
|
in_util = 0.0
|
|
out_util = 0.0
|
|
m_util = _RE_UTIL.search(blob)
|
|
if m_util:
|
|
in_util = float(m_util.group(1))
|
|
out_util = float(m_util.group(2))
|
|
|
|
return DetailRates(
|
|
ifname=ifname,
|
|
admin_oper=admin_oper,
|
|
description=desc,
|
|
bw_bps=bw_bps,
|
|
rate_period_sec=rate_period,
|
|
in_bps=in_bps,
|
|
out_bps=out_bps,
|
|
in_util_pct=in_util,
|
|
out_util_pct=out_util,
|
|
)
|
|
|
|
|
|
def brief_port_to_dict(row: BriefPort) -> dict[str, Any]:
|
|
return {
|
|
"ifname": row.ifname,
|
|
"attribute": row.attribute,
|
|
"mode": row.mode,
|
|
"bw_raw": row.bw_raw,
|
|
"bw_bps": row.bw_bps,
|
|
"admin": row.admin,
|
|
"phy": row.phy,
|
|
"prot": row.prot,
|
|
"description": row.description,
|
|
}
|
|
|
|
|
|
def detail_to_dict(row: DetailRates) -> dict[str, Any]:
|
|
return {
|
|
"ifname": row.ifname,
|
|
"admin_oper": row.admin_oper,
|
|
"description": row.description,
|
|
"bw_bps": row.bw_bps,
|
|
"rate_period_sec": row.rate_period_sec,
|
|
"in_bps": row.in_bps,
|
|
"out_bps": row.out_bps,
|
|
"in_util_pct": row.in_util_pct,
|
|
"out_util_pct": row.out_util_pct,
|
|
}
|
|
|
|
|
|
def _norm_updown(token: str) -> str:
|
|
"""Normalize Huawei tokens like up(s) / *down to up|down|''."""
|
|
text = (token or "").strip().lower()
|
|
if text.startswith("*"):
|
|
text = text[1:]
|
|
m = _RE_UPDOWN_TOKEN.match(text)
|
|
return m.group(1).lower() if m else ""
|
|
|
|
|
|
def parse_huawei_interface_brief(text: str) -> list[BriefPort]:
|
|
"""Parse Huawei `display interface brief` into port rows (BW ignored / 0)."""
|
|
lines = (text or "").replace("\r\n", "\n").replace("\r", "\n").split("\n")
|
|
out: list[BriefPort] = []
|
|
started = False
|
|
for raw in lines:
|
|
line = raw.rstrip()
|
|
if not line.strip():
|
|
continue
|
|
low = line.lstrip().lower()
|
|
if low.startswith("interface") and "phy" in low and ("inuti" in low or "protocol" in low):
|
|
started = True
|
|
continue
|
|
if not started:
|
|
continue
|
|
stripped = line.strip()
|
|
if stripped.startswith("<") and stripped.endswith(">"):
|
|
break
|
|
if "#" in stripped and " " not in stripped.split("#", 1)[0]:
|
|
break
|
|
m = _RE_HW_BRIEF_ROW.match(stripped)
|
|
if not m:
|
|
continue
|
|
ifname = m.group(1)
|
|
phy = _norm_updown(m.group(2))
|
|
prot = _norm_updown(m.group(3))
|
|
if not ifname or ifname.lower() == "interface":
|
|
continue
|
|
if "#" in ifname or ">" in ifname:
|
|
continue
|
|
if not _RE_IFNAME.match(ifname):
|
|
continue
|
|
if not phy or not prot:
|
|
continue
|
|
out.append(
|
|
BriefPort(
|
|
ifname=ifname,
|
|
attribute="",
|
|
mode="",
|
|
bw_raw="",
|
|
bw_bps=0,
|
|
admin=phy, # Huawei brief has no separate Admin; PHY is closest.
|
|
phy=phy,
|
|
prot=prot,
|
|
description="",
|
|
)
|
|
)
|
|
return out
|
|
|
|
|
|
def parse_huawei_interface_detail(text: str) -> DetailRates:
|
|
"""Parse Huawei `display interface {if}` Last N seconds rate / utility."""
|
|
blob = text or ""
|
|
ifname = ""
|
|
admin_oper = ""
|
|
m_state = _RE_HW_STATE.search(blob)
|
|
if m_state:
|
|
ifname = m_state.group(1)
|
|
st = m_state.group(2).lower()
|
|
admin_oper = "down" if "down" in st else "up"
|
|
desc = ""
|
|
m_desc = _RE_DESC.search(blob)
|
|
if m_desc:
|
|
desc = m_desc.group(1).strip()
|
|
|
|
rate_period = 0
|
|
in_bps = 0.0
|
|
out_bps = 0.0
|
|
m_in = _RE_HW_IN_RATE.search(blob)
|
|
if m_in:
|
|
rate_period = int(m_in.group(1))
|
|
in_bps = float(m_in.group(2))
|
|
m_out = _RE_HW_OUT_RATE.search(blob)
|
|
if m_out:
|
|
if not rate_period:
|
|
rate_period = int(m_out.group(1))
|
|
out_bps = float(m_out.group(2))
|
|
|
|
in_util = 0.0
|
|
out_util = 0.0
|
|
m_iu = _RE_HW_IN_UTIL.search(blob)
|
|
if m_iu:
|
|
if not rate_period:
|
|
rate_period = int(m_iu.group(1))
|
|
in_util = float(m_iu.group(2))
|
|
m_ou = _RE_HW_OUT_UTIL.search(blob)
|
|
if m_ou:
|
|
if not rate_period:
|
|
rate_period = int(m_ou.group(1))
|
|
out_util = float(m_ou.group(2))
|
|
|
|
return DetailRates(
|
|
ifname=ifname,
|
|
admin_oper=admin_oper,
|
|
description=desc,
|
|
bw_bps=0, # sample has no BW; leave 0 for now
|
|
rate_period_sec=rate_period,
|
|
in_bps=in_bps,
|
|
out_bps=out_bps,
|
|
in_util_pct=in_util,
|
|
out_util_pct=out_util,
|
|
)
|
|
|
|
|
|
def parse_interface_brief(text: str, vendor_key: str = "zte") -> list[BriefPort]:
|
|
key = str(vendor_key or "zte").strip().lower()
|
|
if key == "huawei":
|
|
return parse_huawei_interface_brief(text)
|
|
if key == "cisco":
|
|
return parse_cisco_interface_brief(text)
|
|
return parse_zte_interface_brief(text)
|
|
|
|
|
|
def parse_interface_detail(text: str, vendor_key: str = "zte") -> DetailRates:
|
|
key = str(vendor_key or "zte").strip().lower()
|
|
if key == "huawei":
|
|
return parse_huawei_interface_detail(text)
|
|
if key == "cisco":
|
|
return parse_cisco_interface_detail(text)
|
|
return parse_zte_interface_detail(text)
|
|
|
|
|
|
def _cisco_period_to_sec(n: int, unit: str) -> int:
|
|
u = (unit or "").lower()
|
|
if u.startswith("second"):
|
|
return int(n)
|
|
if u.startswith("minute"):
|
|
return int(n) * 60
|
|
if u.startswith("hour"):
|
|
return int(n) * 3600
|
|
return int(n)
|
|
|
|
|
|
def parse_cisco_interface_brief(text: str) -> list[BriefPort]:
|
|
"""Parse Cisco `show ip interface brief` into port rows."""
|
|
lines = (text or "").replace("\r\n", "\n").replace("\r", "\n").split("\n")
|
|
out: list[BriefPort] = []
|
|
started = False
|
|
for raw in lines:
|
|
line = raw.rstrip()
|
|
if not line.strip():
|
|
continue
|
|
low = line.lstrip().lower()
|
|
if low.startswith("interface") and "status" in low and "protocol" in low:
|
|
started = True
|
|
continue
|
|
if not started:
|
|
continue
|
|
stripped = line.strip()
|
|
if stripped.endswith("#") or (stripped.endswith(">") and stripped.startswith("<")):
|
|
break
|
|
if "#" in stripped and " " not in stripped.split("#", 1)[0]:
|
|
break
|
|
m = _RE_CISCO_BRIEF_ROW.match(stripped)
|
|
if not m:
|
|
continue
|
|
ifname = m.group(1)
|
|
status = (m.group(5) or "").strip().lower()
|
|
prot = (m.group(6) or "").strip().lower()
|
|
if not _RE_IFNAME.match(ifname):
|
|
continue
|
|
if prot not in ("up", "down"):
|
|
continue
|
|
admin_down = "administratively" in status
|
|
if "down" in status:
|
|
phy = "down"
|
|
elif "up" in status:
|
|
phy = "up"
|
|
else:
|
|
continue
|
|
admin = "down" if admin_down else phy
|
|
out.append(
|
|
BriefPort(
|
|
ifname=ifname,
|
|
attribute="",
|
|
mode="",
|
|
bw_raw="",
|
|
bw_bps=0,
|
|
admin=admin,
|
|
phy=phy,
|
|
prot=prot,
|
|
description="",
|
|
)
|
|
)
|
|
return out
|
|
|
|
|
|
def parse_cisco_interface_detail(text: str) -> DetailRates:
|
|
"""Parse Cisco `show interfaces {if}` BW + N minute/second rate (util ignored)."""
|
|
blob = text or ""
|
|
ifname = ""
|
|
admin_oper = ""
|
|
m_state = _RE_CISCO_IF_STATE.search(blob)
|
|
if m_state:
|
|
ifname = m_state.group(1)
|
|
admin_oper = "down" if m_state.group(2) or m_state.group(3).lower() == "down" else "up"
|
|
if not ifname:
|
|
m_up = _RE_IF_UP.search(blob)
|
|
if m_up:
|
|
ifname = m_up.group(1)
|
|
admin_oper = m_up.group(2).lower()
|
|
|
|
desc = ""
|
|
m_desc = _RE_DESC.search(blob)
|
|
if m_desc:
|
|
desc = m_desc.group(1).strip()
|
|
|
|
bw_bps = 0
|
|
for line in blob.splitlines():
|
|
if "BW" in line.upper() and ("bit" in line.lower()):
|
|
bw_bps = parse_bw_to_bps(line)
|
|
if bw_bps:
|
|
break
|
|
|
|
rate_period = 0
|
|
in_bps = 0.0
|
|
out_bps = 0.0
|
|
for m in _RE_CISCO_RATE.finditer(blob):
|
|
period = _cisco_period_to_sec(int(m.group(1)), m.group(2))
|
|
direction = m.group(3).lower()
|
|
rate = float(m.group(4))
|
|
if not rate_period:
|
|
rate_period = period
|
|
if direction == "input":
|
|
in_bps = rate
|
|
else:
|
|
out_bps = rate
|
|
|
|
return DetailRates(
|
|
ifname=ifname,
|
|
admin_oper=admin_oper,
|
|
description=desc,
|
|
bw_bps=bw_bps,
|
|
rate_period_sec=rate_period,
|
|
in_bps=in_bps,
|
|
out_bps=out_bps,
|
|
in_util_pct=0.0, # sample has no util %; ignore for now
|
|
out_util_pct=0.0,
|
|
)
|