"""Read-only grouped review contract for nutrition-to-histamine mappings.

The contract deliberately derives its state from existing rows.  It neither creates
schema nor updates queue/mapping state, which makes it safe to use against an older
HealthManager database without running a migration first.
"""
from __future__ import annotations

import hashlib
import json
import re
import sqlite3
import unicodedata
from collections import defaultdict
from typing import Any, Iterable

CONTRACT_VERSION = "nutrition_mapping_review_v1"
MAX_ITEM_ROWS = 250_000
MAX_SUPPORT_ROWS = 20_000
MAX_CATALOG_ROWS = 5_000
MAX_PUBLIC_EXAMPLES = 8
MAX_PUBLIC_DAYS = 31
_PRIVATE_METADATA = re.compile(
    r"(?:/home/|/tmp/|/Users/|/(?:var|etc|usr|opt|srv|root|run|mnt|media|private)/|"
    r"~/|file:|https?://|[A-Za-z]:[\\/]|(?<![A-Za-z0-9])/(?:[^/\s]+/)+[^/\s]+|"
    r"\.\.[\\/]|\\\\[^\\/\s]+[\\/]|\.hermes(?:/|$))",
    re.IGNORECASE,
)
_OPAQUE_IDENTIFIER = re.compile(r"(?<![A-Za-z0-9_-])[A-Za-z0-9_-]{25,}(?![A-Za-z0-9_-])")


class ReviewLimitError(RuntimeError):
    """Raised before an unbounded review source is materialized."""


class UnsafeReviewMetadataError(RuntimeError):
    """Raised before private or technical metadata can enter a public bundle."""


_STATES = frozenset(
    {
        "unassigned",
        "suggestion_available",
        "review_required",
        "mapped_unverified",
        "confirmed",
        "deferred",
        "not_assignable",
        "irrelevant",
        "conflict",
    }
)
_OPEN_STATES = frozenset(
    {"unassigned", "suggestion_available", "review_required", "deferred", "conflict"}
)
_OPEN_FILTER_STATES = frozenset({"unassigned", "suggestion_available", "review_required"})
_TERMINAL_DECISIONS = {
    "defer": "deferred",
    "deferred": "deferred",
    "not_assignable": "not_assignable",
    "not-assignable": "not_assignable",
    "irrelevant": "irrelevant",
    "ignore": "irrelevant",
    "conflict": "conflict",
}


def exact_identity(value: Any) -> str:
    """NFC + casefold + collapsed whitespace; punctuation is data."""
    return " ".join(unicodedata.normalize("NFC", str(value or "")).casefold().split())


def _display(value: Any) -> str:
    return " ".join(unicodedata.normalize("NFC", str(value or "")).split())


def aggressive_identity(value: Any) -> str:
    """Return the legacy punctuation-erasing queue identity.

    This is public because the existing worker must continue validating legacy
    queue rows with exactly the same normalization while the review UI uses the
    stricter :func:`exact_identity` for grouping.
    """
    normalized = unicodedata.normalize("NFC", str(value or "")).casefold().translate(
        str.maketrans(
            {"ä": "ae", "ö": "oe", "ü": "ue", "é": "e", "è": "e", "à": "a", "ß": "ss"}
        )
    )
    return re.sub(r"\s+", " ", re.sub(r"[^a-z0-9]+", " ", normalized)).strip()


# Private spelling retained to keep the collision code visually explicit.
_exact_identity = exact_identity
_legacy_identity = aggressive_identity


def _digest(domain: str, payload: Any, length: int = 32) -> str:
    serialized = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(f"{domain}\0{serialized}".encode("utf-8")).hexdigest()[:length]


def _group_key(identity: str) -> str:
    return _digest("healthmanager:nutrition-mapping-review:group:v1", identity)


