io.py

"""Structured file and JSON utilities shared across the TNFR engine."""

from __future__ import annotations

import json
import os
import tempfile
from dataclasses import dataclass
from functools import partial
from pathlib import Path
from typing import Any, Callable

from .._exact_time import finite_represented_real
from ..errors import TNFRValueError
from .init import LazyImportProxy, cached_import, get_logger

logger = get_logger(__name__)


@dataclass(frozen=True)
class JsonDumpsParams:
    """Container describing the parameters used by :func:`json_dumps`."""

    sort_keys: bool = False
    default: Callable[[Any], Any] | None = None
    ensure_ascii: bool = True
    separators: tuple[str, str] = (",", ":")
    cls: type[json.JSONEncoder] | None = None
    to_bytes: bool = False


DEFAULT_PARAMS = JsonDumpsParams()


def _reject_duplicate_json_keys(pairs: list[tuple[str, Any]]) -> None:
    """Reject colliding decoded names without retaining a second object tree."""
    names: set[str] = set()
    for name, _ in pairs:
        if name in names:
            raise ValueError(f"JSON object keys collide after serialization: {name!r}")
        names.add(name)


class _JSONValueError(ValueError):
    """Syntactically decodable JSON with ambiguous or unrepresentable values."""


def _decoded_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    _reject_duplicate_json_keys(pairs)
    return dict(pairs)


def _reject_nonfinite_constant(token: str) -> None:
    raise ValueError(f"JSON numbers must be finite; found {token}")


def _represented_json_float(token: str) -> float:
    value = float(token)
    finite_represented_real(value, "JSON number")
    # The decoder admits the grammar. Only the mantissa decides exact zero,
    # independently of the exponent's magnitude.
    mantissa = token.lower().split("e", 1)[0]
    if value == 0.0 and any(digit in "123456789" for digit in mantissa):
        raise ValueError("JSON number is nonzero but underflows to represented zero")
    return value


def json_loads(text: str | bytes | bytearray) -> Any:
    """Decode JSON using the shared report and configuration admission policy.

    Reject duplicate decoded names, nonfinite numbers and nonzero fractional
    literals that underflow to binary64 zero. Integers retain Python's integer
    representation; finite fractions retain normal float rounding. Object,
    array and scalar roots are supported. This does not validate model fields
    or authenticate a report's provenance.
    """
    try:
        return json.loads(
            text,
            object_pairs_hook=_decoded_object,
            parse_constant=_reject_nonfinite_constant,
            parse_float=_represented_json_float,
        )
    except (json.JSONDecodeError, UnicodeDecodeError):
        raise
    except ValueError as exc:
        raise _JSONValueError(str(exc)) from exc


def _json_dumps_std(
    obj: Any,
    params: JsonDumpsParams,
    **kwargs: Any,
) -> bytes | str:
    """Serialize using the standard library :func:`json.dumps`."""

    result = json.dumps(
        obj,
        sort_keys=params.sort_keys,
        ensure_ascii=params.ensure_ascii,
        separators=params.separators,
        cls=params.cls,
        default=params.default,
        **kwargs,
    )
    return result if not params.to_bytes else result.encode("utf-8")


def json_dumps(
    obj: Any,
    *,
    sort_keys: bool = False,
    default: Callable[[Any], Any] | None = None,
    ensure_ascii: bool = True,
    separators: tuple[str, str] = (",", ":"),
    cls: type[json.JSONEncoder] | None = None,
    to_bytes: bool = False,
    **kwargs: Any,
) -> bytes | str:
    """Serialize with stable standard-library JSON semantics.

    Optional encoders must not change cache/signature bytes, turn nonfinite
    values into null, reject otherwise supported integers/keys, or ignore
    requested formatting. One encoder keeps those contracts independent of
    installed dependencies. Set allow_nan=False for strict finite JSON.
    """

    if not isinstance(sort_keys, bool):
        raise TypeError("sort_keys must be a boolean")
    if default is not None and not callable(default):
        raise TypeError("default must be callable when provided")
    if not isinstance(ensure_ascii, bool):
        raise TypeError("ensure_ascii must be a boolean")
    if not isinstance(separators, tuple) or len(separators) != 2:
        raise TypeError("separators must be a tuple of two strings")
    if not all(isinstance(part, str) for part in separators):
        raise TypeError("separators must be a tuple of two strings")
    if cls is not None:
        if not isinstance(cls, type) or not issubclass(cls, json.JSONEncoder):
            raise TypeError("cls must be a subclass of json.JSONEncoder")
    if not isinstance(to_bytes, bool):
        raise TypeError("to_bytes must be a boolean")

    if (
        sort_keys is False
        and default is None
        and ensure_ascii is True
        and separators == (",", ":")
        and cls is None
        and to_bytes is False
    ):
        params = DEFAULT_PARAMS
    else:
        params = JsonDumpsParams(
            sort_keys=sort_keys,
            default=default,
            ensure_ascii=ensure_ascii,
            separators=separators,
            cls=cls,
            to_bytes=to_bytes,
        )
    return _json_dumps_std(obj, params, **kwargs)


