"""Prometheus widget templates: the YAML a person writes, read and checked.

A template says which of a target's series a dashboard widget shows and
how. It is data only: it names series and lays them out in blocks, and
Steward draws them (template_view.py). It can't add a visual, a colour or a
formula; a new block is a Steward release (System 181's ruling).

    name: Fabled Archives
    version: 1
    expects: fabledarchives_
    blocks:
      - states:
          title: Backup targets
          metric: fabledarchives_backup_target_state
          name: target
          healthy: [fresh]
          quiet: [disabled, waiting]
          warning: [late]
      - values:
          items:
            - {label: Disk free, metric: fabledarchives_data_disk_free_bytes,
               of: fabledarchives_data_disk_size_bytes, warn_below: 10%}

check(text) returns the Template, or every problem found, each with its
line and a sentence saying what to change.
"""
from __future__ import annotations

import re
from dataclasses import dataclass, field

import yaml

FORMAT = 1
MAX_BYTES = 64 * 1024
MAX_BLOCKS = 20
MAX_ITEMS = 30

_METRIC = re.compile(r"^[a-zA-Z_:][a-zA-Z0-9_:]*$")
_LABEL = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]*$")
_DURATION = re.compile(r"^(\d+(?:\.\d+)?)\s*([smhd]?)$")
_SECONDS = {"": 1, "s": 1, "m": 60, "h": 3600, "d": 86400}

# What a unit shows a value as; the name is the suffix metric_value reads.
UNITS = {"bytes": "_bytes", "seconds": "_secs", "percent": "_pct", "per_sec": "_per_sec",
         "count": "", "timestamp": "_timestamp_seconds"}


@dataclass
class Problem:
    line: int
    text: str

    def __str__(self) -> str:
        return f"Line {self.line}: {self.text}" if self.line else self.text


@dataclass
class Block:
    kind: str        # states | values | times | table | info
    title: str
    line: int
    spec: dict = field(default_factory=dict)


@dataclass
class Template:
    name: str
    version: int
    description: str = ""
    expects: str = ""
    blocks: list[Block] = field(default_factory=list)
    text: str = ""


class TemplateError(ValueError):
    def __init__(self, problems: list[Problem]):
        self.problems = problems
        super().__init__("; ".join(str(p) for p in problems))


# ── YAML with line numbers ───────────────────────────────────────────────────

class _Map(dict):
    line = 0
    lines: dict


class _List(list):
    line = 0
    lines: list


class _Loader(yaml.SafeLoader):
    pass


def _construct_map(loader, node):
    loader.flatten_mapping(node)
    out = _Map()
    out.line, out.lines = node.start_mark.line + 1, {}
    for key_node, value_node in node.value:
        key = loader.construct_object(key_node, deep=True)
        if not isinstance(key, str):
            raise yaml.constructor.ConstructorError(
                None, None, "a key here should be a word, like `metric`", key_node.start_mark)
        if key in out:
            raise yaml.constructor.ConstructorError(
                None, None, f"`{key}` is given twice", key_node.start_mark)
        out[key] = loader.construct_object(value_node, deep=True)
        out.lines[key] = key_node.start_mark.line + 1
    return out


def _construct_list(loader, node):
    out = _List(loader.construct_object(n, deep=True) for n in node.value)
    out.line, out.lines = node.start_mark.line + 1, [n.start_mark.line + 1 for n in node.value]
    return out


_Loader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, _construct_map)
_Loader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_SEQUENCE_TAG, _construct_list)


def _load(text: str) -> tuple[object, list[Problem]]:
    if len(text.encode()) > MAX_BYTES:
        return None, [Problem(0, f"A template is at most {MAX_BYTES // 1024} KB.")]
    try:
        # An alias can make a small file expand without end; a template has no use for one.
        for event in yaml.parse(text, Loader=yaml.SafeLoader):
            if isinstance(event, yaml.AliasEvent):
                return None, [Problem(event.start_mark.line + 1,
                                      "Anchors and aliases (&name, *name) aren't allowed; write the value out.")]
        return yaml.load(text, Loader=_Loader), []
    except yaml.constructor.ConstructorError as exc:      # ours: a key twice, or not a word
        mark = exc.problem_mark
        return None, [Problem(mark.line + 1 if mark else 0, exc.problem[:1].upper() + exc.problem[1:] + ".")]
    except yaml.MarkedYAMLError as exc:
        mark = exc.problem_mark or exc.context_mark
        return None, [Problem(mark.line + 1 if mark else 0, f"This isn't valid YAML: {exc.problem}.")]
    except yaml.YAMLError as exc:
        return None, [Problem(0, f"This isn't valid YAML: {exc}.")]


