Byted Sol Stability Observability Asset Governance

MIT License

Copyright (c) 2026 ByteDance

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

[build-system]
requires = ["setuptools>=68", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "byted-sol-stablity-observability-asset-governance"
version = "0.1.0"
description = "Govern observability assets with drift detection, usage analysis, and incremental update planning"
readme = "SKILL.md"
requires-python = ">=3.10"
dependencies = ["pyyaml>=6.0"]

[project.scripts]
byted-sol-stablity-observability-asset-governance = "observability_asset_governance_skill.cli:main"

[tool.setuptools.packages.find]
where = ["src"]

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from .pipeline import PipelineOptions, run_pipeline

__all__ = ["PipelineOptions", "run_pipeline"]

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from __future__ import annotations

from typing import Dict, List

from .models import NormalizedGovernanceInputs, RegistryEntry


def build_asset_registry(inputs: NormalizedGovernanceInputs) -> Dict[str, object]:
    merged: Dict[str, RegistryEntry] = {}

    for entry in inputs.registry_entries:
        key = f"{entry.asset_type}:{entry.asset_id or entry.name}"
        merged[key] = entry.with_last_seen()

    dashboard_uid = str(inputs.dashboard.get("uid") or "dashboard")
    dashboard_title = str(inputs.dashboard.get("title") or "dashboard")
    dashboard_entry = RegistryEntry(
        asset_id=dashboard_uid,
        asset_type="dashboard",
        name=dashboard_title,
        service=inputs.repo_slug,
        owner="unassigned",
        source="governance_detected",
    ).with_last_seen()
    merged[f"dashboard:{dashboard_uid}"] = dashboard_entry

    for panel in inputs.panel_targets:
        panel_entry = RegistryEntry(
            asset_id=f"{dashboard_uid}:{panel.panel_id}:{panel.target_index}",
            asset_type="query",
            name=f"panel-{panel.panel_id}-{panel.ref_id}",
            service=inputs.repo_slug,
            owner="unassigned",
            source="governance_detected",
        ).with_last_seen()
        merged[f"query:{panel_entry.asset_id}"] = panel_entry

    for sli in inputs.sli_items:
        sli_entry = RegistryEntry(
            asset_id=sli.sli_name,
            asset_type="sli",
            name=sli.sli_name,
            service=inputs.repo_slug,
            owner="unassigned",
            source="governance_detected",
        ).with_last_seen()
        merged[f"sli:{sli_entry.asset_id}"] = sli_entry

    entries: List[RegistryEntry] = [merged[key] for key in sorted(merged)]
    return {
        "summary": {
            "total_assets": len(entries),
            "unassigned_owners": sum(1 for item in entries if item.owner == "unassigned"),
        },
        "assets": [
            {
                "asset_id": item.asset_id,
                "asset_type": item.asset_type,
                "name": item.name,
                "service": item.service,
                "owner": item.owner,
                "status": item.status,
                "source": item.source,
                "last_seen": item.last_seen,
            }
            for item in entries
        ],
    }

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from __future__ import annotations

import argparse
import json
from pathlib import Path
from typing import List

from .pipeline import PipelineOptions, run_pipeline


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Run observability asset governance pipeline")
    parser.add_argument("--sli-spec", required=True, help="Path to SLI spec JSON")
    parser.add_argument("--architecture-spec", required=True, help="Path to architecture spec JSON or directory")
    parser.add_argument("--metric-mapping-spec", required=True, help="Path to metric mapping spec JSON")
    parser.add_argument("--existing-dashboard", required=True, help="Path to existing dashboard JSON")
    parser.add_argument("--metrics-catalog", default="", help="Optional path to metrics catalog JSON")
    parser.add_argument("--usage-stats", default="", help="Optional path to usage stats JSON")
    parser.add_argument("--asset-registry", default="", help="Optional path to existing asset registry JSON")
    parser.add_argument("--grafana-url", default="", help="Optional Grafana base URL")
    parser.add_argument("--grafana-token", default="", help="Optional Grafana token")
    parser.add_argument("--prom-url", default="", help="Optional Prometheus base URL")
    parser.add_argument("--prom-bearer", default="", help="Optional Prometheus bearer token")
    parser.add_argument("--prom-username", default="", help="Optional Prometheus basic auth username")
    parser.add_argument("--prom-password", default="", help="Optional Prometheus basic auth password")
    parser.add_argument("--out-dir", default="output", help="Output base directory")
    parser.add_argument("--focus-service", default="", help="Optional focus service")
    parser.add_argument("--offline", action="store_true", help="Run with offline-only semantics")
    return parser


def _assert_exists(path_value: str, label: str) -> None:
    if not Path(path_value).exists():
        raise SystemExit(f"{label} not found: {path_value}")


def run_cli(argv: List[str] | None = None) -> int:
    args = _parser().parse_args(argv)

    _assert_exists(args.sli_spec, "--sli-spec")
    _assert_exists(args.architecture_spec, "--architecture-spec")
    _assert_exists(args.metric_mapping_spec, "--metric-mapping-spec")
    _assert_exists(args.existing_dashboard, "--existing-dashboard")

    if args.metrics_catalog:
        _assert_exists(args.metrics_catalog, "--metrics-catalog")
    if args.usage_stats:
        _assert_exists(args.usage_stats, "--usage-stats")
    if args.asset_registry:
        _assert_exists(args.asset_registry, "--asset-registry")

    result = run_pipeline(
        sli_spec=args.sli_spec,
        architecture_spec=args.architecture_spec,
        metric_mapping_spec=args.metric_mapping_spec,
        existing_dashboard=args.existing_dashboard,
        metrics_catalog=args.metrics_catalog or None,
        usage_stats=args.usage_stats or None,
        asset_registry=args.asset_registry or None,
        options=PipelineOptions(
            out_dir=args.out_dir,
            focus_service=args.focus_service or None,
            offline=args.offline,
            grafana_url=args.grafana_url,
            grafana_token=args.grafana_token,
            prom_url=args.prom_url,
            prom_bearer=args.prom_bearer,
            prom_username=args.prom_username,
            prom_password=args.prom_password,
        ),
    )

    print(json.dumps(result.to_dict(), ensure_ascii=False, indent=2))
    return 0


def main() -> None:
    raise SystemExit(run_cli())


if __name__ == "__main__":
    main()

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from __future__ import annotations

import re
from typing import Dict, List

from .models import GovernanceFinding, NormalizedGovernanceInputs, PanelTarget, SLIItem


TOKEN_PATTERN = re.compile(r"[a-z0-9_]+")


def _tokens(text: str) -> set[str]:
    return {token for token in TOKEN_PATTERN.findall(text.lower()) if len(token) >= 3}


def _match_score(sli: SLIItem, panel: PanelTarget) -> float:
    sli_tokens = _tokens(f"{sli.sli_name} {sli.sli_type} {sli.measurement}")
    panel_tokens = _tokens(f"{panel.panel_title} {panel.query}")
    if not sli_tokens:
        return 0.0
    overlap = sli_tokens & panel_tokens
    return len(overlap) / len(sli_tokens)


def detect_dashboard_drift(inputs: NormalizedGovernanceInputs) -> Dict[str, object]:
    findings: List[GovernanceFinding] = []

    for sli in inputs.sli_items:
        best = 0.0
        for panel in inputs.panel_targets:
            best = max(best, _match_score(sli, panel))
        if best < 0.2:
            findings.append(
                GovernanceFinding(
                    category="dashboard_drift",
                    finding_type="missing_sli_panel_binding",
                    severity="error",
                    message=f"no panel binding found for SLI: {sli.sli_name}",
                    recommendation="add or update panel/query to reflect current SLI intent",
                    owner="unassigned",
                    asset_refs=[sli.sli_name],
                )
            )

    for panel in inputs.panel_targets:
        query = panel.query.lower()
        if "unknown_" in query or "placeholder" in query:
            findings.append(
                GovernanceFinding(
                    category="dashboard_drift",
                    finding_type="placeholder_query_detected",
                    severity="warning",
                    message=f"panel {panel.panel_id} still uses placeholder query",
                    recommendation="replace with mapped query_template from latest metric mapping",
                    owner="unassigned",
                    asset_refs=[f"panel:{panel.panel_id}"],
                )
            )
        if panel.datasource not in {"prometheus", "loki", "tempo", "internal_tsdb", "grafana"}:
            findings.append(
                GovernanceFinding(
                    category="dashboard_drift",
                    finding_type="unknown_datasource",
                    severity="error",
                    message=f"panel {panel.panel_id} uses unsupported datasource {panel.datasource}",
                    recommendation="normalize datasource to supported value and rebind panel target",
                    owner="unassigned",
                    asset_refs=[f"panel:{panel.panel_id}"],
                )
            )

    return {
        "summary": {
            "total_findings": len(findings),
            "errors": sum(1 for item in findings if item.severity == "error"),
            "warnings": sum(1 for item in findings if item.severity == "warning"),
        },
        "findings": [item.to_dict() for item in findings],
    }

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from __future__ import annotations

from collections import defaultdict
from typing import Dict, List, Set

from .models import GovernanceFinding, NormalizedGovernanceInputs


def detect_definition_conflicts(inputs: NormalizedGovernanceInputs) -> Dict[str, object]:
    findings: List[GovernanceFinding] = []
    dims_by_metric: Dict[str, Set[tuple[str, ...]]] = defaultdict(set)

    for mapping in inputs.mapping_items:
        if mapping.chosen_metric:
            dims_by_metric[mapping.chosen_metric].add(tuple(sorted(set(mapping.dimensions))))

    for metric in inputs.catalog_metrics:
        dims_by_metric[metric.name].add(tuple(sorted(set(metric.dimensions))))

    for metric_name, schemas in dims_by_metric.items():
        if len(schemas) < 2:
            continue
        findings.append(
            GovernanceFinding(
                category="definition_conflict",
                finding_type="metric_label_contract_conflict",
                severity="warning",
                message=f"metric {metric_name} has conflicting label definitions across assets",
                recommendation="standardize metric contract and align mappings/dashboard queries",
                owner="unassigned",
                asset_refs=[metric_name],
            )
        )

    return {
        "summary": {
            "total_findings": len(findings),
            "warnings": len(findings),
            "errors": 0,
        },
        "findings": [item.to_dict() for item in findings],
    }

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from __future__ import annotations

import hashlib
from collections import defaultdict
from typing import Dict, List

from .models import GovernanceFinding, NormalizedGovernanceInputs


def _query_fingerprint(query: str) -> str:
    normalized = " ".join(query.lower().split())
    return hashlib.sha1(normalized.encode("utf-8")).hexdigest()


def detect_duplicate_dashboards(inputs: NormalizedGovernanceInputs) -> Dict[str, object]:
    findings: List[GovernanceFinding] = []
    by_fingerprint: Dict[str, List[int]] = defaultdict(list)

    for panel in inputs.panel_targets:
        by_fingerprint[_query_fingerprint(panel.query)].append(panel.panel_id)

    for fingerprint, panel_ids in by_fingerprint.items():
        if len(panel_ids) < 3:
            continue
        findings.append(
            GovernanceFinding(
                category="duplicate_dashboard",
                finding_type="high_query_duplication",
                severity="warning",
                message=f"multiple panels share identical query fingerprint ({len(panel_ids)} panels)",
                recommendation="consolidate duplicated panels or split by distinct dimensions",
                owner="unassigned",
                asset_refs=[f"fingerprint:{fingerprint}"] + [f"panel:{item}" for item in sorted(set(panel_ids))],
            )
        )

    return {
        "summary": {
            "total_findings": len(findings),
            "warnings": len(findings),
            "errors": 0,
        },
        "findings": [item.to_dict() for item in findings],
    }

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from __future__ import annotations

import json
from pathlib import Path

from .models import PipelineArtifacts


def _slug(text: str) -> str:
    chars = [c.lower() if c.isalnum() else "-" for c in text.strip()]
    slug = "".join(chars)
    while "--" in slug:
        slug = slug.replace("--", "-")
    return slug.strip("-") or "governance"


def write_outputs(out_base_dir: str, repo_slug: str, artifacts: PipelineArtifacts) -> str:
    outdir = Path(out_base_dir) / _slug(repo_slug)
    outdir.mkdir(parents=True, exist_ok=True)

    outputs = {
        "governance-summary.json": artifacts.governance_summary,
        "dashboard-drift-report.json": artifacts.dashboard_drift_report,
        "metric-drift-report.json": artifacts.metric_drift_report,
        "usage-analysis-report.json": artifacts.usage_analysis_report,
        "stale-panel-report.json": artifacts.stale_panel_report,
        "duplicate-dashboard-report.json": artifacts.duplicate_dashboard_report,
        "definition-conflict-report.json": artifacts.definition_conflict_report,
        "incremental-update-plan.json": artifacts.incremental_update_plan,
        "asset-registry.json": artifacts.asset_registry,
        "validation-report.json": artifacts.validation_report.to_dict(),
        "traceability.json": artifacts.traceability,
        "evidence-index.enriched.json": artifacts.evidence_enriched,
    }

    for filename, payload in outputs.items():
        (outdir / filename).write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")

    return str(outdir)

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from __future__ import annotations

from typing import Dict, List

from .models import MetricMappingItem, NormalizedGovernanceInputs


def build_incremental_update_plan(inputs: NormalizedGovernanceInputs) -> Dict[str, object]:
    existing_titles = {panel.panel_title.lower() for panel in inputs.panel_targets}
    actions: List[Dict[str, object]] = []

    capability_tokens = [item.name.lower() for item in inputs.architecture_capabilities]

    for sli in inputs.sli_items:
        name_lower = sli.sli_name.lower()
        mapped: MetricMappingItem | None = next((item for item in inputs.mapping_items if item.sli_name == sli.sli_name), None)

        has_panel = any(name_lower[:12] in title for title in existing_titles)
        if has_panel:
            continue

        path_hint = next((token for token in capability_tokens if token in name_lower), "default_path")
        actions.append(
            {
                "sli_name": sli.sli_name,
                "path_hint": path_hint,
                "suggested_panel_title": f"[AUTO] {sli.sli_name}",
                "suggested_query": (mapped.query_template if mapped and mapped.query_template else "sum(rate(request_success_total[5m]))"),
                "datasource": (mapped.datasource if mapped else "prometheus"),
                "confidence": round((mapped.confidence if mapped else 0.4), 3),
            }
        )

    return {
        "summary": {
            "total_actions": len(actions),
        },
        "actions": actions,
    }

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from __future__ import annotations

import json
from pathlib import Path
from typing import Any, Dict, Iterable, List

from .models import (
    ArchitectureCapability,
    CatalogMetric,
    EvidenceItem,
    MetricMappingItem,
    NormalizedGovernanceInputs,
    PanelTarget,
    RegistryEntry,
    RuntimeConfig,
    SLIItem,
    UsageStat,
)


def _as_list(value: Any) -> List[Any]:
    if value is None:
        return []
    if isinstance(value, list):
        return value
    return [value]


def _pick_text(*values: Any, default: str = "") -> str:
    for value in values:
        if isinstance(value, str) and value.strip():
            return value.strip()
    return default


def _to_dimensions(value: Any) -> List[str]:
    if isinstance(value, list):
        return [str(item).strip() for item in value if str(item).strip()]
    if isinstance(value, str):
        return [item.strip() for item in value.split(",") if item.strip()]
    return []


def _load_json(path: str) -> Any:
    return json.loads(Path(path).read_text(encoding="utf-8"))


def _normalize_sli_items(payload: Any) -> List[SLIItem]:
    records: List[Dict[str, Any]] = []
    if isinstance(payload, list):
        records.extend(item for item in payload if isinstance(item, dict))
    elif isinstance(payload, dict):
        for key in ["sli_indicators", "indicators", "slis", "items"]:
            value = payload.get(key)
            if isinstance(value, list):
                records.extend(item for item in value if isinstance(item, dict))
        if not records and payload:
            records.append(payload)

    result: List[SLIItem] = []
    for idx, item in enumerate(records, start=1):
        result.append(
            SLIItem(
                sli_name=_pick_text(item.get("sli_name"), item.get("name"), item.get("id"), default=f"sli-{idx}"),
                sli_type=_pick_text(item.get("sli_type"), item.get("type"), default="availability"),
                measurement=_pick_text(item.get("measurement"), item.get("formula"), item.get("query")),
                target=_pick_text(item.get("target"), item.get("target_slo"), item.get("objective"), item.get("slo")),
                dimensions=_to_dimensions(item.get("dimensions") or item.get("dimension")),
            )
        )
    return result


def _normalize_mapping_items(payload: Any) -> List[MetricMappingItem]:
    records = payload if isinstance(payload, list) else _as_list(payload.get("items") if isinstance(payload, dict) else payload)
    result: List[MetricMappingItem] = []
    for item in records:
        if not isinstance(item, dict):
            continue
        datasource = _pick_text(item.get("datasource"), default="prometheus").lower().replace(" ", "_")
        if datasource in {"internal", "internal-tsdb", "internaltsdb"}:
            datasource = "internal_tsdb"
        result.append(
            MetricMappingItem(
                sli_name=_pick_text(item.get("sli_name"), item.get("name")),
                chosen_metric=_pick_text(item.get("chosen_metric"), item.get("metric")),
                datasource=datasource,
                query_template=_pick_text(item.get("query_template"), item.get("query")),
                dimensions=_to_dimensions(item.get("dimensions") or item.get("labels")),
                confidence=float(item.get("confidence") or 0.0),
                missing_gap=_pick_text(item.get("missing_gap"), default="none"),
            )
        )
    return result


def _normalize_catalog_metrics(payload: Any, mappings: List[MetricMappingItem]) -> List[CatalogMetric]:
    records: List[Dict[str, Any]] = []
    if isinstance(payload, list):
        records.extend(item for item in payload if isinstance(item, dict))
    elif isinstance(payload, dict):
        values = payload.get("metrics") or payload.get("items")
        if isinstance(values, list):
            records.extend(item for item in values if isinstance(item, dict))
        elif payload:
            records.append(payload)

    result: List[CatalogMetric] = []
    for item in records:
        name = _pick_text(item.get("name"), item.get("metric"), item.get("metric_name"))
        if not name:
            continue
        datasource = _pick_text(item.get("datasource"), item.get("source"), default="prometheus").lower().replace(" ", "_")
        if datasource in {"internal", "internal-tsdb", "internaltsdb"}:
            datasource = "internal_tsdb"
        result.append(CatalogMetric(name=name, datasource=datasource, dimensions=_to_dimensions(item.get("dimensions") or item.get("labels"))))

    known = {item.name for item in result}
    for mapping in mappings:
        if mapping.chosen_metric and mapping.chosen_metric not in known:
            result.append(CatalogMetric(name=mapping.chosen_metric, datasource=mapping.datasource, dimensions=mapping.dimensions))
            known.add(mapping.chosen_metric)

    return result


def _iter_panels(panels: Iterable[Dict[str, Any]]) -> Iterable[Dict[str, Any]]:
    for panel in panels:
        yield panel
        nested = panel.get("panels")
        if isinstance(nested, list):
            for sub in _iter_panels(nested):
                yield sub


def _extract_panel_targets(dashboard: Dict[str, Any], fallback_datasource: str = "prometheus") -> List[PanelTarget]:
    result: List[PanelTarget] = []
    for panel in _iter_panels(dashboard.get("panels", [])):
        panel_id = int(panel.get("id") or -1)
        panel_title = _pick_text(panel.get("title"), default="Untitled")
        panel_type = _pick_text(panel.get("type"), default="timeseries")
        panel_ds = panel.get("datasource")
        panel_ds_name = fallback_datasource
        if isinstance(panel_ds, dict):
            panel_ds_name = _pick_text(panel_ds.get("type"), panel_ds.get("uid"), default=fallback_datasource)
        elif isinstance(panel_ds, str) and panel_ds.strip():
            panel_ds_name = panel_ds.strip()

        targets = panel.get("targets") if isinstance(panel.get("targets"), list) else []
        for index, target in enumerate(targets):
            if not isinstance(target, dict):
                continue
            query = _pick_text(target.get("expr"), target.get("query"))
            ref_id = _pick_text(target.get("refId"), default=chr(ord("A") + index))
            target_ds = target.get("datasource")
            datasource = panel_ds_name
            if isinstance(target_ds, dict):
                datasource = _pick_text(target_ds.get("type"), target_ds.get("uid"), default=datasource)
            elif isinstance(target_ds, str) and target_ds.strip():
                datasource = target_ds.strip()

            result.append(
                PanelTarget(
                    panel_id=panel_id,
                    panel_title=panel_title,
                    panel_type=panel_type,
                    target_index=index,
                    ref_id=ref_id,
                    datasource=datasource.lower().replace(" ", "_"),
                    query=query,
                )
            )
    return result


def _normalize_usage_stats(payload: Any, dashboard: Dict[str, Any]) -> List[UsageStat]:
    records: List[Dict[str, Any]] = []
    if isinstance(payload, list):
        records.extend(item for item in payload if isinstance(item, dict))
    elif isinstance(payload, dict):
        values = payload.get("dashboards") or payload.get("items")
        if isinstance(values, list):
            records.extend(item for item in values if isinstance(item, dict))
        elif payload:
            records.append(payload)

    result: List[UsageStat] = []
    for item in records:
        result.append(
            UsageStat(
                dashboard_uid=_pick_text(item.get("dashboard_uid"), item.get("uid"), default=_pick_text(dashboard.get("uid"), default="dashboard")),
                dashboard_title=_pick_text(item.get("dashboard_title"), item.get("title"), default=_pick_text(dashboard.get("title"), default="dashboard")),
                views=int(item.get("views") or 0),
                favorites=int(item.get("favorites") or 0),
                oncall_visits=int(item.get("oncall_visits") or 0),
                panel_views={str(k): int(v) for k, v in (item.get("panel_views") or {}).items()} if isinstance(item.get("panel_views"), dict) else {},
            )
        )

    if not result:
        result.append(
            UsageStat(
                dashboard_uid=_pick_text(dashboard.get("uid"), default="dashboard"),
                dashboard_title=_pick_text(dashboard.get("title"), default="dashboard"),
            )
        )

    return result


def _normalize_registry_entries(payload: Any) -> List[RegistryEntry]:
    records: List[Dict[str, Any]] = []
    if isinstance(payload, list):
        records.extend(item for item in payload if isinstance(item, dict))
    elif isinstance(payload, dict):
        values = payload.get("assets") or payload.get("items")
        if isinstance(values, list):
            records.extend(item for item in values if isinstance(item, dict))
        elif payload:
            records.append(payload)

    result: List[RegistryEntry] = []
    for item in records:
        result.append(
            RegistryEntry(
                asset_id=_pick_text(item.get("asset_id"), item.get("id"), default=""),
                asset_type=_pick_text(item.get("asset_type"), item.get("type"), default="dashboard"),
                name=_pick_text(item.get("name"), item.get("title"), default="unknown"),
                service=_pick_text(item.get("service"), default="unknown"),
                owner=_pick_text(item.get("owner"), default="unassigned"),
                status=_pick_text(item.get("status"), default="active"),
                source=_pick_text(item.get("source"), default="input_registry"),
                last_seen=_pick_text(item.get("last_seen")),
            )
        )
    return result


def _load_architecture_capabilities(path: str) -> List[ArchitectureCapability]:
    p = Path(path)
    if p.is_file():
        return [ArchitectureCapability(name=p.stem, source=str(p))]
    if not p.exists() or not p.is_dir():
        return []

    result: List[ArchitectureCapability] = []
    for item in sorted(p.iterdir()):
        if item.name.startswith("."):
            continue
        if item.is_dir():
            result.append(ArchitectureCapability(name=item.name, source=str(item)))
        elif item.suffix.lower() in {".md", ".json", ".yaml", ".yml"}:
            result.append(ArchitectureCapability(name=item.stem, source=str(item)))
    return result


def _slug(text: str) -> str:
    chars = [c.lower() if c.isalnum() else "-" for c in text.strip()]
    slug = "".join(chars)
    while "--" in slug:
        slug = slug.replace("--", "-")
    return slug.strip("-") or "governance"


def _derive_repo_slug(existing_dashboard_path: str, dashboard: Dict[str, Any], focus_service: str | None) -> str:
    if focus_service and focus_service.strip():
        return _slug(focus_service)
    title = _pick_text(dashboard.get("title"))
    if title:
        return _slug(title)
    return _slug(Path(existing_dashboard_path).stem)


def normalize_inputs(
    sli_spec_path: str,
    architecture_spec_path: str,
    metric_mapping_spec_path: str,
    existing_dashboard_path: str,
    metrics_catalog_path: str | None,
    usage_stats_path: str | None,
    asset_registry_path: str | None,
    runtime: RuntimeConfig,
    focus_service: str | None = None,
) -> NormalizedGovernanceInputs:
    sli_payload = _load_json(sli_spec_path)
    mapping_payload = _load_json(metric_mapping_spec_path)
    dashboard_payload = _load_json(existing_dashboard_path)
    catalog_payload = _load_json(metrics_catalog_path) if metrics_catalog_path else {}
    usage_payload = _load_json(usage_stats_path) if usage_stats_path else {}
    registry_payload = _load_json(asset_registry_path) if asset_registry_path else {}

    sli_items = _normalize_sli_items(sli_payload)
    mapping_items = _normalize_mapping_items(mapping_payload)
    catalog_metrics = _normalize_catalog_metrics(catalog_payload, mapping_items)
    dashboard = dashboard_payload if isinstance(dashboard_payload, dict) else {}
    panel_targets = _extract_panel_targets(dashboard)
    usage_stats = _normalize_usage_stats(usage_payload, dashboard)
    registry_entries = _normalize_registry_entries(registry_payload)
    architecture_capabilities = _load_architecture_capabilities(architecture_spec_path)

    evidence_items: List[EvidenceItem] = []
    for idx, item in enumerate(sli_items, start=1):
        evidence_items.append(EvidenceItem(f"ev-sli-{idx}", "sli_spec", sli_spec_path, f"sli[{idx-1}]", item.sli_name))
    for idx, item in enumerate(mapping_items, start=1):
        evidence_items.append(EvidenceItem(f"ev-map-{idx}", "metric_mapping_spec", metric_mapping_spec_path, f"mapping[{idx-1}]", f"{item.sli_name}->{item.chosen_metric}"))
    for idx, item in enumerate(catalog_metrics, start=1):
        evidence_items.append(EvidenceItem(f"ev-metric-{idx}", "metrics_catalog", metrics_catalog_path or "derived", item.name, item.name))
    for idx, item in enumerate(panel_targets, start=1):
        evidence_items.append(EvidenceItem(f"ev-panel-{idx}", "existing_dashboard", existing_dashboard_path, f"panel={item.panel_id}/target={item.target_index}", item.panel_title))
    for idx, item in enumerate(architecture_capabilities, start=1):
        evidence_items.append(EvidenceItem(f"ev-arch-{idx}", "architecture_spec", item.source, item.name, item.name))

    return NormalizedGovernanceInputs(
        repo_slug=_derive_repo_slug(existing_dashboard_path, dashboard, focus_service),
        sli_items=sli_items,
        architecture_capabilities=architecture_capabilities,
        mapping_items=mapping_items,
        catalog_metrics=catalog_metrics,
        dashboard=dashboard,
        panel_targets=panel_targets,
        usage_stats=usage_stats,
        registry_entries=registry_entries,
        evidence_items=evidence_items,
        runtime=runtime,
    )

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from __future__ import annotations

import re
from typing import Dict, List, Set

from .models import GovernanceFinding, MetricMappingItem, NormalizedGovernanceInputs


METRIC_PATTERN = re.compile(r"\b([a-zA-Z_:][a-zA-Z0-9_:]*)\b")
RESERVED = {
    "sum",
    "rate",
    "irate",
    "increase",
    "clamp_min",
    "clamp_max",
    "histogram_quantile",
    "avg",
    "min",
    "max",
    "count",
    "by",
    "without",
    "topk",
    "bottomk",
}


def _extract_metric_names(query: str) -> Set[str]:
    names = set()
    for token in METRIC_PATTERN.findall(query):
        if token not in RESERVED:
            names.add(token)
    return names


def _catalog_lookup(inputs: NormalizedGovernanceInputs) -> Dict[str, Set[str]]:
    lookup: Dict[str, Set[str]] = {}
    for item in inputs.catalog_metrics:
        lookup[item.name] = set(item.dimensions)
    return lookup


def detect_metric_drift(inputs: NormalizedGovernanceInputs) -> Dict[str, object]:
    findings: List[GovernanceFinding] = []
    catalog = _catalog_lookup(inputs)

    for mapping in inputs.mapping_items:
        if not mapping.chosen_metric:
            continue

        if mapping.chosen_metric not in catalog:
            findings.append(
                GovernanceFinding(
                    category="metric_drift",
                    finding_type="metric_sunset_or_missing",
                    severity="error",
                    message=f"mapped metric not found in catalog: {mapping.chosen_metric}",
                    recommendation="replace metric using schema-aware repair and refresh mapping",
                    owner="unassigned",
                    asset_refs=[mapping.sli_name, mapping.chosen_metric],
                )
            )
            continue

        required = set(mapping.dimensions)
        available = catalog[mapping.chosen_metric]
        if required and not required.issubset(available):
            missing = sorted(required - available)
            findings.append(
                GovernanceFinding(
                    category="metric_drift",
                    finding_type="label_schema_drift",
                    severity="warning",
                    message=f"label drift on {mapping.chosen_metric}: missing {', '.join(missing)}",
                    recommendation="update query label set or instrumentation labels to restore contract",
                    owner="unassigned",
                    asset_refs=[mapping.chosen_metric],
                )
            )

        if mapping.confidence < 0.5 and mapping.missing_gap == "none":
            findings.append(
                GovernanceFinding(
                    category="metric_drift",
                    finding_type="confidence_gap_conflict",
                    severity="warning",
                    message=f"low confidence mapping marked as none gap: {mapping.sli_name}",
                    recommendation="mark drift gap explicitly and trigger query repair suggestion",
                    owner="unassigned",
                    asset_refs=[mapping.sli_name, mapping.chosen_metric],
                )
            )

    for panel in inputs.panel_targets:
        query = panel.query.lower()
        if any(token in query for token in ["absent(", "missing_metric", "no_data_metric"]):
            findings.append(
                GovernanceFinding(
                    category="metric_drift",
                    finding_type="no_data_query_pattern",
                    severity="warning",
                    message=f"panel {panel.panel_id} query indicates no-data risk",
                    recommendation="validate metric liveliness window and migrate to active metric",
                    owner="unassigned",
                    asset_refs=[f"panel:{panel.panel_id}"],
                )
            )
        unknown_metrics = [name for name in _extract_metric_names(panel.query) if name not in catalog]
        if unknown_metrics:
            findings.append(
                GovernanceFinding(
                    category="metric_drift",
                    finding_type="query_uses_unknown_metric",
                    severity="warning",
                    message=f"panel {panel.panel_id} references unknown metrics: {', '.join(sorted(set(unknown_metrics)))}",
                    recommendation="re-generate query from latest mapping/catalog",
                    owner="unassigned",
                    asset_refs=[f"panel:{panel.panel_id}"],
                )
            )

    return {
        "summary": {
            "total_findings": len(findings),
            "errors": sum(1 for item in findings if item.severity == "error"),
            "warnings": sum(1 for item in findings if item.severity == "warning"),
        },
        "findings": [item.to_dict() for item in findings],
    }

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from __future__ import annotations

from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional


KNOWN_DATASOURCES = {"prometheus", "loki", "tempo", "internal_tsdb", "grafana"}
SEVERITY_VALUES = {"info", "warning", "error"}


@dataclass
class SLIItem:
    sli_name: str
    sli_type: str
    measurement: str = ""
    target: str = ""
    dimensions: List[str] = field(default_factory=list)


@dataclass
class ArchitectureCapability:
    name: str
    source: str


@dataclass
class MetricMappingItem:
    sli_name: str
    chosen_metric: str
    datasource: str
    query_template: str
    dimensions: List[str] = field(default_factory=list)
    confidence: float = 0.0
    missing_gap: str = "none"


@dataclass
class CatalogMetric:
    name: str
    datasource: str
    dimensions: List[str] = field(default_factory=list)


@dataclass
class PanelTarget:
    panel_id: int
    panel_title: str
    panel_type: str
    target_index: int
    ref_id: str
    datasource: str
    query: str


@dataclass
class UsageStat:
    dashboard_uid: str
    dashboard_title: str
    views: int = 0
    favorites: int = 0
    oncall_visits: int = 0
    panel_views: Dict[str, int] = field(default_factory=dict)


@dataclass
class RegistryEntry:
    asset_id: str
    asset_type: str
    name: str
    service: str
    owner: str = "unassigned"
    status: str = "active"
    source: str = "generated"
    last_seen: str = ""

    def with_last_seen(self) -> "RegistryEntry":
        if self.last_seen:
            return self
        self.last_seen = datetime.now(timezone.utc).isoformat()
        return self


@dataclass
class EvidenceItem:
    evidence_id: str
    source_type: str
    source_path: str
    locator: str
    summary: str


@dataclass
class RuntimeConfig:
    offline: bool = False
    grafana_url: str = ""
    grafana_token: str = ""
    prom_url: str = ""
    prom_bearer: str = ""
    prom_username: str = ""
    prom_password: str = ""


@dataclass
class NormalizedGovernanceInputs:
    repo_slug: str
    sli_items: List[SLIItem] = field(default_factory=list)
    architecture_capabilities: List[ArchitectureCapability] = field(default_factory=list)
    mapping_items: List[MetricMappingItem] = field(default_factory=list)
    catalog_metrics: List[CatalogMetric] = field(default_factory=list)
    dashboard: Dict[str, Any] = field(default_factory=dict)
    panel_targets: List[PanelTarget] = field(default_factory=list)
    usage_stats: List[UsageStat] = field(default_factory=list)
    registry_entries: List[RegistryEntry] = field(default_factory=list)
    evidence_items: List[EvidenceItem] = field(default_factory=list)
    runtime: RuntimeConfig = field(default_factory=RuntimeConfig)


@dataclass
class GovernanceFinding:
    category: str
    finding_type: str
    severity: str
    message: str
    recommendation: str
    owner: str = "unassigned"
    asset_refs: List[str] = field(default_factory=list)

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)


