onGuard24/onguard24/modules/grafana_catalog.py

"""Кэш иерархии Grafana (инстанс → org → папки → правила) через API + БД."""

from __future__ import annotations

import html
import json
import logging
from dataclasses import dataclass

import asyncpg
from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi.responses import HTMLResponse
from pydantic import BaseModel, Field

from onguard24.config import get_settings
from onguard24.deps import get_pool
from onguard24.domain.events import EventBus
from onguard24.grafana_sources import iter_grafana_sources
from onguard24.integrations import grafana_topology as gt
from onguard24.modules.ui_support import wrap_module_html_page

log = logging.getLogger(__name__)

router = APIRouter(tags=["module-grafana-catalog"])
ui_router = APIRouter(tags=["web-grafana-catalog"], include_in_schema=False)


def register_events(_bus: EventBus, _pool: asyncpg.Pool | None = None) -> None:
    pass


class SyncBody(BaseModel):
    instance_slug: str | None = Field(
        default=None,
        description="Slug из GRAFANA_SOURCES_JSON; пусто — все источники с api_token",
    )


@dataclass
class PullOutcome:
    org_id: int
    org_name: str
    folder_rows: list[tuple[str, str, str | None]]
    rules: list[gt.ParsedRuleRow]
    warnings: list[str]


async def pull_topology(api_url: str, token: str) -> tuple[PullOutcome | None, str | None]:
    org, oerr = await gt.fetch_org(api_url, token)
    if oerr or not org:
        return None, oerr or "no org"
    oid = org.get("id")
    if oid is None:
        return None, "org response without id"
    oname = str(org.get("name") or "")

    warnings: list[str] = []
    folders_raw, ferr = await gt.fetch_all_folders(api_url, token)
    if ferr:
        warnings.append(ferr)

    ruler_raw, rerr = await gt.fetch_ruler_rules_raw(api_url, token)
    if rerr:
        warnings.append(rerr)
        rules: list[gt.ParsedRuleRow] = []
        namespaces: set[str] = set()
    else:
        rules = gt.parse_ruler_rules(ruler_raw or {})
        namespaces = {r.namespace_uid for r in rules}

    merged = gt.merge_folder_rows(folders_raw, namespaces)
    return (
        PullOutcome(
            org_id=int(oid),
            org_name=oname,
            folder_rows=merged,
            rules=rules,
            warnings=warnings,
        ),
        None,
    )


async def persist_topology(
    conn: asyncpg.Connection,
    instance_slug: str,
    outcome: PullOutcome,
) -> None:
    oid = outcome.org_id
    await conn.execute(
        """
        DELETE FROM grafana_catalog_rules
        WHERE instance_slug = $1 AND grafana_org_id = $2
        """,
        instance_slug,
        oid,
    )
    await conn.execute(
        """
        DELETE FROM grafana_catalog_folders
        WHERE instance_slug = $1 AND grafana_org_id = $2
        """,
        instance_slug,
        oid,
    )

    for uid, title, parent in outcome.folder_rows:
        await conn.execute(
            """
            INSERT INTO grafana_catalog_folders
            (instance_slug, grafana_org_id, folder_uid, title, parent_uid)
            VALUES ($1, $2, $3, $4, $5)
            """,
            instance_slug,
            oid,
            uid,
            title,
            parent,
        )

    for r in outcome.rules:
        await conn.execute(
            """
            INSERT INTO grafana_catalog_rules (
                instance_slug, grafana_org_id, namespace_uid, rule_group_name,
                rule_uid, title, rule_group_interval, labels
            )
            VALUES ($1, $2, $3, $4, $5, $6, $7, $8::jsonb)
            """,
            instance_slug,
            oid,
            r.namespace_uid,
            r.rule_group_name,
            r.rule_uid,
            r.title,
            r.rule_group_interval,
            json.dumps(r.labels),
        )

    fc = len(outcome.folder_rows)
    rc = len(outcome.rules)
    warn_txt = "; ".join(outcome.warnings) if outcome.warnings else None
    if warn_txt and len(warn_txt) > 1900:
        warn_txt = warn_txt[:1900] + "…"
    await conn.execute(
        """
        INSERT INTO grafana_catalog_meta (
            instance_slug, grafana_org_id, org_name, synced_at,
            folder_count, rule_count, error_text
        )
        VALUES ($1, $2, $3, now(), $4, $5, $6)
        ON CONFLICT (instance_slug, grafana_org_id) DO UPDATE SET
            org_name = EXCLUDED.org_name,
            synced_at = EXCLUDED.synced_at,
            folder_count = EXCLUDED.folder_count,
            rule_count = EXCLUDED.rule_count,
            error_text = EXCLUDED.error_text
        """,
        instance_slug,
        oid,
        outcome.org_name,
        fc,
        rc,
        warn_txt,
    )


