knowledge-base/scripts/kb-objects-map.py

#!/usr/bin/env python3
"""kb-objects-map — собирает машиночитаемый реестр объектов и хостов.

Источники:
  projects/dttb/netbird-inventory.md   — netbird-пиры (источник правды по железу)
  projects/<dir>/README.md             — frontmatter каждого проекта (aliases, tags, status)
  projects/<file>.md                   — singleton-проекты

Output:
  audit/objects-map.json   — структура для бота / structured-поиска
  projects/_index.md       — человеко-читаемый индекс с wiki-ссылками

Запускать вручную или cron:
  cd ~/knowledge-base && python3 scripts/kb-objects-map.py
"""
from datetime import date, datetime
from pathlib import Path
import json
import re

VAULT = Path(__file__).resolve().parent.parent
INV = VAULT / "projects/dttb/netbird-inventory.md"
JSON_OUT = VAULT / "audit/objects-map.json"
MD_OUT = VAULT / "projects/_index.md"


def parse_frontmatter(text: str) -> dict:
    m = re.match(r"^---\n(.+?)\n---\n", text, re.S)
    if not m:
        return {}
    fm = {}
    for line in m.group(1).splitlines():
        if ":" not in line or line.startswith("#"):
            continue
        k, _, v = line.partition(":")
        k, v = k.strip(), v.strip()
        if v.startswith("[") and v.endswith("]"):
            v = [x.strip().strip("\"'") for x in v[1:-1].split(",") if x.strip()]
        elif v.startswith('"') and v.endswith('"'):
            v = v[1:-1]
        fm[k] = v
    return fm


ROW_RE = re.compile(
    r"^\|\s*([^|]+?)\s*\|\s*(\d+\.\d+\.\d+\.\d+)\s*\|\s*([^|]*?)\s*\|\s*([^|]*?)\s*\|\s*([^|]*?)\s*\|"
)


def parse_netbird(path: Path) -> list[dict]:
    """Парсит online + offline-таблицы netbird-inventory.md."""
    if not path.exists():
        return []
    text = path.read_text()
    parts = re.split(r"^##\s+(?:Оффлайн|Offline)\b.*$", text, maxsplit=1, flags=re.M)
    online_text, offline_text = parts[0], parts[1] if len(parts) > 1 else ""

    peers = []
    for line in online_text.splitlines():
        m = ROW_RE.match(line)
        if not m:
            continue
        name = m.group(1).strip()
        if name in ("Имя", "---"):
            continue
        peers.append({
            "name": name, "ip": m.group(2).strip(),
            "os": m.group(3).strip(), "city": m.group(4).strip(),
            "version": m.group(5).strip(), "online": True,
        })
    # offline-таблица: | Имя | IP | ОС | Последний раз онлайн | Город |
    for line in offline_text.splitlines():
        m = ROW_RE.match(line)
        if not m:
            continue
        name = m.group(1).strip()
        if name in ("Имя", "---"):
            continue
        peers.append({
            "name": name, "ip": m.group(2).strip(),
            "os": m.group(3).strip(),
            "last_seen": m.group(4).strip(), "city": m.group(5).strip(),
            "version": "", "online": False,
        })
    return peers


def _norm(s: str) -> str:
    """Нормализация для нечёткого сравнения. ye→e уравнивает Знаменское/Znamenskoe."""
    return (s or "").lower().replace("ye", "e").replace(" ", "").replace("_", "").replace("-", "")


def name_match(peer_name: str, candidates: list[str]) -> bool:
    """Exact match с нормализацией (ye→e, без пробелов/_/−).
    Чтобы избежать FP типа `cloud` ⊂ `Cloud-NIIKN`, требуется точное равенство."""
    pn = _norm(peer_name)
    if not pn:
        return False
    return any(_norm(c) == pn for c in candidates if c)


