"""``tabular-pinned@v1`` π* — declared-schema tabular canonicalizer.

Domain: ``tabular``. The last reserved-stub π* graduates: structured
2D data (rows × columns) gets first-class equivalence-class identity.

This closes the π* registry chapter — every modality the substrate
paper reserved (text, claim_lattice, code, arithmetic, logic,
time-series, tabular, plus the math-substrate extras
algebra-symbolic, calculus-derivative) is now real.

Input format
------------
UTF-8 JSON object::

    {
      "schema": [
        {"name": "<col_name>", "type": "str|int|rational|bool"},
        ...
      ],
      "key_columns": ["<col_name>", ...],
      "rows": [
        [<cell>, <cell>, ...],
        ...
      ]
    }

- ``schema`` declares column order + per-cell type. Types fold cell
  values: ``int`` and ``rational`` go through SQD §14.1 rational
  arithmetic (1, 1.0, "1.0" all collapse to ``1/1``). ``str`` keeps
  the value verbatim; ``bool`` normalizes to ``true``/``false``.
- ``key_columns`` names the primary-key columns. Rows are sorted by
  the (key_column_1, key_column_2, ...) tuple before serialization;
  ties within key are stable.
- Empty ``rows`` is valid — canonicalizes to an empty body.

Canonical output
----------------
UTF-8 text of shape::

    schema=col1:type1,col2:type2,...
    key=col1,col2,...
    n=<row_count>
    rows:
    cell11|cell12|...
    cell21|cell22|...
    ...

Type codes are the schema's declared types (``str`` / ``int`` /
``rational`` / ``bool``). Cells are joined with ``|``; rows are
joined with ``\n``. Header always present even on empty tables.

Equivalence classes preserved
-----------------------------
- Row order: rows are sorted by ``key_columns``; differently-ordered
  inputs collapse.
- Numeric formatting: ``1`` ≡ ``1.0`` ≡ ``"1.0"`` for ``int`` / ``rational``
  columns (same rational identity through arithmetic@v1).
- JSON whitespace / formatting: same logical content, same canonical.
- Boolean spelling: ``true``/``True``/``"true"`` all normalize for
  ``bool`` columns.

Equivalence classes kept distinct
---------------------------------
- Schema (column order, types, key_columns) is part of identity.
  Two tables with different declared schema → different canonical.
- Different cell values → different canonical.
- Different ``key_columns`` → different canonical (changes the sort).
- Header case: pinned exact. ``Name`` ≠ ``name``. PostgreSQL +
  Excel both care about case; defaulting to lowercase-fold would
  break operator expectations.

Round-trip property
-------------------
Projective. The canonical text form is not valid JSON input —
re-canonicalizing the canonical bytes raises PiStarError.

Versioning
----------
``tabular-pinned@v1`` pins this serialization, sort policy, and
type-folding rules. Any change requires a new version.
"""

from __future__ import annotations

import json
from dataclasses import dataclass
from fractions import Fraction
from decimal import Decimal, InvalidOperation

from arborist.pi_star.protocol import PiStarError
from arborist.pi_star.registry import register


_VALID_TYPES = ("str", "int", "rational", "bool")


@dataclass
class TabularPinnedV1:
    name: str = "tabular-pinned"
    version: str = "v1"
    domain: str = "tabular"

    def canonicalize(self, raw: bytes) -> bytes:
        if not isinstance(raw, (bytes, bytearray)):
            raise PiStarError(
                "tabular-pinned@v1 expects bytes; got "
                f"{type(raw).__name__}"
            )
        try:
            text = raw.decode("utf-8", errors="surrogatepass")
        except UnicodeDecodeError as exc:  # pragma: no cover
            raise PiStarError(f"input not valid UTF-8: {exc}") from exc

        try:
            obj = json.loads(text)
        except json.JSONDecodeError as exc:
            raise PiStarError(
                f"tabular-pinned@v1 input is not valid JSON: {exc}"
            ) from exc

        if not isinstance(obj, dict):
            raise PiStarError(
                "tabular-pinned@v1 input must be a JSON object"
            )
        for required in ("schema", "key_columns", "rows"):
            if required not in obj:
                raise PiStarError(
                    f"tabular-pinned@v1 missing required field "
                    f"{required!r}"
                )

        schema = obj["schema"]
        key_columns = obj["key_columns"]
        rows = obj["rows"]

        if not isinstance(schema, list) or not schema:
            raise PiStarError(
                "tabular-pinned@v1 schema must be a non-empty JSON array"
            )
        if not isinstance(key_columns, list):
            raise PiStarError(
                "tabular-pinned@v1 key_columns must be a JSON array"
            )
        if not isinstance(rows, list):
            raise PiStarError(
                "tabular-pinned@v1 rows must be a JSON array"
            )

        # Validate schema entries.
        col_names: list[str] = []
        col_types: list[str] = []
        for i, entry in enumerate(schema):
            if not isinstance(entry, dict):
                raise PiStarError(
                    f"schema[{i}] must be a JSON object with name + type"
                )
            name = entry.get("name")
            ctype = entry.get("type")
            if not isinstance(name, str) or not name:
                raise PiStarError(
                    f"schema[{i}].name must be a non-empty string"
                )
            if ctype not in _VALID_TYPES:
                raise PiStarError(
                    f"schema[{i}].type must be one of "
                    f"{_VALID_TYPES}; got {ctype!r}"
                )
            col_names.append(name)
            col_types.append(ctype)

        # Validate key_columns reference real schema columns.
        col_index = {n: i for i, n in enumerate(col_names)}
        if len(set(col_names)) != len(col_names):
            raise PiStarError(
                "tabular-pinned@v1 schema column names must be unique"
            )
        key_indices: list[int] = []
        for k in key_columns:
            if not isinstance(k, str):
                raise PiStarError("key_columns entries must be strings")
            if k not in col_index:
                raise PiStarError(
                    f"key_column {k!r} not in schema columns "
                    f"{col_names!r}"
                )
            key_indices.append(col_index[k])

        # Validate + fold each row.
        n_cols = len(col_names)
        folded: list[list[str]] = []
        for ri, row in enumerate(rows):
            if not isinstance(row, list):
                raise PiStarError(
                    f"rows[{ri}] must be a JSON array"
                )
            if len(row) != n_cols:
                raise PiStarError(
                    f"rows[{ri}] has {len(row)} cells; schema declares "
                    f"{n_cols}"
                )
            folded_row: list[str] = []
            for ci, cell in enumerate(row):
                folded_row.append(_fold_cell(cell, col_types[ci], ri, ci))
            folded.append(folded_row)

        # Sort by key tuple (stable). Empty key_columns → preserve
        # input order (some tables have no PK; the operator declares
        # that explicitly).
        if key_indices:
            folded.sort(
                key=lambda r: tuple(r[i] for i in key_indices)
            )

        # Serialize.
        schema_str = ",".join(
            f"{n}:{t}" for n, t in zip(col_names, col_types)
        )
        key_str = ",".join(key_columns)
        body = "\n".join("|".join(r) for r in folded)
        n = len(folded)
        out = (
            f"schema={schema_str}\n"
            f"key={key_str}\n"
            f"n={n}\n"
            f"rows:\n"
            f"{body}"
        )
        # Trailing newline only when rows non-empty so the empty-table
        # canonical is exactly: "schema=...\nkey=...\nn=0\nrows:".
        return out.encode("utf-8")


