azk/chatbot_v3
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(ingestion): add DOCX/CSV/XLSX parsing and fixed-size chunking (ADR-0018)

Browse Source 
Adds src/application/ingestion/ -- Persian normalization, DOCX body
walk with structural data/layout table classification, CSV/XLSX row
rendering, and fixed-size token chunking (cl100k_base, 400/60/512) --
as pure functions per ADR-0015, tested against real production
documents (asia_data_sample, kept out of the repo). ADR-0018 records
where this diverges from ADR-0004 (fixed-size default, no invented
headings/tree, structural table classification, header-provable
labeling only). Plan 001's scope line is corrected from CSV-only to
DOCX/XLSX/CSV, and CLAUDE.md's stale project-status paragraph is
updated to match current implementation state.
This commit is contained in:
Ali Zarin Kolah
2026-08-18 10:22:17 +03:30
parent 80ed5b1577
commit 5cdfb70085
26 changed files with 2438 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
tests/support/documents.json Normal file
View File

@@ -0,0 +1,30 @@
{
"prose_docx": {
"filename": "bimeh_havades.docx",
"why": "38 paragraphs, all Normal, no tables: the plain-prose path, and the proof that no heading is invented where the document declares none."
},
"list_docx": {
"filename": "طرح ها-عمروحوادث-14050220.docx",
"why": "109 paragraphs mixing Normal and List Paragraph: prose carrying list styling."
},
"table_docx": {
"filename": "شماره تماس معاونت و مدیریتها14050309.docx",
"why": "One 35x5 contact table, no body prose. A data table with a real header row and a vertically merged first column."
},
"mixed_docx": {
"filename": "شرایط عمومی بیمه حوادث اشخاص14041205.docx",
"why": "Prose interleaved with two tables: a 30x3 list of injury compensations with NO header row, and an 11x2 table that has one."
},
"layout_docx": {
"filename": "چت بات-مهندسی.docx",
"why": "A 3x2 layout table whose single merged cell holds an entire 77k-character sub-document across 738 paragraphs and 5 nested tables."
},
"qa_xlsx": {
"filename": "fire14050319.xlsx",
"why": "Two-column Q&A sheet (q/a) plus a dead second sheet."
},
"branches_xlsx": {
"filename": "مشخصات شعب 28 اردیبهشت 1405.xlsx",
"why": "Branch directory: a merged title row above a two-row-tall merged header, and a vertically merged province column spanning each province's branches."
}
}

52
tests/support/documents.py Normal file
View File

@@ -0,0 +1,52 @@
"""Load real production documents as test fixtures.
Real files rather than generated ones: a fixture built with the same
understanding of the format that produced the parser cannot catch a wrong
mental model -- it is wrong in the same way, so the test passes and production
breaks. Every table shape this parser handles was found by reading these
files, not by reasoning about what DOCX can contain.
The documents live outside the repository so no customer-facing file enters
git history. Point `TEST_DOCUMENTS_DIR` at the directory holding them; tests
that need one skip when it is absent.
These skips cover a missing *external input*, never a failing assertion -- but
a machine without the directory does run a smaller suite than CI should.
Filenames live in `documents.json` rather than in this module: they are mostly
Persian, and Arabic letterforms in Python source are flagged as ambiguous with
Latin lookalikes (RUF001). They are data, so they belong in a data file.
"""
import json
import os
from pathlib import Path
import pytest
DOCUMENTS_DIR = Path(
os.environ.get("TEST_DOCUMENTS_DIR", "~/Documents/asia_data_sample")
).expanduser()
_MANIFEST = json.loads((Path(__file__).parent / "documents.json").read_text("utf-8"))
# Each entry is the only sample exercising some structure; `why` in the
# manifest records what would go untested without it.
PROSE_DOCX = _MANIFEST["prose_docx"]["filename"]
LIST_DOCX = _MANIFEST["list_docx"]["filename"]
TABLE_DOCX = _MANIFEST["table_docx"]["filename"]
MIXED_DOCX = _MANIFEST["mixed_docx"]["filename"]
LAYOUT_DOCX = _MANIFEST["layout_docx"]["filename"]
QA_XLSX = _MANIFEST["qa_xlsx"]["filename"]
BRANCHES_XLSX = _MANIFEST["branches_xlsx"]["filename"]
def load_document(filename: str) -> bytes:
"""Return a real document's bytes, skipping the test when it is unavailable."""
path = DOCUMENTS_DIR / filename
if not path.is_file():
pytest.skip(
f"{filename} not found in {DOCUMENTS_DIR}; "
f"set TEST_DOCUMENTS_DIR to the directory holding the sample documents"
)
return path.read_bytes()