def _dict_rows(
    connection: sqlite3.Connection,
    sql: str,
    parameters: Iterable[Any] = (),
    *,
    limit: int = MAX_SUPPORT_ROWS,
) -> list[dict[str, Any]]:
    cursor = connection.execute(sql, tuple(parameters))
    columns = [description[0] for description in cursor.description or ()]
    rows = cursor.fetchmany(limit + 1)
    if len(rows) > limit:
        raise ReviewLimitError("nutrition_mapping_review_row_limit_exceeded")
    return [dict(zip(columns, row)) for row in rows]


def _table_exists(connection: sqlite3.Connection, name: str) -> bool:
    return connection.execute(
        "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (name,)
    ).fetchone() is not None


def _load_optional(
    connection: sqlite3.Connection, table: str, sql: str, *, limit: int = MAX_SUPPORT_ROWS
) -> list[dict[str, Any]]:
    return _dict_rows(connection, sql, limit=limit) if _table_exists(connection, table) else []


def _valid_score(value: Any) -> int | None:
    # bool is intentionally excluded even though it is an int subclass.
    return int(value) if type(value) is int and 0 <= value <= 3 else None


def _target_revision(
    rows: list[dict[str, Any]], latest_action: dict[str, Any] | None = None
) -> str:
    target = []
    for row in rows:
        target.append(
            {
                "item_hash": str(row.get("item_hash") or ""),
                "day": str(row.get("datum") or ""),
                "canonical_food": str(row.get("canonical_food") or ""),
                "sighi_score": row.get("sighi_score"),
                "traffic_light": str(row.get("traffic_light") or ""),
                "tags": str(row.get("tags") or ""),
                "confidence": str(row.get("confidence") or ""),
                "reason": str(row.get("reason") or ""),
            }
        )
    target.sort(key=lambda item: (item["item_hash"], item["day"], json.dumps(item, sort_keys=True)))
    state_revision = None
    if latest_action is not None:
        state_revision = {
            "action_id": str(latest_action.get("action_id") or ""),
            "action_hash": str(latest_action.get("action_hash") or ""),
            "decision": str(latest_action.get("decision") or ""),
            "applied_at": str(latest_action.get("applied_at") or ""),
            "rowid": int(latest_action.get("_rowid") or 0),
        }
    return _digest(
        "healthmanager:nutrition-mapping-review:target:v2",
        {"items": target, "review_state": state_revision},
        64,
    )


def assignment_evidence_token(
    rows: list[dict[str, Any]] | list[sqlite3.Row], canonical_food: str, sighi_score: int
) -> str:
    """Commit an explicit assignment to its exact immutable item/day target set."""
    target = sorted(
        (
            str(row["item_hash"] or ""),
            str(row["datum"] or ""),
            canonical_food,
            int(sighi_score),
        )
        for row in rows
    )
    digest = _digest(
        "healthmanager:nutrition-mapping-review:assignment-evidence:v1",
        target,
        64,
    )
    return f"mapping_assignment_v1|{len(target)}|{digest}"


def safe_mapping_metadata(value: Any, maximum: int, *, allow_empty: bool = False) -> str | None:
    """Return bounded public mapping text or fail closed on technical metadata."""
    text = _display(value)
    opaque = any(
        any(character.isalpha() for character in token)
        and any(character.isdigit() for character in token)
        for token in _OPAQUE_IDENTIFIER.findall(text)
    )
    if (
        (not text and not allow_empty)
        or len(text) > maximum
        or any(ord(character) < 32 or ord(character) == 127 for character in text)
        or _PRIVATE_METADATA.search(text)
        or opaque
    ):
        return None
    return text