def _raise_import_error(name: str, *_: Any, **__: Any) -> Any:
    raise ImportError(f"{name} is not installed")


_MISSING_TOML_ERROR = type(
    "MissingTOMLDependencyError",
    (Exception,),
    {"__doc__": "Fallback error used when tomllib/tomli is missing."},
)

_MISSING_YAML_ERROR = type(
    "MissingPyYAMLDependencyError",
    (Exception,),
    {"__doc__": "Fallback error used when pyyaml is missing."},
)


def _resolve_lazy(value: Any) -> Any:
    if isinstance(value, LazyImportProxy):
        return value.resolve()
    return value


class _LazyBool:
    __slots__ = ("_value",)

    def __init__(self, value: Any) -> None:
        self._value = value

    def __bool__(self) -> bool:
        return _resolve_lazy(self._value) is not None


_TOMLI_MODULE = cached_import("tomli", emit="log", lazy=True)
tomllib = cached_import(
    "tomllib",
    emit="log",
    lazy=True,
    fallback=_TOMLI_MODULE,
)
has_toml = _LazyBool(tomllib)

_TOMLI_TOML_ERROR = cached_import(
    "tomli",
    "TOMLDecodeError",
    emit="log",
    lazy=True,
    fallback=_MISSING_TOML_ERROR,
)
TOMLDecodeError = cached_import(
    "tomllib",
    "TOMLDecodeError",
    emit="log",
    lazy=True,
    fallback=_TOMLI_TOML_ERROR,
)

_TOMLI_LOADS = cached_import(
    "tomli",
    "loads",
    emit="log",
    lazy=True,
    fallback=partial(_raise_import_error, "tomllib/tomli"),
)
_TOML_LOADS: Callable[[str], Any] = cached_import(
    "tomllib",
    "loads",
    emit="log",
    lazy=True,
    fallback=_TOMLI_LOADS,
)

yaml = cached_import("yaml", emit="log", lazy=True)

YAMLError = cached_import(
    "yaml",
    "YAMLError",
    emit="log",
    lazy=True,
    fallback=_MISSING_YAML_ERROR,
)

_YAML_SAFE_LOAD: Callable[[str], Any] = cached_import(
    "yaml",
    "safe_load",
    emit="log",
    lazy=True,
    fallback=partial(_raise_import_error, "pyyaml"),
)


def _parse_yaml(text: str) -> Any:
    """Parse YAML ``text`` using ``safe_load`` if available."""

    return _YAML_SAFE_LOAD(text)


def _parse_toml(text: str) -> Any:
    """Parse TOML ``text`` using ``tomllib`` or ``tomli``."""

    return _TOML_LOADS(text)


PARSERS = {
    ".json": json_loads,
    ".yaml": _parse_yaml,
    ".yml": _parse_yaml,
    ".toml": _parse_toml,
}


def _get_parser(suffix: str) -> Callable[[str], Any]:
    try:
        return PARSERS[suffix]
    except KeyError as exc:
        raise TNFRValueError(
            f"Unsupported suffix: {suffix}",
            context={"suffix": suffix, "supported": list(PARSERS.keys())},
            suggestion="Use a supported file extension.",
        ) from exc


_BASE_ERROR_MESSAGES: dict[type[BaseException], str] = {
    OSError: "Could not read {path}: {e}",
    UnicodeDecodeError: "Encoding error while reading {path}: {e}",
    json.JSONDecodeError: "Error parsing JSON file at {path}: {e}",
    _JSONValueError: "Invalid JSON value at {path}: {e}",
    ImportError: "Missing dependency parsing {path}: {e}",
}


def _resolve_exception_type(candidate: Any) -> type[BaseException] | None:
    resolved = _resolve_lazy(candidate)
    if isinstance(resolved, type) and issubclass(resolved, BaseException):
        return resolved
    return None


