"""Enterprise-grade amount / price parser (pure stdlib, zero dependencies). Extracts a monetary amount from noisy or distorted text fields: OCR output, scraped HTML, invoice lines, emails, database columns, etc. Design goals ------------ * **Pure and dependency-free** — stdlib only (``re``, ``decimal``, ``bisect``, ``dataclasses``). No external AI or parsing libraries. * **Deterministic and explainable** — every candidate carries a confidence in [0, 1] so callers can set their own acceptance threshold. * **Locale-aware separators**: - US: ``1,234.56`` - EU: ``22,90 €``, ``1.234,56``, ``1.234.567,89`` - Indian: ``1,50,087.99`` (lakh grouping), ``₹5 Cr`` - Space: ``15 130 Р``, ``75 990,00 Kč`` * **Distortion tolerance** (OCR / data-entry noise): - Junk between currency and digits: ``$iom89.00`` -> ``89.00`` - Stray letters inside a number: ``8a9.00`` -> ``89.00``, ``1i0,000`` -> ``10000`` * **Negatives**: ``-1,234.56``, ``1,234.56-``, unicode minus ``−89.00``, accounting parentheses ``(78,000)``. * **Magnitude suffixes**: ``K``/``k`` (thousand), ``M`` (million), ``B``, ``T``, ``Cr``/``crore`` (10M), ``L``/``lakh``/``lac`` (100k). * **Currency detection**: symbols (``$ € £ ¥ ₹ ...``), ISO codes (``USD``, ``INR``), short forms (``US$``, ``R$``, ``Rp``, ``Rs``, ``Kč`` ...) and words (``dollars``, ``rupees`` ...), on either side of the number. * **Anti-pattern rejection** — dates, times, phone numbers, IPs, reference numbers, percentages, versions and quantities are filtered or penalised so they never win over a real amount. * **Exact arithmetic** — values are ``decimal.Decimal``, never binary floats. Quick start ----------- >>> from app.services.price_parser import parse_amount, extract_amounts >>> parse_amount("$iom89.00") Amount(amount=Decimal('89.00'), currency='$', currency_code=None, ...) >>> parse_amount("USD 78,000").amount Decimal('78000') >>> parse_amount("1,50,087.99").amount # Indian lakh grouping Decimal('150087.99') >>> parse_amount("Page 3 of 10") is None True NOTE: this module intentionally does NOT vendor any third-party parser (see ``price-parser``, ``number-parser``, ``money-parser`` on PyPI) because none of them combine OCR-junk tolerance, Indian lakh/crore grouping, magnitude suffixes, negatives and multi-candidate disambiguation in a single dependency-free implementation. """ from __future__ import annotations import re from bisect import bisect_right from dataclasses import dataclass from decimal import Decimal from typing import Iterator, List, Optional MAX_INPUT_LENGTH = 1_000_000 """Hard cap on input size (characters) to bound worst-case work.""" CONFIDENCE_FLOOR = 0.20 """Minimum confidence for a candidate to count as a real amount. ``parse_amount`` returns ``None`` when the best candidate scores below this, and API consumers use it to decide which candidates to report. """ _CONFIDENCE_FLOOR = CONFIDENCE_FLOOR # --------------------------------------------------------------------------- # Currency tables # --------------------------------------------------------------------------- #: Symbol -> ISO 4217 code when the symbol is unambiguous. _SYMBOL_TO_ISO = { "€": "EUR", "£": "GBP", "₹": "INR", "₩": "KRW", "₽": "RUB", "฿": "THB", "₫": "VND", "₦": "NGN", "₱": "PHP", "₴": "UAH", "₪": "ILS", "₸": "KZT", "₲": "PYG", "₡": "CRC", "₾": "GEL", "৳": "BDT", "₵": "GHS", "₼": "AZN", "៛": "KHR", "₭": "LAK", "₮": "MNT", "₺": "TRY", } #: Symbols that are ambiguous without extra context (no fixed ISO code). _AMBIGUOUS_SYMBOLS = frozenset("$¥₨") #: Common ISO 4217 codes we recognise when written as 3-letter codes. _ISO_CODES = frozenset( { "USD", "EUR", "GBP", "INR", "JPY", "CNY", "AUD", "CAD", "CHF", "SGD", "HKD", "NZD", "SEK", "NOK", "DKK", "PLN", "CZK", "HUF", "RON", "BGN", "HRK", "RUB", "TRY", "ZAR", "BRL", "MXN", "ARS", "CLP", "COP", "PEN", "UYU", "VES", "IDR", "MYR", "PHP", "THB", "VND", "KRW", "TWD", "AED", "SAR", "QAR", "KWD", "BHD", "OMR", "JOD", "LBP", "IQD", "IRR", "EGP", "NGN", "KES", "GHS", "ZMW", "TZS", "UGX", "MAD", "TND", "DZD", "PKR", "BDT", "LKR", "NPR", "MMK", "KZT", "UZS", "AZN", "GEL", "AMD", "BYN", "MNT", "KHR", "LAK", "MOP", "BND", "XOF", "XAF", "MUR", "MVR", "MWK", "MZN", "NAD", "BWP", "GMD", "GNF", "HTG", "ISK", "JMD", "KMF", "LSL", "LYD", "MGA", "MKD", "PGK", "RSD", "SOS", "SRD", "SZL", "TOP", "TTD", "WST", "XCD", "FJD", "ALL", "BAM", "ETB", "GIP", "GYD", "KGS", "KPW", "MRO", "SCR", "SYP", "TJS", "TMT", "VUV", "YER", } ) #: Short/compound currency forms -> ISO code (None when still ambiguous). _SHORT_TO_ISO = { "US$": "USD", "R$": "BRL", "Rp": "IDR", "Rs": "INR", "Re": "INR", "Kč": "CZK", "zł": "PLN", "Ft": "HUF", "руб": "RUB", "р.": "RUB", "lei": "RON", "лв": "BGN", "S/": "PEN", "kr": None, } #: Currency words -> ISO code (None when ambiguous). _WORD_TO_ISO = { "dollars": "USD", "dollar": "USD", "usd": "USD", "euros": "EUR", "euro": "EUR", "pounds": "GBP", "pound": "GBP", "rupees": "INR", "rupee": "INR", "inr": "INR", "yen": "JPY", "yuan": "CNY", "rand": "ZAR", "reais": "BRL", "real": "BRL", "pesos": None, "peso": None, "won": "KRW", "baht": "THB", "dong": "VND", "naira": "NGN", "ringgit": "MYR", "rupiah": "IDR", "zloty": "PLN", "forint": "HUF", "leu": "RON", "lev": "BGN", "dirham": "AED", "riyal": "SAR", "lari": "GEL", "tenge": "KZT", "hryvnia": "UAH", "shekel": "ILS", } # --------------------------------------------------------------------------- # Amount-indicating and non-amount word lists (scored context) # --------------------------------------------------------------------------- _STRONG_AMOUNT_RE = re.compile( r"\b(?:grand total|total|payable|balance)\b", re.IGNORECASE ) _MEDIUM_AMOUNT_RE = re.compile( r"\b(?:subtotal|sub-total|amount|price|charge|cost|value|sum|fee|deposit|" r"paid|received|amt|due|net|gross)\b", re.IGNORECASE, ) _REJECT_WORD_RE = re.compile( r"\b(?:qty|quantity|page|no\.?|number|ref\.?|date|tel|phone|mobile|fax|" r"pin|zip|gst|pan|ifsc|swift|vat|tin|weight|height|width|length|depth|" r"pcs|unit|units|kg|gms?|ml|ltr|est\.?|reg\.?|sr\.?|sl\.?|po\b|a/c|acct|" r"account|contact|inv\.?)\b", re.IGNORECASE, ) _OF_RE = re.compile(r"\bof\b", re.IGNORECASE) _FREE_RE = re.compile(r"(?i)^\s*(?:free|complimentary|n/?a|na|no charge|no cost|zero)\s*$") # --------------------------------------------------------------------------- # Tokenisation # --------------------------------------------------------------------------- #: A digit run with optional grouping/separator characters. #: The space-grouped alternative ("15 130", "1 234.56") is tried first so a #: bare "500.00 600.00" still tokenises as two separate numbers. _NUMBER_TOKEN_RE = re.compile( r"(? float: """Float convenience view of ``amount`` (prefer ``amount`` for money math).""" return float(self.amount) @property def is_zero(self) -> bool: return self.amount == 0 # --------------------------------------------------------------------------- # Number-body interpretation (separator disambiguation) # --------------------------------------------------------------------------- def _clean_grouped_int(int_part: str, sep: str) -> Optional[str]: """Validate a grouped integer part; return clean digits or ``None``. ``sep`` is the grouping separator actually used (``","`` or ``"."``). Accepts Western (``1,234``), Indian (``1,50,087``) and dot-grouped (``1.234``) conventions. """ if int_part.isdigit(): return int_part if sep == ",": if _GROUPED_COMMA_WEST.fullmatch(int_part) or _GROUPED_COMMA_IND.fullmatch(int_part): return int_part.replace(",", "") else: if _GROUPED_DOT.fullmatch(int_part): return int_part.replace(".", "") return None def _frac_decimal(int_clean: str, frac: str) -> Decimal: return Decimal(f"{int_clean}.{frac}") if frac else Decimal(int_clean) def _parse_decimal(body: str) -> Optional[Decimal]: """Interpret a raw digit/separator token as a ``Decimal``, or ``None``. Rules (mirroring the well-tested ``price-parser`` heuristics, extended for Indian lakh/crore grouping): * both separators present -> the *last* one is the decimal separator; * a single separator whose trailing group is 1-2 digits -> decimal (``22,90`` -> 22.90, ``78.90`` -> 78.90); * a single separator whose trailing group is exactly 3 digits -> thousands (``78,000`` -> 78000, ``1.234.567`` -> 1234567); * invalid grouping (``1,23,45``, ``1.2.3``, ``12 34``) -> ``None``. """ body = body.rstrip(".,").strip() if not body: return None if not re.fullmatch(r"[0-9 .,]+", body): return None if body.count(",") + body.count(".") > 6: return None has_comma = "," in body has_dot = "." in body has_space = " " in body if has_space: if has_comma or has_dot: body = body.replace(" ", "") has_comma = "," in body has_dot = "." in body else: groups = body.split(" ") if ( len(groups) >= 2 and all(len(g) == 3 for g in groups[1:]) and 1 <= len(groups[0]) <= 3 ): return Decimal("".join(groups)) return None if has_comma and has_dot: dec = body[max(body.rfind(","), body.rfind("."))] int_part, _, frac = body.rpartition(dec) if not (1 <= len(frac) <= 2 and frac.isdigit()): return None clean = _clean_grouped_int(int_part, "," if dec == "." else ".") if clean is None: return None return _frac_decimal(clean, frac) if has_comma: return _parse_single_sep(body, ",") if has_dot: return _parse_single_sep(body, ".") return Decimal(body) def _parse_single_sep(body: str, sep: str) -> Optional[Decimal]: if body.startswith(sep): frac = body[1:] if 1 <= len(frac) <= 2 and frac.isdigit(): return Decimal(f"0.{frac}") return None groups = body.split(sep) trailing = groups[-1] if 1 <= len(trailing) <= 2: int_part = sep.join(groups[:-1]) if int_part.isdigit(): return _frac_decimal(int_part, trailing) clean = _clean_grouped_int(int_part, sep) if clean is not None: return _frac_decimal(clean, trailing) return None if len(trailing) == 3: clean = _clean_grouped_int(body, sep) return Decimal(clean) if clean is not None else None return None # --------------------------------------------------------------------------- # Context helpers # --------------------------------------------------------------------------- def _resolve_currency(raw: str) -> tuple[Optional[str], Optional[str]]: """Map a raw currency token to ``(display, iso_code)``; unknown -> (None, None).""" r = raw.strip() if r in _SYMBOL_TO_ISO: return r, _SYMBOL_TO_ISO[r] if r in _AMBIGUOUS_SYMBOLS: return r, None up = r.upper() if up in _ISO_CODES: return r, up if r in _SHORT_TO_ISO: return r, _SHORT_TO_ISO[r] low = r.lower() if low in _WORD_TO_ISO: return low, _WORD_TO_ISO[low] return None, None def _nearest_currency( text: str, raw_start: int, raw_end: int ) -> tuple[Optional[str], Optional[str], Optional[int], Optional[int], Optional[int]]: """Find the currency marker nearest to the number span. Returns ``(currency, iso_code, gap, abs_start, abs_end)`` or all-``None``. ``gap`` is the number of characters between the marker and the number; OCR junk between them is tolerated up to ``_CURRENCY_GAP_TOLERANCE``. """ prefix = text[max(0, raw_start - _CURRENCY_WINDOW):raw_start] suffix = text[raw_end:raw_end + _CURRENCY_WINDOW] best = None # (currency, code, gap, abs_start, abs_end) for m in _CURRENCY_FINDER.finditer(prefix): cur, code = _resolve_currency(m.group(0)) if cur is None: continue gap = len(prefix) - m.end() if gap <= _CURRENCY_GAP_TOLERANCE and (best is None or gap < best[2]): base = raw_start - len(prefix) best = (cur, code, gap, base + m.start(), base + m.end()) for m in _CURRENCY_FINDER.finditer(suffix): cur, code = _resolve_currency(m.group(0)) if cur is None: continue gap = m.start() if gap <= _CURRENCY_GAP_TOLERANCE and (best is None or gap < best[2]): best = (cur, code, gap, raw_end + m.start(), raw_end + m.end()) if best is None: return None, None, None, None, None return best def _detect_magnitude(text: str, pos: int) -> tuple[Optional[Decimal], int]: """Detect a magnitude suffix right after the number; return (multiplier, end). Single letters must NOT be followed by another letter so currency codes and units are never eaten: "50K" -> x1000 but "Kč" and "Kg" are not magnitudes. """ i = pos if i < len(text) and text[i] == " ": i += 1 rest = text[i:i + 10] m = re.match(r"(?i)(?:crore|cr|lakh|lac)\b", rest) if m: word = m.group(0).lower() mult = Decimal("10000000") if word in ("crore", "cr") else Decimal("100000") return mult, i + m.end() if not rest: return None, pos c = rest[0] if len(rest) > 1 and rest[1].isalpha(): return None, pos # "Kč", "Kg", "MOP", "LKR" ... not magnitudes if c in "Kk": return Decimal("1000"), i + 1 if c == "M": return Decimal("1000000"), i + 1 if c == "B": return Decimal("1000000000"), i + 1 if c == "T": return Decimal("1000000000000"), i + 1 if c == "L": return Decimal("100000"), i + 1 return None, pos def _merge_stray_letters(text: str, start: int, end: int) -> tuple[int, str]: """Extend a digit token across <=2 stray letters followed by digits. Handles OCR noise like ``8a9.00`` -> ``89.00``. Returns ``(new_end, letters)`` or ``(end, "")`` when there is nothing to merge. Never merges across ``e``/``E`` (scientific) or ``x``/``X`` (hex) markers. """ i = end letters = 0 while i < len(text) and text[i].isalpha() and letters < 3: letters += 1 i += 1 if letters == 0 or letters > 2: return end, "" if any(c in "eExX" for c in text[end:i]): return end, "" j = i if j >= len(text) or not text[j].isdigit(): return end, "" k = j while k < len(text) and (text[k].isdigit() or text[k] in " .,"): k += 1 return k, text[end:i] def _merge_space_continuation(text: str, start: int, end: int) -> int: """Merge a wrapped decimal continuation across a space. OCR / text-extraction artefacts sometimes split a number as ``"896,009.0 0"`` -> ``"896,009.00"``. We merge only when the current token is already a decimal (1-2 fraction digits after the last separator) and the run right after the space is 1-2 digits, so ``"500.00 600.00"`` and ``"10 20"`` stay untouched. """ i = end if i >= len(text) or text[i] != " ": return end j = i + 1 k = j while k < len(text) and text[k].isdigit(): k += 1 run_len = k - j if not 1 <= run_len <= 2: return end if k < len(text) and (text[k].isdigit() or text[k] in ".,"): return end # the run continues into another number body = text[start:end] last_sep = max(body.rfind(","), body.rfind(".")) if last_sep < 0: return end frac = body[last_sep + 1:] if not (1 <= len(frac) <= 2 and frac.isdigit()): return end return k def _sci_guards(text: str, start: int, end: int) -> bool: """Skip tokens that are part of scientific notation or hex literals.""" lo = max(0, start - 3) hi = min(len(text), end + 3) for m in _SCI_RE.finditer(text, lo, hi): if m.end() > start and (m.start() < end or m.start() <= end + 2): return True if text[max(0, start - 1):end + 2].lower().startswith("0x"): return True return False def _has_percent(text: str, raw_start: int, raw_end: int) -> bool: if "%" in text[max(0, raw_start - 1):raw_end + 1]: return True after = text[raw_end:raw_end + 8] return bool(re.search(r"(?i)\bpercent\b|\bper cent\b", after)) def _skip_regions(text: str) -> list[tuple[int, int]]: regions: list[tuple[int, int]] = [] for pat in (_DATE_RE, _TIME_RE, _IP_RE, _ID_RE, _PHONE_RE, _AREA_CODE_RE): regions.extend((m.start(), m.end()) for m in pat.finditer(text)) regions.sort() return regions def _in_skip_region(start: int, end: int, regions: list[tuple[int, int]], starts: list[int]) -> bool: i = bisect_right(starts, end - 1) return i > 0 and regions[i - 1][1] > start # --------------------------------------------------------------------------- # Candidate generation # --------------------------------------------------------------------------- def _iter_candidates( text: str, regions: list[tuple[int, int]], starts: list[int], currency_hint: Optional[str], ) -> Iterator[Amount]: n = len(text) consumed = 0 hint_cur, hint_code = _resolve_currency(currency_hint) if currency_hint else (None, None) for m in _NUMBER_TOKEN_RE.finditer(text): start, end = m.start(), m.end() if start < consumed: continue if _sci_guards(text, start, end): continue merged_end, letters = _merge_stray_letters(text, start, end) merged_end = _merge_space_continuation(text, start, merged_end) if merged_end > consumed: consumed = merged_end # Number body with stray OCR letters ("8a9.00" -> "89.00") and wrapped # decimal continuations ("896,009.0 0" -> "896,009.00") cleaned up. if merged_end > end: seg = text[end + len(letters):merged_end].replace(" ", "") body = text[start:end] + seg else: body = text[start:merged_end] # --- sign --- negative = False raw_start = start if raw_start > 0 and text[raw_start - 1] in "-−+": negative = text[raw_start - 1] in "-−" raw_start -= 1 raw_end = merged_end if raw_end < n and text[raw_end] in "-−": negative = True raw_end += 1 if raw_start > 0 and text[raw_start - 1] == "(": close = text.find(")", raw_end, raw_end + 5) if close != -1 and not any(c.isdigit() for c in text[close + 1:close + 4]): negative = True raw_start -= 1 raw_end = close + 1 value = _parse_decimal(body) if value is None: continue if negative: value = -value # --- magnitude --- mult, mag_end = _detect_magnitude(text, raw_end) magnitude = mult is not None if mult is not None: value *= mult raw_end = mag_end # --- currency --- cur, code, gap, cur_start, cur_end = _nearest_currency(text, raw_start, raw_end) if cur_start is not None: raw_start = min(raw_start, cur_start) raw_end = max(raw_end, cur_end) # --- context words --- # Amount-indicating words only count BEFORE the number ("Total: 550.00"), # so "50.00, Total 550.00" never boosts the wrong candidate. Reject words # (quantities, references, units) count on both sides. prefix = text[max(0, raw_start - _WORD_WINDOW_BEFORE):raw_start] suffix = text[raw_end:raw_end + _WORD_WINDOW_AFTER] amount_strong = bool(_STRONG_AMOUNT_RE.search(prefix)) amount_medium = bool(_MEDIUM_AMOUNT_RE.search(prefix)) reject = bool(_REJECT_WORD_RE.search(prefix + " " + suffix)) of_penalty = bool(_OF_RE.search(prefix[-4:])) int_digits = len(str(abs(value).to_integral_value())) frac_digits = max(-value.as_tuple().exponent, 0) seg = text[raw_start:raw_end] # --- hard skips (region / pattern based) --- if _in_skip_region(start, merged_end, regions, starts): continue if _has_percent(text, raw_start, raw_end): continue if ( not magnitude and cur is None and not amount_strong and not amount_medium and re.fullmatch(r"\d{4}", body) and 1900 <= value <= 2099 ): continue # year if ( cur is None and not amount_strong and not amount_medium and not magnitude and int_digits >= 8 and not any(c in seg for c in " .,") ): continue # long bare digit run (IDs, phone numbers) if ( cur is None and not amount_strong and not amount_medium and not magnitude and int_digits >= 7 and (" " in seg or "-" in seg) ): continue # phone-like number with separators # --- scoring --- score = 0.5 if cur is not None: score += 0.35 if gap == 0 else 0.30 if code is not None: score += 0.05 elif hint_cur is not None: cur, code = hint_cur, hint_code score += 0.10 if amount_strong: score += 0.25 elif amount_medium: score += 0.15 if magnitude: score += 0.10 if frac_digits == 2: score += 0.05 if negative: score += 0.02 if of_penalty: score -= 0.40 # A quantity/reference word near the number ("Qty: 3", "Invoice No: 00125") # kills the candidate unless an amount word is also present ("Unit price: 25"). if reject and not (amount_strong or amount_medium): score -= 0.60 if cur is None and not amount_strong and not amount_medium and int_digits <= 3: score -= 0.15 score = max(0.0, min(1.0, score)) raw = text[raw_start:raw_end] yield Amount( amount=value, currency=cur, currency_code=code, amount_text=body, raw=raw, confidence=round(score, 4), is_negative=negative, position=start, ) # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- def extract_amounts( text, *, limit: Optional[int] = None, currency_hint: Optional[str] = None, ) -> List[Amount]: """Return every candidate amount in ``text``, ranked by confidence. Accepts ``str`` (or a numeric input such as ``int``/``float``/``Decimal``). ``None`` and non-string inputs return an empty list. Raises ``ValueError`` for inputs longer than ``MAX_INPUT_LENGTH`` characters. """ if text is None: return [] if isinstance(text, (bool, int, float, Decimal)): text = str(text) if not isinstance(text, str): return [] if len(text) > MAX_INPUT_LENGTH: raise ValueError( f"input exceeds MAX_INPUT_LENGTH={MAX_INPUT_LENGTH:,} characters" ) regions = _skip_regions(text) starts = [s for s, _ in regions] results = list(_iter_candidates(text, regions, starts, currency_hint)) results.sort(key=lambda a: (-a.confidence, a.position)) if limit is not None: results = results[:limit] return results def parse_amount( text, *, currency_hint: Optional[str] = None, default_currency: Optional[str] = None, ) -> Optional[Amount]: """Return the single best amount in ``text``, or ``None``. * ``currency_hint`` labels candidates that have no currency marker of their own (e.g. ``parse_amount("34.99", currency_hint="руб")``). * ``default_currency`` labels the final result when no currency was found (and no hint supplied). * Strings that mean "no charge" (``"Free"``, ``"N/A"``, ``"no charge"`` ...) return ``Amount(0)`` with high confidence. * Returns ``None`` when no candidate clears the confidence floor — e.g. dates, phone numbers, percentages, invoice references. """ if text is None: return None if isinstance(text, (bool, int, float, Decimal)): text = str(text) if not isinstance(text, str): return None stripped = text.strip() if not stripped: return None if len(text) > MAX_INPUT_LENGTH: raise ValueError( f"input exceeds MAX_INPUT_LENGTH={MAX_INPUT_LENGTH:,} characters" ) if _FREE_RE.fullmatch(stripped): return Amount( amount=Decimal("0"), currency=None, currency_code=None, amount_text="0", raw=stripped, confidence=0.95, is_negative=False, position=0, ) results = extract_amounts(text, currency_hint=currency_hint) if not results: return None best = results[0] if best.confidence < _CONFIDENCE_FLOOR: return None if best.currency is None and default_currency is not None: cur, code = _resolve_currency(default_currency) if cur is not None: return Amount( amount=best.amount, currency=cur, currency_code=code, amount_text=best.amount_text, raw=best.raw, confidence=best.confidence, is_negative=best.is_negative, position=best.position, ) return best #: Alias mirroring the familiar ``price_parser.parse_price`` name. parse_price = parse_amount __all__ = [ "Amount", "extract_amounts", "parse_amount", "parse_price", "MAX_INPUT_LENGTH", "CONFIDENCE_FLOOR", ]