"""Versioned, deterministic health scoring primitives."""
from __future__ import annotations

import hashlib
import json
from collections.abc import Iterable
from dataclasses import dataclass
from datetime import date, datetime
from decimal import ROUND_HALF_UP, Decimal
from enum import Enum, StrEnum
from typing import Any

# v2 (2026-08-18): two lifecycle definition fixes that could not be applied
# in place — a metric口径 change never rewrites history, it takes a new
# version (see DECISIONS.md 2026-08-18).
#   * ``lifecycle.leaver_failure_rate`` counted a leaver as failed on any
#     historical retry (``error_count`` never resets) and on the ordinary
#     in-flight ``blocked`` status, so it was pinned to the worst band
#     regardless of how offboarding actually ran.
#   * ``lifecycle.transfer_recovery`` was retired: its only signal source
#     (``provisioning_tasks.source='role_change'``) has never produced a row
#     in production, so the metric was permanently unavailable.
# v3 (2026-08-18): three asset definition fixes plus one lifecycle
# definition fix, all landed the same day as v2 and for the same reason — a
# metric口径 change never rewrites history, it takes a new version (see
# DECISIONS.md 2026-08-18). Both batches share v3 because no v3 snapshot had
# been written yet when the second landed, so the version has one consistent
# meaning from its first row onward.
#   * ``assets.inventory_accuracy`` scored an unanswered stocktake item as a
#     wrong record, so a cycle opened at its worst score and stayed pinned
#     there after closing (closing freezes the remainder at ``pending``).
#     Response rate moved to confidence, where it belongs.
#   * ``assets.idle_rate`` dated idleness from ``asset_items.updated_at`` — a
#     row-mutation stamp that *answering a stocktake* resets — and counted
#     monitors, printers and office network gear as reclaimable capital.
#   * ``assets.warranty_coverage`` was retired: ``warranty_until`` has never
#     been written in production and no UI control produces it.
#   * ``lifecycle.onboarding_sla`` stopped reading the CURRENT
#     ``sla_breach_level`` — a flag ``evaluate_sla`` clears on every terminal
#     process and resets when the last blocker closes, so it scored "not red
#     right now", not the attainment its label promises. It now scores the
#     shipped 72h SLA against history over the processes whose window has
#     closed.
# The same release also carries three software/API fact-read fixes that do
# NOT touch definitions (adopted console keys and not-yet-issued approvals
# out of the key-expiry denominator; KNOWN 备案保留 external holders netted
# into reconciliation, mirroring the panel that owns those rows).
# ``health-v4`` (2026-09-01) renames ten metric KEYS and changes no
# definition, threshold, weight or reading. It is a version bump because
# the key IS the metric's identity: ``intelligence_snapshots`` has a unique
# key carrying ``score_version``, historical rows keep their old spellings
# untouched, and a reader must never join a ``people.org_coverage`` from
# last week to a ``people_org.org_coverage`` from this one as though they
# were the same series.
#
# Why the rename: 30 of the 40 catalogue metrics repeated their dimension
# name in their key (``cloud_network.resource_waste_rate`` under
# ``cloud_network``), and the citation flattener strips that repeat, so the
# name published to the model equalled the address it had to type. The
# other ten did not repeat it — ``people_org`` keyed its metrics
# ``people.*`` and ``platform_ops`` keyed its ``platform_operations.*`` —
# so their addresses carried an extra segment
# (``people_org.metrics.people.prediction_data_coverage``). Two shapes for
# one kind of record: the model learned ``<dimension>.<metric>`` from the
# 30-metric majority and could not cite the other ten CORRECTLY, which is
# how a real catalogue metric came back as "Unknown fact reference"
# (production, 2026-09-01, probe run 2 of 3). Aligning the ten makes the
# rule the model already infers true for all forty.
#
# ``validate_metric_key_shape_is_uniform`` now enforces it at import
# time, so the shape cannot drift back in on the next metric someone adds.
SCORE_VERSION = "health-v4"

DIMENSION_WEIGHTS: dict[str, Decimal] = {
    "people_org": Decimal("0.10"),
    "lifecycle": Decimal("0.15"),
    "assets": Decimal("0.10"),
    "software_api": Decimal("0.10"),
    "cloud_network": Decimal("0.15"),
    "cost_efficiency": Decimal("0.15"),
    "security_compliance": Decimal("0.15"),
    "platform_ops": Decimal("0.10"),
}