_OPTIONAL_ERROR_MESSAGE_FACTORIES: tuple[
    tuple[Callable[[], type[BaseException] | None], str],
    ...,
] = (
    (
        lambda: _resolve_exception_type(YAMLError),
        "Error parsing YAML file at {path}: {e}",
    ),
    (
        lambda: _resolve_exception_type(TOMLDecodeError),
        "Error parsing TOML file at {path}: {e}",
    ),
)

_BASE_STRUCTURED_EXCEPTIONS = (
    OSError,
    UnicodeDecodeError,
    json.JSONDecodeError,
    _JSONValueError,
    ImportError,
)


def _iter_optional_exceptions() -> list[type[BaseException]]:
    errors: list[type[BaseException]] = []
    for resolver, _ in _OPTIONAL_ERROR_MESSAGE_FACTORIES:
        exc_type = resolver()
        if exc_type is not None:
            errors.append(exc_type)
    return errors


def _is_structured_error(exc: Exception) -> bool:
    if isinstance(exc, _BASE_STRUCTURED_EXCEPTIONS):
        return True
    for optional_exc in _iter_optional_exceptions():
        if isinstance(exc, optional_exc):
            return True
    return False


def _format_structured_file_error(path: Path, e: Exception) -> str:
    for exc, msg in _BASE_ERROR_MESSAGES.items():
        if isinstance(e, exc):
            return msg.format(path=path, e=e)

    for resolver, msg in _OPTIONAL_ERROR_MESSAGE_FACTORIES:
        exc_type = resolver()
        if exc_type is not None and isinstance(e, exc_type):
            return msg.format(path=path, e=e)

    return f"Error parsing {path}: {e}"


class StructuredFileError(Exception):
    """Error while reading or parsing a structured file."""

    def __init__(self, path: Path, original: Exception) -> None:
        super().__init__(_format_structured_file_error(path, original))
        self.path = path


def read_structured_file(
    path: Path | str,
    *,
    base_dir: Path | str | None = None,
    allowed_extensions: tuple[str, ...] | None = (".json", ".yaml", ".yml", ".toml"),
) -> Any:
    """Read a JSON, YAML or TOML file and return parsed data.

    This function includes path traversal protection. When ``base_dir`` is
    provided, the resolved path must stay within that directory.
    JSON uses :func:`json_loads`'s strict value admission. YAML and TOML retain
    their respective parser semantics; consumers validate their model fields.

    Parameters
    ----------
    path : Path | str
        Path to the structured file to read.
    base_dir : Path | str | None, optional
        Base directory to restrict file access. If provided, the resolved
        path must stay within this directory (prevents path traversal).
    allowed_extensions : tuple[str, ...] | None, optional
        tuple of allowed file extensions. Default is JSON, YAML, and TOML.
        Pass None to allow any extension (not recommended for user input).

    Returns
    -------
    Any
        Parsed data from the file.

    Raises
    ------
    StructuredFileError
        If the file cannot be read or parsed.
    ValueError
        If the path is invalid or contains unsafe patterns.
    PathTraversalError
        If path traversal is detected.

    Examples
    --------
    >>> from pathlib import Path
    >>> # Read config file with path validation
    >>> data = read_structured_file("config.json")  # doctest: +SKIP

    >>> # Read with base directory restriction
    >>> data = read_structured_file(
    ...     "settings.yaml",
    ...     base_dir="/home/user/configs"
    ... )  # doctest: +SKIP
    """
    # Import here to avoid circular dependency
    from ..security import PathTraversalError, resolve_safe_path, validate_file_path

    # Validate and resolve the path
    try:
        if base_dir is not None:
            # Resolve path within base directory (prevents traversal)
            validated_path = resolve_safe_path(
                path,
                base_dir,
                must_exist=True,
                allowed_extensions=allowed_extensions,
            )
        else:
            # Validate path without base directory restriction
            path_obj = Path(path) if not isinstance(path, Path) else path
            validated_path = validate_file_path(
                path_obj,
                allow_absolute=True,
                allowed_extensions=allowed_extensions,
            ).resolve()

            # Check existence
            if not validated_path.exists():
                raise FileNotFoundError(f"File not found: {validated_path}")
    except (ValueError, PathTraversalError) as e:
        raise StructuredFileError(Path(path), e) from e
    except FileNotFoundError as e:
        raise StructuredFileError(Path(path), e) from e

    suffix = validated_path.suffix.lower()
    try:
        parser = _get_parser(suffix)
    except ValueError as e:
        raise StructuredFileError(validated_path, e) from e
    try:
        text = validated_path.read_text(encoding="utf-8")
        return parser(text)
    except Exception as e:
        if _is_structured_error(e):
            raise StructuredFileError(validated_path, e) from e
        raise


