Spaces:
Sleeping
Sleeping
| from datetime import date, datetime, timedelta | |
| from zoneinfo import ZoneInfo | |
| def run_date_sanity_checks( | |
| *, | |
| document, | |
| extracted_fields, | |
| platform_configs=None, | |
| ): | |
| """ | |
| Runs Date Sanity Checks after date normalisation. | |
| Expected document fields: | |
| document.id | |
| document.tenant_id | |
| Optional: document.tenant_timezone | |
| Expected extracted_fields fields: | |
| invoice_date | |
| due_date | |
| due_date_origin | |
| payment_terms | |
| service_start_date | |
| service_end_date | |
| Returns: | |
| dict with advisory risk flags and audit-friendly result. | |
| """ | |
| platform_configs = platform_configs or {} | |
| default_timezone = platform_configs.get( | |
| "validation.default_timezone", | |
| getattr(document, "tenant_timezone", "UTC") or "UTC", | |
| ) | |
| max_invoice_age_days = int( | |
| platform_configs.get("validation.max_invoice_age_days", 365) | |
| ) | |
| max_invoice_age_days = max(30, min(max_invoice_age_days, 1825)) | |
| service_end_grace_days = int( | |
| platform_configs.get("validation.service_end_post_invoice_grace_days", 0) | |
| ) | |
| def as_date(value): | |
| if value is None or value == "": | |
| return None | |
| if isinstance(value, datetime): | |
| return value.date() | |
| if isinstance(value, date): | |
| return value | |
| return datetime.strptime(str(value), "%Y-%m-%d").date() | |
| def as_int(value): | |
| if value is None or value == "": | |
| return None | |
| try: | |
| return int(value) | |
| except (ValueError, TypeError): | |
| return None | |
| def tenant_current_date(): | |
| try: | |
| return datetime.now(ZoneInfo(default_timezone)).date() | |
| except Exception: | |
| return datetime.now(ZoneInfo("UTC")).date() | |
| def add_flag(code, *, weight, field, detail=None): | |
| risk_flags.append( | |
| { | |
| "code": code, | |
| "severity": "advisory", | |
| "weight": weight, | |
| "field": field, | |
| "detail": detail or {}, | |
| } | |
| ) | |
| risk_flags = [] | |
| advisory_flags = [] | |
| skipped_steps = [] | |
| invoice_date = as_date(getattr(extracted_fields, "invoice_date", None)) | |
| due_date = as_date(getattr(extracted_fields, "due_date", None)) | |
| service_start_date = as_date(getattr(extracted_fields, "service_start_date", None)) | |
| service_end_date = as_date(getattr(extracted_fields, "service_end_date", None)) | |
| due_date_origin = getattr(extracted_fields, "due_date_origin", None) | |
| payment_terms = as_int(getattr(extracted_fields, "payment_terms", None)) | |
| current_date = tenant_current_date() | |
| # 1. Payment terms must be non-negative. | |
| # Zero is allowed: due-on-receipt. | |
| if payment_terms is not None and payment_terms < 0: | |
| add_flag( | |
| "date_anomaly_payment_terms_negative", | |
| weight=0.20, | |
| field="payment_terms", | |
| detail={ | |
| "payment_terms": payment_terms, | |
| }, | |
| ) | |
| # 2. Due date must be on or after invoice date. | |
| if invoice_date is not None and due_date is not None: | |
| if due_date < invoice_date: | |
| add_flag( | |
| "date_anomaly_due_before_invoice", | |
| weight=0.20, | |
| field="due_date", | |
| detail={ | |
| "invoice_date": str(invoice_date), | |
| "due_date": str(due_date), | |
| }, | |
| ) | |
| if invoice_date is None: | |
| skipped_steps.extend([ | |
| {"step": "date_anomaly_invoice_future_dated", "skip_reason": "no_invoice_date"}, | |
| {"step": "date_anomaly_invoice_too_old", "skip_reason": "no_invoice_date"}, | |
| ]) | |
| # 3. Invoice must not be future-dated. | |
| if invoice_date is not None and invoice_date > current_date: | |
| add_flag( | |
| "date_anomaly_invoice_future_dated", | |
| weight=0.20, | |
| field="invoice_date", | |
| detail={ | |
| "invoice_date": str(invoice_date), | |
| "current_date": str(current_date), | |
| "timezone": default_timezone, | |
| }, | |
| ) | |
| # 4. Invoice must not be too old. | |
| if invoice_date is not None: | |
| oldest_allowed_date = current_date - timedelta(days=max_invoice_age_days) | |
| if invoice_date < oldest_allowed_date: | |
| add_flag( | |
| "date_anomaly_invoice_too_old", | |
| weight=0.15, | |
| field="invoice_date", | |
| detail={ | |
| "invoice_date": str(invoice_date), | |
| "oldest_allowed_date": str(oldest_allowed_date), | |
| "max_invoice_age_days": max_invoice_age_days, | |
| }, | |
| ) | |
| # 5. Service period must not be reversed. | |
| if service_start_date is not None and service_end_date is not None: | |
| if service_end_date < service_start_date: | |
| add_flag( | |
| "date_anomaly_service_period_reversed", | |
| weight=0.15, | |
| field="service_end_date", | |
| detail={ | |
| "service_start_date": str(service_start_date), | |
| "service_end_date": str(service_end_date), | |
| }, | |
| ) | |
| # 6. Service end must not be too far after invoice date. | |
| if service_end_date is not None: | |
| if invoice_date is None: | |
| skipped_steps.append( | |
| {"step": "date_anomaly_service_end_after_invoice", "skip_reason": "no_invoice_date"} | |
| ) | |
| else: | |
| allowed_service_end = invoice_date + timedelta(days=service_end_grace_days) | |
| if service_end_date > allowed_service_end: | |
| add_flag( | |
| "date_anomaly_service_end_after_invoice", | |
| weight=0.15, | |
| field="service_end_date", | |
| detail={ | |
| "invoice_date": str(invoice_date), | |
| "service_end_date": str(service_end_date), | |
| "allowed_service_end": str(allowed_service_end), | |
| "grace_days": service_end_grace_days, | |
| }, | |
| ) | |
| # 7. Due date could not be derived. | |
| if due_date is None and due_date_origin == "unknown": | |
| add_flag( | |
| "due_date_not_derivable", | |
| weight=0.10, | |
| field="due_date", | |
| detail={ | |
| "due_date_origin": due_date_origin, | |
| }, | |
| ) | |
| return { | |
| "risk_flags": risk_flags, | |
| "advisory_flags": advisory_flags, | |
| "skipped_steps": skipped_steps, | |
| "flag_codes": [flag["code"] for flag in risk_flags], | |
| "checked_fields": { | |
| "invoice_date": str(invoice_date) if invoice_date else None, | |
| "due_date": str(due_date) if due_date else None, | |
| "due_date_origin": due_date_origin, | |
| "payment_terms": payment_terms, | |
| "service_start_date": str(service_start_date) if service_start_date else None, | |
| "service_end_date": str(service_end_date) if service_end_date else None, | |
| "current_date": str(current_date), | |
| "timezone": default_timezone, | |
| "max_invoice_age_days": max_invoice_age_days, | |
| "service_end_post_invoice_grace_days": service_end_grace_days, | |
| }, | |
| } |