@dataclass
class ValidationIssue:
    level: str
    rule: str
    message: str


@dataclass
class ValidationReport:
    passed: bool
    summary: Dict[str, int]
    issues: List[ValidationIssue] = field(default_factory=list)

    def to_dict(self) -> Dict[str, Any]:
        return {
            "passed": self.passed,
            "summary": self.summary,
            "issues": [asdict(item) for item in self.issues],
        }


@dataclass
class PipelineArtifacts:
    governance_summary: Dict[str, Any]
    dashboard_drift_report: Dict[str, Any]
    metric_drift_report: Dict[str, Any]
    usage_analysis_report: Dict[str, Any]
    stale_panel_report: Dict[str, Any]
    duplicate_dashboard_report: Dict[str, Any]
    definition_conflict_report: Dict[str, Any]
    incremental_update_plan: Dict[str, Any]
    asset_registry: Dict[str, Any]
    validation_report: ValidationReport
    traceability: Dict[str, Any]
    evidence_enriched: List[Dict[str, Any]]


@dataclass
class PipelineResult:
    output_dir: str
    finding_count: int
    overall_pass: bool

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Dict, Optional

from .asset_registry import build_asset_registry
from .dashboard_drift_detector import detect_dashboard_drift
from .definition_conflict_detector import detect_definition_conflicts
from .duplicate_dashboard_detector import detect_duplicate_dashboards
from .exporter import write_outputs
from .incremental_updater import build_incremental_update_plan
from .input_normalizer import normalize_inputs
from .metric_drift_detector import detect_metric_drift
from .models import PipelineArtifacts, PipelineResult, RuntimeConfig
from .stale_panel_detector import detect_stale_panels
from .traceability import build_traceability
from .usage_analyzer import analyze_usage
from .validator import validate_outputs