@router.post("/sync", status_code=200)
async def sync_catalog_api(
    body: SyncBody,
    pool: asyncpg.Pool | None = Depends(get_pool),
):
    if pool is None:
        raise HTTPException(status_code=503, detail="database disabled")
    sources = iter_grafana_sources(get_settings())
    if body.instance_slug:
        sl = body.instance_slug.strip().lower()
        sources = [s for s in sources if s.slug == sl]
        if not sources:
            raise HTTPException(status_code=404, detail="unknown instance slug")
    to_run = [s for s in sources if s.api_token.strip()]
    if not to_run:
        raise HTTPException(
            status_code=400,
            detail="нет источников с api_token; задайте GRAFANA_SOURCES_JSON или GRAFANA_SERVICE_ACCOUNT_TOKEN",
        )

    results: list[dict] = []
    for src in to_run:
        outcome, err = await pull_topology(src.api_url, src.api_token)
        if err or outcome is None:
            results.append({"slug": src.slug, "ok": False, "error": err})
            log.warning("grafana_catalog sync failed %s: %s", src.slug, err)
            continue
        async with pool.acquire() as conn:
            async with conn.transaction():
                await persist_topology(conn, src.slug, outcome)
        results.append(
            {
                "slug": src.slug,
                "ok": True,
                "org_id": outcome.org_id,
                "org_name": outcome.org_name,
                "folders": len(outcome.folder_rows),
                "rules": len(outcome.rules),
                "warnings": outcome.warnings,
            }
        )
    return {"results": results}


@router.get("/meta")
async def list_meta_api(pool: asyncpg.Pool | None = Depends(get_pool)):
    if pool is None:
        return {"items": [], "database": "disabled"}
    async with pool.acquire() as conn:
        rows = await conn.fetch(
            """
            SELECT instance_slug, grafana_org_id, org_name, synced_at,
                   folder_count, rule_count, error_text
            FROM grafana_catalog_meta
            ORDER BY instance_slug, grafana_org_id
            """
        )
    items = []
    for r in rows:
        items.append(
            {
                "instance_slug": r["instance_slug"],
                "grafana_org_id": r["grafana_org_id"],
                "org_name": r["org_name"],
                "synced_at": r["synced_at"].isoformat() if r["synced_at"] else None,
                "folder_count": r["folder_count"],
                "rule_count": r["rule_count"],
                "error_text": r["error_text"],
            }
        )
    return {"items": items}


async def _build_catalog_tree_dict(conn: asyncpg.Connection, instance_slug: str) -> dict | None:
    """Сборка дерева из БД (общая логика для API и UI)."""
    slug = instance_slug.strip().lower()
    meta = await conn.fetchrow(
        """
        SELECT * FROM grafana_catalog_meta
        WHERE instance_slug = $1 AND grafana_org_id > 0
        ORDER BY synced_at DESC LIMIT 1
        """,
        slug,
    )
    if not meta:
        return None
    oid = meta["grafana_org_id"]
    folders = await conn.fetch(
        """
        SELECT folder_uid, title, parent_uid
        FROM grafana_catalog_folders
        WHERE instance_slug = $1 AND grafana_org_id = $2
        ORDER BY title
        """,
        slug,
        oid,
    )
    rules = await conn.fetch(
        """
        SELECT namespace_uid, rule_group_name, rule_uid, title,
               rule_group_interval, labels
        FROM grafana_catalog_rules
        WHERE instance_slug = $1 AND grafana_org_id = $2
        ORDER BY namespace_uid, rule_group_name, title
        """,
        slug,
        oid,
    )

    by_ns: dict[str, list[dict]] = {}
    for r in rules:
        ns = r["namespace_uid"]
        by_ns.setdefault(ns, []).append(
            {
                "rule_uid": r["rule_uid"],
                "title": r["title"],
                "rule_group": r["rule_group_name"],
                "interval": r["rule_group_interval"],
                "labels": r["labels"] if isinstance(r["labels"], dict) else {},
            }
        )

    folder_uids = {f["folder_uid"] for f in folders}
    orphans = sorted(set(by_ns.keys()) - folder_uids)
    orphan_blocks = [{"namespace_uid": ns, "rules": by_ns[ns]} for ns in orphans]

    folder_nodes = []
    for f in folders:
        uid = f["folder_uid"]
        folder_nodes.append(
            {
                "folder_uid": uid,
                "title": f["title"],
                "parent_uid": f["parent_uid"],
                "rules": by_ns.get(uid, []),
            }
        )

    return {
        "instance_slug": slug,
        "grafana_org_id": oid,
        "org_name": meta["org_name"],
        "synced_at": meta["synced_at"].isoformat() if meta["synced_at"] else None,
        "folders": folder_nodes,
        "orphan_rule_namespaces": orphans,
        "orphan_rule_groups": orphan_blocks,
    }