def validate_metric_weight_totals(
    definitions: Iterable[tuple[str, Decimal]],
    dimension_weights: dict[str, Decimal],
) -> None:
    """Fail startup when versioned dimension or metric weights drift."""
    if any(weight <= 0 for weight in dimension_weights.values()):
        raise ValueError("Dimension weights must be positive")
    if sum(dimension_weights.values(), Decimal("0")) != Decimal("1.00"):
        raise ValueError("Dimension weights must total 1")
    totals = {dimension: Decimal("0") for dimension in dimension_weights}
    for dimension, weight in definitions:
        if dimension not in totals:
            raise ValueError(f"Unknown metric dimension: {dimension}")
        if weight <= 0:
            raise ValueError(f"Metric weight in {dimension!r} must be positive")
        totals[dimension] += weight
    invalid = {key: value for key, value in totals.items() if value != Decimal("1.00")}
    if invalid:
        raise ValueError(f"Metric weights must total 1 per dimension: {invalid}")


def validate_metric_key_shape_is_uniform(
    definitions: Iterable[tuple[str, str]],
) -> None:
    """One catalogue must address all its metrics the same way.

    ``definitions`` is ``(metric_key, dimension)`` pairs. The rule is not
    "every key must start with its dimension" — it is that a catalogue may
    not contain BOTH spellings, because the citation flattener treats them
    differently and the model then has a majority rule that a minority
    breaks.

    Production, 2026-09-01: the platform catalogue had 30 metrics whose key
    repeated the dimension (``cloud_network.resource_waste_rate`` under
    ``cloud_network``) and 10 whose key did not (``people_org`` keyed its
    metrics ``people.*``, ``platform_ops`` keyed its
    ``platform_operations.*``). ``_list_item_segment`` strips a repeated
    namespace and keeps a foreign one, so the first 30 published the name
    ``<dimension>.<leaf>`` at address ``<dimension>.metrics.<leaf>`` while
    the other 10 published ``<other>.<leaf>`` at
    ``<dimension>.metrics.<other>.<leaf>``. The model generalised the
    majority and cited ``/facts/people_org.prediction_data_coverage.value``
    — a real catalogue metric, refused as an unknown reference, discarding
    a whole grounded brief.

    Uniformity is what makes a citation mechanical rather than a guess: the
    cloud-subscription catalogue keys every one of its 18 metrics
    ``cloud.<group>.<leaf>`` and never had this failure, because there is
    no competing majority to infer a different rule from. So both shapes
    are legal; mixing them within one catalogue is not.
    """
    repeats: list[str] = []
    foreign: list[str] = []
    for key, dimension in definitions:
        (repeats if key.startswith(f"{dimension}.") else foreign).append(key)
    if repeats and foreign:
        raise ValueError(
            "Metric keys must address every metric the same way; this "
            "catalogue mixes both shapes, which is what made a real metric "
            "uncitable in production on 2026-09-01. Keys repeating their "
            f"dimension: {sorted(repeats)[:3]}... ({len(repeats)}); keys "
            f"that do not: {sorted(foreign)[:3]}... ({len(foreign)}). Pick "
            "one shape for the whole catalogue."
        )


class MetricKind(StrEnum):
    COVERAGE = "coverage"
    FAILURE_RATE = "failure_rate"
    FRESHNESS = "freshness"
    DIRECT = "direct"