# ── Values ───────────────────────────────────────────────────────────────────

def duration(value) -> float | None:
    """Seconds from 90, "90s", "15m", "26h" or "2d"; None if it isn't one."""
    if isinstance(value, bool):
        return None
    if isinstance(value, (int, float)):
        return float(value) if value >= 0 else None
    m = _DURATION.match(str(value).strip().lower())
    return float(m.group(1)) * _SECONDS[m.group(2)] if m else None


@dataclass
class Limit:
    """A threshold: a number, or a share of the item's `of` ("10%")."""
    value: float
    share: bool = False


def limit(value) -> Limit | None:
    if isinstance(value, bool):
        return None
    if isinstance(value, (int, float)):
        return Limit(float(value))
    text = str(value).strip()
    if text.endswith("%"):
        try:
            return Limit(float(text[:-1]), share=True)
        except ValueError:
            return None
    return None


# ── The checker ──────────────────────────────────────────────────────────────

class _Check:
    def __init__(self):
        self.problems: list[Problem] = []

    def add(self, line: int, text: str) -> None:
        self.problems.append(Problem(line, text))

    def keys(self, m: _Map, allowed: tuple, required: tuple, where: str) -> bool:
        ok = True
        for key in m:
            if key not in allowed:
                self.add(m.lines[key], f"{where} has no `{key}`. It takes: {', '.join(allowed)}.")
                ok = False
        for key in required:
            if key not in m:
                self.add(m.line, f"{where} needs `{key}`.")
                ok = False
        return ok

    def text(self, m: _Map, key: str, where: str, default: str = "") -> str:
        if key not in m:
            return default
        value = m[key]
        if isinstance(value, (dict, list)) or value is None:
            self.add(m.lines[key], f"`{key}` in {where} should be text.")
            return default
        return str(value)

    def metric(self, m: _Map, key: str, where: str) -> str:
        value = m.get(key)
        if not isinstance(value, str) or not _METRIC.match(value):
            self.add(m.lines.get(key, m.line),
                     f"`{key}` in {where} should be a metric name as Steward stores it, like "
                     "fabledarchives_files (a counter is stored as <name>_per_sec).")
            return ""
        return value

    def label(self, m: _Map, key: str, where: str) -> str:
        if key not in m:
            return ""
        value = m[key]
        if not isinstance(value, str) or not _LABEL.match(value):
            self.add(m.lines[key], f"`{key}` in {where} should be a label name, like target.")
            return ""
        return value

    def words(self, m: _Map, key: str, where: str) -> list[str]:
        if key not in m:
            return []
        value = m[key]
        if not isinstance(value, list) or not all(isinstance(v, (str, int)) for v in value):
            self.add(m.lines[key], f"`{key}` in {where} should be a list, like [fresh, waiting].")
            return []
        if any(isinstance(v, bool) for v in value):
            self.add(m.lines[key], f"`{key}` in {where} has a word YAML reads as true or false "
                                   "(on, off, yes, no); put it in quotes, like 'off'.")
            return []
        return [str(v) for v in value]

    def where(self, m: _Map, ctx: str) -> dict[str, str]:
        if "where" not in m:
            return {}
        value = m["where"]
        if not isinstance(value, dict) or not all(
                isinstance(k, str) and _LABEL.match(k) and isinstance(v, (str, int, float))
                and not isinstance(v, bool) for k, v in value.items()):
            self.add(m.lines["where"], f"`where` in {ctx} should map labels to values, like "
                                       "{kind: files}. Quote a value YAML would read as true or false.")
            return {}
        return {k: str(v) for k, v in value.items()}

    def unit(self, m: _Map, ctx: str) -> str:
        if "unit" not in m:
            return ""
        value = m["unit"]
        if value not in UNITS:
            self.add(m.lines["unit"], f"`unit` in {ctx} is one of: {', '.join(UNITS)}.")
            return ""
        return value

    def span(self, m: _Map, key: str, ctx: str) -> float | None:
        if key not in m:
            return None
        value = duration(m[key])
        if value is None:
            self.add(m.lines[key], f"`{key}` in {ctx} should be a span, like 26h, 90m or 2d.")
        return value

    def items(self, m: _Map, ctx: str) -> list[_Map]:
        value = m.get("items")
        if not isinstance(value, list) or not value:
            self.add(m.lines.get("items", m.line), f"`items` in {ctx} should be a list of one or more items.")
            return []
        if len(value) > MAX_ITEMS:
            self.add(m.lines["items"], f"{ctx} has {len(value)} items; at most {MAX_ITEMS}.")
        out = []
        for i, item in enumerate(value):
            if not isinstance(item, dict):
                self.add(value.lines[i], f"Each item in {ctx} should be a set of keys, like "
                                         "{label: Disk free, metric: …}.")
            else:
                out.append(item)
        return out


