"""Small typed tables in CSV, XLSX and ODS, without importing engines eagerly.
Writers create new export workbooks; they are not editors of existing workbook
styles, charts or formulas. Empty trailing worksheet padding is discarded,
while interior blank cells/rows remain explicit. Formulas are never evaluated.
"""
from __future__ import annotations
import csv
import math
import re
from collections.abc import Mapping, Sequence
from itertools import zip_longest
from numbers import Integral, Real
from pathlib import Path
from typing import Any, Literal
from knowledgespaces.structures.attribution import _positive_limit
Cell = str | int | float | bool | None
TableFormat = Literal["auto", "csv", "xlsx", "ods"]
FormulaPolicy = Literal["reject", "cached"]
def _format(path: str | Path, format: str) -> str:
selected = Path(path).suffix[1:].lower() if format == "auto" else format.lower()
if selected not in ("csv", "xlsx", "ods"):
raise ValueError("Table format must be csv, xlsx or ods (or auto with such an extension).")
return selected
def _cell(value: Any) -> Cell:
if value is None or isinstance(value, (str, bool)):
return value
if isinstance(value, Integral):
if abs(int(value)) > 2**53 - 1:
raise ValueError("Large integer cells must be stored as text to preserve precision.")
return int(value)
if isinstance(value, Real) and math.isfinite(value):
return float(value)
raise ValueError("Table cells must be text, finite numbers, booleans or None.")
def _sheet(names: Sequence[str], sheet: str | int) -> str:
if isinstance(sheet, str) and sheet in names:
return sheet
if isinstance(sheet, int) and not isinstance(sheet, bool) and 0 <= sheet < len(names):
return names[sheet]
raise ValueError("Unknown sheet; use its exact name or a zero-based index.")
class _Rows:
"""Defer blank padding and repeated ODS rows until actual content follows."""
def __init__(self, limit: int) -> None:
self.rows: list[list[Cell]] = []
self.width = 0
self.pending = 0
self.limit = limit
def add(self, row: list[Cell], repeat: int = 1) -> None:
while row and row[-1] is None:
row.pop()
if not row:
self.pending += repeat
return
width = max(self.width, len(row))
if (len(self.rows) + self.pending + repeat) * width > self.limit:
raise ValueError("Table exceeds max_cells; no truncated table is returned.")
self.rows.extend([] for _ in range(self.pending))
self.rows.extend(row.copy() for _ in range(repeat))
self.pending = 0
self.width = width
def result(self) -> list[list[Cell]]:
return [r + [None] * (self.width - len(r)) for r in self.rows]
[docs]
def read_table(
path: str | Path,
*,
sheet: str | int = 0,
format: TableFormat = "auto",
delimiter: str = ",",
formula_policy: FormulaPolicy = "reject",
max_cells: int = 1_000_000,
) -> list[list[Cell]]:
"""Read one rectangular table, including its header if present.
Only the selected sheet is interpreted. ``formula_policy='cached'`` uses
stored formula results explicitly; their freshness is the caller's
responsibility. Missing caches raise. The default rejects formulas.
CSV has no formula type and its fields are always read as literal text.
CSV sheet selection must be 0. ``max_cells`` bounds the returned rectangular
area, including interior blanks; trailing empty padding is ignored.
XLSX/ODS require the optional ``spreadsheets`` extra.
"""
_positive_limit(max_cells, "max_cells")
if formula_policy not in ("reject", "cached"):
raise ValueError("formula_policy must be reject or cached.")
kind = _format(path, format)
rows = _Rows(max_cells)
if kind == "csv":
if sheet != 0 or isinstance(sheet, bool):
raise ValueError("CSV has one table; sheet must be 0.")
with open(path, encoding="utf-8-sig", newline="") as stream:
for row in csv.reader(stream, delimiter=delimiter):
rows.add(list(row))
elif kind == "xlsx":
_read_xlsx(path, sheet, formula_policy, rows)
else:
_read_ods(path, sheet, formula_policy, rows)
return rows.result()
def _read_xlsx(path: str | Path, sheet: str | int, policy: str, rows: _Rows) -> None:
try:
from openpyxl import load_workbook
except ImportError as exc:
raise ImportError('XLSX requires pip install "knowledgespaces[spreadsheets]".') from exc
# File objects also support an explicitly declared format with a nonstandard suffix.
with open(path, "rb") as source, open(path, "rb") as cached_source:
book = load_workbook(source, read_only=True, data_only=False, keep_links=False)
cached = None
try:
name = _sheet(book.sheetnames, sheet)
worksheet = book[name]
worksheet.reset_dimensions()
cached_rows: Any = ()
if policy == "cached":
cached = load_workbook(
cached_source, read_only=True, data_only=True, keep_links=False
)
cached[name].reset_dimensions()
cached_rows = cached[name].iter_rows()
# Empty rows may be implicit in the XML. Do not rely on declared dimensions.
for cells, cache in zip_longest(worksheet.iter_rows(), cached_rows, fillvalue=()):
values = []
for i, cell in enumerate(cells):
value = cell.value
if cell.data_type == "f":
if policy == "reject":
raise ValueError(
"Formula cell encountered; request cached values explicitly."
)
value = cache[i].value if i < len(cache) else None
if value is None:
raise ValueError("Formula cell has no cached result.")
if cache[i].data_type == "e":
raise ValueError("Cached spreadsheet error cell encountered.")
if cell.data_type == "e":
raise ValueError("Spreadsheet error cell encountered.")
values.append(_cell(value))
rows.add(values)
finally:
book.close()
if cached is not None:
cached.close()
def _read_ods(path: str | Path, sheet: str | int, policy: str, rows: _Rows) -> None:
try:
from odf import teletype
from odf.namespaces import TABLENS
from odf.opendocument import load
from odf.table import Table, TableRow
from odf.text import P
except ImportError as exc:
raise ImportError('ODS requires pip install "knowledgespaces[spreadsheets]".') from exc
book = load(str(path))
sheets = book.spreadsheet.getElementsByType(Table)
names = [t.getAttribute("name") for t in sheets]
table = sheets[names.index(_sheet(names, sheet))]
for row in table.getElementsByType(TableRow):
values: list[Cell] = []
pending = 0
for cell in row.childNodes:
if getattr(cell, "qname", None) not in (
(TABLENS, "table-cell"),
(TABLENS, "covered-table-cell"),
):
continue
repeat = _repeat(cell, "numbercolumnsrepeated")
formula = cell.getAttribute("formula")
if formula and policy == "reject":
raise ValueError("Formula cell encountered; request cached values explicitly.")
kind = cell.getAttribute("valuetype")
value: Cell = None
if kind == "string":
value = cell.getAttribute("stringvalue")
if value is None:
value = "\n".join(teletype.extractText(p) for p in cell.getElementsByType(P))
elif kind in ("float", "percentage", "currency"):
value = _cell(float(cell.getAttribute("value")))
elif kind == "boolean":
raw = cell.getAttribute("booleanvalue")
if raw not in ("true", "false"):
raise ValueError("Invalid ODS boolean cell.")
value = raw == "true"
elif kind:
raise ValueError(f"Unsupported ODS cell type: {kind}.")
if formula and value is None:
raise ValueError("Formula cell has no cached result.")
if value is None:
pending += repeat
else:
if len(values) + pending + repeat > rows.limit:
raise ValueError("Table exceeds max_cells.")
values.extend([None] * pending)
values.extend([value] * repeat)
pending = 0
rows.add(values, _repeat(row, "numberrowsrepeated"))
def _repeat(element: Any, name: str) -> int:
try:
count = int(element.getAttribute(name) or "1")
except ValueError as exc:
raise ValueError("Invalid ODS repetition count.") from exc
if count <= 0:
raise ValueError("Invalid ODS repetition count.")
return count
[docs]
def write_tables(
tables: Mapping[str, Sequence[Sequence[Cell]]],
path: str | Path,
*,
format: TableFormat = "auto",
delimiter: str = ",",
max_cells: int = 1_000_000,
) -> None:
"""Create an export file from named tables (headers are ordinary first rows).
CSV accepts exactly one table. XLSX and ODS preserve literal string labels,
including strings starting with '='; they are never written as formulas.
Interior blanks remain empty cells. All input cells and dimensions are
checked before opening the output. No existing workbook content is retained.
"""
_positive_limit(max_cells, "max_cells")
kind = _format(path, format)
if not tables or any(not isinstance(n, str) or not n for n in tables):
raise ValueError("Export requires at least one named table.")
prepared = {}
for name, table in tables.items():
if kind == "xlsx" and (any(c in name for c in "[]:*?/\\") or len(name) > 31):
raise ValueError("Sheet names must be portable to XLSX (31 characters, no []:*?/\\).")
rows = _Rows(max_cells)
for row in table:
values = [_cell(v) for v in row]
for value in values:
if isinstance(value, str):
if kind == "xlsx" and len(value) > 32767:
raise ValueError(
"XLSX text cells cannot exceed 32767 characters; no truncation is allowed."
)
if kind != "csv" and re.search(
r"[\x00-\x08\x0b\x0c\x0e-\x1f\ud800-\udfff\ufffe\uffff]", value
):
raise ValueError("Workbook text contains characters disallowed by XML.")
rows.add(values)
prepared[name] = rows.result()
if kind == "xlsx" and len({n.lower() for n in prepared}) != len(prepared):
raise ValueError("Sheet names must be unique ignoring case.")
if kind == "csv":
if len(prepared) != 1:
raise ValueError("CSV cannot contain multiple sheets.")
with open(path, "w", encoding="utf-8", newline="") as stream:
csv.writer(stream, delimiter=delimiter).writerows(next(iter(prepared.values())))
elif kind == "xlsx":
_write_xlsx(prepared, path)
else:
_write_ods(prepared, path)
def _write_xlsx(tables: dict[str, list[list[Cell]]], path: str | Path) -> None:
try:
from openpyxl import Workbook
from openpyxl.cell import WriteOnlyCell
except ImportError as exc:
raise ImportError('XLSX requires pip install "knowledgespaces[spreadsheets]".') from exc
book = Workbook(write_only=True)
for name, rows in tables.items():
sheet = book.create_sheet(name)
for row in rows:
cells = []
for value in row:
cell = WriteOnlyCell(sheet, value=value)
if isinstance(value, str):
cell.data_type = "s"
cells.append(cell)
sheet.append(cells)
if any(
isinstance(v, str) and "\r" in v for rows in tables.values() for row in rows for v in row
):
# et_xmlfile emits literal CRs in XML text; XML readers normalize them.
# Escape those characters in this newly generated archive to preserve labels.
from tempfile import TemporaryFile
from zipfile import ZipFile
with TemporaryFile() as temporary:
book.save(temporary)
temporary.seek(0)
with ZipFile(temporary) as source, ZipFile(path, "w") as target:
for entry in source.infolist():
data = source.read(entry.filename)
if entry.filename.endswith(".xml"):
data = data.replace(b"\r", b" ")
target.writestr(entry, data)
else:
book.save(path)
def _write_ods(tables: dict[str, list[list[Cell]]], path: str | Path) -> None:
try:
from odf import teletype
from odf.element import Text
from odf.opendocument import OpenDocumentSpreadsheet
from odf.table import Table, TableCell, TableRow
from odf.text import P
except ImportError as exc:
raise ImportError('ODS requires pip install "knowledgespaces[spreadsheets]".') from exc
class CarriageReturn(Text):
# XML normalizes literal CRs; a character reference preserves the label.
def toXml(self, level: int, stream: Any) -> None:
stream.write(" ")
book = OpenDocumentSpreadsheet()
for name, rows in tables.items():
sheet = Table(name=name)
for row in rows:
target = TableRow()
for value in row:
if value is None:
cell = TableCell()
elif isinstance(value, str):
cell = TableCell(valuetype="string")
paragraph = P()
for i, part in enumerate(value.split("\r")):
if i:
paragraph.appendChild(CarriageReturn("\r"))
teletype.addTextToElement(paragraph, part)
cell.addElement(paragraph)
elif isinstance(value, bool):
cell = TableCell(valuetype="boolean", booleanvalue=str(value).lower())
cell.addElement(P(text=str(value).upper()))
else:
cell = TableCell(valuetype="float", value=str(value))
cell.addElement(P(text=str(value)))
target.addElement(cell)
sheet.addElement(target)
book.spreadsheet.addElement(sheet)
book.save(str(path), addsuffix=False)