explanations.py

"""Versioned plain-language explanations for deterministic screening results.

The explanation layer is deliberately separate from the rule engine. Rules
decide which records match; this module only validates and loads display copy
linked to those records by stable ``rule_id``. A missing or invalid
explanation must never change a screening result.
"""

from __future__ import annotations

import hashlib
import json
import re
from dataclasses import dataclass
from datetime import date
from pathlib import Path
from typing import Any

from .screening import Rule

SCHEMA_VERSION = 1
DISPLAY_GROUPS = ("route", "standard", "local_process")
REVIEW_STATUSES = (
    "prototype_review_pending",
    "human_reviewed",
    "jurisdiction_approved",
)
TRANSLATION_STATUSES = (
    "machine_draft",
    "human_reviewed",
    "jurisdiction_approved",
)
_SEMVER = re.compile(r"^\d+\.\d+\.\d+$")


@dataclass(frozen=True)
class Review:
    status: str
    reviewer: str | None
    reviewed_on: str | None
    method: str | None
    reviewed_version: str | None


@dataclass(frozen=True)
class Highlight:
    label: str
    text: str


@dataclass(frozen=True)
class HighlightGroup:
    title: str
    items: tuple[Highlight, ...]


@dataclass(frozen=True)
class LocalizedExplanation:
    summary: str
    next_steps: tuple[str, ...]
    confirm_with_staff: tuple[str, ...]
    highlights: HighlightGroup | None = None
    translation_status: str | None = None
    reviewer: str | None = None
    reviewed_on: str | None = None
    method: str | None = None
    reviewed_version: str | None = None


@dataclass(frozen=True)
class PlainLanguageExplanation:
    version: str
    source_rule_id: str
    source_verified_on: str | None
    citation_fingerprint: str
    rule_fingerprint: str
    display_group: str
    drafted_by: str
    updated_on: str
    review: Review
    en: LocalizedExplanation
    es: LocalizedExplanation | None

    def localized(self, language: str) -> LocalizedExplanation:
        """Return requested display copy, falling back to English."""

        return self.es if language == "es" and self.es is not None else self.en

    def localized_language(self, language: str) -> str:
        """Return the language actually used by :meth:`localized`."""

        return "es" if language == "es" and self.es is not None else "en"


def citation_fingerprint(rule: Rule) -> str:
    """Hash the normalized citation fields an explanation was checked against."""

    citation = rule.citation
    payload = json.dumps(
        {
            "excerpt": citation.excerpt,
            "excerpt_sha256": citation.excerpt_sha256,
            "source": citation.source,
            "url": citation.url,
            "verified_on": citation.verified_on,
        },
        ensure_ascii=False,
        separators=(",", ":"),
        sort_keys=True,
    )
    return "sha256:" + hashlib.sha256(payload.encode("utf-8")).hexdigest()


def rule_fingerprint(rule: Rule) -> str:
    """Hash every rule field that can affect explanation meaning."""

    payload = json.dumps(
        {
            "citation": {
                "excerpt": rule.citation.excerpt,
                "excerpt_sha256": rule.citation.excerpt_sha256,
                "source": rule.citation.source,
                "url": rule.citation.url,
                "verified_on": rule.citation.verified_on,
            },
            "criteria": rule.criteria,
            "jurisdiction_scope": rule.jurisdiction_scope,
            "notes": rule.notes,
            "pathway": rule.pathway,
            "required_documents": rule.required_documents,
            "route_class": rule.route_class,
            "rule_id": rule.rule_id,
        },
        ensure_ascii=False,
        separators=(",", ":"),
        sort_keys=True,
    )
    return "sha256:" + hashlib.sha256(payload.encode("utf-8")).hexdigest()


def _iso_date(value: Any, field: str, *, optional: bool = False) -> str | None:
    if value is None and optional:
        return None
    if not isinstance(value, str):
        raise ValueError(f"{field}: expected ISO date string")
    try:
        date.fromisoformat(value)
    except ValueError as error:
        raise ValueError(f"{field}: invalid ISO date {value!r}") from error
    return value


def _optional_text(value: Any, field: str) -> str | None:
    if value is None:
        return None
    if not isinstance(value, str) or not value.strip():
        raise ValueError(f"{field}: expected non-blank text or null")
    return value.strip()


def _required_text(value: Any, field: str) -> str:
    text = _optional_text(value, field)
    if text is None:
        raise ValueError(f"{field}: expected non-blank text")
    return text


def _text_list(value: Any, field: str) -> tuple[str, ...]:
    if not isinstance(value, list) or not value:
        raise ValueError(f"{field}: expected a non-empty list")
    return tuple(
        _required_text(item, f"{field}[{index}]")
        for index, item in enumerate(value)
    )


