Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| import gzip | |
| import io | |
| import json | |
| from collections.abc import Iterator | |
| from pathlib import Path | |
| from typing import TYPE_CHECKING, Final | |
| from redstack.ports._types import RawMapping, SourceMalformed, SourceOk, SourceRecord | |
| from redstack.ports.candidate_source import CandidateSourceError | |
| #: gzip magic prefix; used for suffix-independent auto-detection of compression. | |
| _GZIP_MAGIC: Final[bytes] = b"\x1f\x8b" | |
| class JsonlCandidateSourceAdapter: | |
| """Lazy, single-pass, order-preserving JSONL/JSONL.GZ candidate source. | |
| Constructed by the pipeline composition root with a resolved path; never | |
| self-constructs. Each :meth:`stream` call opens a fresh handle and returns | |
| an independent one-pass iterator; the handle closes on iterator exhaustion | |
| or generator close. Not re-entrant; not shared across threads. | |
| """ | |
| __slots__ = ("_path", "_gzipped") | |
| def __init__(self, path: Path, *, gzipped: bool | None = None) -> None: | |
| """Bind the adapter to a resolved input path. | |
| Args: | |
| path: Filesystem path to the ``.jsonl`` / ``.jsonl.gz`` source. | |
| gzipped: Force gzip handling (``True``) or plain (``False``); when | |
| ``None`` (default) compression is auto-detected from the | |
| ``.gz`` suffix and confirmed by the file's magic bytes. | |
| """ | |
| self._path: Final[Path] = path | |
| self._gzipped: Final[bool] = ( | |
| self._detect_gzip(path) if gzipped is None else gzipped | |
| ) | |
| def path(self) -> Path: | |
| """The bound source path (audit-only).""" | |
| return self._path | |
| def gzipped(self) -> bool: | |
| """Whether the source is read through gzip decompression (audit-only).""" | |
| return self._gzipped | |
| def _detect_gzip(path: Path) -> bool: | |
| """Auto-detect gzip by ``.gz`` suffix, confirmed by the magic prefix. | |
| Suffix is authoritative for naming; the magic-byte read is a cheap guard | |
| against a mislabeled file. A missing/unreadable file is left for | |
| :meth:`stream` to surface as ``CandidateSourceError`` (detection is | |
| best-effort and never raises). | |
| """ | |
| if path.suffix != ".gz": | |
| return False | |
| try: | |
| with open(path, "rb") as probe: | |
| return probe.read(2) == _GZIP_MAGIC | |
| except OSError: | |
| # Defer the IO failure to stream(); assume gzip per the suffix. | |
| return True | |
| def _open(self) -> gzip.GzipFile | io.BufferedReader: | |
| """Open a fresh binary handle, transparently gzip-wrapping when needed. | |
| Raises: | |
| CandidateSourceError: the source cannot be opened or decompressed. | |
| """ | |
| try: | |
| if self._gzipped: | |
| return gzip.open(self._path, "rb") | |
| return open(self._path, "rb") | |
| except (OSError, EOFError) as exc: | |
| raise CandidateSourceError( | |
| f"cannot open source {self._path!s}: {exc}" | |
| ) from exc | |
| def stream(self) -> Iterator[SourceRecord]: | |
| """Yield records lazily in file order, one at a time. | |
| ``SourceOk`` carries the undecoded JSON object plus its ``source_index`` | |
| (enumeration order over emitted records) and physical ``line_no``; | |
| ``SourceMalformed`` carries ``line_no`` and an error string. Calling | |
| ``stream`` again starts a fresh, independent pass. | |
| Raises: | |
| CandidateSourceError: the source cannot be opened/decompressed, or a | |
| low-level IO/decompression error occurs mid-stream. | |
| """ | |
| handle = self._open() | |
| try: | |
| line_no = 0 | |
| source_index = 0 | |
| while True: | |
| try: | |
| raw_line = handle.readline() | |
| except (OSError, EOFError) as exc: | |
| raise CandidateSourceError( | |
| f"IO error reading {self._path!s} at line {line_no + 1}: {exc}" | |
| ) from exc | |
| if not raw_line: | |
| break | |
| line_no += 1 | |
| if not raw_line.strip(): | |
| # Blank / whitespace-only line: not a record (validator row | |
| # semantics). Consumes a physical line, no source_index. | |
| continue | |
| record = self._decode_line(raw_line, line_no, source_index) | |
| source_index += 1 | |
| yield record | |
| finally: | |
| handle.close() | |
| def _decode_line( | |
| raw_line: bytes, line_no: int, source_index: int | |
| ) -> SourceRecord: | |
| """Decode one non-blank physical line into a ``SourceRecord``. | |
| UTF-8 strict, then ``json.loads``, then a top-level-object structural | |
| check. Any failure is reported as ``SourceMalformed`` (data, never an | |
| exception); the schema boundary is downstream. | |
| """ | |
| try: | |
| text = raw_line.decode("utf-8") | |
| except UnicodeDecodeError as exc: | |
| return SourceMalformed(line_no=line_no, error=f"utf-8 decode: {exc}") | |
| try: | |
| decoded: object = json.loads(text) | |
| except json.JSONDecodeError as exc: | |
| return SourceMalformed(line_no=line_no, error=f"json decode: {exc.msg}") | |
| if not isinstance(decoded, dict): | |
| return SourceMalformed( | |
| line_no=line_no, | |
| error=f"top-level JSON value is {type(decoded).__name__}, not object", | |
| ) | |
| raw: RawMapping = {str(key): value for key, value in decoded.items()} | |
| return SourceOk(raw=raw, line_no=line_no, source_index=source_index) | |
| def count(self) -> int | None: | |
| """Return ``None``: the record count is unknown without a full pass. | |
| Counting would require streaming the entire source, which would violate | |
| the single-pass / O(1) contract, so the honest cheap answer is ``None``. | |
| """ | |
| return None | |
| if TYPE_CHECKING: | |
| from redstack.ports.candidate_source import CandidateSourcePort | |
| # Compile-time structural conformance to the frozen port surface. | |
| _PORT_CONFORMANCE: type[CandidateSourcePort] = JsonlCandidateSourceAdapter | |
| __all__: tuple[str, ...] = ("JsonlCandidateSourceAdapter",) |