# Each block: the keys it takes, those it needs, and its title when none is given.
_BLOCKS = {
    "states": (("title", "metric", "where", "name", "healthy", "quiet", "warning", "warn_past",
                "crit_past", "columns"),
               ("metric", "healthy"), "States"),
    "values": (("title", "items"), ("items",), "Figures"),
    "times": (("title", "items"), ("items",), "Times"),
    "table": (("title", "metric", "where", "rows", "columns", "unit"), ("metric", "rows", "columns"),
              "Table"),
    "info": (("title", "metric", "where", "labels"), ("metric",), "About"),
}
_VALUE_ITEM = ("label", "metric", "where", "name", "of", "unit", "warn_above", "crit_above",
               "warn_below", "crit_below", "sparkline")
_TIME_ITEM = ("label", "metric", "where", "name", "warn_after", "crit_after")
_COLUMN = ("label", "metric", "unit")


def _states(c: _Check, m: _Map, ctx: str) -> dict:
    spec = {"metric": c.metric(m, "metric", ctx), "where": c.where(m, ctx), "name": c.label(m, "name", ctx),
            "healthy": c.words(m, "healthy", ctx), "quiet": c.words(m, "quiet", ctx),
            "warning": c.words(m, "warning", ctx), "columns": [],
            # A deadline per row (a *_timestamp_seconds): once it has passed, a
            # healthy row reads as a warning or a problem. A job that says
            # "running" for days is still overdue.
            "warn_past": c.metric(m, "warn_past", ctx) if "warn_past" in m else "",
            "crit_past": c.metric(m, "crit_past", ctx) if "crit_past" in m else ""}
    if "healthy" in m and not spec["healthy"] and isinstance(m["healthy"], list):
        c.add(m.lines["healthy"], f"`healthy` in {ctx} needs at least one state.")
    seen: dict[str, str] = {}
    for key in ("healthy", "quiet", "warning"):
        for state in spec[key]:
            if state in seen:
                c.add(m.lines[key], f"“{state}” is in both `{seen[state]}` and `{key}` in {ctx}.")
            seen[state] = key
    if "columns" in m:
        cols = m["columns"]
        if not isinstance(cols, list):
            c.add(m.lines["columns"], f"`columns` in {ctx} should be a list of {{label, metric}}.")
        else:
            for i, col in enumerate(cols):
                where = f"a column of {ctx}"
                if not isinstance(col, dict):
                    c.add(cols.lines[i], f"Each of `columns` in {ctx} should be {{label, metric}}.")
                    continue
                if c.keys(col, _COLUMN, ("label", "metric"), where):
                    spec["columns"].append({"label": c.text(col, "label", where),
                                            "metric": c.metric(col, "metric", where),
                                            "unit": c.unit(col, where)})
    return spec


def _value_item(c: _Check, item: _Map, ctx: str) -> dict | None:
    where = f"an item of {ctx}"
    if not c.keys(item, _VALUE_ITEM, ("label", "metric"), where):
        return None
    spec = {"label": c.text(item, "label", where), "metric": c.metric(item, "metric", where),
            "where": c.where(item, where), "name": c.label(item, "name", where),
            "of": c.metric(item, "of", where) if "of" in item else "", "unit": c.unit(item, where),
            "sparkline": False}
    if "sparkline" in item:
        if not isinstance(item["sparkline"], bool):
            c.add(item.lines["sparkline"], f"`sparkline` in {where} is true or false.")
        spec["sparkline"] = item["sparkline"] is True
    for key in ("warn_above", "crit_above", "warn_below", "crit_below"):
        spec[key] = None
        if key in item:
            lim = limit(item[key])
            if lim is None:
                c.add(item.lines[key], f"`{key}` in {where} should be a number, or a share like 10%.")
            elif lim.share and not spec["of"]:
                c.add(item.lines[key], f"`{key}` in {where} is a share ({item[key]}), which needs `of`: "
                                       "the metric it is a share of.")
            spec[key] = lim
    return spec