270
tests/unit/application/test_chunking.py Normal file
View File

@@ -0,0 +1,270 @@
"""Fixed-size chunking, chunk identity, and neighbor linking (ADR-0001, ADR-0018)."""
import uuid
import pytest
from pydantic import ValidationError
from src.application.ingestion import (
ContentType,
ParsedDocument,
StructuralUnit,
chunk_document,
chunk_id_for,
count_tokens,
split_by_tokens,
)
from src.config import ChunkingSettings
pytestmark = pytest.mark.unit
ENCODING = "cl100k_base"
FILE_ID = uuid.UUID("11111111-1111-1111-1111-111111111111")
OTHER_FILE_ID = uuid.UUID("22222222-2222-2222-2222-222222222222")
@pytest.fixture
def settings() -> ChunkingSettings:
return ChunkingSettings()
def _long_text(tokens: int) -> str:
"""Text of roughly `tokens` cl100k tokens: one common word per token."""
return " ".join(["word"] * tokens)
def _rows(rows: list[str]) -> ParsedDocument:
"""A parsed spreadsheet: every row an atomic table_row unit."""
return ParsedDocument(
units=[StructuralUnit(text=row, content_type=ContentType.TABLE_ROW) for row in rows],
block_count=len(rows),
)
def _prose(text: str) -> ParsedDocument:
"""A parsed document consisting of one flowing prose block."""
return ParsedDocument(
units=[StructuralUnit(text=text, content_type=ContentType.PARAGRAPH)],
block_count=1,
)
# ── split_by_tokens ───────────────────────────────────────────────
def test_split_by_tokens_text_under_limit_returns_single_chunk() -> None:
text = _long_text(50)
assert split_by_tokens(text, chunk_size=400, overlap=60, encoding_name=ENCODING) == [text]
def test_split_by_tokens_long_text_splits_into_multiple_windows() -> None:
windows = split_by_tokens(_long_text(1000), chunk_size=400, overlap=60, encoding_name=ENCODING)
assert len(windows) > 1
assert all(count_tokens(window, ENCODING) <= 400 for window in windows)
def test_split_by_tokens_consecutive_windows_share_overlapping_tokens() -> None:
windows = split_by_tokens(
" ".join(str(number) for number in range(1000)),
chunk_size=100,
overlap=20,
encoding_name=ENCODING,
)
# The tail of one window must reappear at the head of the next, or a
# sentence cut at the boundary is unrecoverable.
tail_tokens = windows[0].split()[-5:]
assert " ".join(tail_tokens) in windows[1]
def test_split_by_tokens_overlap_at_least_chunk_size_raises() -> None:
"""Guards an infinite loop: `start = end - overlap` never advances."""
with pytest.raises(ValueError, match="smaller than chunk_size"):
split_by_tokens("text", chunk_size=100, overlap=100, encoding_name=ENCODING)
# ── chunk identity ────────────────────────────────────────────────
def test_chunk_id_for_is_deterministic_across_calls() -> None:
assert chunk_id_for(FILE_ID, 7) == chunk_id_for(FILE_ID, 7)
def test_chunk_id_for_differs_across_files_and_indices() -> None:
assert chunk_id_for(FILE_ID, 0) != chunk_id_for(OTHER_FILE_ID, 0)
assert chunk_id_for(FILE_ID, 0) != chunk_id_for(FILE_ID, 1)
def test_chunk_document_identical_content_in_different_files_gets_different_ids(
settings: ChunkingSettings,
) -> None:
"""The evaluation repo derived ids from the filename with non-ASCII stripped,
so every Farsi filename collapsed to underscores and collided."""
parsed = _rows(["a: 1"])
first = chunk_document(parsed, file_id=FILE_ID, settings=settings)
second = chunk_document(parsed, file_id=OTHER_FILE_ID, settings=settings)
assert first[0].chunk_id != second[0].chunk_id
def test_chunk_document_reingesting_same_file_yields_same_ids(
settings: ChunkingSettings,
) -> None:
"""Deterministic ids are what make a re-upload upsert instead of duplicate."""
parsed = _rows(["a: 1", "a: 2"])
first = chunk_document(parsed, file_id=FILE_ID, settings=settings)
second = chunk_document(parsed, file_id=FILE_ID, settings=settings)
assert [chunk.chunk_id for chunk in first] == [chunk.chunk_id for chunk in second]
# ── ordering and neighbor links ───────────────────────────────────
def test_chunk_document_assigns_contiguous_indices_and_one_based_order(
settings: ChunkingSettings,
) -> None:
parsed = _rows(["a: 1", "a: 2", "a: 3"])
chunks = chunk_document(parsed, file_id=FILE_ID, settings=settings)
assert [chunk.chunk_index for chunk in chunks] == [0, 1, 2]
assert [chunk.order_id for chunk in chunks] == [1.0, 2.0, 3.0]
def test_chunk_document_blank_rows_do_not_leave_index_gaps(
settings: ChunkingSettings,
) -> None:
"""An index gap breaks the previous/next chain retrieval expansion walks."""
parsed = _rows(["a: 1", " ", "a: 2"])
chunks = chunk_document(parsed, file_id=FILE_ID, settings=settings)
assert [chunk.chunk_index for chunk in chunks] == [0, 1]
def test_chunk_document_links_neighbors_with_null_at_both_ends(
settings: ChunkingSettings,
) -> None:
parsed = _rows(["a: 1", "a: 2", "a: 3"])
chunks = chunk_document(parsed, file_id=FILE_ID, settings=settings)
assert chunks[0].previous_chunk_id is None
assert chunks[-1].next_chunk_id is None
for position, chunk in enumerate(chunks[:-1]):
assert chunk.next_chunk_id == chunks[position + 1].chunk_id
assert chunks[position + 1].previous_chunk_id == chunk.chunk_id
def test_chunk_document_single_chunk_has_no_neighbors(settings: ChunkingSettings) -> None:
parsed = _rows(["a: 1"])
(chunk,) = chunk_document(parsed, file_id=FILE_ID, settings=settings)
assert chunk.previous_chunk_id is None
assert chunk.next_chunk_id is None
# ── content types and the token cap ───────────────────────────────
def test_chunk_document_spreadsheet_row_becomes_one_atomic_chunk(
settings: ChunkingSettings,
) -> None:
rows = ["branch: one\ncity: two", "branch: three\ncity: four"]
parsed = _rows(rows)
chunks = chunk_document(parsed, file_id=FILE_ID, settings=settings)
assert [chunk.content for chunk in chunks] == rows
assert all(chunk.content_type is ContentType.TABLE_ROW for chunk in chunks)
def test_chunk_document_oversized_row_splits_rather_than_truncating(
settings: ChunkingSettings,
) -> None:
"""Silent truncation is the failure mode: the model drops the tail, not raises."""
oversized = _long_text(settings.max_chunk_tokens * 2)
parsed = _rows([oversized])
chunks = chunk_document(parsed, file_id=FILE_ID, settings=settings)
assert len(chunks) > 1
assert all(chunk.token_count <= settings.max_chunk_tokens for chunk in chunks)
def test_chunk_document_markdown_produces_paragraph_chunks(
settings: ChunkingSettings,
) -> None:
parsed = _prose("# Heading\n\nBody text.")
chunks = chunk_document(parsed, file_id=FILE_ID, settings=settings)
assert all(chunk.content_type is ContentType.PARAGRAPH for chunk in chunks)
def test_chunk_document_no_chunk_exceeds_the_model_sequence_length(
settings: ChunkingSettings,
) -> None:
parsed = _prose(_long_text(5000))
chunks = chunk_document(parsed, file_id=FILE_ID, settings=settings)
assert chunks
assert all(chunk.token_count <= settings.max_chunk_tokens for chunk in chunks)
def test_chunk_document_persian_text_never_exceeds_the_cap(
settings: ChunkingSettings,
) -> None:
"""Persian is the case where the cap can actually be breached.
Slicing a token list can cut a multi-byte character in half; decoding then
re-encoding that window does not always round-trip to the same token count.
`ChunkTooLargeError` is the guard for exactly this, and it must not fire.
"""
persian_word = "".join(chr(code) for code in (0x0628, 0x06CC, 0x0645, 0x0647))
parsed = _prose(" ".join([persian_word] * 4000))
chunks = chunk_document(parsed, file_id=FILE_ID, settings=settings)
assert len(chunks) > 1
assert all(chunk.token_count <= settings.max_chunk_tokens for chunk in chunks)
def test_chunk_document_records_token_and_character_counts(
settings: ChunkingSettings,
) -> None:
parsed = _rows(["branch: one"])
(chunk,) = chunk_document(parsed, file_id=FILE_ID, settings=settings)
assert chunk.character_count == len("branch: one")
assert chunk.token_count == count_tokens("branch: one", ENCODING)
# ── settings validation ───────────────────────────────────────────
def test_chunking_settings_overlap_not_smaller_than_size_is_rejected() -> None:
with pytest.raises(ValidationError, match="must be smaller than"):
ChunkingSettings(chunk_size=100, chunk_overlap=100)
def test_chunking_settings_size_over_model_limit_is_rejected() -> None:
with pytest.raises(ValidationError, match="must not exceed"):
ChunkingSettings(chunk_size=600, max_chunk_tokens=512)
def test_chunking_settings_defaults_match_adr_0018() -> None:
settings = ChunkingSettings()
assert (settings.chunk_size, settings.chunk_overlap) == (400, 60)
assert settings.max_chunk_tokens == 512
assert settings.strategy == "fixed_size"

