Netpulse_SasS/db/import/extract_noc.py

#!/usr/bin/env python3
"""
Витяг команд збору конфігу з профілів NOC.

Файли розбираються модулем ast — без виконання. Це принципово: NOC-скрипт
імпортує половину фреймворку, і спроба його імпортувати означала б або
тягнути весь NOC, або підсовувати заглушки, які мовчки змінюють поведінку.
AST бачить рівно те, що написано в тексті.
"""
import ast
import json
import pathlib
import re
import sys
from collections import Counter

ROOT = pathlib.Path(sys.argv[1] if len(sys.argv) > 1 else ".")


def literal(node):
    """Рядковий літерал або None. f-рядки й конкатенації свідомо не беремо."""
    if isinstance(node, ast.Constant) and isinstance(node.value, str):
        return node.value
    return None


class Walker(ast.NodeVisitor):
    """Збирає виклики self.cli(...) і strip_first_lines(...) у межах методу."""

    def __init__(self):
        self.cli = []          # (команда, під_умовою_startup)
        self.strip = None
        self.snmp = False
        self.dynamic = False   # є cli() з нелітеральним аргументом
        self._startup_depth = 0

    def visit_If(self, node):
        # Гілка policy == "s" — це startup-config.
        src = ast.unparse(node.test) if hasattr(ast, "unparse") else ""
        is_startup = '"s"' in src or "'s'" in src
        if is_startup:
            self._startup_depth += 1
            for n in node.body:
                self.visit(n)
            self._startup_depth -= 1
            for n in node.orelse:
                self.visit(n)
        else:
            self.generic_visit(node)

    def visit_Call(self, node):
        fn = node.func
        name = getattr(fn, "attr", None) or getattr(fn, "id", None)

        # cli_clean — той самий виклик під іншою назвою (Ericsson).
        if name in ("cli", "cli_clean") and node.args:
            cmd = literal(node.args[0])
            if cmd is None:
                self.dynamic = True
            else:
                self.cli.append((cmd, self._startup_depth > 0))
        elif name == "strip_first_lines" and len(node.args) >= 2:
            n = node.args[1]
            if isinstance(n, ast.Constant) and isinstance(n.value, int):
                self.strip = n.value
        elif name in ("get", "set", "getnext") and getattr(fn, "value", None) is not None:
            if getattr(fn.value, "attr", "") == "snmp":
                self.snmp = True

        self.generic_visit(node)


def platform_filter(decorators):
    """Регекс платформи з @BaseScript.match(platform__regex=...)."""
    for d in decorators:
        if not isinstance(d, ast.Call):
            continue
        for kw in d.keywords or []:
            if kw.arg and "platform" in kw.arg:
                v = literal(kw.value)
                if v:
                    return v
    return None


def parse_file(path):
    try:
        tree = ast.parse(path.read_text(encoding="utf-8", errors="replace"))
    except SyntaxError as e:
        return {"error": f"синтаксис: {e}"}

    variants = []
    for cls in [n for n in ast.walk(tree) if isinstance(n, ast.ClassDef)]:
        for fn in [n for n in cls.body if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef))]:
            if not fn.name.startswith("execute"):
                continue
            w = Walker()
            for stmt in fn.body:
                w.visit(stmt)
            if not w.cli and not w.snmp:
                continue
            variants.append({
                "method": fn.name,
                "platform_regex": platform_filter(fn.decorator_list),
                "running": [c for c, s in w.cli if not s],
                "startup": [c for c, s in w.cli if s],
                "strip_first_lines": w.strip,
                "uses_snmp": w.snmp,
                "dynamic_cli": w.dynamic,
            })
    return {"variants": variants}


def main():
    files = sorted(ROOT.rglob("get_config.py"))
    out, stats = [], Counter()

    for f in files:
        rel = f.relative_to(ROOT)
        parts = rel.parts
        vendor = parts[0]
        platform = ".".join(parts[1:-1]) if len(parts) > 2 else ""

        res = parse_file(f)
        if "error" in res:
            stats["помилка розбору"] += 1
            continue

        variants = res["variants"]
        if not variants:
            stats["без команд"] += 1
            continue

        has_cmd = any(v["running"] or v["startup"] for v in variants)
        if not has_cmd:
            stats["лише snmp/динаміка"] += 1
            continue

        stats["витягнуто"] += 1
        if len(variants) > 1:
            stats["  з них із варіантами платформ"] += 1
        if any(v["dynamic_cli"] for v in variants):
            stats["  з них із нелітеральними командами"] += 1

        out.append({
            "vendor": vendor,
            "platform": platform,
            "source": str(rel).replace("\\", "/"),
            "variants": variants,
        })

    print(f"файлів get_config.py: {len(files)}", file=sys.stderr)
    for k, v in stats.most_common():
        print(f"  {k}: {v}", file=sys.stderr)

    json.dump(out, sys.stdout, ensure_ascii=False, indent=1)


if __name__ == "__main__":
    main()