Spaces:
Sleeping
Sleeping
| import re | |
| from datetime import datetime | |
| from typing import Dict, Any, Optional, Tuple | |
| from ..data.validators import validate_extracted_fields, ValidationResult | |
| CURRENCY_SYMBOLS_PATTERN = re.compile(r'[$€£¥₹\s]|usd|eur|idr|rp', re.IGNORECASE) | |
| def normalize_amount(raw_val: Any) -> Optional[float]: | |
| if raw_val is None: | |
| return None | |
| s = str(raw_val).strip() | |
| s = CURRENCY_SYMBOLS_PATTERN.sub('', s) | |
| # Handle European comma decimals vs thousands separator | |
| if ',' in s and '.' in s: | |
| s = s.replace(',', '') | |
| elif ',' in s and '.' not in s: | |
| # Check if comma is decimal (e.g. 50,00) or thousands (e.g. 1,000) | |
| parts = s.split(',') | |
| if len(parts) == 2 and len(parts[1]) == 2: | |
| s = s.replace(',', '.') | |
| else: | |
| s = s.replace(',', '') | |
| cleaned = re.sub(r'[^\d.-]', '', s) | |
| try: | |
| return round(float(cleaned), 2) | |
| except ValueError: | |
| return None | |
| def normalize_date(raw_val: Any) -> Optional[str]: | |
| if raw_val is None: | |
| return None | |
| s = str(raw_val).strip() | |
| # Remove ordinal suffixes | |
| s = re.sub(r'(\d+)(st|nd|rd|th)', r'\1', s, flags=re.IGNORECASE) | |
| date_formats = [ | |
| "%Y-%m-%d", "%d/%m/%Y", "%m/%d/%Y", "%d-%m-%Y", "%Y/%m/%d", | |
| "%B %d, %Y", "%b %d, %Y", "%d %B %Y", "%d %b %Y", "%Y.%m.%d" | |
| ] | |
| for fmt in date_formats: | |
| try: | |
| dt = datetime.strptime(s, fmt) | |
| return dt.strftime("%Y-%m-%d") | |
| except ValueError: | |
| pass | |
| # Regex search for year-month-day or day-month-year | |
| match = re.search(r'(\d{1,4})[/\-.](\d{1,2})[/\-.](\d{1,4})', s) | |
| if match: | |
| p1, p2, p3 = match.groups() | |
| if len(p1) == 4: # YYYY-MM-DD | |
| return f"{p1}-{int(p2):02d}-{int(p3):02d}" | |
| elif len(p3) == 4: # DD-MM-YYYY or MM-DD-YYYY | |
| return f"{p3}-{int(p2):02d}-{int(p1):02d}" | |
| return s | |
| def clean_text_field(raw_val: Any) -> Optional[str]: | |
| if raw_val is None: | |
| return None | |
| s = " ".join(str(raw_val).split()) | |
| return s if s else None | |
| def postprocess_extracted_fields(doc_class: str, fields: Dict[str, Any]) -> Tuple[Dict[str, Any], ValidationResult]: | |
| """ | |
| Applies standardization and normalization to extracted fields, | |
| then runs business rule validation. | |
| """ | |
| normalized = {} | |
| for k, v in fields.items(): | |
| if v is None: | |
| normalized[k] = None | |
| continue | |
| if k in ["total_amount", "amount_due", "total_tax"]: | |
| amt = normalize_amount(v) | |
| normalized[k] = amt if amt is not None else v | |
| elif k in ["date", "invoice_date", "due_date"]: | |
| dt_str = normalize_date(v) | |
| normalized[k] = dt_str if dt_str is not None else v | |
| else: | |
| normalized[k] = clean_text_field(v) | |
| val_res = validate_extracted_fields(doc_class, normalized) | |
| return normalized, val_res | |