225
tests/unit/application/test_docx_parser.py Normal file
View File

@@ -0,0 +1,225 @@
"""DOCX parsing against the real sample corpus (ADR-0004, ADR-0018).
Every assertion here was written after reading the document it covers, so the
numbers are observations rather than predictions.
"""
import pytest
from src.application.ingestion import ContentType, DocumentParseError, parse_docx
from src.application.ingestion.docx_parser import heading_level_from_style
from src.config import ChunkingSettings
from tests.support.documents import (
LAYOUT_DOCX,
LIST_DOCX,
MIXED_DOCX,
PROSE_DOCX,
TABLE_DOCX,
load_document,
)
pytestmark = pytest.mark.unit
@pytest.fixture
def settings() -> ChunkingSettings:
return ChunkingSettings()
def _units(filename: str, settings: ChunkingSettings):
return parse_docx(load_document(filename), settings).units
def _types(units) -> set[ContentType]:
return {unit.content_type for unit in units}
# ── heading styles ────────────────────────────────────────────────
@pytest.mark.parametrize(
("style", "expected"),
[
("Heading1", 1), # the styleId Word actually stores
("Heading 1", 1), # the friendly name python-docx reports
("heading 2", 2),
("HEADING3", 3),
("Normal", None),
("List Paragraph", None),
("Heading", None),
("Headingfoo", None),
],
)
def test_heading_level_from_style_maps_word_styles_only(style: str, expected: int | None) -> None:
assert heading_level_from_style(style) == expected
def test_parse_docx_invents_no_headings_when_document_declares_none(
settings: ChunkingSettings,
) -> None:
"""Every document in the corpus lacks Heading styles, so this is the norm.
The evaluation repo upgraded paragraphs to headings by text pattern; a
wrongly-detected heading silently reshapes the tree, so ADR-0018 drops it.
"""
parsed = parse_docx(load_document(PROSE_DOCX), settings)
assert "#" not in (parsed.markdown or "")
# ── prose documents ───────────────────────────────────────────────
def test_parse_docx_prose_document_becomes_one_flowing_block(
settings: ChunkingSettings,
) -> None:
"""Consecutive paragraphs accumulate into one prose unit (ADR-0004).
One unit per paragraph would hand the splitter 36 fragments instead of
flowing text.
"""
units = _units(PROSE_DOCX, settings)
assert len(units) == 1
assert units[0].content_type is ContentType.PARAGRAPH
assert units[0].text.count("\n\n") == 35
def test_parse_docx_list_styled_paragraphs_stay_prose(settings: ChunkingSettings) -> None:
units = _units(LIST_DOCX, settings)
assert _types(units) == {ContentType.PARAGRAPH}
def test_parse_docx_normalizes_persian_letterforms(settings: ChunkingSettings) -> None:
parsed = parse_docx(load_document(PROSE_DOCX), settings)
# No Arabic kaf or yeh survives normalization.
assert chr(0x0643) not in (parsed.markdown or "")
assert chr(0x064A) not in (parsed.markdown or "")
def test_parse_docx_preserves_paragraph_breaks_in_markdown(
settings: ChunkingSettings,
) -> None:
"""Normalization collapses whitespace, so it must run per block.
Applied to the assembled document it would flatten every paragraph onto a
single line -- the easiest way to get this wrong.
"""
parsed = parse_docx(load_document(PROSE_DOCX), settings)
assert "\n\n" in (parsed.markdown or "")
# ── data tables ───────────────────────────────────────────────────
def test_parse_docx_data_table_becomes_one_unit_per_row(settings: ChunkingSettings) -> None:
"""A 35-row contact table: one header row plus 34 data rows."""
units = _units(TABLE_DOCX, settings)
assert len(units) == 34
assert _types(units) == {ContentType.TABLE_ROW}
def test_parse_docx_data_table_rows_are_labeled_with_headers(
settings: ChunkingSettings,
) -> None:
units = _units(TABLE_DOCX, settings)
assert "\n" in units[0].text
assert all(":" in line for line in units[0].text.split("\n"))
def test_parse_docx_vertically_merged_cell_repeats_on_every_row(
settings: ChunkingSettings,
) -> None:
"""The group column is merged down several rows; each row must carry it."""
units = _units(TABLE_DOCX, settings)
labels = [line.split(":")[0] for unit in units for line in unit.text.split("\n")]
group_label = units[1].text.split(":")[0]
assert labels.count(group_label) > 1
def test_parse_docx_empty_cell_is_omitted_not_rendered_as_bare_label(
settings: ChunkingSettings,
) -> None:
"""The first data row has no group; it should have one fewer field."""
units = _units(TABLE_DOCX, settings)
assert len(units[0].text.split("\n")) < len(units[1].text.split("\n"))
# ── tables without a header ───────────────────────────────────────
def test_parse_docx_headerless_table_keeps_every_row(settings: ChunkingSettings) -> None:
"""A 30-row compensation list plus an 11-row table, all rows preserved.
Regression: scanning forward for a header row discarded the rows above the
match and labeled the rest from a data row (`80: 70`).
"""
units = _units(MIXED_DOCX, settings)
rows = [unit for unit in units if unit.content_type is ContentType.TABLE_ROW]
assert len(rows) == 41
def test_parse_docx_headerless_table_rows_are_unlabeled(settings: ChunkingSettings) -> None:
"""No header is provable, so cells are joined rather than mislabeled."""
units = _units(MIXED_DOCX, settings)
first_row = next(unit for unit in units if unit.content_type is ContentType.TABLE_ROW)
assert " | " in first_row.text
assert first_row.text.startswith("1 | ")
def test_parse_docx_mixed_document_interleaves_prose_and_rows(
settings: ChunkingSettings,
) -> None:
units = _units(MIXED_DOCX, settings)
assert _types(units) == {ContentType.PARAGRAPH, ContentType.TABLE_ROW}
# ── layout tables ─────────────────────────────────────────────────
def test_parse_docx_extracts_document_held_inside_a_table_cell(
settings: ChunkingSettings,
) -> None:
"""One merged cell holds 77k characters across 738 paragraphs.
Regression: a merged cell reported once per spanned grid position was
deduplicated by `id()`, which lxml reuses across element proxies, so the
whole sub-document vanished and the file yielded 70 tokens.
"""
units = _units(LAYOUT_DOCX, settings)
assert sum(len(unit.text) for unit in units) > 70_000
def test_parse_docx_layout_table_cells_become_prose_not_rows(
settings: ChunkingSettings,
) -> None:
"""A cell that alone exceeds chunk_size holds a document, not a field."""
units = _units(LAYOUT_DOCX, settings)
assert ContentType.PARAGRAPH in _types(units)
# ── failure modes ─────────────────────────────────────────────────
def test_parse_docx_unopenable_bytes_raises_parse_error(settings: ChunkingSettings) -> None:
with pytest.raises(DocumentParseError, match="Could not open DOCX"):
parse_docx(b"not a docx at all", settings)
def test_parse_docx_legacy_doc_binary_raises_parse_error(settings: ChunkingSettings) -> None:
"""`.doc` is rejected rather than converted (ADR-0018); python-docx cannot open it."""
with pytest.raises(DocumentParseError):
parse_docx(b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1" + b"\x00" * 512, settings)

95
tests/unit/application/test_normalization.py Normal file
View File

@@ -0,0 +1,95 @@
"""Persian normalization (ADR-0018).
Characters are written as escapes rather than literals: Arabic letterforms are
visually indistinguishable in a monospace editor, and literals would render
right-to-left and visually reorder each assertion.
"""
import pytest
from src.application.ingestion import normalize_persian_text
pytestmark = pytest.mark.unit
ARABIC_KAF = chr(0x0643)
PERSIAN_KEHEH = chr(0x06A9)
ARABIC_YEH = chr(0x064A)
ALEF_MAKSURA = chr(0x0649)
PERSIAN_YEH = chr(0x06CC)
ALEF_HAMZA_ABOVE = chr(0x0623)
ALEF_HAMZA_BELOW = chr(0x0625)
ALEF = chr(0x0627)
ALEF_FINAL_FORM = chr(0xFE8E)
FATHA = chr(0x064E)
SHADDA = chr(0x0651)
TATWEEL = chr(0x0640)
SUPERSCRIPT_ALEF = chr(0x0670)
PERSIAN_DIGITS = "".join(chr(0x06F1 + offset) for offset in range(3))
ARABIC_SEMICOLON = chr(0x061B)
ARABIC_THOUSANDS_SEPARATOR = chr(0x066C)
@pytest.mark.parametrize(
("source", "expected"),
[
(ARABIC_KAF, PERSIAN_KEHEH),
(ARABIC_YEH, PERSIAN_YEH),
(ALEF_MAKSURA, PERSIAN_YEH),
(ALEF_HAMZA_ABOVE, ALEF),
(ALEF_HAMZA_BELOW, ALEF),
],
)
def test_normalize_arabic_letterform_folds_to_persian(source: str, expected: str) -> None:
assert normalize_persian_text(source) == expected
def test_normalize_mixed_keyboard_spellings_converge_to_one_form() -> None:
"""The defect this module exists for: one word, two spellings, two embeddings."""
arabic_spelling = f"{ARABIC_KAF}{ARABIC_YEH}"
persian_spelling = f"{PERSIAN_KEHEH}{PERSIAN_YEH}"
assert normalize_persian_text(arabic_spelling) == normalize_persian_text(persian_spelling)
@pytest.mark.parametrize("mark", [FATHA, SHADDA, TATWEEL, SUPERSCRIPT_ALEF])
def test_normalize_diacritic_is_removed(mark: str) -> None:
assert normalize_persian_text(f"{ALEF}{mark}{ALEF}") == f"{ALEF}{ALEF}"
def test_normalize_not_sign_becomes_space() -> None:
assert normalize_persian_text(f"{ALEF}¬{ALEF}") == f"{ALEF} {ALEF}"
def test_normalize_applies_nfkc_compatibility_forms() -> None:
# U+FE8E is the final presentation form of alef; NFKC maps it to the base
# letter, so a document pasted from a PDF renderer matches a typed one.
assert normalize_persian_text(ALEF_FINAL_FORM) == ALEF
def test_normalize_collapses_whitespace_runs_and_strips() -> None:
assert normalize_persian_text(" a \t\n b ") == "a b"
@pytest.mark.parametrize(
"preserved",
[PERSIAN_DIGITS, ARABIC_SEMICOLON, ARABIC_THOUSANDS_SEPARATOR],
)
def test_normalize_digits_and_punctuation_survive_unchanged(preserved: str) -> None:
"""ADR-0018 deliberately folds letters only.
Rewriting Persian digits to Western ones would render citations wrong to a
Persian reader, so this guards against a later 'helpful' addition.
"""
assert normalize_persian_text(preserved) == preserved
def test_normalize_already_normalized_text_is_unchanged() -> None:
normalized = normalize_persian_text(f"{ARABIC_KAF}{FATHA} {ALEF_HAMZA_ABOVE}")
assert normalize_persian_text(normalized) == normalized
def test_normalize_empty_string_returns_empty() -> None:
assert normalize_persian_text("") == ""

183
tests/unit/application/test_spreadsheet_parser.py Normal file
View File

@@ -0,0 +1,183 @@
"""CSV and XLSX parsing against the real sample corpus (ADR-0004, ADR-0018)."""
import pytest
from src.application.ingestion import ContentType, DocumentParseError, parse_csv, parse_xlsx
from tests.support.documents import BRANCHES_XLSX, QA_XLSX, load_document
pytestmark = pytest.mark.unit
def _field(text: str, label: str) -> str | None:
for line in text.split("\n"):
name, _, value = line.partition(":")
if name == label:
return value.strip()
return None
# ── the two real sheet shapes ─────────────────────────────────────
def test_parse_xlsx_qa_sheet_renders_question_and_answer_labels() -> None:
"""A two-column q/a sheet needs no special case; the generic renderer serves it."""
parsed = parse_xlsx(load_document(QA_XLSX))
assert parsed.units[0].content_type is ContentType.TABLE_ROW
assert _field(parsed.units[0].text, "q")
assert _field(parsed.units[0].text, "a")
def test_parse_xlsx_branch_sheet_renders_every_column_with_its_header() -> None:
"""The same renderer, a completely different schema, no shape detection."""
parsed = parse_xlsx(load_document(BRANCHES_XLSX))
first = parsed.units[0].text
assert len(first.split("\n")) > 5
assert all(":" in line for line in first.split("\n"))
def test_parse_xlsx_dead_second_sheet_is_skipped() -> None:
"""The q/a workbook carries an empty Sheet2 seen throughout the corpus."""
parsed = parse_xlsx(load_document(QA_XLSX))
assert parsed.block_count == 150
# ── merged cells ──────────────────────────────────────────────────
def test_parse_xlsx_merged_title_row_is_not_treated_as_header() -> None:
"""A merged banner spans the sheet; the real header sits below it."""
parsed = parse_xlsx(load_document(BRANCHES_XLSX))
assert _field(parsed.units[0].text, "استان") is not None
def test_parse_xlsx_two_row_tall_merged_header_is_not_emitted_as_data() -> None:
"""Regression: forward fill duplicates the header into the row beneath it.
Without skipping that duplicate the first chunk was the header labeling
itself (`استان: استان`).
"""
parsed = parse_xlsx(load_document(BRANCHES_XLSX))
assert _field(parsed.units[0].text, "استان") != "استان"
def test_parse_xlsx_vertically_merged_column_fills_every_row_in_its_range() -> None:
"""openpyxl stores a merged range's value only in its top-left cell.
The province is merged across each of its branches, so without forward
filling every branch but the first loses the field entirely.
"""
parsed = parse_xlsx(load_document(BRANCHES_XLSX))
provinces = [_field(unit.text, "استان") for unit in parsed.units]
assert all(province for province in provinces)
assert len(set(provinces)) == 31
def test_parse_xlsx_branch_rows_are_self_contained() -> None:
"""Each branch is its own chunk carrying its province, not just its name."""
parsed = parse_xlsx(load_document(BRANCHES_XLSX))
second = parsed.units[1].text
assert _field(second, "استان") == _field(parsed.units[0].text, "استان")
assert _field(second, "شعبه") != _field(parsed.units[0].text, "شعبه")
# ── csv ───────────────────────────────────────────────────────────
def test_parse_csv_renders_rows_with_headers() -> None:
data = b"branch,city,phone\nvali asr,tehran,021\nazadi,karaj,026\n"
parsed = parse_csv(data)
assert parsed.block_count == 2
assert _field(parsed.units[0].text, "branch") == "vali asr"
assert _field(parsed.units[1].text, "city") == "karaj"
def test_parse_csv_sniffs_semicolon_delimiter() -> None:
data = b"branch;city;phone\nvali asr;tehran;021\nazadi;karaj;026\n"
parsed = parse_csv(data)
assert _field(parsed.units[0].text, "city") == "tehran"
def test_parse_csv_unprovable_header_falls_back_to_joined_cells() -> None:
"""Short text over short text is genuinely ambiguous, so nothing is labeled.
The cost of this rule is unlabeled rows; the cost of guessing is labeling
every row from a data row, which is how a headerless compensation table
became `80: 70`. Losing a label is recoverable, mislabeling is not.
"""
parsed = parse_csv(b"branch,city\nvali asr,tehran\nazadi,karaj\n")
assert parsed.units[0].text == "branch | city"
assert parsed.block_count == 3
def test_parse_csv_empty_cells_are_omitted() -> None:
data = b"branch,city,phone\nvali asr,,021\n"
parsed = parse_csv(data)
assert _field(parsed.units[0].text, "city") is None
assert _field(parsed.units[0].text, "phone") == "021"
def test_parse_csv_blank_rows_are_skipped() -> None:
data = b"branch,city,phone\nvali asr,tehran,021\n\n,,\nazadi,karaj,026\n"
parsed = parse_csv(data)
assert parsed.block_count == 2
@pytest.mark.parametrize("encoding", ["utf-8", "utf-8-sig", "cp1256"])
def test_parse_csv_decodes_farsi_in_each_supported_encoding(encoding: str) -> None:
"""cp1256 is common for Farsi exported by older Excel."""
tehran = "تهران"
parsed = parse_csv(f"branch,city,phone\nvali asr,{tehran},021\n".encode(encoding))
assert _field(parsed.units[0].text, "city") == tehran
def test_parse_csv_normalizes_persian_letterforms() -> None:
arabic_yeh, persian_yeh = chr(0x064A), chr(0x06CC)
parsed = parse_csv(f"branch,city,phone\nvali asr,{arabic_yeh},021\n".encode())
assert _field(parsed.units[0].text, "city") == persian_yeh
# ── failure modes ─────────────────────────────────────────────────
def test_parse_csv_cp1256_fallback_always_decodes_rather_than_raising() -> None:
"""cp1256 is single-byte, so it maps every byte and never fails.
That makes the "could not decode" branch unreachable in practice: bytes
that are not valid UTF-8 come back as mojibake instead of an error. The
guard stays as defence, but this documents the real behaviour so nobody
relies on a decode failure to reject a bad upload.
"""
parsed = parse_csv(b"branch,city,phone\nvali asr,\xff\xfe,021\n")
assert parsed.block_count == 1
def test_parse_csv_empty_file_raises_parse_error() -> None:
with pytest.raises(DocumentParseError, match="no rows"):
parse_csv(b"")
def test_parse_xlsx_unopenable_bytes_raise_parse_error() -> None:
with pytest.raises(DocumentParseError, match="Could not open XLSX"):
parse_xlsx(b"not a workbook")