@router.get("/tree")
async def tree_api(
    instance_slug: str,
    pool: asyncpg.Pool | None = Depends(get_pool),
):
    if pool is None:
        raise HTTPException(status_code=503, detail="database disabled")
    async with pool.acquire() as conn:
        data = await _build_catalog_tree_dict(conn, instance_slug)
    if not data:
        raise HTTPException(status_code=404, detail="no catalog for this slug; run POST /sync first")
    return data


def _rules_subtable_html(rules: list[dict]) -> str:
    if not rules:
        return "<p class='gc-muted'>Нет правил в этой группе/папке (по данным Ruler).</p>"
    rows = []
    for ru in rules:
        title = html.escape(str(ru.get("title") or "—"))
        uid = html.escape(str(ru.get("rule_uid") or "—"))
        grp = html.escape(str(ru.get("rule_group") or "—"))
        interval = html.escape(str(ru.get("interval") or "—"))
        labels = ru.get("labels") or {}
        lab_s = html.escape(json.dumps(labels, ensure_ascii=False)[:240])
        if len(json.dumps(labels, ensure_ascii=False)) > 240:
            lab_s += "…"
        rows.append(
            f"<tr><td>{title}</td><td><code>{uid}</code></td>"
            f"<td>{grp}</td><td>{interval}</td><td><code>{lab_s}</code></td></tr>"
        )
    head = "<thead><tr><th>Название правила</th><th>rule_uid</th><th>Группа</th><th>Интервал</th><th>Labels (фрагмент)</th></tr></thead>"
    return f"<table class='gc-subtable irm-table'>{head}<tbody>{''.join(rows)}</tbody></table>"


def _catalog_tree_html(tree: dict) -> str:
    """HTML: папки и правила под ними + «осиротевшие» namespace из Ruler."""
    parts: list[str] = [
        "<div class='gc-tree'><h2 style='font-size:1.1rem;margin:0 0 0.5rem'>Папки и правила</h2>"
        "<p class='gc-muted' style='margin:0 0 0.75rem'>Папки — из API Grafana; правила — из Ruler, привязка по UID папки (namespace). "
        "Название колонки «Правил» наверху — число всех записей с типом grafana_alert.</p>"
    ]
    for f in tree.get("folders") or []:
        uid = html.escape(str(f.get("folder_uid") or ""))
        title = html.escape(str(f.get("title") or "—"))
        parent = f.get("parent_uid")
        parent_s = html.escape(str(parent)) if parent else "—"
        rules: list = f.get("rules") or []
        n = len(rules)
        summ = (
            f"<strong>{title}</strong> "
            f"<span class='gc-muted'>· UID папки <code>{uid}</code> · родитель <code>{parent_s}</code> · правил: {n}</span>"
        )
        parts.append(
            f"<details class='gc-folder' open><summary>{summ}</summary>{_rules_subtable_html(rules)}</details>"
        )
    for block in tree.get("orphan_rule_groups") or []:
        ns = block.get("namespace_uid")
        rules = block.get("rules") or []
        ns_e = html.escape(str(ns))
        parts.append(
            f"<div class='gc-orphan'><strong>Namespace Ruler без папки в API</strong> "
            f"(<code>{ns_e}</code>) — часто системная группа. Правил: {len(rules)}."
            f"{_rules_subtable_html(rules)}</div>"
        )
    parts.append("</div>")
    return "".join(parts)


_SYNC_SCRIPT = """
<script>
(function () {
  var btn = document.getElementById('og-sync-btn');
  var st = document.getElementById('og-sync-status');
  if (!btn || !st) return;
  btn.addEventListener('click', function () {
    st.textContent = 'Синхронизация…';
    btn.disabled = true;
    fetch('/api/v1/modules/grafana-catalog/sync', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: '{}'
    }).then(function (r) {
      return r.text().then(function (t) {
        if (!r.ok) {
          st.textContent = 'Ошибка HTTP ' + r.status + ': ' + (t.slice(0, 400) || r.statusText);
          btn.disabled = false;
          return;
        }
        st.textContent = 'Готово, обновляем страницу…';
        location.reload();
      });
    }).catch(function (e) {
      st.textContent = 'Сбой сети: ' + e;
      btn.disabled = false;
    });
  });
})();
</script>
"""