def _rules_and_aliases(
    connection: sqlite3.Connection,
) -> tuple[dict[str, dict[str, Any]], dict[str, dict[str, Any]], set[str]]:
    rules: dict[str, dict[str, Any]] = {}
    ambiguous: set[str] = set()
    for row in _load_optional(
        connection,
        "histamine_food_rules",
        "SELECT canonical_food,sighi_score,category,confidence,source,updated_at FROM histamine_food_rules ORDER BY canonical_food",
        limit=MAX_CATALOG_ROWS,
    ):
        canonical = safe_mapping_metadata(row.get("canonical_food"), 120)
        category = safe_mapping_metadata(row.get("category") or "unbekannt", 80)
        provenance_source = safe_mapping_metadata(row.get("source") or "Lokaler Katalog", 80)
        provenance_version = safe_mapping_metadata(row.get("updated_at") or "unbekannt", 40)
        score = _valid_score(row.get("sighi_score"))
        if canonical and category and provenance_source and provenance_version and score is not None:
            identity = _exact_identity(canonical)
            candidate = {
                "canonical_food": canonical,
                "sighi_score": score,
                "confidence": str(row.get("confidence") or "low")
                if str(row.get("confidence") or "") in {"low", "medium", "high"}
                else "low",
                "category": category,
                "source_label": "Lokaler Lebensmittelkatalog",
                "provenance_source": provenance_source,
                "provenance_version": provenance_version,
            }
            if identity in ambiguous:
                continue
            if identity in rules and rules[identity] != candidate:
                rules.pop(identity, None)
                ambiguous.add(identity)
            else:
                rules[identity] = candidate
    aliases: dict[str, dict[str, Any]] = {}
    alias_targets: dict[str, str] = {}
    for row in _load_optional(
        connection,
        "histamine_food_aliases",
        "SELECT alias,canonical_food FROM histamine_food_aliases ORDER BY alias,canonical_food",
        limit=MAX_CATALOG_ROWS,
    ):
        canonical_identity = _exact_identity(row.get("canonical_food"))
        rule = rules.get(canonical_identity)
        alias_identity = _exact_identity(row.get("alias"))
        if alias_identity and rule is not None:
            if alias_identity in ambiguous:
                continue
            if alias_identity in rules and alias_identity != canonical_identity:
                aliases.pop(alias_identity, None)
                alias_targets.pop(alias_identity, None)
                rules.pop(alias_identity, None)
                ambiguous.add(alias_identity)
                continue
            if alias_identity in alias_targets and alias_targets[alias_identity] != canonical_identity:
                aliases.pop(alias_identity, None)
                ambiguous.add(alias_identity)
            else:
                alias_targets[alias_identity] = canonical_identity
                aliases[alias_identity] = rule
    return rules, aliases, ambiguous


def resolve_catalog_assignment(
    connection: sqlite3.Connection,
    canonical_food: str,
    *,
    alias: str | None = None,
    require_alias: bool = False,
) -> dict[str, Any] | None:
    """Resolve one assignment through the collision-aware public catalog truth."""
    rules, aliases, ambiguous = _rules_and_aliases(connection)
    canonical_identity = _exact_identity(canonical_food)
    if canonical_identity in ambiguous:
        return None
    rule = rules.get(canonical_identity)
    if rule is None or rule["canonical_food"] != canonical_food:
        return None
    if require_alias:
        alias_identity = _exact_identity(alias)
        if alias_identity in ambiguous or aliases.get(alias_identity) != rule:
            return None
    return dict(rule)


def _recommendation(
    identity: str,
    queue_rows: list[dict[str, Any]],
    rules: dict[str, dict[str, Any]],
    aliases: dict[str, dict[str, Any]],
) -> tuple[dict[str, Any] | None, str | None]:
    # A queue suggestion is advice, never proof.  It is released only when its
    # canonical and score agree with an existing local rule.
    matching_queue = [
        row
        for row in queue_rows
        if str(row.get("status") or "") == "open"
        and _exact_identity(row.get("example_name")) == identity
    ]
    for row in sorted(matching_queue, key=lambda item: str(item.get("updated_at") or ""), reverse=True):
        rule = rules.get(_exact_identity(row.get("suggested_canonical_food")))
        score = _valid_score(row.get("suggested_score"))
        if rule is not None and score == rule["sighi_score"]:
            return (
                {
                    "canonical_food": rule["canonical_food"],
                    "sighi_score": rule["sighi_score"],
                    "source": "queue_suggestion",
                    "method": "sighi_reference",
                    "category": rule["category"],
                    "source_label": rule["source_label"],
                },
                rule["confidence"],
            )
    exact = aliases.get(identity)
    source = "exact_local_alias"
    if exact is None:
        exact = rules.get(identity)
        source = "exact_canonical_rule"
    if exact is None:
        return None, None
    return (
        {
            "canonical_food": exact["canonical_food"],
            "sighi_score": exact["sighi_score"],
            "source": source,
            "method": "local_alias" if source == "exact_local_alias" else "sighi_reference",
            "category": exact["category"],
            "source_label": exact["source_label"],
        },
        exact["confidence"],
    )