@dataclass
class PipelineOptions:
    out_dir: str = "output"
    focus_service: Optional[str] = None
    offline: bool = False
    grafana_url: str = ""
    grafana_token: str = ""
    prom_url: str = ""
    prom_bearer: str = ""
    prom_username: str = ""
    prom_password: str = ""


def _findings_count(report: Dict[str, Any]) -> int:
    findings = report.get("findings")
    if isinstance(findings, list):
        return len(findings)
    return 0


def _severity_count(report: Dict[str, Any], severity: str) -> int:
    findings = report.get("findings")
    if not isinstance(findings, list):
        return 0
    return sum(1 for item in findings if isinstance(item, dict) and str(item.get("severity") or "").lower() == severity)


def run_pipeline(
    sli_spec: str,
    architecture_spec: str,
    metric_mapping_spec: str,
    existing_dashboard: str,
    metrics_catalog: str | None = None,
    usage_stats: str | None = None,
    asset_registry: str | None = None,
    options: PipelineOptions | None = None,
) -> PipelineResult:
    opts = options or PipelineOptions()

    runtime = RuntimeConfig(
        offline=opts.offline,
        grafana_url=opts.grafana_url,
        grafana_token=opts.grafana_token,
        prom_url=opts.prom_url,
        prom_bearer=opts.prom_bearer,
        prom_username=opts.prom_username,
        prom_password=opts.prom_password,
    )

    inputs = normalize_inputs(
        sli_spec_path=sli_spec,
        architecture_spec_path=architecture_spec,
        metric_mapping_spec_path=metric_mapping_spec,
        existing_dashboard_path=existing_dashboard,
        metrics_catalog_path=metrics_catalog,
        usage_stats_path=usage_stats,
        asset_registry_path=asset_registry,
        runtime=runtime,
        focus_service=opts.focus_service,
    )

    dashboard_drift_report = detect_dashboard_drift(inputs)
    metric_drift_report = detect_metric_drift(inputs)
    usage_analysis_report = analyze_usage(inputs)
    stale_panel_report = detect_stale_panels(inputs)
    duplicate_dashboard_report = detect_duplicate_dashboards(inputs)
    definition_conflict_report = detect_definition_conflicts(inputs)
    incremental_update_plan = build_incremental_update_plan(inputs)
    registry_report = build_asset_registry(inputs)

    category_reports = {
        "dashboard_drift": dashboard_drift_report,
        "metric_drift": metric_drift_report,
        "usage_analysis": usage_analysis_report,
        "stale_panel": stale_panel_report,
        "duplicate_dashboard": duplicate_dashboard_report,
        "definition_conflict": definition_conflict_report,
    }

    total_findings = sum(_findings_count(report) for report in category_reports.values())
    governance_summary = {
        "repo_slug": inputs.repo_slug,
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "total_findings": total_findings,
        "errors": sum(_severity_count(report, "error") for report in category_reports.values()),
        "warnings": sum(_severity_count(report, "warning") for report in category_reports.values()),
        "infos": sum(_severity_count(report, "info") for report in category_reports.values()),
        "category_counts": {name: _findings_count(report) for name, report in category_reports.items()},
        "total_assets": int((registry_report.get("summary") or {}).get("total_assets") or 0),
        "total_incremental_actions": int((incremental_update_plan.get("summary") or {}).get("total_actions") or 0),
    }

    validation_report = validate_outputs(
        governance_summary=governance_summary,
        dashboard_drift_report=dashboard_drift_report,
        metric_drift_report=metric_drift_report,
        usage_analysis_report=usage_analysis_report,
        stale_panel_report=stale_panel_report,
        duplicate_dashboard_report=duplicate_dashboard_report,
        definition_conflict_report=definition_conflict_report,
        asset_registry=registry_report,
    )

    traceability = build_traceability(
        inputs=inputs,
        dashboard_drift_report=dashboard_drift_report,
        metric_drift_report=metric_drift_report,
        stale_panel_report=stale_panel_report,
    )

    evidence_enriched = [
        {
            "evidence_id": item.evidence_id,
            "source_type": item.source_type,
            "source_path": item.source_path,
            "locator": item.locator,
            "summary": item.summary,
            "repo_slug": inputs.repo_slug,
        }
        for item in inputs.evidence_items
    ]

    artifacts = PipelineArtifacts(
        governance_summary=governance_summary,
        dashboard_drift_report=dashboard_drift_report,
        metric_drift_report=metric_drift_report,
        usage_analysis_report=usage_analysis_report,
        stale_panel_report=stale_panel_report,
        duplicate_dashboard_report=duplicate_dashboard_report,
        definition_conflict_report=definition_conflict_report,
        incremental_update_plan=incremental_update_plan,
        asset_registry=registry_report,
        validation_report=validation_report,
        traceability=traceability,
        evidence_enriched=evidence_enriched,
    )

    outdir = write_outputs(opts.out_dir, inputs.repo_slug, artifacts)
    return PipelineResult(output_dir=outdir, finding_count=total_findings, overall_pass=validation_report.passed)

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from __future__ import annotations

