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, }, }