feat(raw): raw derive writes an Excel file's sheets as TSV files with a formula list (#202)
CI / verify (push) Successful in 6m47s
CI / pwsh (push) Successful in 2m25s
Release / release (push) Successful in 34s
Release / wiki-mirror (push) Skipped

Each sheet becomes NN-<sheet>.tsv (RFC 4180, tab-delimited, A1-aligned) in a
<stem>.sheets/ folder beside the .md; sheets with formulas also get a
.formulas.tsv. derived.md is the overview. Adds testfiles/ with a CI test.

Files changed:
- .gitattributes
- CHANGES.md
- VERSION
- docs/dev/development.md
- docs/how-to/ingest-a-source.md
- docs/reference/commands/raw-material-and-uploads.md
- raw/CONTRACT.md
- testfiles/README.md
- tools/CONTRACT.md
- tools/chemenu/commands/raw_cmd.py
- tools/chemenu/filters/__init__.py
- tools/chemenu/filters/archive.py
- tools/chemenu/filters/ooxml.py
- tools/chemenu/tests/test_raw_derive.py
- tools/chemenu/tests/test_raw_derive_testfiles.py

Co-Authored-By: Claude Code <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SnAJ7Z3CpVD3PRbN73QtU2
This commit is contained in:
torbenandClaude Code committed 2026-10-09 13:35:03 +02:00
1 parent 8ea2795d88
commit 9868fce3d5
15 files changed
+719 -50

No files matched your search