from typing import Dict, List

from .models import GovernanceFinding, NormalizedGovernanceInputs


def detect_stale_panels(inputs: NormalizedGovernanceInputs) -> Dict[str, object]:
    findings: List[GovernanceFinding] = []
    usage = {item.dashboard_uid: item for item in inputs.usage_stats}
    dashboard_uid = str(inputs.dashboard.get("uid") or "dashboard")
    dashboard_usage = usage.get(dashboard_uid)
    panel_views = dashboard_usage.panel_views if dashboard_usage else {}

    for panel in inputs.panel_targets:
        panel_key = str(panel.panel_id)
        panel_view_count = int(panel_views.get(panel_key, 0))
        query = panel.query.lower()

        if panel_view_count == 0:
            findings.append(
                GovernanceFinding(
                    category="stale_panel",
                    finding_type="unvisited_panel",
                    severity="warning",
                    message=f"panel {panel.panel_id} has no usage records",
                    recommendation="review panel relevance and archive if redundant",
                    owner="unassigned",
                    asset_refs=[f"panel:{panel.panel_id}"],
                )
            )

        if any(token in query for token in ["absent(", "vector(0)", "no_data_metric", "missing_metric"]):
            findings.append(
                GovernanceFinding(
                    category="stale_panel",
                    finding_type="long_empty_pattern",
                    severity="warning",
                    message=f"panel {panel.panel_id} query indicates persistent empty data",
                    recommendation="replace with active metric or remove stale panel",
                    owner="unassigned",
                    asset_refs=[f"panel:{panel.panel_id}"],
                )
            )

    return {
        "summary": {
            "total_findings": len(findings),
            "warnings": len(findings),
            "errors": 0,
        },
        "findings": [item.to_dict() for item in findings],
    }

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from __future__ import annotations