def _latest_action(identity: str, key: str, actions: list[dict[str, Any]]) -> dict[str, Any] | None:
    matching = [
        row
        for row in actions
        if str(row.get("queue_key") or "") == key
        or (_exact_identity(row.get("alias")) and _exact_identity(row.get("alias")) == identity)
    ]
    if not matching:
        return None
    return max(matching, key=lambda row: (str(row.get("applied_at") or ""), int(row.get("_rowid") or 0)))


def _mapping_state(rows: list[dict[str, Any]]) -> tuple[str | None, dict[str, Any] | None, str | None]:
    valid: list[tuple[str, int, str]] = []
    invalid_or_missing = False
    for row in rows:
        canonical = str(row.get("canonical_food") or "").strip()
        score = _valid_score(row.get("sighi_score"))
        if not canonical or score is None:
            invalid_or_missing = True
            continue
        confidence = str(row.get("confidence") or "low")
        valid.append((canonical, score, confidence if confidence in {"low", "medium", "high"} else "low"))
    pairs = {(canonical, score) for canonical, score, _confidence in valid}
    if len(pairs) > 1:
        return "conflict", None, None
    if valid and not invalid_or_missing and len(valid) == len(rows):
        canonical, score = next(iter(pairs))
        confidences = {confidence for _canonical, _score, confidence in valid}
        confidence = next(iter(confidences)) if len(confidences) == 1 else "mixed"
        return "confirmed", {"canonical_food": canonical, "sighi_score": score}, confidence
    if valid or any(row.get("canonical_food") is not None or row.get("sighi_score") is not None for row in rows):
        return "review_required", None, None
    return None, None, None


