netx/netx_api/biz_state/command_match.py

"""Match concrete CLI commands to ParseProfiles (longest / most specific wins)."""

from __future__ import annotations

import re
from dataclasses import dataclass
from typing import Any

from .profiles import ParseProfile, all_profiles, get_profile


@dataclass(frozen=True)
class MatchResult:
    profile: ParseProfile
    params: dict[str, str]
    matched_length: int


def normalize_command(command: str) -> str:
    text = str(command or "").strip()
    text = re.sub(r"\s+", " ", text)
    return text


def match_command(
    *,
    vendor_key: str,
    command: str,
    profiles: list[ParseProfile] | None = None,
) -> MatchResult | None:
    cmd = normalize_command(command)
    if not cmd:
        return None
    key = str(vendor_key or "").strip().lower()
    cands = profiles if profiles is not None else [
        p
        for p in all_profiles()
        if p.enabled
        and p.kind == "collect"
        and (p.vendor_key == key or p.vendor_key == "*")
    ]
    best: MatchResult | None = None
    for p in cands:
        try:
            m = re.match(p.match, cmd)
        except re.error:
            continue
        if not m:
            continue
        params = {k: str(v or "").strip() for k, v in (m.groupdict() or {}).items()}
        scored = MatchResult(profile=p, params=params, matched_length=len(p.match))
        if best is None or scored.matched_length > best.matched_length:
            best = scored
    return best


def normalize_binding_dicts(
    bindings: list[dict[str, str]] | None,
    *,
    placeholders: list | None = None,
) -> list[dict[str, str]]:
    """Accept ``{vrf: X}`` or ``{placeholder, value}`` rows; expand multi-value placeholders."""
    raw = list(bindings or [])
    converted: list[dict[str, str]] = []
    for b in raw:
        d = dict(b or {})
        if "placeholder" in d or ("name" in d and "value" in d):
            ph = str(d.get("placeholder") or d.get("name") or "").strip()
            val = str(d.get("value") or "").strip()
            if ph and val:
                converted.append({ph: val})
            continue
        converted.append({str(k): str(v).strip() for k, v in d.items() if str(v).strip()})

    phs = list(placeholders or [])
    if len(phs) == 1:
        name = str(getattr(phs[0], "name", "") or "")
        schema = str(getattr(phs[0], "schema_field", "") or name)
        out: list[dict[str, str]] = []
        for c in converted:
            val = c.get(name) or c.get(schema) or ""
            if val:
                out.append({name: val})
        return out
    return converted


def expand_from_bindings(
    *,
    profile: ParseProfile,
    bindings: list[dict[str, str]] | None = None,
    command_override: str = "",
) -> list[tuple[str, dict[str, str]]]:
    """Return list of (concrete_command, params). Reject leftover placeholders."""
    override = normalize_command(command_override)
    if override:
        if "<" in override and ">" in override:
            raise ValueError(f"command still has placeholders: {override}")
        return [(override, {})]

    tmpl = profile.command_template
    if not profile.placeholders:
        concrete = normalize_command(tmpl)
        if "<" in concrete and ">" in concrete:
            raise ValueError(f"template has placeholders but profile defines none: {concrete}")
        return [(concrete, {})]

    binds = normalize_binding_dicts(bindings, placeholders=profile.placeholders)
    if not binds:
        raise ValueError(f"profile {profile.profile_id} requires parameter bindings")

    out: list[tuple[str, dict[str, str]]] = []
    for params in binds:
        rendered = tmpl
        for ph in profile.placeholders:
            val = params.get(ph.name) or params.get(ph.schema_field) or ""
            if ph.required and not val:
                raise ValueError(f"missing placeholder {ph.name} for {profile.profile_id}")
            rendered = rendered.replace(f"<{ph.name}>", val)
        concrete = normalize_command(rendered)
        if re.search(r"<[^>]+>", concrete):
            raise ValueError(f"unresolved placeholders in: {concrete}")
        out.append((concrete, dict(params)))
    return out


def preview_task_item(
    *,
    vendor_key: str,
    profile_id: str = "",
    command: str = "",
    bindings: list[dict[str, str]] | None = None,
    kind: str = "catalog",
) -> dict[str, Any]:
    """Dry-run expand + match for UI preview."""
    if kind == "custom_raw":
        cmd = normalize_command(command)
        return {
            "ok": bool(cmd),
            "kind": "custom_raw",
            "commands": [cmd] if cmd else [],
            "parse": "skipped_custom",
            "message": "custom row: collect only, no parse",
        }

    profile = get_profile(profile_id) if profile_id else None
    if profile is None and command:
        hit = match_command(vendor_key=vendor_key, command=command)
        if hit:
            profile = hit.profile
    if profile is None:
        return {
            "ok": False,
            "kind": kind,
            "commands": [],
            "parse": "unmatched",
            "message": "no profile matched",
        }

    try:
        pairs = expand_from_bindings(
            profile=profile,
            bindings=bindings,
            command_override=command if command and command != profile.command_template else "",
        )
    except ValueError as exc:
        return {
            "ok": False,
            "kind": kind,
            "profile_id": profile.profile_id,
            "commands": [],
            "parse": "invalid",
            "message": str(exc),
        }

    previews = []
    for concrete, params in pairs:
        hit = match_command(vendor_key=vendor_key, command=concrete)
        previews.append(
            {
                "command": concrete,
                "params": params,
                "matched_profile_id": hit.profile.profile_id if hit else "",
                "metric_id": hit.profile.metric_id if hit else "",
                "parser_id": hit.profile.parser_id if hit else "",
                "match_params": hit.params if hit else {},
            }
        )
    return {
        "ok": all(bool(x["matched_profile_id"]) for x in previews),
        "kind": kind,
        "profile_id": profile.profile_id,
        "commands": previews,
        "parse": "ok" if previews and all(x["matched_profile_id"] for x in previews) else "partial",
        "message": "",
    }