import re
from typing import Any, Dict, List

from .models import NormalizedGovernanceInputs


METRIC_PATTERN = re.compile(r"\b([a-zA-Z_:][a-zA-Z0-9_:]*)\b")
RESERVED = {
    "sum",
    "rate",
    "irate",
    "increase",
    "clamp_min",
    "clamp_max",
    "histogram_quantile",
    "avg",
    "min",
    "max",
    "count",
    "by",
    "without",
    "topk",
    "bottomk",
}


def _query_metrics(query: str) -> List[str]:
    metrics = []
    for token in METRIC_PATTERN.findall(query):
        if token in RESERVED:
            continue
        metrics.append(token)
    return sorted(set(metrics))


def build_traceability(
    inputs: NormalizedGovernanceInputs,
    dashboard_drift_report: Dict[str, Any],
    metric_drift_report: Dict[str, Any],
    stale_panel_report: Dict[str, Any],
) -> Dict[str, Any]:
    sli_links = []
    for mapping in inputs.mapping_items:
        sli_links.append(
            {
                "sli_name": mapping.sli_name,
                "metric": mapping.chosen_metric,
                "datasource": mapping.datasource,
                "confidence": mapping.confidence,
            }
        )

    panel_links = []
    for panel in inputs.panel_targets:
        panel_links.append(
            {
                "panel_id": panel.panel_id,
                "panel_title": panel.panel_title,
                "query": panel.query,
                "metrics": _query_metrics(panel.query),
            }
        )

    finding_links = []
    for category, report in [
        ("dashboard_drift", dashboard_drift_report),
        ("metric_drift", metric_drift_report),
        ("stale_panel", stale_panel_report),
    ]:
        for item in report.get("findings", []):
            finding_links.append(
                {
                    "category": category,
                    "severity": item.get("severity"),
                    "asset_refs": item.get("asset_refs", []),
                    "message": item.get("message"),
                }
            )

    return {
        "sli_to_metric": sli_links,
        "panel_to_query": panel_links,
        "finding_links": finding_links,
        "evidence_refs": [
            {
                "evidence_id": item.evidence_id,
                "source_type": item.source_type,
                "source_path": item.source_path,
                "locator": item.locator,
            }
            for item in inputs.evidence_items
        ],
    }

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from __future__ import annotations