def _all_groups(connection: sqlite3.Connection) -> list[dict[str, Any]]:
    if not _table_exists(connection, "nutrition_items"):
        return []
    rows = _dict_rows(
        connection,
        """SELECT i.datum,i.name,i.item_hash,
                  h.canonical_food,h.sighi_score,h.traffic_light,h.tags,
                  h.confidence,h.reason
             FROM nutrition_items i
             LEFT JOIN nutrition_histamine_scores h ON h.item_id=i.id""",
        limit=MAX_ITEM_ROWS,
    )
    grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for row in rows:
        identity = _exact_identity(row.get("name"))
        if identity:
            grouped[identity].append(row)

    legacy_members: dict[str, set[str]] = defaultdict(set)
    for identity in grouped:
        legacy_members[_legacy_identity(identity)].add(identity)

    rules, aliases, ambiguous_identities = _rules_and_aliases(connection)
    queue_rows = _load_optional(
        connection,
        "nutrition_review_queue",
        """SELECT normalized_name,example_name,suggested_canonical_food,
                  suggested_score,status,updated_at
             FROM nutrition_review_queue""",
    )
    actions = _load_optional(
        connection,
        "nutrition_mapping_action_log",
        """SELECT rowid AS _rowid,action_id,action_hash,queue_key,decision,alias,canonical_food,
                  sighi_score,confidence,ingredient_review_required,definition_version,applied_at
             FROM nutrition_mapping_action_log""",
    )
    composites = {
        str(row.get("normalized_name") or ""): str(row.get("status") or "")
        for row in _load_optional(
            connection,
            "nutrition_composite_product_review",
            "SELECT normalized_name,status FROM nutrition_composite_product_review",
        )
    }

    result = []
    for identity, members in grouped.items():
        members.sort(key=lambda row: (str(row.get("datum") or ""), str(row.get("item_hash") or "")))
        key = _group_key(identity)
        mapping_status, current_mapping, mapping_confidence = _mapping_state(members)
        safe_names = [safe_mapping_metadata(row.get("name"), 160) for row in members]
        if any(name is None for name in safe_names):
            raise UnsafeReviewMetadataError("unsafe_nutrition_mapping_metadata")
        public_names = [name for name in safe_names if name is not None]
        if current_mapping is not None and safe_mapping_metadata(
            current_mapping.get("canonical_food"), 120
        ) is None:
            raise UnsafeReviewMetadataError("unsafe_nutrition_mapping_metadata")
        recommendation, recommendation_confidence = _recommendation(identity, queue_rows, rules, aliases)
        latest = _latest_action(identity, key, actions)
        collision = len(legacy_members[_legacy_identity(identity)]) > 1
        queue_open = any(
            str(row.get("status") or "") == "open"
            and _exact_identity(row.get("example_name")) == identity
            for row in queue_rows
        )
        composite_open = any(
            status == "needs_ingredient_review"
            and (normalized == _legacy_identity(identity) or normalized == identity)
            for normalized, status in composites.items()
        )

        latest_decision = str(latest.get("decision") or "") if latest else ""
        if identity in ambiguous_identities or collision or mapping_status == "conflict":
            state = "conflict"
        elif latest_decision in _TERMINAL_DECISIONS:
            state = _TERMINAL_DECISIONS[latest_decision]
        elif latest_decision == "composite" or composite_open:
            state = "review_required"
        elif (
            mapping_status == "confirmed"
            and latest_decision == "assign"
            and current_mapping is not None
            and latest is not None
            and _exact_identity(latest.get("canonical_food"))
                == _exact_identity(current_mapping.get("canonical_food"))
            and _valid_score(latest.get("sighi_score")) == current_mapping.get("sighi_score")
            and str(latest.get("definition_version") or "").endswith(
                "|" + assignment_evidence_token(
                    members,
                    current_mapping["canonical_food"],
                    current_mapping["sighi_score"],
                )
            )
        ):
            state = "confirmed"
        elif mapping_status == "confirmed":
            state = "mapped_unverified"
        elif mapping_status == "review_required" or composite_open:
            state = "review_required"
        elif recommendation is not None:
            state = "suggestion_available"
        elif queue_open:
            state = "review_required"
        else:
            state = "unassigned"

        confidence = mapping_confidence
        if confidence is None and latest is not None:
            action_confidence = str(latest.get("confidence") or "")
            confidence = action_confidence if action_confidence in {"low", "medium", "high"} else None
        if confidence is None:
            confidence = recommendation_confidence or "unknown"
        if state == "mapped_unverified" and current_mapping is not None:
            current_rule = rules.get(_exact_identity(current_mapping["canonical_food"]))
            if current_rule is not None and current_rule["sighi_score"] == current_mapping["sighi_score"]:
                recommendation = {
                    "canonical_food": current_rule["canonical_food"],
                    "sighi_score": current_rule["sighi_score"],
                    "source": "existing_technical_mapping",
                    "method": "sighi_reference",
                    "category": current_rule["category"],
                    "source_label": "Lokaler Lebensmittelkatalog",
                }
        # Terminal/conflict state must not accidentally expose advice as a decision.
        public_recommendation = (
            recommendation if state in {"suggestion_available", "mapped_unverified"} else None
        )
        all_examples = sorted(set(public_names), key=lambda value: (value.casefold(), value))
        all_days = sorted({str(row.get("datum") or "") for row in members if row.get("datum")})
        public_examples = all_examples[:MAX_PUBLIC_EXAMPLES]
        public_days = all_days[:MAX_PUBLIC_DAYS]
        result.append(
            {
                "key": key,
                "queue_key": key,
                "status": state,
                "confidence": confidence,
                "entry_count": len(members),
                "occurrences": len(members),
                "day_count": len(all_days),
                "target_count": len(members),
                "target_day_count": len(all_days),
                "days": public_days,
                "days_truncated": len(public_days) < len(all_days),
                "examples": public_examples,
                "example_count": len(all_examples),
                "examples_truncated": len(public_examples) < len(all_examples),
                "name": public_names[0],
                "example_name": public_names[0],
                "first_seen": all_days[0] if all_days else None,
                "last_seen": all_days[-1] if all_days else None,
                "current_mapping": current_mapping,
                "recommendation": public_recommendation,
                "suggested_canonical_food": (
                    public_recommendation["canonical_food"] if public_recommendation else ""
                ),
                "suggested_score": (
                    public_recommendation["sighi_score"] if public_recommendation else "unknown"
                ),
                "target_revision": _target_revision(members, latest),
            }
        )
    return sorted(result, key=lambda group: (_exact_identity(group["examples"][0]), group["key"]))


