Spaces:
Running
Running
| """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"(?<![0-9.,])(?:" | |
| r"[0-9]{1,3}(?: [0-9]{3})+(?:[.,][0-9]{1,2})?|" # space-grouped thousands | |
| r"[0-9][0-9.,]*[0-9]|" # grouped/plain with dots/commas | |
| r"[0-9]+|" # single digit | |
| r"\.[0-9]+" # ".99" | |
| r")" | |
| ) | |
| #: Patterns whose matches mark whole regions that must never yield an amount. | |
| _DATE_RE = re.compile(r"\b\d{1,4}[-/.]\d{1,2}[-/.]\d{1,4}\b") | |
| _TIME_RE = re.compile(r"\b\d{1,2}:\d{2}(?::\d{2})?\b") | |
| _IP_RE = re.compile(r"\b\d{1,3}(?:\.\d{1,3}){3}\b") | |
| _ID_RE = re.compile(r"\b\d{3,4}(?:[- ]\d{3,4}){2,}\b") | |
| _PHONE_RE = re.compile(r"\+[\d\s()\-]{7,}\d") | |
| _AREA_CODE_RE = re.compile(r"\(\d{3}\)") | |
| _SCI_RE = re.compile(r"[eE][+-]?\d+") | |
| _GROUPED_COMMA_WEST = re.compile(r"\d{1,3}(?:,\d{3})+") | |
| _GROUPED_COMMA_IND = re.compile(r"\d{1,2}(?:,\d{2})*,\d{3}") | |
| _GROUPED_DOT = re.compile(r"\d{1,3}(?:\.\d{3})+") | |
| _CURRENCY_FINDER = re.compile( | |
| r"US\$|R\$|Rp\.?|Rs\.?|Re\.?|Kč|zł|Ft|руб|р\.|lei|лв|S/|" | |
| r"[$€£¥₹₩₽฿₫₦₱₴₪₸₲₡₾৳₵₼៛₭₮₺₨₧]|" | |
| r"\b(?:dollars?|euros?|pounds?|rupees?|yen|yuan|rand|reais|real|pesos?|" | |
| r"won|baht|dong|naira|ringgit|rupiah|zloty|forint|leu|lev|dirham|riyal|" | |
| r"lari|tenge|hryvnia|shekel)\b|" | |
| r"[A-Za-z]{3}" | |
| ) | |
| # Currency / magnitude scan windows (chars). Generous for OCR junk. | |
| _CURRENCY_WINDOW = 14 | |
| _CURRENCY_GAP_TOLERANCE = 10 | |
| _WORD_WINDOW_BEFORE = 10 | |
| _WORD_WINDOW_AFTER = 12 | |
| class Amount: | |
| """A single extracted monetary amount with its context and confidence.""" | |
| amount: Decimal | |
| """Numeric value (always ``Decimal`` — never a binary float).""" | |
| currency: Optional[str] | |
| """Currency marker as found in the text (symbol, code or word), or hint/default.""" | |
| currency_code: Optional[str] | |
| """ISO 4217 code when determinable, else ``None`` (e.g. bare ``$``).""" | |
| amount_text: str | |
| """The number part as it appeared (junk letters removed, separators kept).""" | |
| raw: str | |
| """Full matched substring: currency marker + sign + number + magnitude suffix.""" | |
| confidence: float | |
| """Heuristic confidence in [0, 1] — higher is more likely a real amount.""" | |
| is_negative: bool = False | |
| """True when a leading/trailing minus or accounting parentheses were found.""" | |
| position: int = 0 | |
| """Character offset of the digit token in the source text.""" | |
| def amount_float(self) -> float: | |
| """Float convenience view of ``amount`` (prefer ``amount`` for money math).""" | |
| return float(self.amount) | |
| 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", | |
| ] | |