from typing import Dict, List

from .models import GovernanceFinding, NormalizedGovernanceInputs


def analyze_usage(inputs: NormalizedGovernanceInputs) -> Dict[str, object]:
    findings: List[GovernanceFinding] = []

    for stat in inputs.usage_stats:
        if stat.views == 0:
            findings.append(
                GovernanceFinding(
                    category="usage_analysis",
                    finding_type="unused_dashboard",
                    severity="warning",
                    message=f"dashboard has zero views: {stat.dashboard_title}",
                    recommendation="mark for archive review or validate routing/oncall links",
                    owner="unassigned",
                    asset_refs=[stat.dashboard_uid],
                )
            )
        if stat.favorites == 0 and stat.oncall_visits > 0:
            findings.append(
                GovernanceFinding(
                    category="usage_analysis",
                    finding_type="oncall_low_subscription",
                    severity="info",
                    message=f"oncall relies on dashboard but favorites remain low: {stat.dashboard_title}",
                    recommendation="improve discoverability and pin dashboard in oncall runbook",
                    owner="unassigned",
                    asset_refs=[stat.dashboard_uid],
                )
            )
        if stat.oncall_visits == 0 and stat.views > 0:
            findings.append(
                GovernanceFinding(
                    category="usage_analysis",
                    finding_type="non_oncall_usage_only",
                    severity="info",
                    message=f"dashboard appears used outside oncall workflows: {stat.dashboard_title}",
                    recommendation="verify incident-response ownership and escalation linkage",
                    owner="unassigned",
                    asset_refs=[stat.dashboard_uid],
                )
            )

    return {
        "summary": {
            "total_dashboards": len(inputs.usage_stats),
            "total_findings": len(findings),
            "errors": 0,
            "warnings": sum(1 for item in findings if item.severity == "warning"),
            "infos": sum(1 for item in findings if item.severity == "info"),
        },
        "findings": [item.to_dict() for item in findings],
    }

