"""Checks for independent OCR line persistence and item association.""" import sys import unittest from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parents[3] sys.path.insert(0, str(PROJECT_ROOT)) from apps.wp_invoices.services.ocr_layout import _parse_tsv, associate_items, associate_monetary_values, classify_lines from apps.wp_invoices.services.pipeline import (_apply_monetary_recovery_patches, _apply_row_completeness, _materialize_unrepresented_primary_rows, _monetary_patch_acceptance) TSV = """level\tpage_num\tblock_num\tpar_num\tline_num\tword_num\tleft\ttop\twidth\theight\tconf\ttext 5\t1\t1\t1\t1\t1\t10\t20\t80\t20\t95\tTuna 5\t1\t1\t1\t1\t2\t100\t20\t60\t20\t94\t1.09 5\t1\t1\t1\t2\t1\t10\t50\t100\t20\t92\tCapsicum 5\t1\t1\t1\t2\t2\t120\t50\t60\t20\t93\t1.96 """ class OcrLayoutChecks(unittest.TestCase): def test_layout_lines_keep_coordinates_confidence_and_candidates(self): lines = _parse_tsv(TSV, 200, 100) self.assertEqual("ocr-line-1", lines[0]["source_line_id"]) self.assertEqual(["1.09"], lines[0]["monetary_candidates"]) self.assertEqual("tesseract_observed", lines[0]["bounding_region_origin"]) def test_unconsumed_visual_line_is_explicit(self): lines = _parse_tsv(TSV, 200, 100) association = associate_items( {"item_line_candidates": lines}, [{"description": "Tuna", "source_text": "Tuna 1.09"}], ) self.assertEqual("ocr-line-2", association["unconsumed_source_lines"][0]["source_line_id"]) self.assertIsNone(association["unconsumed_source_lines"][0]["consumed_by_item_id"]) def test_product_code_tax_and_weight_rows_are_classified_without_deduplication(self): source = [ {"source_line_id": "line-1", "raw_text": "59496 Tuna Flav 95g 1.09 A", "monetary_candidates": ["1.09"]}, {"source_line_id": "line-2", "raw_text": "59496 Tuna Flav 95g 1.09 A", "monetary_candidates": ["1.09"]}, {"source_line_id": "line-3", "raw_text": "76384 Grape White per kg 4.11 A", "monetary_candidates": ["4.11"]}, {"source_line_id": "line-4", "raw_text": "0.824kg Net @ 4.99 $/kg", "monetary_candidates": ["4.99"]}, ] lines = classify_lines(source) self.assertEqual(["59496", "59496"], [lines[0]["product_code"], lines[1]["product_code"]]) self.assertEqual(["A", "A"], [lines[0]["tax_code"], lines[1]["tax_code"]]) self.assertEqual("item_weight_continuation", lines[3]["classification"]) self.assertEqual("line-3", lines[3]["attached_to_source_line_id"]) def test_code_bearing_row_without_money_remains_incomplete_primary(self): line = classify_lines([{ "source_line_id": "ocr-line-22", "raw_text": "713738 CrushedGarlic 250g A", "monetary_candidates": [], }])[0] self.assertEqual("item_primary", line["classification"]) self.assertFalse(line["complete"]) self.assertIn("line_total", line["missing_fields"]) item = {"description": "Crushed Garlic", "line_total": 1.29} _apply_row_completeness(item, line) self.assertFalse(item["complete"]) self.assertEqual(["line_total"], item["missing_fields"]) def test_code_bearing_row_without_tax_remains_incomplete_primary(self): line = classify_lines([{ "source_line_id": "line-1", "raw_text": "713738 CrushedGarlic 250g 1.29", "monetary_candidates": ["1.29"], }])[0] self.assertEqual("item_primary", line["classification"]) self.assertFalse(line["complete"]) self.assertIn("tax_code", line["missing_fields"]) def test_weight_continuation_never_becomes_primary(self): line = classify_lines([{ "source_line_id": "line-1", "raw_text": "0.824kg Net @ 4.99 $/kg", "monetary_candidates": ["4.99"], }])[0] self.assertEqual("item_weight_continuation", line["classification"]) def test_repeated_product_codes_are_distinct_source_items(self): lines = classify_lines([ {"source_line_id": "line-1", "raw_text": "59496 Tuna 1.09 A", "monetary_candidates": ["1.09"]}, {"source_line_id": "line-2", "raw_text": "59496 Tuna 1.09 A", "monetary_candidates": ["1.09"]}, ]) self.assertEqual(2, len([line for line in lines if line["classification"] == "item_primary"])) self.assertEqual(["line-1", "line-2"], [line["source_line_id"] for line in lines]) def test_25_primary_rows_plus_two_weight_continuations_make_25_items(self): source = [] for index in range(25): source.append({"source_line_id": f"line-{index + 1}", "raw_text": f"{700000 + index} Product {index + 1} 1.00 A", "monetary_candidates": ["1.00"], "confidence": 0.9, "estimated_bounding_region": [0, index / 30, 1, (index + 1) / 30]}) if index in (5, 11): source.append({"source_line_id": f"weight-{index}", "raw_text": "0.824kg Net @ 4.99 $/kg", "monetary_candidates": ["4.99"], "confidence": 0.9, "estimated_bounding_region": [0, 0, 1, 1]}) classified = classify_lines(source) layout = {"item_primary_lines": [line for line in classified if line["classification"] == "item_primary"], "item_weight_continuations": [line for line in classified if line["classification"] == "item_weight_continuation"]} result = {"items": []} _materialize_unrepresented_primary_rows(result, layout) self.assertEqual(25, len(result["items"])) def test_monetary_column_uses_only_tokens_inside_each_row(self): lines = classify_lines([ {"source_line_id": "line-1", "raw_text": "700001 First 3.49 A", "monetary_candidates": ["3.49"], "monetary_tokens": [{"raw": "3.49", "bbox": [0.8, 0.1, 0.9, 0.2], "center_x": 0.85}], "confidence": 0.9, "estimated_bounding_region": [0, 0.1, 1, 0.2]}, {"source_line_id": "line-2", "raw_text": "700002 Second 5.99 A", "monetary_candidates": ["5.99"], "monetary_tokens": [{"raw": "5.99", "bbox": [0.8, 0.2, 0.9, 0.3], "center_x": 0.85}], "confidence": 0.9, "estimated_bounding_region": [0, 0.2, 1, 0.3]}, ]) monetary = associate_monetary_values(lines) self.assertEqual(0.85, monetary["line_total_column_x"]) self.assertEqual("3.49", lines[0]["selected_line_total_token"]["raw"]) self.assertEqual("5.99", lines[1]["selected_line_total_token"]["raw"]) def test_weight_continuation_rate_is_attached_not_added_as_line_total(self): lines = classify_lines([ {"source_line_id": "line-1", "raw_text": "76384 Grapes per kg 4.11 A", "monetary_candidates": ["4.11"], "monetary_tokens": [{"raw": "4.11", "bbox": [0.8, 0.1, 0.9, 0.2], "center_x": 0.85}], "confidence": 0.9}, {"source_line_id": "line-2", "raw_text": "0.824kg Net @ 4.99 $/kg", "monetary_candidates": ["4.99"], "monetary_tokens": [{"raw": "4.99", "bbox": [0.6, 0.2, 0.7, 0.3], "center_x": 0.65}], "confidence": 0.9}, ]) associate_monetary_values(lines) self.assertEqual(4.99, float(lines[0]["weight_continuation"]["selected_unit_rate_token"]["raw"])) self.assertIsNone(lines[1].get("selected_line_total_token")) def test_recovery_cannot_replace_confirmed_ocr_total(self): primary = {"items": [{"source_line": "ocr-line-1", "line_total": 3.49, "evidence": {"line_total": {"status": "confirmed"}}}]} candidate, operations = _apply_monetary_recovery_patches( primary, [{"operation": "replace_conflicting_line_total", "source_line_id": "ocr-line-1", "line_total": 9.99}], {"item_primary_lines": []}) self.assertEqual(3.49, candidate["items"][0]["line_total"]) self.assertEqual("entire_response_outside_immutable_target_set", operations[0]["reason"]) def test_patch_acceptance_requires_reduced_delta_and_stable_identity(self): primary = {"subtotal": 10, "items": [{"source_line": "ocr-line-1", "line_total": 12, "evidence": {"line_total": {"status": "conflicting"}}}], "reconciliation": {"items_sum": 12}} candidate = {"subtotal": 10, "items": [{"source_line": "ocr-line-1", "line_total": 10, "evidence": {"line_total": {"status": "observed_unverified"}}}], "reconciliation": {"items_sum": 10}} accepted, _ = _monetary_patch_acceptance(primary, candidate, [{"accepted": True}]) self.assertTrue(accepted) def test_capsicum_rate_is_exclusive_and_milk_uses_own_449(self): lines = classify_lines([ {"source_line_id": "ocr-line-18", "raw_text": "77111 Capsicum per kg 1.96 A", "monetary_candidates": ["1.96"], "monetary_tokens": [{"raw": "1.96", "bbox": [.8,.10,.9,.12], "center_x": .85, "center_y": .11}], "confidence": .9, "estimated_bounding_region": [0,.10,1,.12]}, {"source_line_id": "ocr-line-19", "raw_text": "0.207kg Net @ 9.49 $/kg", "monetary_candidates": ["9.49"], "monetary_tokens": [{"raw": "9.49", "bbox": [.7,.19,.8,.21], "center_x": .75, "center_y": .20}], "confidence": .9, "estimated_bounding_region": [0,.19,1,.21]}, {"source_line_id": "ocr-line-20", "raw_text": "9372 Milk Full Cream 3L 9.49 4.49 A", "monetary_candidates": ["9.49","4.49"], "monetary_tokens": [{"raw": "9.49", "bbox": [.7,.19,.8,.21], "center_x": .75, "center_y": .20}, {"raw": "4.49", "bbox": [.8,.29,.9,.31], "center_x": .85, "center_y": .30}], "confidence": .9, "estimated_bounding_region": [0,.29,1,.31]}, ]) associate_monetary_values(lines) self.assertEqual("ocr-line-19", lines[2]["monetary_tokens"][0]["owner_source_line_id"]) self.assertEqual("4.49", lines[2]["selected_line_total_token"]["raw"]) self.assertTrue(lines[1]["selected_unit_rate_token"]["exclusive_to_weight_continuation"]) def test_missing_crushed_garlic_candidate_is_noncanonical_and_contributes_zero(self): from apps.wp_invoices.services.pipeline import _apply_layout_and_refresh item = {"description": "Crushed Garlic", "product_code": "713738", "line_total": .95, "quantity": 1, "unit_price": .95, "tax_code": "A", "evidence": {"line_total": {"status": "observed_unverified"}}} line = {"source_line_id": "ocr-line-22", "raw_text": "713738 CrushedGarlic 250g A", "product_code": "713738", "tax_code": "A", "classification": "item_primary", "complete": False, "missing_fields": ["line_total"], "selected_line_total_token": None, "monetary_candidates": [], "confidence": .9, "estimated_bounding_region": [0,.3,1,.31]} result = {"items": [item], "adjustments": [], "tax_summary": {"lines": []}, "subtotal": 1.19, "total": 1.19, "supplier": {"name": "ALDI"}, "invoice_date": "2024-04-06", "tax": None, "tax_inclusive": True, "observed_item_line_count": 1} _apply_layout_and_refresh(result, {"item_primary_lines": [line]}) self.assertIsNone(item["line_total"]) self.assertEqual(0.0, item["subtotal_contribution"]) self.assertEqual(.95, item["evidence"]["line_total"]["semantic_candidate"]) self.assertFalse(item["complete"]) def test_recovery_rejects_non_target_and_unsupported_crop_value(self): primary = {"items": [{"source_line": "ocr-line-22", "line_total": None, "evidence": {"line_total": {"status": "missing"}}}]} outside, operations = _apply_monetary_recovery_patches( primary, [{"operation": "fill_missing_line_total", "source_line_id": "ocr-line-28", "line_total": 2.99}], {"item_primary_lines": []}, allowed_targets={"ocr-line-22": "fill_missing_line_total"}, crop_candidates={"ocr-line-22": [1.19]}) self.assertIsNone(outside["items"][0]["line_total"]) self.assertEqual("entire_response_outside_immutable_target_set", operations[0]["reason"]) unsupported, operations = _apply_monetary_recovery_patches( primary, [{"operation": "fill_missing_line_total", "source_line_id": "ocr-line-22", "line_total": 1.09}], {"item_primary_lines": [{"source_line_id": "ocr-line-22"}]}, allowed_targets={"ocr-line-22": "fill_missing_line_total"}, crop_candidates={"ocr-line-22": [1.19]}) self.assertIsNone(unsupported["items"][0]["line_total"]) self.assertEqual("value_not_supported_by_isolated_crop", operations[0]["reason"]) def test_expected_fixture_primary_sum_before_recovery_is_8827(self): values = [3.49,5.99,7.99,2.19,7.49,3.49,4.11,2.49,3.99,3.59,1.96,4.49, .95,None,2.29,1.09,1.09,1.09,2.99,.95,1.39,10.69,5.99,4.99,3.49] self.assertEqual(88.27, round(sum(value or 0 for value in values), 2)) if __name__ == "__main__": unittest.main(verbosity=2)