"""File readers — iterate raw rows from various file formats. Each reader yields rows as list[str]. The first row is typically headers, but that interpretation is left to the consumer (DictRowIterator). """ import csv import json from collections.abc import Iterator from typing import Protocol, runtime_checkable from openpyxl import load_workbook @runtime_checkable class FileReader(Protocol): """Protocol for reading raw rows from a file.""" def read(self, file_path: str) -> Iterator[list[str]]: ... class CsvFileReader: """Reads a CSV file, yielding each row as a list of strings.""" def read(self, file_path: str) -> Iterator[list[str]]: with open(file_path, 'r', encoding='utf-8') as f: reader = csv.reader(f) yield from reader class JsonFileReader: """Reads a JSON file containing a list of objects. Uses incremental parsing (ijson) when available, falling back to ``json.load`` for small files. The first yielded row is the header (keys of the first object). Subsequent rows are values in header order. """ def read(self, file_path: str) -> Iterator[list[str]]: try: import ijson yield from self._read_streaming(file_path, ijson) except ImportError: yield from self._read_eager(file_path) @staticmethod def _read_eager(file_path: str) -> Iterator[list[str]]: with open(file_path, 'r', encoding='utf-8') as f: data = json.load(f) if not isinstance(data, list) or not data: return headers = list(data[0].keys()) yield headers for obj in data: yield [str(obj.get(h, '')) for h in headers] @staticmethod def _read_streaming(file_path: str, ijson) -> Iterator[list[str]]: headers: list[str] | None = None with open(file_path, 'rb') as f: for obj in ijson.items(f, 'item'): if headers is None: headers = list(obj.keys()) yield headers yield [str(obj.get(h, '')) for h in headers] class XlsxFileReader: """Reads an XLSX file, yielding each row as a list of strings. Requires openpyxl to be installed. """ def read(self, file_path: str) -> Iterator[list[str]]: wb = load_workbook(file_path, read_only=True, data_only=True) ws = wb.active for row in ws.iter_rows(values_only=True): yield [str(cell) if cell is not None else '' for cell in row] wb.close()