# Copyright (c) 2026 ByteDance
# SPDX-License-Identifier: MIT

from __future__ import annotations

from typing import Any, Dict, List

from .models import SEVERITY_VALUES, ValidationIssue, ValidationReport


def _findings(report: Dict[str, Any]) -> List[Dict[str, Any]]:
    values = report.get("findings")
    if isinstance(values, list):
        return [item for item in values if isinstance(item, dict)]
    return []


def validate_outputs(
    governance_summary: Dict[str, Any],
    dashboard_drift_report: Dict[str, Any],
    metric_drift_report: Dict[str, Any],
    usage_analysis_report: Dict[str, Any],
    stale_panel_report: Dict[str, Any],
    duplicate_dashboard_report: Dict[str, Any],
    definition_conflict_report: Dict[str, Any],
    asset_registry: Dict[str, Any],
) -> ValidationReport:
    issues: List[ValidationIssue] = []

    reports = [
        dashboard_drift_report,
        metric_drift_report,
        usage_analysis_report,
        stale_panel_report,
        duplicate_dashboard_report,
        definition_conflict_report,
    ]

    flattened: List[Dict[str, Any]] = []
    for report in reports:
        flattened.extend(_findings(report))

    for item in flattened:
        severity = str(item.get("severity") or "").lower()
        if severity not in SEVERITY_VALUES:
            issues.append(ValidationIssue(level="error", rule="severity_enum", message=f"invalid severity: {severity or 'empty'}"))
        recommendation = str(item.get("recommendation") or "").strip()
        if not recommendation:
            issues.append(ValidationIssue(level="error", rule="recommendation_required", message="finding missing recommendation"))
        owner = str(item.get("owner") or "").strip()
        if not owner:
            issues.append(ValidationIssue(level="error", rule="owner_required", message="finding missing owner"))

    expected_total = len(flattened)
    actual_total = int(governance_summary.get("total_findings") or 0)
    if expected_total != actual_total:
        issues.append(
            ValidationIssue(
                level="error",
                rule="summary_recount",
                message=f"governance_summary total_findings mismatch expected={expected_total} actual={actual_total}",
            )
        )

    assets = asset_registry.get("assets")
    if not isinstance(assets, list):
        issues.append(ValidationIssue(level="error", rule="asset_registry_shape", message="asset_registry.assets must be a list"))
    else:
        for item in assets:
            if not isinstance(item, dict):
                issues.append(ValidationIssue(level="error", rule="asset_registry_entry_shape", message="asset entry must be object"))
                continue
            if not str(item.get("asset_type") or "").strip():
                issues.append(ValidationIssue(level="error", rule="asset_type_required", message="asset entry missing asset_type"))
            if not str(item.get("name") or "").strip():
                issues.append(ValidationIssue(level="error", rule="asset_name_required", message="asset entry missing name"))

    summary = {
        "errors": sum(1 for item in issues if item.level == "error"),
        "warnings": sum(1 for item in issues if item.level == "warning"),
        "total_findings": expected_total,
    }

    return ValidationReport(passed=summary["errors"] == 0, summary=summary, issues=issues)