def _highlights(value: Any, field: str) -> HighlightGroup | None:
    if value is None:
        return None
    if not isinstance(value, dict):
        raise ValueError(f"{field}: expected an object or null")
    title = _required_text(value.get("title"), f"{field}.title")
    items = value.get("items")
    if not isinstance(items, list) or not items:
        raise ValueError(f"{field}.items: expected a non-empty list")
    parsed: list[Highlight] = []
    for index, item in enumerate(items):
        item_field = f"{field}.items[{index}]"
        if not isinstance(item, dict):
            raise ValueError(f"{item_field}: expected an object")
        parsed.append(
            Highlight(
                label=_required_text(
                    item.get("label"), f"{item_field}.label"
                ),
                text=_required_text(item.get("text"), f"{item_field}.text"),
            )
        )
    return HighlightGroup(title=title, items=tuple(parsed))


def _review(
    record: Any,
    rule_id: str,
    version: str,
    updated_on: str,
) -> Review:
    field = f"{rule_id}.review"
    if not isinstance(record, dict):
        raise ValueError(f"{field}: expected an object")
    status = _required_text(record.get("status"), f"{field}.status")
    if status not in REVIEW_STATUSES:
        raise ValueError(f"{field}.status: unknown value {status!r}")
    reviewer = _optional_text(record.get("reviewer"), f"{field}.reviewer")
    reviewed_on = _iso_date(
        record.get("reviewed_on"), f"{field}.reviewed_on", optional=True
    )
    method = _optional_text(record.get("method"), f"{field}.method")
    reviewed_version = _optional_text(
        record.get("reviewed_version"), f"{field}.reviewed_version"
    )
    metadata = (reviewer, reviewed_on, method, reviewed_version)
    if status == "prototype_review_pending" and any(metadata):
        raise ValueError(f"{field}: pending review cannot claim reviewer metadata")
    if status != "prototype_review_pending" and not all(metadata):
        raise ValueError(
            f"{field}: completed review requires reviewer, date, method, "
            f"and reviewed_version"
        )
    if status != "prototype_review_pending":
        if reviewed_version != version:
            raise ValueError(
                f"{field}: reviewed_version must match explanation version"
            )
        if reviewed_on < updated_on:
            raise ValueError(
                f"{field}: review date predates the explanation update date"
            )
    return Review(status, reviewer, reviewed_on, method, reviewed_version)


def _localized(
    record: Any,
    rule_id: str,
    language: str,
    version: str,
    updated_on: str,
) -> LocalizedExplanation:
    field = f"{rule_id}.{language}"
    if not isinstance(record, dict):
        raise ValueError(f"{field}: expected an object")
    translation_status = None
    reviewer = None
    reviewed_on = None
    method = None
    reviewed_version = None
    if language == "es":
        translation_status = _required_text(
            record.get("translation_status"), f"{field}.translation_status"
        )
        if translation_status not in TRANSLATION_STATUSES:
            raise ValueError(
                f"{field}.translation_status: unknown value "
                f"{translation_status!r}"
            )
        reviewer = _optional_text(record.get("reviewer"), f"{field}.reviewer")
        reviewed_on = _iso_date(
            record.get("reviewed_on"), f"{field}.reviewed_on", optional=True
        )
        method = _optional_text(record.get("method"), f"{field}.method")
        reviewed_version = _optional_text(
            record.get("reviewed_version"), f"{field}.reviewed_version"
        )
        metadata = (reviewer, reviewed_on, method, reviewed_version)
        if translation_status == "machine_draft" and any(metadata):
            raise ValueError(
                f"{field}: machine draft cannot claim translation review metadata"
            )
        if translation_status != "machine_draft" and not all(metadata):
            raise ValueError(
                f"{field}: reviewed translation requires reviewer, date, method, "
                f"and reviewed_version"
            )
        if translation_status != "machine_draft":
            if reviewed_version != version:
                raise ValueError(
                    f"{field}: reviewed_version must match explanation version"
                )
            if reviewed_on < updated_on:
                raise ValueError(
                    f"{field}: review date predates the explanation update date"
                )
    return LocalizedExplanation(
        summary=_required_text(record.get("summary"), f"{field}.summary"),
        next_steps=_text_list(record.get("next_steps"), f"{field}.next_steps"),
        confirm_with_staff=_text_list(
            record.get("confirm_with_staff"), f"{field}.confirm_with_staff"
        ),
        highlights=_highlights(record.get("highlights"), f"{field}.highlights"),
        translation_status=translation_status,
        reviewer=reviewer,
        reviewed_on=reviewed_on,
        method=method,
        reviewed_version=reviewed_version,
    )