METRIC_RULE_LABELS: dict[MetricKind, dict[str, str]] = {
    MetricKind.COVERAGE: {
        "zh-CN": "覆盖率达到 95% 得 100 分，90–95% 得 80 分，80–90% 得 60 分，低于 80% 得 30 分。",
        "en-US": "Coverage scores 100 at 95% or above, 80 at 90–95%, 60 at 80–90%, and 30 below 80%.",
        "ja-JP": "カバレッジは95%以上で100点、90–95%で80点、80–90%で60点、80%未満で30点です。",
    },
    MetricKind.FAILURE_RATE: {
        "zh-CN": "失败率不高于 1% 得 100 分，1–3% 得 80 分，3–10% 得 50 分，高于 10% 得 20 分。",
        "en-US": "Failure rate scores 100 at 1% or below, 80 at 1–3%, 50 at 3–10%, and 20 above 10%.",
        "ja-JP": "失敗率は1%以下で100点、1–3%で80点、3–10%で50点、10%超で20点です。",
    },
    MetricKind.FRESHNESS: {
        "zh-CN": "在任务周期内或仅错过一个宽限周期得 100 分，连续错过两期得 60 分，三期及以上得 20 分。",
        "en-US": "On-schedule data or one missed grace cycle scores 100; two missed cycles score 60, and three or more score 20.",
        "ja-JP": "予定内または猶予として1周期のみ未実行は100点、2周期連続未実行は60点、3周期以上は20点です。",
    },
    MetricKind.DIRECT: {
        "zh-CN": "确定性来源直接提供 0–100 分，并在保存前限制到有效范围。",
        "en-US": "The deterministic source supplies a 0–100 score, bounded to the valid range before persistence.",
        "ja-JP": "決定論的データソースが0–100点を提供し、保存前に有効範囲へ制限します。",
    },
}


@dataclass(frozen=True)
class MetricInput:
    key: str
    kind: MetricKind
    value: Decimal | int | float
    weight: Decimal | int | float = Decimal("1")
    confidence: Decimal | int | float = Decimal("100")
    evidence_refs: tuple[str, ...] = ()
    data_as_of: datetime | None = None


def _decimal(value: Decimal | int | float) -> Decimal:
    # str avoids importing binary floating-point noise into persisted scores.
    return value if isinstance(value, Decimal) else Decimal(str(value))


def _bounded(value: Decimal | int | float) -> Decimal:
    return max(Decimal("0"), min(Decimal("100"), _decimal(value)))


def score_metric(metric: MetricInput) -> Decimal:
    """Apply the immutable v1 threshold catalogue to a metric."""
    value = _decimal(metric.value)
    if metric.kind is MetricKind.DIRECT:
        score = _bounded(value)
    elif metric.kind is MetricKind.COVERAGE:
        score = Decimal("100") if value >= 95 else (
            Decimal("80") if value >= 90 else Decimal("60") if value >= 80
            else Decimal("30")
        )
    elif metric.kind is MetricKind.FAILURE_RATE:
        score = Decimal("100") if value <= 1 else (
            Decimal("80") if value <= 3 else Decimal("50") if value <= 10
            else Decimal("20")
        )
    elif metric.kind is MetricKind.FRESHNESS:
        # Value is the number of consecutively missed expected runs/cycles.
        score = Decimal("100") if value <= 1 else (
            Decimal("60") if value == 2 else Decimal("20")
        )
    else:  # defensive if a non-enum value leaks in from an untyped caller
        raise ValueError(f"Unsupported metric kind: {metric.kind!r}")
    return score.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)


def score_dimension(key: str, metrics: Iterable[MetricInput]) -> dict[str, Any]:
    if key not in DIMENSION_WEIGHTS:
        raise ValueError(f"Unknown health dimension: {key}")
    rows = list(metrics)
    if not rows:
        return {
            "key": key, "score": None, "confidence": Decimal("0.00"),
            "weight": DIMENSION_WEIGHTS[key], "metrics": [],
        }
    total_weight = sum((_decimal(row.weight) for row in rows), Decimal("0"))
    if total_weight <= 0:
        raise ValueError(f"Dimension {key} has no positive metric weight")
    metric_rows: list[dict[str, Any]] = []
    weighted_score = Decimal("0")
    weighted_confidence = Decimal("0")
    scored_weight = Decimal("0")
    for row in rows:
        weight = _decimal(row.weight)
        if weight < 0:
            raise ValueError(f"Metric {row.key} has a negative weight")
        confidence = _bounded(row.confidence)
        weighted_confidence += confidence * weight
        score = score_metric(row) if confidence > 0 else None
        if score is not None:
            weighted_score += score * weight
            scored_weight += weight
        metric_rows.append({
            "key": row.key,
            "kind": row.kind.value,
            "value": _decimal(row.value),
            "weight": weight,
            "score": score,
            "confidence": confidence,
            "evidence_refs": list(row.evidence_refs),
            "data_as_of": row.data_as_of,
        })
    quant = Decimal("0.01")
    dimension_score = (
        (weighted_score / scored_weight).quantize(quant, rounding=ROUND_HALF_UP)
        if scored_weight > 0 else None
    )
    return {
        "key": key,
        "score": dimension_score,
        "confidence": (weighted_confidence / total_weight).quantize(
            quant, rounding=ROUND_HALF_UP,
        ),
        "weight": DIMENSION_WEIGHTS[key],
        "metrics": metric_rows,
    }