def main() -> None:
    peers = parse_netbird(INV)
    objects = []
    seen_files = set()

    project_dirs = sorted(p for p in (VAULT / "projects").iterdir() if p.is_dir())
    for d in project_dirs:
        readme = d / "README.md"
        fm = parse_frontmatter(readme.read_text()) if readme.exists() else {}
        rel_file = f"projects/{d.name}/README.md" if readme.exists() else None
        aliases = fm.get("aliases", [])
        if isinstance(aliases, str):
            aliases = [aliases]
        names = sorted({d.name, *aliases})
        nb = [p for p in peers if name_match(p["name"], names)]
        objects.append({
            "id": d.name,
            "type": "project",
            "names": names,
            "netbird_peers": nb,
            "tags": fm.get("tags", []),
            "owner": fm.get("owner") or fm.get("client"),
            "region": fm.get("region"),
            "status": fm.get("status", "unknown"),
            "file": rel_file,
        })
        if rel_file:
            seen_files.add(rel_file)

    for f in sorted((VAULT / "projects").glob("*.md")):
        if f.name == "_index.md":
            continue
        rel = f"projects/{f.name}"
        if rel in seen_files:
            continue
        fm = parse_frontmatter(f.read_text())
        objects.append({
            "id": f.stem,
            "type": "project-note",
            "names": [f.stem],
            "netbird_peers": [],
            "tags": fm.get("tags", []),
            "owner": fm.get("owner"),
            "region": fm.get("region"),
            "status": fm.get("status", "unknown"),
            "file": rel,
        })

    matched_peer_names = {p["name"] for o in objects for p in o["netbird_peers"]}
    for p in peers:
        if p["name"] in matched_peer_names:
            continue
        objects.append({
            "id": p["name"].replace(" ", "_"),
            "type": "netbird-only",
            "names": [p["name"]],
            "netbird_peers": [p],
            "tags": [],
            "owner": None,
            "region": p["city"] or None,
            "status": "no-project-page",
            "file": None,
        })

    JSON_OUT.parent.mkdir(parents=True, exist_ok=True)
    JSON_OUT.write_text(json.dumps(objects, indent=2, ensure_ascii=False) + "\n")

    today = date.today().isoformat()
    now = datetime.now().isoformat(timespec="minutes")
    n_proj = sum(1 for o in objects if o["type"] in ("project", "project-note"))
    n_with_nb = sum(1 for o in objects if o["type"] == "project" and o["netbird_peers"])
    n_orphan = sum(1 for o in objects if o["type"] == "netbird-only")

    md = [
        "---",
        f"date: {today}",
        "type: index",
        "source: scripts/kb-objects-map.py",
        "tags: [index, registry, objects, netbird]",
        "---",
        "",
        "# Реестр объектов и netbird-пиров",
        "",
        f"Авто-сгенерировано `{now}` из [[dttb/netbird-inventory]] + frontmatter в `projects/`.",
        "**Не править вручную** — перепишется. Источник правды — frontmatter в каждом README.",
        "",
        f"- Проектов: **{n_proj}**, из них с netbird-привязкой: **{n_with_nb}**",
        f"- NetBird-пиров без projects-страницы: **{n_orphan}** (TODO — создать стабы)",
        "",
        "## Проекты с netbird-привязкой",
        "",
        "| ID | Имена | NetBird IP | OS | Город | Файл | Статус |",
        "|---|---|---|---|---|---|---|",
    ]
    for o in sorted(objects, key=lambda x: x["id"]):
        if o["type"] == "project" and o["netbird_peers"]:
            nb_ip = ", ".join(p["ip"] for p in o["netbird_peers"])
            nb_os = ", ".join(sorted({p["os"] for p in o["netbird_peers"] if p["os"]}))
            nb_city = ", ".join(sorted({p["city"] for p in o["netbird_peers"] if p["city"]}))
            file_link = f"[[{o['file'][:-3]}]]" if o["file"] else "—"
            names = ", ".join(o["names"][:5])
            md.append(f"| {o['id']} | {names} | {nb_ip} | {nb_os} | {nb_city} | {file_link} | {o['status']} |")

    md.extend([
        "",
        "## Проекты без netbird-привязки",
        "",
        "| ID | Тип | Файл | Статус |",
        "|---|---|---|---|",
    ])
    for o in sorted(objects, key=lambda x: x["id"]):
        if o["type"] in ("project", "project-note") and not o["netbird_peers"]:
            file_link = f"[[{o['file'][:-3]}]]" if o["file"] else "—"
            md.append(f"| {o['id']} | {o['type']} | {file_link} | {o['status']} |")

    md.extend([
        "",
        "## NetBird-пиры без projects-страницы — TODO",
        "",
        "Эти пиры есть в инвентаре, но у них нет своей карточки в `projects/`. Бот не сможет ответить на «найди мне X» — нет файла. Нужно создать стабы (Фаза 4 плана).",
        "",
        "| Имя в NetBird | IP | OS | Город | Версия |",
        "|---|---|---|---|---|",
    ])
    for o in sorted(objects, key=lambda x: x["id"]):
        if o["type"] == "netbird-only":
            p = o["netbird_peers"][0]
            md.append(f"| `{p['name']}` | {p['ip']} | {p['os']} | {p['city']} | {p['version']} |")

    md.append("")
    MD_OUT.write_text("\n".join(md))

    print(f"Wrote {JSON_OUT.relative_to(VAULT)} ({len(objects)} entries: "
          f"{n_proj} projects, {n_orphan} netbird-only)")
    print(f"Wrote {MD_OUT.relative_to(VAULT)}")


if __name__ == "__main__":
    main()