def load_explanations(
    path: Path,
    rules: list[Rule],
    *,
    require_complete: bool = True,
    strict: bool = True,
) -> dict[str, PlainLanguageExplanation]:
    """Load and validate display copy against the canonical rule set.

    Strict mode catches duplicate, orphaned, missing, citation-drifted, and
    source-date-drifted records. Display runtimes may use ``strict=False`` to
    discard invalid records individually and fall back from invalid Spanish
    copy to English. Neither mode participates in matching.
    """

    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as error:
        if strict:
            raise ValueError(
                f"plain-language data could not be loaded: {error}"
            ) from error
        return {}
    if not isinstance(payload, dict) or payload.get("schema_version") != SCHEMA_VERSION:
        got = payload.get("schema_version") if isinstance(payload, dict) else None
        error = ValueError(
            f"plain-language schema_version must be {SCHEMA_VERSION}; got {got!r}"
        )
        if strict:
            raise error
        return {}
    records = payload.get("entries")
    if not isinstance(records, list):
        if strict:
            raise ValueError("plain-language entries: expected a list")
        return {}

    rules_by_id = {rule.rule_id: rule for rule in rules}
    if len(rules_by_id) != len(rules):
        if strict:
            raise ValueError("canonical rule set contains duplicate rule IDs")
        return {}

    explanations: dict[str, PlainLanguageExplanation] = {}
    seen: set[str] = set()
    blocked: set[str] = set()
    for index, record in enumerate(records):
        try:
            if not isinstance(record, dict):
                raise ValueError(f"entries[{index}]: expected an object")
            rule_id = _required_text(
                record.get("source_rule_id"),
                f"entries[{index}].source_rule_id",
            )
        except ValueError:
            if strict:
                raise
            continue

        if rule_id in seen:
            if strict:
                raise ValueError(
                    f"{rule_id}: duplicate plain-language explanation"
                )
            explanations.pop(rule_id, None)
            blocked.add(rule_id)
            continue
        seen.add(rule_id)
        if rule_id in blocked:
            continue

        try:
            rule = rules_by_id.get(rule_id)
            if rule is None:
                raise ValueError(
                    f"{rule_id}: explanation references an unknown rule"
                )

            version = _required_text(record.get("version"), f"{rule_id}.version")
            if not _SEMVER.fullmatch(version):
                raise ValueError(f"{rule_id}.version: expected semantic version")
            source_verified_on = _iso_date(
                record.get("source_verified_on"),
                f"{rule_id}.source_verified_on",
                optional=True,
            )
            if source_verified_on != rule.citation.verified_on:
                raise ValueError(
                    f"{rule_id}: explanation source date {source_verified_on!r} "
                    f"does not match rule source date "
                    f"{rule.citation.verified_on!r}"
                )
            fingerprint = _required_text(
                record.get("citation_fingerprint"),
                f"{rule_id}.citation_fingerprint",
            )
            expected_fingerprint = citation_fingerprint(rule)
            if fingerprint != expected_fingerprint:
                raise ValueError(
                    f"{rule_id}: citation fingerprint does not match linked rule"
                )
            full_rule_fingerprint = _required_text(
                record.get("rule_fingerprint"),
                f"{rule_id}.rule_fingerprint",
            )
            expected_rule_fingerprint = rule_fingerprint(rule)
            if full_rule_fingerprint != expected_rule_fingerprint:
                raise ValueError(
                    f"{rule_id}: rule fingerprint does not match linked rule"
                )
            display_group = _required_text(
                record.get("display_group"), f"{rule_id}.display_group"
            )
            if display_group not in DISPLAY_GROUPS:
                raise ValueError(
                    f"{rule_id}.display_group: unknown value {display_group!r}"
                )
            drafted_by = _required_text(
                record.get("drafted_by"), f"{rule_id}.drafted_by"
            )
            if drafted_by != "ai_assisted":
                raise ValueError(
                    f"{rule_id}.drafted_by: expected 'ai_assisted', "
                    f"got {drafted_by!r}"
                )

            updated_on = _iso_date(
                record.get("updated_on"), f"{rule_id}.updated_on"
            )
            if source_verified_on and updated_on < source_verified_on:
                raise ValueError(
                    f"{rule_id}: explanation update date {updated_on!r} "
                    f"predates linked source date {source_verified_on!r}"
                )
            review = _review(
                record.get("review"), rule_id, version, updated_on
            )
            english = _localized(
                record.get("en"), rule_id, "en", version, updated_on
            )
            try:
                spanish = _localized(
                    record.get("es"), rule_id, "es", version, updated_on
                )
            except ValueError:
                if strict:
                    raise
                spanish = None

            explanation = PlainLanguageExplanation(
                version=version,
                source_rule_id=rule_id,
                source_verified_on=source_verified_on,
                citation_fingerprint=fingerprint,
                rule_fingerprint=full_rule_fingerprint,
                display_group=display_group,
                drafted_by=drafted_by,
                updated_on=updated_on,
                review=review,
                en=english,
                es=spanish,
            )
        except ValueError:
            if strict:
                raise
            continue
        explanations[rule_id] = explanation

    if require_complete and strict:
        missing = sorted(set(rules_by_id) - set(explanations))
        if missing:
            raise ValueError(
                "plain-language explanations missing rule IDs: "
                + ", ".join(missing)
            )
    return explanations