def _fold_cell(cell, col_type: str, ri: int, ci: int) -> str:
    """Fold one cell value into its canonical string form per type."""
    if col_type == "str":
        if not isinstance(cell, str):
            raise PiStarError(
                f"rows[{ri}][{ci}] expected str cell; got "
                f"{type(cell).__name__}"
            )
        return cell

    if col_type == "bool":
        if isinstance(cell, bool):
            return "true" if cell else "false"
        if isinstance(cell, str):
            normalized = cell.strip().lower()
            if normalized in ("true", "1", "yes"):
                return "true"
            if normalized in ("false", "0", "no"):
                return "false"
        raise PiStarError(
            f"rows[{ri}][{ci}] expected bool-shape cell; got {cell!r}"
        )

    if col_type == "int":
        # Accept int, float that's integer-valued, or string that parses.
        if isinstance(cell, bool):
            # bool is a subtype of int in Python; reject to avoid
            # silently treating True as 1.
            raise PiStarError(
                f"rows[{ri}][{ci}] expected int cell; got bool"
            )
        if isinstance(cell, int):
            return str(cell)
        if isinstance(cell, float):
            if not cell.is_integer():
                raise PiStarError(
                    f"rows[{ri}][{ci}] expected int cell; got "
                    f"non-integer float {cell!r}"
                )
            return str(int(cell))
        if isinstance(cell, str):
            try:
                f = Fraction(Decimal(cell.strip()))
            except (ValueError, InvalidOperation) as exc:
                raise PiStarError(
                    f"rows[{ri}][{ci}] cannot parse int from {cell!r}"
                ) from exc
            if f.denominator != 1:
                raise PiStarError(
                    f"rows[{ri}][{ci}] expected int cell; got "
                    f"non-integer rational {cell!r}"
                )
            return str(f.numerator)
        raise PiStarError(
            f"rows[{ri}][{ci}] cannot fold {cell!r} as int"
        )

    if col_type == "rational":
        # Fold to num/den lowest-terms form, sharing arithmetic@v1's
        # contract (Decimal(str(...)) for floats avoids drift).
        if isinstance(cell, bool):
            raise PiStarError(
                f"rows[{ri}][{ci}] expected rational cell; got bool"
            )
        try:
            if isinstance(cell, int):
                f = Fraction(cell, 1)
            elif isinstance(cell, float):
                f = Fraction(Decimal(str(cell)))
            elif isinstance(cell, str):
                stripped = cell.strip()
                if "/" in stripped:
                    f = Fraction(stripped)
                else:
                    f = Fraction(Decimal(stripped))
            else:
                raise PiStarError(
                    f"rows[{ri}][{ci}] cannot fold {cell!r} as rational"
                )
        except (ValueError, InvalidOperation, ZeroDivisionError) as exc:
            raise PiStarError(
                f"rows[{ri}][{ci}] cannot parse rational from {cell!r}: "
                f"{exc}"
            ) from exc
        return f"{f.numerator}/{f.denominator}"

    raise PiStarError(  # pragma: no cover — _VALID_TYPES exhausts above
        f"rows[{ri}][{ci}] unknown type {col_type!r}"
    )


register(TabularPinnedV1())