def safe_write(
    path: str | Path,
    write: Callable[[Any], Any],
    *,
    mode: str = "w",
    encoding: str | None = "utf-8",
    atomic: bool = True,
    sync: bool | None = None,
    base_dir: str | Path | None = None,
    **open_kwargs: Any,
) -> None:
    """Write to ``path`` ensuring parent directory exists and handle errors.

    This function includes path traversal protection. When ``base_dir`` is
    provided, the resolved path must stay within that directory.

    Parameters
    ----------
    path:
        Destination file path.
    write:
        Callback receiving the opened file object and performing the actual
        write.
    mode:
        File mode passed to :func:`open`. Text modes (default) use UTF-8
        encoding unless ``encoding`` is ``None``. When a binary mode is used
        (``'b'`` in ``mode``) no encoding parameter is supplied so
        ``write`` may write bytes. Atomic writing supports replacement modes
        based on 'w' only. Use atomic=False for append, update or exclusive
        creation; a fresh temporary file cannot supply those semantics.
    encoding:
        Encoding for text modes. Ignored for binary modes.
    atomic:
        When ``True`` (default) writes to a temporary file and atomically
        replaces the destination after flushing to disk. When ``False``
        writes directly to ``path`` without any atomicity guarantee.
    sync:
        When ``True`` flushes and fsyncs the file descriptor after writing.
        ``None`` uses ``atomic`` to determine syncing behaviour.
    base_dir:
        Optional base directory to restrict file writes. If provided, the
        resolved path must stay within this directory (prevents path traversal).

    Raises
    ------
    ValueError
        If the path is invalid or contains unsafe patterns.
    PathTraversalError
        If path traversal is detected when base_dir is provided.
    """
    # Import here to avoid circular dependency
    from ..security import PathTraversalError, resolve_safe_path, validate_file_path

    # Validate and resolve the path
    try:
        if base_dir is not None:
            # Resolve path within base directory (prevents traversal)
            validated_path = resolve_safe_path(
                path,
                base_dir,
                must_exist=False,
            )
        else:
            # Validate path without base directory restriction
            path_obj = Path(path) if not isinstance(path, Path) else path
            validated_path = validate_file_path(
                path_obj,
                allow_absolute=True,
            ).resolve()
    except (ValueError, PathTraversalError) as e:
        raise type(e)(f"Invalid path {path!r}: {e}") from e

    if not isinstance(mode, str):
        raise TypeError("mode must be a string")
    if atomic and mode.replace("b", "").replace("t", "").replace("+", "") != "w":
        raise ValueError("Atomic writing requires a replacement mode based on 'w'")

    path = validated_path
    path.parent.mkdir(parents=True, exist_ok=True)
    open_params = dict(mode=mode, **open_kwargs)
    if "b" not in mode and encoding is not None:
        open_params["encoding"] = encoding
    if sync is None:
        sync = atomic
    tmp_path: Path | None = None
    try:
        if atomic:
            tmp_fd = tempfile.NamedTemporaryFile(dir=path.parent, delete=False)
            tmp_path = Path(tmp_fd.name)
            tmp_fd.close()
            with open(tmp_path, **open_params) as fd:
                write(fd)
                if sync:
                    fd.flush()
                    os.fsync(fd.fileno())
            try:
                os.replace(tmp_path, path)
            except OSError as e:
                logger.error(
                    "Atomic replace failed for %s -> %s: %s", tmp_path, path, e
                )
                raise
        else:
            with open(path, **open_params) as fd:
                write(fd)
                if sync:
                    fd.flush()
                    os.fsync(fd.fileno())
    except (OSError, ValueError, TypeError) as e:
        # Preserve structured exception arguments, including encoding details
        # and OSError.errno. add_note is available on Python 3.11+.
        if hasattr(e, "add_note"):
            e.add_note(f"Failed to write file {path}")
        raise
    finally:
        if tmp_path is not None:
            tmp_path.unlink(missing_ok=True)


__all__ = (
    "JsonDumpsParams",
    "DEFAULT_PARAMS",
    "json_dumps",
    "json_loads",
    "read_structured_file",
    "safe_write",
    "StructuredFileError",
    "TOMLDecodeError",
    "YAMLError",
)