sam
dfdbd85bf7
SNMP walk parser (snmp-parse.py) with MIB resolution, structured LLDP neighbor extraction, IP address parsing, and comprehensive table reconstruction for Accedian AMN-1000-GT-S NIDs. HTML viewer generator (build_nid_viewer.py) with dark-themed dashboard including LLDP topology diagram, SFP optics, traffic stats, alarms, port config, coverage matrix, and policy/filter/regulator sections. Includes 15 Accedian MIB files and sample walk data from 10.13.60.102. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
130 lines
4.7 KiB
Python
130 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
SNMP Walk Parser - Combined Nested JSON + Flat CSV + Basic Analysis
|
|
|
|
Reads a snmpwalk output file (net-snmp style with -On -OQ format)
|
|
Creates:
|
|
- nested JSON tree
|
|
- flat CSV with oid + value columns
|
|
- shows basic statistics and example filters
|
|
"""
|
|
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
import pandas as pd
|
|
|
|
# ────────────────────────────────────────────────
|
|
# CONFIG - change these as needed
|
|
# ────────────────────────────────────────────────
|
|
|
|
INPUT_FILE = Path("~/snmp-walks/10-13-60-102_2026-02-27_11-23-07_walk.txt").expanduser()
|
|
|
|
# Where to save results (same folder as input by default)
|
|
OUTPUT_JSON = INPUT_FILE.with_suffix('.tree.json')
|
|
OUTPUT_CSV = INPUT_FILE.with_suffix('.flat.csv')
|
|
|
|
# ────────────────────────────────────────────────
|
|
# Helper functions
|
|
# ────────────────────────────────────────────────
|
|
|
|
def parse_line(line: str) -> tuple[str, str] | None:
|
|
"""Split oid and value, clean value quotes"""
|
|
line = line.strip()
|
|
if not line or 'No more variables left' in line:
|
|
return None
|
|
|
|
# Matches: .1.3.6.1.... "value with \"escaped\" quotes" or .1.3.6.1.... 12345
|
|
match = re.match(r'^(\.\d+(?:\.\d+)*)\s+(.*)$', line)
|
|
if not match:
|
|
return None
|
|
|
|
oid, value = match.groups()
|
|
value = value.strip()
|
|
|
|
# Remove surrounding quotes if present (simple case)
|
|
if value.startswith('"') and value.endswith('"'):
|
|
value = value[1:-1]
|
|
|
|
# Optional: you could add more cleaning here (unescape \", etc.)
|
|
return oid, value
|
|
|
|
|
|
def build_nested_tree(flat_data: dict[str, str]) -> dict:
|
|
"""Convert flat oid → value into nested dictionary"""
|
|
tree = {}
|
|
for oid, value in flat_data.items():
|
|
parts = oid.lstrip('.').split('.')
|
|
current = tree
|
|
for part in parts[:-1]:
|
|
current = current.setdefault(part, {})
|
|
current[parts[-1]] = value
|
|
return tree
|
|
|
|
|
|
def main():
|
|
if not INPUT_FILE.is_file():
|
|
print(f"File not found: {INPUT_FILE}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
print(f"Reading: {INPUT_FILE}")
|
|
print(f"Output: {OUTPUT_JSON}")
|
|
print(f" {OUTPUT_CSV}\n")
|
|
|
|
# ── Parse file ───────────────────────────────────────
|
|
flat_data = {} # oid → value
|
|
skipped = 0
|
|
|
|
with INPUT_FILE.open(encoding='utf-8', errors='replace') as f:
|
|
for line in f:
|
|
result = parse_line(line)
|
|
if result:
|
|
oid, value = result
|
|
flat_data[oid] = value
|
|
else:
|
|
skipped += 1
|
|
|
|
print(f"Parsed {len(flat_data):,} OIDs (skipped {skipped:,} lines)\n")
|
|
|
|
# ── 1. Nested JSON tree ──────────────────────────────
|
|
tree = build_nested_tree(flat_data)
|
|
|
|
with OUTPUT_JSON.open('w', encoding='utf-8') as f:
|
|
json.dump(tree, f, indent=2, ensure_ascii=False)
|
|
|
|
print(f"→ Nested tree saved: {OUTPUT_JSON}")
|
|
|
|
# ── 2. Flat DataFrame + CSV ──────────────────────────
|
|
df = pd.DataFrame.from_dict(flat_data, orient='index', columns=['value'])
|
|
df.index.name = 'oid'
|
|
df = df.reset_index()
|
|
|
|
df.to_csv(OUTPUT_CSV, index=False, encoding='utf-8')
|
|
print(f"→ Flat table saved: {OUTPUT_CSV}")
|
|
|
|
# ── Quick console summary & examples ─────────────────
|
|
print("\nQuick summary:")
|
|
print(f"Total entries: {len(df):,}")
|
|
print(f"Unique OID prefixes (top 8):")
|
|
df['prefix'] = df['oid'].str.replace(r'\.\d+$', '', regex=True)
|
|
print(df['prefix'].value_counts().head(8))
|
|
|
|
print("\nExample: System group (.1.3.6.1.2.1.1)")
|
|
system = df[df['oid'].str.startswith('.1.3.6.1.2.1.1')]
|
|
if not system.empty:
|
|
print(system[['oid', 'value']].to_string(index=False))
|
|
else:
|
|
print("→ No system group entries found")
|
|
|
|
print("\nExample: Interface descriptions (.1.3.6.1.2.1.2.2.1.2)")
|
|
ifaces = df[df['oid'].str.contains(r'\.1\.3\.6\.1\.2\.1\.2\.2\.1\.2\b')]
|
|
if not ifaces.empty:
|
|
print(ifaces[['oid', 'value']].to_string(index=False))
|
|
else:
|
|
print("→ No interface descriptions found")
|
|
|
|
print("\nDone. Open the .json or .csv file in VS Code / Excel / pandas.")
|
|
|
|
if __name__ == '__main__':
|
|
main() |