def _time_item(c: _Check, item: _Map, ctx: str) -> dict | None:
    where = f"an item of {ctx}"
    if not c.keys(item, _TIME_ITEM, ("label", "metric"), where):
        return None
    spec = {"label": c.text(item, "label", where), "metric": c.metric(item, "metric", where),
            "where": c.where(item, where), "name": c.label(item, "name", where),
            "warn_after": c.span(item, "warn_after", where), "crit_after": c.span(item, "crit_after", where)}
    if spec["warn_after"] and spec["crit_after"] and spec["crit_after"] < spec["warn_after"]:
        c.add(item.lines["crit_after"], f"`crit_after` in {where} should be later than `warn_after`.")
    return spec


def _block(c: _Check, raw, line: int, n: int) -> Block | None:
    if not isinstance(raw, dict) or len(raw) != 1:
        c.add(line, f"Block {n} should be one kind with its settings under it, like "
                    f"`- states: {{metric: …}}`. The kinds are: {', '.join(_BLOCKS)}.")
        return None
    kind, body = next(iter(raw.items()))
    if kind not in _BLOCKS:
        c.add(line, f"Block {n} is “{kind}”; the kinds are: {', '.join(_BLOCKS)}.")
        return None
    if not isinstance(body, dict):
        c.add(line, f"Block {n} ({kind}) should have its settings under it, like `metric: …`.")
        return None
    allowed, required, default_title = _BLOCKS[kind]
    ctx = f"block {n} ({kind})"
    if not c.keys(body, allowed, required, ctx.capitalize()):
        return None
    title = c.text(body, "title", ctx, default_title)
    if kind == "states":
        spec = _states(c, body, ctx)
    elif kind in ("values", "times"):
        reader = _value_item if kind == "values" else _time_item
        spec = {"items": [s for s in (reader(c, i, ctx) for i in c.items(body, ctx)) if s]}
    elif kind == "table":
        spec = {"metric": c.metric(body, "metric", ctx), "where": c.where(body, ctx),
                "rows": c.label(body, "rows", ctx), "columns": c.label(body, "columns", ctx),
                "unit": c.unit(body, ctx)}
    else:
        spec = {"metric": c.metric(body, "metric", ctx), "where": c.where(body, ctx),
                "labels": c.words(body, "labels", ctx)}
    return Block(kind, title, line, spec)


_TOP = ("name", "description", "version", "format", "expects", "blocks")


def check(text: str) -> Template:
    """The template `text` describes. Raises TemplateError with every problem
    found, each with its line, in the order they appear."""
    raw, problems = _load(text)
    if problems:
        raise TemplateError(problems)
    c = _Check()
    if not isinstance(raw, dict):
        raise TemplateError([Problem(1, "A template starts with its keys: name, version and blocks.")])
    c.keys(raw, _TOP, ("name", "version", "blocks"), "A template")
    name = c.text(raw, "name", "the template").strip()
    if "name" in raw and not name:
        c.add(raw.lines["name"], "Give the template a name.")
    if len(name) > 80:
        c.add(raw.lines["name"], "A name is at most 80 characters.")
    version = raw.get("version")
    if "version" in raw and (isinstance(version, bool) or not isinstance(version, int) or version < 1):
        c.add(raw.lines["version"], "`version` is the template's own version: a whole number from 1. "
                                    "Raise it when you change a template you have shared.")
    if "format" in raw and raw["format"] != FORMAT:
        c.add(raw.lines["format"], f"This Steward reads template format {FORMAT}.")
    expects = c.text(raw, "expects", "the template")
    if expects and not re.match(r"^[a-zA-Z_:][a-zA-Z0-9_:]*$", expects):
        c.add(raw.lines["expects"], "`expects` should be the start of the metric names it reads, "
                                    "like fabledarchives_.")
    blocks: list[Block] = []
    raw_blocks = raw.get("blocks")
    if "blocks" in raw:
        if not isinstance(raw_blocks, list) or not raw_blocks:
            c.add(raw.lines["blocks"], "`blocks` should be a list of one or more blocks.")
        else:
            if len(raw_blocks) > MAX_BLOCKS:
                c.add(raw.lines["blocks"], f"A template has at most {MAX_BLOCKS} blocks.")
            for i, b in enumerate(raw_blocks):
                block = _block(c, b, raw_blocks.lines[i], i + 1)
                if block:
                    blocks.append(block)
    if c.problems:
        raise TemplateError(sorted(c.problems, key=lambda p: p.line))
    return Template(name=name, version=version, description=c.text(raw, "description", "the template"),
                    expects=expects, blocks=blocks, text=text)


def problems(text: str) -> list[Problem]:
    """Every problem with `text`; empty when it is a template."""
    try:
        check(text)
    except TemplateError as exc:
        return exc.problems
    return []