""" Stage 2: Parse raw OCR text blocks into structured receipt data. Uses spatial layout (bounding box coordinates) to reconstruct receipt rows, rather than relying on single-line regex matching. Receipts have a consistent column layout: - Far left: quantity (single digit) - Middle: item description - Far right: price The parser groups blocks into rows by y-coordinate proximity, classifies columns by x-position, then extracts structured fields. """ import difflib import re from datetime import date from typing import Any # Known UK retailer names (uppercase) for exact matching against header text _KNOWN_RETAILERS = { "TESCO", "ASDA", "ALDI", "SAINSBURY'S", "SAINSBURYS", "MORRISONS", "WAITROSE", "COSTCO", "ICELAND", "SPAR", "NISA", "BOOTHS", "LIDL", "CO-OP", "M&S", "BUDGENS", "LONDIS", "ONE STOP", "HOME BARGAINS", "B&M", "B&M BARGAINS", "POUNDLAND", "POUNDSTRETCHER", "FARMFOODS", "HERON FOODS", "WILKO", "SAVERS", "SUPERDRUG", "BOOTS", "COSTCUTTER", "PREMIER", "MCCOLL'S", "MCCOLLS", "BARGAIN BOOZE", } # Map of common OCR misreads for stylized logos → canonical retailer name _OCR_VARIANTS = { "LODZ": "LIDL", "LIOL": "LIDL", "LDL": "LIDL", "IIDL": "LIDL", "COOP": "CO-OP", "CO OP": "CO-OP", "OWNED BY YOU": "CO-OP", "OWNED BY YOU.": "CO-OP", "RIGHT BY YOU": "CO-OP", "RIGHT BY YOU.": "CO-OP", "M & S": "M&S", "MARKS & SPENCER": "M&S", } # --------------------------------------------------------------------------- # Compiled patterns # --------------------------------------------------------------------------- _DATE_PATTERNS = [ re.compile(r"\b(\d{4}[-/]\d{1,2}[-/]\d{1,2})\b"), re.compile(r"\b(\d{1,2}[/-]\d{1,2}[/-]\d{2,4})\b"), re.compile( r"\b(\d{1,2}\s+" r"(?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)[a-z]*" r"\s+\d{2,4})\b", re.IGNORECASE, ), ] _PRICE_EXTRACT_RE = re.compile(r"(-?)\s*[^\d\s]{0,2}\s*(\d{1,6}[.,\s]\d{2})") _TOTALS_KEYWORDS = re.compile( r"\b(total|sub[\s-]?total|subtotal|savings|promotions|tax|gst|hst|balance" r"|amount\s+due|amount\s+payable|to\s+pay)\b", re.IGNORECASE, ) _HEADER_SKIP = re.compile( r"(www\.|\.com|\.co\.uk|vat\s*(?:no\.?|number)|questions?\s*please|please\s*visit|store.locator" r"|\btel\b|\bphone\b)", re.IGNORECASE, ) # Retailer names as whole words inside a larger header block ("TESCO EXTRA", # "SAINSBURY'S LOCAL"). Custom boundaries because names contain & and '. _RETAILER_WORD_RES = { name: re.compile(rf"(? dict[str, Any]: """ Accept OCR blocks (sorted top-to-bottom) and return structured receipt data. Each block: { "text": str, "confidence": float, "bbox": [...] } """ if not blocks: return _empty_result() clean_blocks = [b for b in blocks if _is_readable(b["text"])] if not clean_blocks: return _empty_result() rows = _build_rows(clean_blocks) receipt_width = _estimate_receipt_width(clean_blocks) date = _extract_date(clean_blocks) header_end, totals_start = _find_sections(rows, receipt_width) merchant_name, store_location = _extract_header(rows[:header_end]) # Correct the price column's drift before reading items off the rows — # only over the item section, since the header and totals have their own # (centred, label-and-value) geometry that this doesn't model. item_rows = _realign_price_column(rows[header_end:totals_start], receipt_width) # Tesco receipts can split the first item's qty+description and price # into two rows when ghost text above the item pulls the row anchor up. # Detect an orphaned qty row immediately before the first-price row and # glue its blocks into the first item row. if header_end > 0 and item_rows: prev_row = rows[header_end - 1] if any(_is_qty_block(b, receipt_width) for b in prev_row): item_rows = [ sorted(prev_row + item_rows[0], key=lambda b: _left_x(b["bbox"])) ] + item_rows[1:] line_items = _extract_line_items(item_rows, receipt_width) _strip_qty_prefixes(line_items, rows) totals = _extract_totals(rows[totals_start:]) return _finalize({ "merchant_name": merchant_name, "store_location": store_location, "date": date, "line_items": line_items, "subtotal": totals.get("subtotal"), "savings": totals.get("savings"), "total": totals.get("total"), }) def is_complete_parse(result: dict[str, Any]) -> bool: """ True when a parse is internally consistent enough to trust on its own: no anomalies, and the line items (net of their discounts) sum to the printed total. An OCR pass that missed rows can't satisfy the arithmetic, so this is a safe gate for skipping a second pass. """ if result.get("anomalies"): return False try: total = float(result["total"]) items_sum = sum( float(item["total_price"]) + (float(item["discount"]) if item.get("discount") else 0.0) for item in result["line_items"] ) except (TypeError, ValueError): return False return abs(items_sum - total) < 0.015 def _empty_result() -> dict[str, Any]: return _finalize({ "merchant_name": None, "store_location": None, "date": None, "line_items": [], "subtotal": None, "savings": None, "total": None, }) def _finalize(result: dict[str, Any]) -> dict[str, Any]: """ Flag parse problems on the result itself (not just server logs), so a consumer (e.g. the mobile app) can prompt the user to double-check specific items instead of silently trusting a bad parse. Sets a per-item "needs_review" bool and a top-level "anomalies" list of {"type", "item_index"} — item_index is 1-based to match human-readable logging, or None for receipt-level anomalies. """ anomalies: list[dict[str, Any]] = [] for idx, item in enumerate(result["line_items"], 1): reasons = [] if not item.get("description"): reasons.append("no_description") if not item.get("total_price"): reasons.append("no_price") item["needs_review"] = bool(reasons) anomalies.extend({"type": reason, "item_index": idx} for reason in reasons) if not result["line_items"]: anomalies.append({"type": "no_items", "item_index": None}) if result.get("total") is None: anomalies.append({"type": "no_total", "item_index": None}) result["anomalies"] = anomalies return result # --------------------------------------------------------------------------- # Text quality filter # --------------------------------------------------------------------------- def _is_readable(text: str) -> bool: """ Filter out garbage text from receipt backs / noise. Ghost text tends to have: random case mixing, no recognizable words, high consonant density. Real receipt text has: prices, normal English words, or standard labels. """ if not text or not text.strip(): return False stripped = text.strip() # Single digits pass (could be quantity column) if len(stripped) == 1: return stripped.isdigit() if len(stripped) < 2: return False # Prices always pass if _PRICE_EXTRACT_RE.search(stripped): return True # Short tokens (2-3 chars) — allow if they look like real text if len(stripped) <= 3: # Allow "Cc", digits, common abbreviations if re.fullmatch(r"[A-Za-z]{2,3}|[0-9]+", stripped): return True return False # For longer text: check if it has a reasonable ratio of lowercase letters # and spaces (real English text). Ghost text is often CamelCase garbage # with no spaces and random character distribution. alpha = sum(1 for c in stripped if c.isalpha()) if alpha == 0: # Pure numbers/symbols — keep if it has recognizable structure return bool(re.search(r"\d", stripped)) # Check for word-like patterns (sequences of letters separated by spaces/punct) words = re.findall(r"[A-Za-z]+", stripped) if not words: return False # Ghost text signature: many words with unusual capitalization mixing # Real text: "Tesco British Whole Milk", "VAT Number", "Subtotal:" # Ghost text: "VIAJeY", "biqemoa ot vlggsanoiibno", "68T3-uoniqAoncguaGAon" # Heuristic: if average word length > 6 and mostly lowercase jumbles, it's noise avg_word_len = sum(len(w) for w in words) / len(words) has_spaces = " " in stripped # Long unbroken text with digits mixed into letters is ghost text # (e.g., "68T3-uoniqAoncguaGAon"). Pure-letter words like "Cornflakes" # or "BakedBeansTomSauce" are valid receipt items. if avg_word_len > 8 and not has_spaces and re.search(r"\d", stripped) and alpha > 5: return False # Check consonant density — ghost text has unusual consonant clusters consonants = sum(1 for c in stripped.lower() if c in "bcdfghjklmnpqrstvwxyz") vowels = sum(1 for c in stripped.lower() if c in "aeiou") if vowels > 0 and consonants / (vowels + consonants) > 0.80: return False if vowels == 0 and alpha > 3: return False return True # --------------------------------------------------------------------------- # Row building — group blocks by y-coordinate proximity # --------------------------------------------------------------------------- def _top_y(bbox: list) -> float: try: return min(pt[1] for pt in bbox) except (TypeError, IndexError): return 0.0 def _left_x(bbox: list) -> float: try: return min(pt[0] for pt in bbox) except (TypeError, IndexError): return 0.0 def _build_rows(blocks: list[dict]) -> list[list[dict]]: """Group blocks into rows by y-coordinate proximity, sorted left-to-right.""" return _group_by_top_y(blocks, _ROW_Y_TOLERANCE) def _group_by_top_y(blocks: list[dict], tolerance: float) -> list[list[dict]]: """Chain blocks into rows while each stays within `tolerance` of the row's first block, sorted left-to-right.""" if not blocks: return [] sorted_blocks = sorted(blocks, key=lambda b: _top_y(b["bbox"])) rows: list[list[dict]] = [] current_row: list[dict] = [sorted_blocks[0]] current_y = _top_y(sorted_blocks[0]["bbox"]) for block in sorted_blocks[1:]: y = _top_y(block["bbox"]) if abs(y - current_y) <= tolerance: current_row.append(block) else: rows.append(sorted(current_row, key=lambda b: _left_x(b["bbox"]))) current_row = [block] current_y = y if current_row: rows.append(sorted(current_row, key=lambda b: _left_x(b["bbox"]))) return rows def _median(values: list[float]) -> float: ordered = sorted(values) mid = len(ordered) // 2 if not ordered: return 0.0 if len(ordered) % 2: return ordered[mid] return (ordered[mid - 1] + ordered[mid]) / 2 def _row_pitch(rows: list[list[dict]]) -> float: """ Median vertical distance between printed lines. Measured over the rows as first banded, not over the description column alone: a price-only line has no description, and missing it inflates the pitch enough to make the drift gate below meaningless. """ tops = sorted(min(_top_y(b["bbox"]) for b in row) for row in rows if row) gaps = [b - a for a, b in zip(tops, tops[1:]) if b - a >= 8] if gaps: return _median(gaps) heights = [_block_height(b["bbox"]) for row in rows for b in row] return _median(heights) if heights else 0.0 def _realign_price_column( item_rows: list[list[dict]], receipt_width: float ) -> list[list[dict]]: """ Re-band the item rows after correcting the price column's vertical drift. A photographed or scanned receipt is never a flat rectangle. Curl and perspective compress one side relative to the other, so the price column's y-coordinates slide away from the description column's as you go down the page — not by a constant (that would be plain rotation) but by an amount that grows. On the Lidl receipt that exposed this, the price started 11px above its own description and ended 32px above, against a 34px line pitch. Fixed-tolerance banding cannot survive that: once the drift approaches a line pitch, prices band with the row above their description, and a description left without a price is read as a continuation line and glued onto the item before it (which is how eight product names ended up as one item priced at the first one's total). So: treat the left-hand description blocks as the anchors, walk the floating blocks (prices, qty columns) top-to-bottom tracking the drift as an EMA, and snap each one onto the description row it actually belongs to. A block whose offset is too far from the running drift to trust stays on a row of its own rather than being forced onto a description. On a well-aligned receipt the drift stays near zero and this changes nothing. The one thing tracking cannot decide for itself is where to start: an already-offset receipt is equally consistent with pairing every price one row late, and the EMA will happily follow that hypothesis all the way down. So the walk is run for a seed a line pitch either side too, and scored — see _score_assignment for what gives the off-by-one away. """ blocks = [b for row in item_rows for b in row] anchors = [b for b in blocks if _left_x(b["bbox"]) < receipt_width * _DESC_COLUMN_X] anchor_ids = {id(b) for b in anchors} floats = sorted( (b for b in blocks if id(b) not in anchor_ids), key=lambda b: _top_y(b["bbox"]), ) pitch = _row_pitch(item_rows) if not anchors or not floats or pitch <= 0: return item_rows # The global 30px tolerance is wider than the line pitch wherever the # receipt is compressed, which bands two products into one row before the # drift is even considered. Scale it to the pitch this receipt actually has. rows = _group_by_top_y(anchors, min(_ROW_Y_TOLERANCE, pitch * _ANCHOR_ROW_RATIO)) row_of_anchor = {id(a): idx for idx, row in enumerate(rows) for a in row} def nearest_anchor(block: dict, drift: float) -> tuple[dict, float]: top = _top_y(block["bbox"]) return min( ((a, _top_y(a["bbox"]) - top) for a in anchors), key=lambda pair: abs(pair[1] - drift), ) gate = pitch * _DRIFT_GATE_PITCH_RATIO def walk(seed: float) -> tuple[list[tuple[int, dict]], list[dict], float]: """Assign every floating block to a description row (or to nothing), tracking the drift from `seed` down the page. Also returns how far the offsets strayed from the tracked drift, as a tiebreak.""" drift = seed matched: list[tuple[int, dict]] = [] unmatched: list[dict] = [] residual = 0.0 for block in floats: anchor, offset = nearest_anchor(block, drift) if abs(offset - drift) > gate: # No description this can plausibly belong to — a price-only # row (a standalone discount, say). Leave it on its own row, # which is what it was before, rather than inventing a match. unmatched.append(block) continue residual += abs(offset - drift) drift = (1 - _DRIFT_ALPHA) * drift + _DRIFT_ALPHA * offset matched.append((row_of_anchor[id(anchor)], block)) return matched, unmatched, residual # Nearest-anchor seed, plus the two off-by-one hypotheses around it. On a # receipt whose price column starts out a whole line from its descriptions # the nearest anchor is the wrong one, and nothing further down the page # will contradict it — the tracking only stays self-consistent. base = _median([nearest_anchor(f, 0.0)[1] for f in floats[:3]]) matched, unmatched, _ = min( (walk(base + offset) for offset in (0.0, -pitch, pitch)), key=lambda result: _score_assignment(result, len(rows), pitch), ) for row_index, block in matched: rows[row_index].append(block) return sorted( ( sorted(row, key=lambda b: _left_x(b["bbox"])) for row in rows + [[b] for b in unmatched] ), key=lambda row: min(_top_y(b["bbox"]) for b in row), ) def _score_assignment( result: tuple[list[tuple[int, dict]], list[dict], float], row_count: int, pitch: float, ) -> float: """ Cost of one drift hypothesis — lower is better. An off-by-one is internally consistent everywhere except at the two ends of the item list, which is the only place it can be caught: pairing every price with the row below its own leaves the first description with nothing on it and the last price with no row left to land on. So charge a line pitch for each unplaced price and for each leading description row that ends up bare, and use the residuals only to separate otherwise equal hypotheses. """ matched, unmatched, residual = result filled = {row_index for row_index, _ in matched} leading_bare = 0 for index in range(row_count): if index in filled: break leading_bare += 1 return (len(unmatched) + leading_bare) * pitch + residual # --------------------------------------------------------------------------- # Receipt width estimation and column classification # --------------------------------------------------------------------------- def _estimate_receipt_width(blocks: list[dict]) -> float: """Estimate the receipt width from the rightmost x-coordinate.""" max_x = 0.0 for block in blocks: for pt in block["bbox"]: try: max_x = max(max_x, float(pt[0])) except (TypeError, IndexError): pass return max_x if max_x > 0 else 1000.0 def _is_price_block(block: dict, receipt_width: float) -> bool: """A price block sits in the right column and contains a price pattern.""" text = block["text"] if _PER_UNIT_PRICE_RE.search(text): return False x = _left_x(block["bbox"]) return x > receipt_width * 0.70 and bool(_PRICE_EXTRACT_RE.search(text)) def _is_qty_block(block: dict, receipt_width: float) -> bool: """A quantity block sits in the left column and is a 1-2 digit count. Two digits covers bulk buys ("12"); three or more is a code, not a qty.""" x = _left_x(block["bbox"]) return x < receipt_width * 0.12 and re.fullmatch(r"\d{1,2}", block["text"].strip()) is not None def _row_has_right_price(row: list[dict], receipt_width: float) -> bool: """Check if any block in the row is a price in the right column.""" return any(_is_price_block(b, receipt_width) for b in row) def _is_price_only_row(row: list[dict], receipt_width: float) -> bool: """A row whose only content is a right-column price, with no description.""" if not _row_has_right_price(row, receipt_width): return False return all( _is_price_block(b, receipt_width) or not b["text"].strip() for b in row ) # --------------------------------------------------------------------------- # Section detection # --------------------------------------------------------------------------- def _find_sections(rows: list[list[dict]], receipt_width: float) -> tuple[int, int]: """ Find where the header ends and the totals section begins. Returns (header_end_index, totals_start_index). """ header_end = 0 totals_start = len(rows) # Header ends at the first row with a price in the right column for i, row in enumerate(rows): if _row_has_right_price(row, receipt_width): header_end = i break # Primary: keyword detection ("Subtotal:", "TOTAL:", etc.) for i in range(header_end, len(rows)): row_text = " ".join(b["text"] for b in rows[i]) if _TOTALS_KEYWORDS.search(row_text): totals_start = i break # Lidl prints the total amount on its own line above the "TOTAL" label. # That lone price has no description, so without pulling the boundary back # it becomes a phantom descriptionless line item. Absorb any price-only # rows directly preceding the keyword row into the totals section. while totals_start > header_end and _is_price_only_row( rows[totals_start - 1], receipt_width ): totals_start -= 1 # Fallback: if no keyword found, look for a vertical gap significantly # larger than normal item spacing. This handles receipts where OCR # misses the "TOTAL" text (e.g., Lidl's dashed separator). if totals_start == len(rows) and len(rows) > header_end + 2: spacings = [] for i in range(header_end + 1, len(rows)): prev_y = max(_top_y(b["bbox"]) for b in rows[i - 1]) curr_y = min(_top_y(b["bbox"]) for b in rows[i]) spacings.append(curr_y - prev_y) if spacings: avg_spacing = sum(spacings) / len(spacings) gap_threshold = max(avg_spacing * 1.8, 60) for i, spacing in enumerate(spacings): if spacing > gap_threshold: totals_start = header_end + 1 + i break return header_end, totals_start # --------------------------------------------------------------------------- # Header extraction # --------------------------------------------------------------------------- def _extract_header(header_rows: list[list[dict]]) -> tuple[str | None, str | None]: """ Extract merchant name and store location from header rows. Matching stages: known retailer list (exact block / OCR-variant map / whole word inside a larger block), fuzzy match against the list for one-glyph misreads, then fallback to the most prominent header text so receipts from shops outside the list still carry a usable merchant. """ retailer, retailer_y = _match_known_retailer(header_rows) guessed = False if not retailer: retailer, retailer_y = _guess_merchant(header_rows) guessed = True if not retailer: return None, None # Only consider rows below the retailer logo. Thermal-printed receipts # can curl and reveal the back's ghost text above the logo; that text # often passes the readability filter and must be excluded here. # A guessed merchant carries no logo anchor, so only look at the rows # directly beneath it — scanning further would pick up item names. rows_below = [ row for row in header_rows if min(_top_y(b["bbox"]) for b in row) > retailer_y ] if guessed: rows_below = rows_below[:2] store_location = None for row in rows_below: meaningful = [b for b in row if len(b["text"].strip()) >= 3 and not _HEADER_SKIP.search(b["text"]) and not any(p.search(b["text"]) for p in _DATE_PATTERNS) and b["text"].strip().upper() != retailer and b["text"].strip().upper() not in _OCR_VARIANTS and b["confidence"] >= 0.85] if not meaningful: continue best = max(meaningful, key=lambda b: b["confidence"]) store_location = best["text"].strip() break return retailer, store_location def _match_known_retailer(header_rows: list[list[dict]]) -> tuple[str | None, float]: """ Check if any header block matches a known retailer or OCR variant. Returns (retailer_name, y_coordinate_of_match) or (None, 0.0). """ for row in header_rows: for block in row: text = block["text"].strip().upper() if text in _KNOWN_RETAILERS: return text, _top_y(block["bbox"]) if text in _OCR_VARIANTS: return _OCR_VARIANTS[text], _top_y(block["bbox"]) for name, word_re in _RETAILER_WORD_RES.items(): if word_re.search(text): return name, _top_y(block["bbox"]) # Fuzzy stage: a single misread glyph ("TESC0", "MORRISON5") defeats # exact matching. Same length ±1 and ratio ≥ 0.8 keeps ordinary header # words ("SAVE" vs "SAVERS") from matching. for row in header_rows: for block in row: text = block["text"].strip().upper() if len(text) < 4: continue for name in _KNOWN_RETAILERS: if len(name) < 4 or abs(len(name) - len(text)) > 1: continue if difflib.SequenceMatcher(None, text, name).ratio() >= 0.8: return name, _top_y(block["bbox"]) return None, 0.0 def _block_height(bbox: list) -> float: try: ys = [pt[1] for pt in bbox] return max(ys) - min(ys) except (TypeError, IndexError): return 0.0 def _guess_merchant(header_rows: list[list[dict]]) -> tuple[str | None, float]: """ Best-effort merchant for shops outside the known list: the most prominent (tallest) plausible text near the top of the receipt. Only the first few header rows are considered — lower rows are addresses, VAT numbers, or items whose price failed to OCR. """ candidates: list[dict] = [] for row in header_rows[:5]: for b in row: text = b["text"].strip() letters = sum(c.isalpha() for c in text) digits = sum(c.isdigit() for c in text) if (len(text) < 4 or b["confidence"] < 0.90 or letters < 3 or digits > letters or _HEADER_SKIP.search(text) or any(p.search(text) for p in _DATE_PATTERNS) or _PRICE_EXTRACT_RE.search(text)): continue candidates.append(b) if not candidates: return None, 0.0 best = max(candidates, key=lambda b: _block_height(b["bbox"])) return best["text"].strip(), _top_y(best["bbox"]) # --------------------------------------------------------------------------- # Date extraction # --------------------------------------------------------------------------- _MONTH_ABBREV = { "jan": 1, "feb": 2, "mar": 3, "apr": 4, "may": 5, "jun": 6, "jul": 7, "aug": 8, "sep": 9, "oct": 10, "nov": 11, "dec": 12, } def _extract_date(blocks: list[dict]) -> str | None: """Scan all blocks for the first date match, normalised to ISO 8601.""" for block in blocks: for pattern in _DATE_PATTERNS: m = pattern.search(block["text"]) if m: return _normalise_date(m.group(1)) return None def _normalise_date(raw: str) -> str: """ Convert a matched date string to ISO 8601 (YYYY-MM-DD). UK receipts print day-first, so ambiguous numeric dates are read as DD/MM. When the parts don't form a valid calendar date the raw string is returned unchanged — a displayable-but-unsorted date beats losing it. """ m = re.fullmatch(r"(\d{4})[-/](\d{1,2})[-/](\d{1,2})", raw) if m: return _iso_or_raw(raw, int(m.group(1)), int(m.group(2)), int(m.group(3))) m = re.fullmatch(r"(\d{1,2})[/-](\d{1,2})[/-](\d{2,4})", raw) if m: d, mo, y = int(m.group(1)), int(m.group(2)), int(m.group(3)) if y < 100: y += 2000 # Day-first assumed; an unambiguous US-style print (month first) # is recoverable when the "month" slot exceeds 12. if mo > 12 and d <= 12: d, mo = mo, d return _iso_or_raw(raw, y, mo, d) m = re.fullmatch(r"(\d{1,2})\s+([A-Za-z]+)\s+(\d{2,4})", raw) if m: mo = _MONTH_ABBREV.get(m.group(2)[:3].lower()) y = int(m.group(3)) if y < 100: y += 2000 if mo: return _iso_or_raw(raw, y, mo, int(m.group(1))) return raw def _iso_or_raw(raw: str, year: int, month: int, day: int) -> str: try: return date(year, month, day).isoformat() except ValueError: return raw # --------------------------------------------------------------------------- # Line item extraction (price-anchored) # --------------------------------------------------------------------------- def _extract_line_items( item_rows: list[list[dict]], receipt_width: float ) -> list[dict[str, Any]]: """ Row-based line-item extraction. Walk the rows produced by _build_rows top-to-bottom: - A row containing a positive right-column price starts a new item using the non-price blocks in that row as its initial description. - A row without a price is a continuation — append its text to the current item. - A row containing a negative price attaches it as a discount to the current item (first discount wins). """ items: list[dict[str, Any]] = [] current: dict[str, Any] | None = None for row in item_rows: price_blk = next( (b for b in reversed(row) if _is_price_block(b, receipt_width)), None, ) desc_blocks = [b for b in row if b is not price_blk] if price_blk is None: if current is not None: _append_desc(current, desc_blocks, receipt_width) continue price_str = _normalise_price(price_blk["text"]) if ( not price_str.startswith("-") and current is not None and current["discount"] is None and any(_is_cc_discount_indicator(b["text"]) for b in desc_blocks) ): price_str = f"-{price_str}" if price_str.startswith("-"): if current is not None: _append_desc(current, desc_blocks, receipt_width) if current["discount"] is None: current["discount"] = price_str continue # No preceding item — standalone negative price (refund/return) current = { "description": None, "quantity": 1, "unit_price": price_str, "total_price": price_str, "discount": None, } _append_desc(current, desc_blocks, receipt_width) items.append(current) return items def _is_desc_block(block: dict, receipt_width: float) -> bool: """ Description blocks start before the price column (70% mark). Exclude blocks in the 65-70% zone that are short fragments — these are typically standalone ghost text from the receipt back. """ x = _left_x(block["bbox"]) return x < receipt_width * 0.65 def _append_desc(item: dict, desc_blocks: list[dict], receipt_width: float) -> None: """Merge extra desc blocks into an item, updating qty and unit_price.""" parts: list[str] = [item["description"]] if item["description"] else [] for b in desc_blocks: text = b["text"].strip() if _is_qty_block(b, receipt_width): item["quantity"] = int(text) continue if _DISCOUNT_PREFIX_RE.match(text): continue if _EACH_RE.search(text): continue iq = _INLINE_QTY_RE.match(text) if iq: item["quantity"] = int(iq.group(1)) continue if not _is_desc_block(b, receipt_width): continue iq_suffix = _INLINE_QTY_SUFFIX_RE.match(text) if iq_suffix: item["quantity"] = int(iq_suffix.group(2)) text = _TRAILING_CODE_RE.sub("", iq_suffix.group(1).strip()) parts.append(text) item["description"] = " ".join(parts).strip() or None item["unit_price"] = _calc_unit_price(item["total_price"], item["quantity"]) def _strip_qty_prefixes(items: list[dict[str, Any]], rows: list[list[dict]]) -> None: """ Pull a leading qty out of descriptions on qty-column receipts. Strips only when the receipt prints a qty column header ("Qty Item"), or when every item (3+) carries the prefix — a lone "2 Pint Milk" on an ordinary receipt must keep its name intact. """ if not items: return matches = [_QTY_PREFIX_RE.match(item["description"] or "") for item in items] has_qty_header = any( _QTY_HEADER_RE.match(b["text"]) for row in rows for b in row ) all_prefixed = len(items) >= 3 and all(matches) if not (has_qty_header or all_prefixed): return for item, m in zip(items, matches): if not m: continue if item["quantity"] == 1: item["quantity"] = int(m.group(1)) item["description"] = m.group(2).strip() item["unit_price"] = _calc_unit_price(item["total_price"], item["quantity"]) # --------------------------------------------------------------------------- # Totals extraction # --------------------------------------------------------------------------- def _extract_totals(totals_rows: list[list[dict]]) -> dict[str, str | None]: """Extract subtotal, savings, and total from the totals section.""" result: dict[str, str | None] = {"subtotal": None, "savings": None, "total": None} # A keyword-less positive price row directly above a label row: on skewed # photos the price column shifts up a line, so a label's true amount sits # in the row before it while the label row grabs the next line's value. prev_free_price: str | None = None total_from_keyword = False for ri, row in enumerate(totals_rows): row_text = " ".join(b["text"] for b in row).strip().lower() # Find the price — prefer rightmost block price = None for block in sorted(row, key=lambda b: _left_x(b["bbox"]), reverse=True): m = _PRICE_EXTRACT_RE.search(block["text"]) if m: price = _normalise_price(block["text"]) break if price is None: continue # Skip payment rows (CASH, CHANGE, CARD, etc.) if _PAYMENT_SKIP.search(row_text): prev_free_price = None continue # For savings/promotions, prefer the negative price if available. # Sometimes OCR splits "Savings: -£6.70" into two rows. if _SAVINGS_RE.search(row_text): if not price.startswith("-"): # Check the next row for a standalone negative price if ri + 1 < len(totals_rows): next_row = totals_rows[ri + 1] for nb in sorted(next_row, key=lambda b: _left_x(b["bbox"]), reverse=True): nm = _PRICE_EXTRACT_RE.search(nb["text"]) if nm: np_ = _normalise_price(nb["text"]) if np_.startswith("-"): price = np_ break result["savings"] = result["savings"] or price prev_free_price = None elif _SUBTOTAL_RE.search(row_text): # A subtotal is never negative; a negative value means the label # row paired with the next line's savings, so recover the amount # from the keyword-less row above. if price.startswith("-") and prev_free_price is not None: price = prev_free_price result["subtotal"] = result["subtotal"] or price prev_free_price = None elif _TOTAL_RE.search(row_text) and "card" not in row_text: if price.startswith("-") and prev_free_price is not None: price = prev_free_price # A labelled total beats one inferred from a standalone price row, # which may have been the subtotal amount shifted onto its own line. if not total_from_keyword: result["total"] = price total_from_keyword = True prev_free_price = None else: if result["total"] is None and not price.startswith("-"): # Standalone positive price with no keyword — treat as total if # not yet set (handles receipts where OCR misses the "TOTAL" text). # Negative standalone prices are savings/discounts, not totals. result["total"] = price prev_free_price = price if not price.startswith("-") else None return result # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _normalise_price(text: str) -> str: """Extract and standardise a price string.""" m = _PRICE_EXTRACT_RE.search(text) if not m: return text.strip() sign = m.group(1) digits = m.group(2).replace(",", ".").replace(" ", ".") cleaned = re.sub(r"[£$€¥\s]", "", digits) return f"{sign}{cleaned}" def _is_cc_discount_indicator(text: str) -> bool: """True when text is a Clubcard discount marker: 'Cc' followed by a price and no other meaningful words. Distinguishes 'Cc £2.25' (discount) from 'Cc Any 3 For 2' (promotion label).""" text = text.strip() if not _DISCOUNT_PREFIX_RE.match(text): return False after_cc = text[2:].strip() if not re.search(r"\d", after_cc): return False cleaned = _PRICE_EXTRACT_RE.sub("", after_cc) cleaned = re.sub(r"[£$€¥\d.,\s]", "", cleaned) return len(cleaned) <= 2 def _calc_unit_price(total_price: str, quantity: int) -> str: """Calculate unit price from total and quantity.""" if quantity <= 1: return total_price try: return str(round(float(total_price) / quantity, 2)) except (ValueError, ZeroDivisionError): return total_price