def _summary(groups: list[dict[str, Any]]) -> dict[str, Any]:
    total_entries = sum(int(group["entry_count"]) for group in groups)
    mapped_entries = sum(
        int(group["entry_count"])
        for group in groups
        if group.get("current_mapping") is not None
    )
    verified_entries = sum(
        int(group["entry_count"]) for group in groups if group["status"] == "confirmed"
    )
    open_entries = sum(
        int(group["entry_count"]) for group in groups if group["status"] in _OPEN_STATES
    )
    return {
        "total_entries": total_entries,
        "mapped_entries": mapped_entries,
        "verified_entries": verified_entries,
        "open_entries": open_entries,
        "coverage": round(mapped_entries / total_entries, 3) if total_entries else 0.0,
        "conflict_groups": sum(group["status"] == "conflict" for group in groups),
        "deferred_groups": sum(group["status"] == "deferred" for group in groups),
        "total_groups": len(groups),
    }


def build_review(
    connection: sqlite3.Connection, status: str = "all", search: str = ""
) -> dict[str, Any]:
    """Build the complete review summary and an optionally filtered group list."""
    if status not in _STATES | {"all", "open", "verified"}:
        raise ValueError("invalid_review_status")
    groups = _all_groups(connection)
    rules, _, _ = _rules_and_aliases(connection)
    catalog = sorted(
        (
            {
                "canonical_food": rule["canonical_food"],
                "sighi_score": rule["sighi_score"],
                "category": rule["category"],
                "source_label": "Lokaler Lebensmittelkatalog",
            }
            for rule in {rule["canonical_food"]: rule for rule in rules.values()}.values()
        ),
        key=lambda item: str(item["canonical_food"]).casefold(),
    )
    summary = _summary(groups)  # Filtering never changes global coverage/counts.
    if status == "open":
        visible = [group for group in groups if group["status"] in _OPEN_FILTER_STATES]
    elif status == "verified":
        visible = [group for group in groups if group["status"] == "confirmed"]
    elif status == "all":
        visible = groups
    else:
        visible = [group for group in groups if group["status"] == status]
    needle = _exact_identity(search)
    if needle:
        visible = [
            group
            for group in visible
            if needle in _exact_identity(" ".join(group["examples"]))
            or needle in _exact_identity(
                (group.get("current_mapping") or group.get("recommendation") or {}).get(
                    "canonical_food", ""
                )
            )
        ]
    return {
        "version": 1,
        "contract_version": CONTRACT_VERSION,
        "summary": summary,
        "groups": visible,
        "catalog": catalog,
    }


def resolve_group(connection: sqlite3.Connection, key: str) -> dict[str, Any] | None:
    """Resolve an opaque key against current targets, returning no raw database ID."""
    if not re.fullmatch(r"[0-9a-f]{32}", str(key or "")):
        return None
    return next((group for group in _all_groups(connection) if group["key"] == key), None)