from __future__ import annotations

import ast
import json
import re
from pathlib import Path
from typing import TypedDict, cast


class OptionalTargetFields(TypedDict, total=False):
    legendFormat: str


class Target(OptionalTargetFields):
    expr: str


class GraphPanel(TypedDict, total=False):
    title: str
    panels: list["GraphPanel"]
    targets: list[Target]


class Row(TypedDict):
    title: str
    panels: list[GraphPanel]


class Variable(TypedDict):
    name: str


class Templating(TypedDict):
    list: list[Variable]


Dashboard = TypedDict(
    "Dashboard",
    {
        "__inputs": "list[dict[str, str]]",
        "panels": "list[Row]",
        "templating": Templating,
    },
)


def load_dashboard(path: Path) -> Dashboard:
    return cast(Dashboard, json.loads(path.read_text(encoding="utf-8")))


def find_row(dashboard: Dashboard, title: str) -> Row:
    matches = [row for row in dashboard["panels"] if row["title"] == title]
    assert len(matches) == 1, f'expected one row titled "{title}", found {len(matches)}'
    return matches[0]


def find_graph_panel(row: Row, title: str) -> GraphPanel:
    matches = [panel for panel in row["panels"] if panel.get("title") == title]
    assert (
        len(matches) == 1
    ), f'expected one panel titled "{title}", found {len(matches)}'
    return matches[0]


def target_expressions(panel: GraphPanel) -> list[str]:
    return [target["expr"] for target in panel.get("targets", [])]


def row_expressions(row: Row) -> list[str]:
    return [
        expression
        for panel in row["panels"]
        for expression in target_expressions(panel)
    ]


def expressions_match(
    expressions: list[str], required: tuple[str, ...], forbidden: tuple[str, ...] = ()
) -> bool:
    return bool(expressions) and all(
        all(fragment in expression for fragment in required)
        and all(fragment not in expression for fragment in forbidden)
        for expression in expressions
    )


def quota_expressions(panel: GraphPanel, legends: set[str]) -> list[str]:
    return [
        target["expr"]
        for target in panel.get("targets", [])
        if target.get("legendFormat") in legends
    ]


def all_targets(dashboard: Dashboard) -> list[Target]:
    return [
        target
        for row in dashboard["panels"]
        for panel in row["panels"]
        for target in panel_targets(panel)
    ]


def panel_targets(panel: GraphPanel) -> list[Target]:
    return panel.get("targets", []) + [
        target for child in panel.get("panels", []) for target in panel_targets(child)
    ]


def thread_cpu_selectors(source_path: Path) -> list[str]:
    source = ast.parse(source_path.read_text(encoding="utf-8"))
    selectors: list[str] = []
    for node in ast.walk(source):
        if not (
            isinstance(node, ast.Call)
            and isinstance(node.func, ast.Name)
            and node.func.id == "thread_cpu_panel_pair"
        ):
            continue
        for keyword in node.keywords:
            if (
                keyword.arg == "label_selector"
                and isinstance(keyword.value, ast.Constant)
                and isinstance(keyword.value.value, str)
            ):
                selectors.append(keyword.value.value)
    return [re.sub(r"\[[^]]*\]", "", selector) for selector in selectors]