def score_health(
    dimensions: Iterable[dict[str, Any]],
    *,
    expected_dimensions: Iterable[str] | None = None,
) -> dict[str, Any]:
    """Combine available dimensions and report missing coverage separately.

    Missing dimensions do not become a fabricated zero. Their configured
    weight reduces data confidence, while the score is normalized across the
    dimensions that actually have evidence.
    """
    by_key = {row["key"]: row for row in dimensions}
    unknown = set(by_key) - set(DIMENSION_WEIGHTS)
    if unknown:
        raise ValueError(f"Unknown dimensions: {sorted(unknown)}")
    expected = tuple(expected_dimensions or DIMENSION_WEIGHTS)
    if not expected or len(set(expected)) != len(expected):
        raise ValueError("Expected health dimensions must be unique and non-empty")
    unknown_expected = set(expected) - set(DIMENSION_WEIGHTS)
    if unknown_expected:
        raise ValueError(f"Unknown expected dimensions: {sorted(unknown_expected)}")
    unexpected = set(by_key) - set(expected)
    if unexpected:
        raise ValueError(f"Unexpected dimensions for score scope: {sorted(unexpected)}")

    score_sum = Decimal("0")
    available_weight = Decimal("0")
    confidence_sum = Decimal("0")
    expected_weight = Decimal("0")
    ordered: list[dict[str, Any]] = []
    for key in expected:
        weight = DIMENSION_WEIGHTS[key]
        expected_weight += weight
        row = by_key.get(key) or {
            "key": key, "score": None, "confidence": Decimal("0"),
            "weight": weight, "metrics": [],
        }
        ordered.append(row)
        if row["score"] is not None:
            score_sum += _bounded(row["score"]) * weight
            available_weight += weight
        confidence_sum += _bounded(row["confidence"]) * weight
    overall = score_sum / available_weight if available_weight else Decimal("0")
    quant = Decimal("0.01")
    overall = overall.quantize(quant, rounding=ROUND_HALF_UP)
    confidence = (confidence_sum / expected_weight).quantize(
        quant, rounding=ROUND_HALF_UP,
    )
    return {
        "score_version": SCORE_VERSION,
        "overall_score": overall,
        "confidence_score": confidence,
        "insufficient_data": confidence < 70,
        "grade": health_grade(overall),
        "dimensions": ordered,
    }


def health_grade(score: Decimal | int | float) -> str:
    value = _bounded(score)
    if value >= 90:
        return "excellent"
    if value >= 75:
        return "healthy"
    if value >= 60:
        return "attention"
    return "high_risk"


def _canonical_default(value: Any) -> str:
    if isinstance(value, Decimal):
        return format(value, "f")
    if isinstance(value, (datetime, date)):
        return value.isoformat()
    if isinstance(value, Enum):
        return str(value.value)
    raise TypeError(f"Cannot canonicalize {type(value).__name__}")


def canonical_json(value: Any) -> str:
    return json.dumps(
        value, default=_canonical_default, ensure_ascii=False,
        sort_keys=True, separators=(",", ":"), allow_nan=False,
    )


def json_value(value: Any, *, decimal_as_string: bool = False) -> Any:
    """Convert scoring output to values accepted by SQLAlchemy JSON.

    Scores retain Decimal arithmetic until this persistence boundary.  The
    stored/public contract uses JSON numbers and ISO-8601 timestamps.
    """
    if isinstance(value, Decimal):
        return format(value, "f") if decimal_as_string else float(value)
    if isinstance(value, (datetime, date)):
        return value.isoformat()
    if isinstance(value, Enum):
        return value.value
    if isinstance(value, dict):
        return {
            str(key): json_value(child, decimal_as_string=decimal_as_string)
            for key, child in value.items()
        }
    if isinstance(value, (list, tuple)):
        return [
            json_value(child, decimal_as_string=decimal_as_string)
            for child in value
        ]
    return value


def facts_hash(facts: dict[str, Any], *, score_version: str = SCORE_VERSION) -> str:
    material = {"score_version": score_version, "facts": facts}
    return hashlib.sha256(canonical_json(material).encode("utf-8")).hexdigest()