+196 -11
View File
@@ -5,9 +5,11 @@ readable in the test that uses it and no binary sits in the repository.
"""
from __future__ import annotations
import csv
import hashlib
import io
import json
import shutil
import zipfile
from email import message_from_bytes, policy
from email.message import EmailMessage
@@ -156,6 +158,30 @@ def make_xlsx() -> bytes:
})
def make_book(sheets: list[tuple[str, str, str]]) -> bytes:
"""A workbook of `(name, state, <row> elements)` with inline strings only."""
rels = "".join(
f'<Relationship Id="rId{n}" Type="{R}/worksheet" Target="worksheets/sheet{n}.xml"/>'
for n in range(1, len(sheets) + 1)
)
listed = "".join(
f'<sheet name="{name}" sheetId="{n}" state="{state}" r:id="rId{n}"/>'
for n, (name, state, _rows) in enumerate(sheets, 1)
)
parts = {
"_rels/.rels": _rels("xl/workbook.xml"),
"xl/workbook.xml": f'<workbook xmlns="{S}" xmlns:r="{R}"><sheets>{listed}</sheets></workbook>',
"xl/_rels/workbook.xml.rels": f'<Relationships xmlns="{REL}">{rels}</Relationships>',
}
for n, (_name, _state, rows) in enumerate(sheets, 1):
parts[f"xl/worksheets/sheet{n}.xml"] = f'<worksheet xmlns="{S}"><sheetData>{rows}</sheetData></worksheet>'
return _zip(parts)
def _text_cell(ref: str, text: str) -> str:
return f'<c r="{ref}" t="inlineStr"><is><t xml:space="preserve">{text.replace(chr(13), "&#13;")}</t></is></c>'
def make_eml(body: str = "Hallo Welt\n", *, html: str = "", attach: bool = False, inline: bool = False,
subject: str = "Angebot") -> bytes:
msg = EmailMessage()
@@ -374,7 +400,11 @@ def test_a_folder_gets_a_md_beside_each_file_with_a_filter(tree, capsys):
_derive("incoming/mappe")
out = capsys.readouterr().out
assert sorted(p.name for p in folder.iterdir()) == ["foto.jpg", "plan.docx", "plan.md", "zahlen.md", "zahlen.xlsx"]
assert sorted(p.name for p in folder.iterdir()) == ["foto.jpg", "plan.docx", "plan.md", "zahlen.md", "zahlen.sheets", "zahlen.xlsx"]
assert sorted(p.name for p in (folder / "zahlen.sheets").iterdir()) == [
"01-budget.formulas.tsv", "01-budget.tsv", "02-geheim.tsv",
]
assert "`zahlen.sheets/01-budget.tsv`" in (folder / "zahlen.md").read_text(encoding="utf-8")
header, _ = _header_and_text(folder / "plan.md")
assert header["derived_from"] == "plan.docx"
assert "original_name" not in header
@@ -487,20 +517,175 @@ def test_pptx_slides_notes_and_footer_placeholders(tree):
# --- XLSX -------------------------------------------------------------------
def test_xlsx_values_hidden_sheet_and_trimmed_edges(tree, capsys):
def _tsv_rows(path: Path) -> list[list[str]]:
with path.open(newline="", encoding="utf-8") as handle:
return list(csv.reader(handle, delimiter="\t"))
def test_xlsx_becomes_an_overview_and_one_tsv_per_sheet(tree, capsys):
_put(tree, "zahlen.xlsx", make_xlsx())
_derive("incoming/zahlen.xlsx")
out = capsys.readouterr().out
header, text = _header_and_text(tree / "incoming/zahlen/derived.md")
folder = tree / "incoming/zahlen"
header, text = _header_and_text(folder / "derived.md")
assert header["sheets"] == 2 and header["hidden_sheets"] == ["Geheim"]
assert "## Budget" in text and "## Geheim" in text and "VERSTECKT-INHALT" in text
assert "| Posten | Wert |" in text
assert "| Miete | 1200 |" in text
assert "2024-01-01" in text
assert "TRUE" in text
assert text.splitlines()[text.splitlines().index("## Budget") + 2].startswith("| Posten")
assert "Values are as last saved" in out and "Hidden sheets" in out
assert header["sheets"] == 2 and header["hidden_sheets"] == ["Geheim"] and header["formulas"] == 1
assert "## Budget" in text and "## Geheim" in text
assert "Sheet 2 of 2, hidden" in text
assert "Range A1:C5 (5 rows x 3 columns), 1 formula cell(s)" in text
assert "`derived.sheets/01-budget.tsv`" in text and "`derived.sheets/01-budget.formulas.tsv`" in text
assert "`derived.sheets/02-geheim.tsv`" in text and "02-geheim.formulas" not in text
assert "Preview" in text and "Posten\tWert" in text
# Row n of the file is row n of the sheet, column k its k-th column: the sheet starts at B2.
assert _tsv_rows(folder / "derived.sheets/01-budget.tsv") == [
["", "", ""],
["", "Posten", "Wert"],
["", "Miete", "1200"],
["", "Stichtag", "2024-01-01"],
["", "Fertig", "TRUE"],
]
assert _tsv_rows(folder / "derived.sheets/01-budget.formulas.tsv") == [
["cell", "formula", "value"],
["C3", "=1200+0", "1200"],
]
assert _tsv_rows(folder / "derived.sheets/02-geheim.tsv") == [["VERSTECKT-INHALT"]]
assert "Values are as last saved" in out and "Hidden sheets" in out and ".formulas.tsv" in out
assert "derived.sheets" in out
def test_tsv_quotes_per_rfc_4180_and_reads_back_cell_for_cell(tree):
cells = ["a\tb", 'say "hi"', "two\nlines", "cr\rhere", " padded ", "plain", "="]
row = "".join(_text_cell(f"{chr(65 + i)}1", c) for i, c in enumerate(cells))
single = '<row r="1"><c r="A1" t="inlineStr"><is><t></t></is></c></row><row r="2">' + _text_cell("A2", "x") + "</row>"
_put(tree, "q.xlsx", make_book([("Quoting", "visible", f'<row r="1">{row}</row>'), ("One column", "visible", single)]))
_derive("incoming/q.xlsx")
sheets = tree / "incoming/q/derived.sheets"
assert _tsv_rows(sheets / "01-quoting.tsv") == [cells]
raw = (sheets / "01-quoting.tsv").read_bytes()
assert raw.endswith(b"\n") and b'"a\tb"' in raw and b'"say ""hi"""' in raw and b'"cr\rhere"' in raw
# A single-column sheet whose first row is empty: the empty row is still a row.
assert (sheets / "02-one-column.tsv").read_bytes() == b'""\nx\n'
only_empty_first = make_book([("S", "visible", '<row r="2">' + _text_cell("A2", "x") + "</row>")])
_put(tree, "e.xlsx", only_empty_first)
_derive("incoming/e.xlsx")
assert _tsv_rows(tree / "incoming/e/derived.sheets/01-s.tsv") == [[""], ["x"]]
assert (tree / "incoming/e/derived.sheets/01-s.tsv").read_bytes() == b'""\nx\n'
def test_shared_array_and_plain_formulas_are_listed_with_their_stored_values(tree, capsys):
master = "A2*$B$1+B2+SUM(A$1:B$1)+LOG10(A2)&amp;\"A2\"&amp;'My Sheet'!A2"
rows = (
'<row r="1"><c r="A1"><v>10</v></c><c r="B1"><v>2</v></c></row>'
'<row r="2"><c r="A2"><v>20</v></c><c r="B2"><v>3</v></c>'
f'<c r="C2"><f t="shared" ref="C2:C4" si="0">{master}</f><v>1</v></c></row>'
'<row r="3"><c r="A3"><v>30</v></c><c r="C3"><f t="shared" si="0"/><v>2</v></c></row>'
'<row r="4"><c r="C4"><f t="shared" si="0"/><v>3</v></c>'
'<c r="D4"><f t="array" ref="D4:D5">SUM(A2:A4*B2:B4)</f><v>7</v></c></row>'
)
_put(tree, "f.xlsx", make_book([("Calc", "visible", rows)]))
_derive("incoming/f.xlsx")
formulas = _tsv_rows(tree / "incoming/f/derived.sheets/01-calc.formulas.tsv")
def shifted(n: int) -> str:
return f"=A{n}*$B$1+B{n}+SUM(A$1:B$1)+LOG10(A{n})&\"A2\"&'My Sheet'!A{n}"
assert formulas == [
["cell", "formula", "value"],
["C2", shifted(2), "1"],
["C3", shifted(3), "2"],
["C4", shifted(4), "3"],
["D4", "=SUM(A2:A4*B2:B4) (array formula, range D4:D5)", "7"],
]
assert "WARN" not in capsys.readouterr().out
assert _header_and_text(tree / "incoming/f/derived.md")[0]["formulas"] == 4
def test_a_shared_formula_that_would_leave_the_sheet_is_named_by_its_master(tree, capsys):
rows = (
'<row r="5"><c r="B5"><f t="shared" ref="B5:B6" si="3">A4+1</f><v>1</v></c></row>'
'<row r="6"><c r="A6"><f t="shared" si="3"/><v>2</v></c><c r="C6"><f t="shared" si="9"/><v>3</v></c></row>'
)
_put(tree, "u.xlsx", make_book([("S", "visible", rows)]))
_derive("incoming/u.xlsx")
out = capsys.readouterr().out
formulas = _tsv_rows(tree / "incoming/u/derived.sheets/01-s.formulas.tsv")
assert formulas[1:] == [["B5", "=A4+1", "1"], ["A6", "shared:B5", "2"], ["C6", "shared:si=9", "3"]]
assert "WARN 2 shared formula cell(s) could not be written out" in out
def test_xlsx_files_are_the_same_on_a_second_run(tree):
data = make_xlsx()
_put(tree, "a.xlsx", data)
_derive("incoming/a.xlsx")
first = {k: v for k, v in _snapshot(tree / "incoming").items() if k.endswith(".tsv")}
shutil.rmtree(tree / "incoming/a")
_put(tree, "a.xlsx", data)
_derive("incoming/a.xlsx")
assert {k: v for k, v in _snapshot(tree / "incoming").items() if k.endswith(".tsv")} == first
assert first
def test_an_empty_and_a_chart_sheet_get_no_file(tree):
data = make_book([("Leer", "visible", ""), ("Daten", "visible", '<row r="1">' + _text_cell("A1", "x") + "</row>")])
_put(tree, "l.xlsx", data)
_derive("incoming/l.xlsx")
_header, text = _header_and_text(tree / "incoming/l/derived.md")
assert "*(empty sheet)*" in text
assert [p.name for p in (tree / "incoming/l/derived.sheets").iterdir()] == ["02-daten.tsv"]
def test_sheet_files_count_against_the_path_budget_and_a_refusal_leaves_incoming_alone(tree, capsys):
stem = "s" * 120
_put(tree, f"{stem}.xlsx", make_xlsx())
before = _snapshot(tree / "incoming")
_refused(f"incoming/{stem}.xlsx")
assert "--name" in capsys.readouterr().out
assert _snapshot(tree / "incoming") == before
_derive(f"incoming/{stem}.xlsx", "kurz")
assert (tree / "incoming/kurz/derived.sheets/01-budget.tsv").is_file()
def test_the_folder_form_never_overwrites_a_sheets_folder_and_changes_nothing(tree):
folder = tree / "incoming/mappe"
(folder / "zahlen.sheets").mkdir(parents=True)
(folder / "zahlen.xlsx").write_bytes(make_xlsx())
before = _snapshot(tree / "incoming")
(folder / "zahlen.sheets/x.txt").write_text("mine", encoding="utf-8")
before = _snapshot(tree / "incoming")
_refused("incoming/mappe")
assert _snapshot(tree / "incoming") == before
def test_a_workbook_in_a_zip_or_a_mail_keeps_its_sheets_beside_its_md(tree):
_put(tree, "s.zip", make_zip({"sub/Budget 2024.xlsx": make_xlsx(), "Budget 2024.sheets": b"in the way"}))
_derive("incoming/s.zip")
folder = tree / "incoming/s"
assert (folder / "sub/Budget 2024.md").is_file()
assert sorted(p.name for p in (folder / "sub/Budget 2024.sheets").iterdir()) == [
"01-budget.formulas.tsv", "01-budget.tsv", "02-geheim.tsv",
]
assert "`Budget 2024.sheets/01-budget.tsv`" in (folder / "sub/Budget 2024.md").read_text(encoding="utf-8")
_put(tree, "m.eml", make_mail([("derived.xlsx", make_xlsx(), "application/octet-stream")]))
_derive("incoming/m.eml")
mail = tree / "incoming/m"
assert (mail / "derived-2.md").is_file() and (mail / "derived-2.sheets/01-budget.tsv").is_file()
assert "`derived-2.sheets/01-budget.tsv`" in (mail / "derived-2.md").read_text(encoding="utf-8")
def test_the_derived_folder_with_its_sheet_files_is_accepted_whole(tree):
_put(tree, "zahlen.xlsx", make_xlsx())
_derive("incoming/zahlen.xlsx")
raw_accept_command(
files=[Path("incoming/zahlen")], fidelity="verbatim", authority="reporting",
page=None, replaces=None, dry_run=False,
)
accepted = sorted(p for p in (tree / "raw").rglob("*.tsv"))
assert [p.name for p in accepted] == ["01-budget.formulas.tsv", "01-budget.tsv", "02-geheim.tsv"]
assert not (tree / "incoming/zahlen").exists()
def test_xlsx_check_is_printed_only_for_the_formats_derived_in_the_run(tree, capsys):
@@ -0,0 +1,104 @@
"""`wikitool raw derive` against the real documents in `testfiles/`.
The synthetic fixtures in `binary_formats.py` and `test_raw_derive.py` prove the filters on
documents built to be easy; the files in `testfiles/` are whatever the world produced. Every file
there is derived, compared with itself on a second run and accepted into `raw/`, and an optional
`<file>.expect` next to it names the text that must come out. A distributed instance has no
`testfiles/`, so the whole module is then skipped.
"""
from __future__ import annotations
import importlib.util
import shutil
from pathlib import Path
import pytest
from chemenu import filters
from chemenu.commands.raw_cmd import raw_accept_command
from chemenu.tests.test_raw_derive import ( # noqa: F401 (tree is a fixture)
_derive,
_header_and_text,
tree,
)
TESTFILES = Path(__file__).resolve().parents[3] / "testfiles"
SIDECAR = ".expect"
pytestmark = pytest.mark.skipif(
importlib.util.find_spec("pypdf") is None or importlib.util.find_spec("olefile") is None,
reason="pypdf and olefile are needed for real PDFs, MSGs and archives holding them",
)
def _documents() -> list[Path]:
if not TESTFILES.is_dir():
return []
return sorted(
path
for path in TESTFILES.rglob("*")
if path.is_file()
and not path.name.startswith(".")
and path.name != "README.md"
and path.suffix != SIDECAR
)
def _id(path: Path) -> str:
return path.relative_to(TESTFILES).as_posix()
def _texts(folder: Path) -> dict[str, str]:
texts = {}
for md in sorted(folder.rglob("*.md")):
lines = md.read_text(encoding="utf-8").splitlines()
texts[md.relative_to(folder).as_posix()] = "\n".join(
line for line in lines if not line.startswith("derived:")
)
return texts
def _expected(document: Path) -> list[str]:
sidecar = document.with_name(document.name + SIDECAR)
if not sidecar.is_file():
return []
lines = sidecar.read_text(encoding="utf-8").splitlines()
return [line for line in lines if line.strip() and not line.startswith("#")]
def _put_and_derive(tree: Path, document: Path) -> Path:
(tree / "incoming" / document.name).write_bytes(document.read_bytes())
_derive(f"incoming/{document.name}")
return tree / "incoming" / document.stem
@pytest.mark.parametrize("document", _documents(), ids=_id)
def test_a_real_document_derives_stably_and_is_accepted(tree, document):
assert filters.filter_for(document.name) is not None, (
f"no import filter takes {document.name}; a file in testfiles/ must be one `raw derive` reads"
)
folder = _put_and_derive(tree, document)
assert not (tree / "incoming" / document.name).exists()
if document.suffix.lower() == ".zip":
assert any(folder.rglob("*.md")), "an unpacked archive gets a .md beside its documents"
else:
original = folder / f"original{document.suffix.lower()}"
assert original.read_bytes() == document.read_bytes()
header, _text = _header_and_text(folder / "derived.md")
assert header["derived_by"] == "wikitool raw derive"
assert header["original_name"] == document.name
first = _texts(folder)
for needle in _expected(document):
assert any(needle in text for text in first.values()), f"{needle!r} not in the derived text"
shutil.rmtree(folder)
assert _texts(_put_and_derive(tree, document)) == first, "the same bytes gave a different text"
raw_accept_command(
files=[Path(f"incoming/{document.stem}")], fidelity="verbatim", authority="reporting",
page=None, replaces=None, dry_run=False,
)
assert not (tree / "incoming" / document.stem).exists()