Spaces:
Sleeping
Sleeping
File size: 6,314 Bytes
feb1b1c e776c3c feb1b1c e776c3c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 |
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
)
@property
def path(self) -> Path:
"""The bound source path (audit-only)."""
return self._path
@property
def gzipped(self) -> bool:
"""Whether the source is read through gzip decompression (audit-only)."""
return self._gzipped
@staticmethod
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()
@staticmethod
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",) |