@ui_router.get("/", response_class=HTMLResponse)
async def grafana_catalog_ui(request: Request):
    pool = get_pool(request)
    inner = ""
    tree_html = ""
    if pool is None:
        inner = "<p>База не настроена.</p>"
    else:
        try:
            async with pool.acquire() as conn:
                rows = await conn.fetch(
                    """
                    SELECT instance_slug, org_name, synced_at, folder_count, rule_count, error_text
                    FROM grafana_catalog_meta
                    ORDER BY instance_slug
                    """
                )
            sync_bar = """<div class="og-sync-bar">
<button type="button" class="og-btn og-btn-primary" id="og-sync-btn">Синхронизировать с Grafana</button>
<span class="og-sync-status" id="og-sync-status"></span>
</div>"""
            if not rows:
                inner = (
                    sync_bar
                    + "<p>Каталог пуст — нажмите кнопку выше (нужны <code>GRAFANA_URL</code> и токен в .env).</p>"
                )
            else:
                inner = (
                    sync_bar
                    + "<p class='gc-muted'>Строка ниже — сводка по последней синхронизации. "
                    "Число <strong>Правил</strong> — все правила алертинга (Grafana-managed) из Ruler для этой org.</p>"
                    + "<table class='irm-table'><thead><tr><th>Slug</th><th>Org</th><th>Синхр.</th>"
                    "<th>Папок (API)</th><th>Правил (Ruler)</th><th>Ошибка</th></tr></thead><tbody>"
                )
                seen_slugs: set[str] = set()
                for r in rows:
                    err = html.escape(str(r["error_text"] or "—"))[:200]
                    st = r["synced_at"].isoformat() if r["synced_at"] else "—"
                    inner += (
                        f"<tr><td>{html.escape(r['instance_slug'])}</td>"
                        f"<td>{html.escape(str(r['org_name']))}</td>"
                        f"<td>{html.escape(st)}</td>"
                        f"<td>{r['folder_count']}</td><td>{r['rule_count']}</td>"
                        f"<td>{err}</td></tr>"
                    )
                    slug = str(r["instance_slug"])
                    if slug not in seen_slugs:
                        seen_slugs.add(slug)
                inner += "</tbody></table>"
                async with pool.acquire() as conn:
                    for slug in sorted(seen_slugs):
                        tree = await _build_catalog_tree_dict(conn, slug)
                        if tree:
                            tree_html += (
                                f"<h2 style='font-size:1.15rem;margin:1.25rem 0 0.35rem'>Инстанс "
                                f"<code>{html.escape(slug)}</code> · {html.escape(str(tree['org_name']))}</h2>"
                                + _catalog_tree_html(tree)
                            )
        except Exception as e:
            inner = f"<p class='module-err'>{html.escape(str(e))}</p>"
    page = f"""<h1>Каталог Grafana</h1>
<p>Иерархия: инстанс (slug) → организация → папки (имена и UID) → правила (название, UID, группа, интервал, labels).</p>
{inner}
{tree_html}
<p><small>API: <code>POST /api/v1/modules/grafana-catalog/sync</code>, <code>GET …/tree?instance_slug=…</code></small></p>
{_SYNC_SCRIPT}"""
    return HTMLResponse(
        wrap_module_html_page(
            document_title="Каталог Grafana — onGuard24",
            current_slug="grafana-catalog",
            main_inner_html=page,
        )
    )


async def render_home_fragment(request: Request) -> str:
    pool = get_pool(request)
    if pool is None:
        return '<p class="module-note">Нужна БД для каталога Grafana.</p>'
    try:
        async with pool.acquire() as conn:
            n = await conn.fetchval("SELECT count(*)::int FROM grafana_catalog_meta WHERE grafana_org_id >= 0")
            last = await conn.fetchrow(
                "SELECT max(synced_at) AS m FROM grafana_catalog_meta WHERE grafana_org_id >= 0"
            )
    except Exception:
        return '<p class="module-note">Таблицы каталога недоступны (миграции?).</p>'
    ts = last["m"].isoformat() if last and last["m"] else "никогда"
    return (
        f'<div class="module-fragment"><p>Источников с синхронизацией: <strong>{int(n)}</strong>. '
        f"Последняя синхр.: <strong>{html.escape(ts)}</strong></p></div>"
    )