FabledSteward/plugins/prometheus/library.py

"""The widget templates a Prometheus widget can draw with.

Bundled templates ship in plugins/prometheus/widgets/*.yaml, read once and
read-only; their key is "bundled:<file name>". Templates an admin writes or
copies in are rows (PrometheusWidgetTemplate, #5692); their key is
"custom:<id>". Steward fetches none from anywhere (System 181's ruling).
"""
from __future__ import annotations

import json
import re
from dataclasses import dataclass, field
from datetime import datetime
from functools import lru_cache
from pathlib import Path

from sqlalchemy import select

from .models import PrometheusTarget, PrometheusWidgetTemplate
from .template_format import Problem, Template, TemplateError, check

BUNDLED = Path(__file__).parent / "widgets"
CUSTOM = "custom:"


@dataclass
class Entry:
    key: str
    name: str
    bundled: bool
    text: str
    template: Template | None = None
    problems: list[Problem] = field(default_factory=list)
    id: str | None = None                   # a custom template's row
    updated_at: datetime | None = None


def _entry(key: str, fallback_name: str, text: str, bundled: bool, **row) -> Entry:
    try:
        template = check(text)
    except TemplateError as exc:
        return Entry(key, fallback_name, bundled, text, None, exc.problems, **row)
    return Entry(key, template.name, bundled, text, template, **row)


@lru_cache(maxsize=1)
def bundled() -> tuple[Entry, ...]:
    return tuple(_entry(f"bundled:{p.stem}", p.stem, p.read_text(encoding="utf-8"), True)
                 for p in sorted(BUNDLED.glob("*.yaml")))


def custom_entry(row: PrometheusWidgetTemplate) -> Entry:
    # Saved only once it passed, but a later Steward may read the format
    # differently: it is checked again, and says so if it no longer passes.
    return _entry(CUSTOM + row.id, row.name, row.text, False, id=row.id, updated_at=row.updated_at)


async def entries(db) -> list[Entry]:
    """Every template, by name."""
    rows = (await db.execute(select(PrometheusWidgetTemplate))).scalars().all()
    return sorted([*bundled(), *(custom_entry(r) for r in rows)], key=lambda e: e.name.lower())


async def find(db, key: str) -> Entry | None:
    if key.startswith(CUSTOM):
        row = await db.get(PrometheusWidgetTemplate, key[len(CUSTOM):])
        return custom_entry(row) if row else None
    return next((e for e in bundled() if e.key == key), None)


async def name_taken(db, name: str, other_than: str | None = None) -> bool:
    """Another saved template has this name; the widget picker lists by name."""
    q = select(PrometheusWidgetTemplate.id).where(PrometheusWidgetTemplate.name == name)
    if other_than:
        q = q.where(PrometheusWidgetTemplate.id != other_than)
    return (await db.execute(q)).first() is not None


async def used_by(db) -> dict[str, int]:
    """How many dashboard widgets draw with each template key."""
    from steward.models.dashboard import DashboardWidget
    counts: dict[str, int] = {}
    rows = (await db.execute(select(DashboardWidget.config_json)
                             .where(DashboardWidget.widget_key == "prometheus"))).scalars()
    for raw in rows:
        try:
            key = (json.loads(raw or "{}") or {}).get("template")
        except (ValueError, AttributeError):
            continue
        if key:
            counts[key] = counts.get(key, 0) + 1
    return counts


# ── Copying ──────────────────────────────────────────────────────────────────

_NAME_LINE = re.compile(r"^name:[^\n]*$", re.MULTILINE)


def renamed(text: str, name: str) -> str:
    """`text` with its top-level `name:` set to `name`, the rest as written
    (its comments and layout kept)."""
    line = "name: " + json.dumps(name, ensure_ascii=False)
    if _NAME_LINE.search(text):
        return _NAME_LINE.sub(lambda _: line, text, count=1)
    return f"{line}\n{text}"


def copy_name(name: str, taken: set[str]) -> str:
    """"Fabled Archives (copy)", then "(copy 2)", … a name nobody has."""
    for n in range(1, 1000):
        candidate = f"{name} (copy)" if n == 1 else f"{name} (copy {n})"
        if candidate not in taken:
            return candidate[:80]
    return name[:80]


def file_name(name: str) -> str:
    """A download's file name: "Fabled Archives" → fabled-archives.yaml."""
    slug = re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-")
    return f"{slug or 'template'}.yaml"


# ── The widget's settings ────────────────────────────────────────────────────

async def target_options(db) -> list[dict]:
    rows = (await db.execute(select(PrometheusTarget.id, PrometheusTarget.name)
                             .order_by(PrometheusTarget.name))).all()
    return [{"value": tid, "label": name} for tid, name in rows]


async def template_options(db) -> list[dict]:
    return [{"value": e.key, "label": f"{e.name} (built in)" if e.bundled else e.name}
            for e in await entries(db) if e.template is not None]


async def default_title(db, config: dict) -> str | None:
    """A new widget is named after its template: "Fabled Archives"."""
    entry = await find(db, str(config.get("template") or ""))
    return entry.name if entry and entry.template else None