from __future__ import annotations from calendar import monthrange from datetime import date, datetime import re from typing import Any OPEN_STATUSES = { "em aberto", "a vencer", "em atraso", "atrasada", "atrasado", "vencida", "vencido", "aberto", "aberto deb aut", "aberto cc", "aberto pix", "open", "unpaid", } PAID_STATUSES = {"pago", "paga", "paid", "quitada", "quitado"} CONTESTED_STATUSES = {"contestada", "contestado", "em contestacao", "em contestação", "disputed"} PAYMENT_TYPE_LABELS = {1: "debito_automatico", 2: "fatura", 3: "cartao_credito", 4: "incobraveis"} PAYMENT_TYPES_REQUIRE_SMS = {1, 3} def _norm(value: Any) -> str: return " ".join(str(value or "").strip().casefold().replace("_", " ").split()) def _invoice_id(item: dict[str, Any]) -> str: return str(item.get("invoiceId") or item.get("invoice_id") or item.get("invoiceNumber") or item.get("number") or "").strip() def payment_items(payload: Any) -> list[dict[str, Any]]: if not isinstance(payload, dict): return [] rows = payload.get("paymentItems") or payload.get("payment_items") or payload.get("invoices") or [] return [dict(x) for x in rows if isinstance(x, dict)] if isinstance(rows, (list, tuple)) else [] def select_invoice(payload: Any, invoice_id: Any = "") -> dict[str, Any]: rows = payment_items(payload) wanted = str(invoice_id or "").strip() if wanted: for row in rows: if _invoice_id(row) == wanted: return row return {} return rows[0] if rows else {} def invoice_status(item: dict[str, Any]) -> str: return str(item.get("invoiceStatus") or item.get("invoice_status") or item.get("status") or "").strip() def is_unpaid_status(value: Any) -> bool: text = _norm(value) return text in OPEN_STATUSES or any(k in text for k in ("aberto", "atras", "vencid", "unpaid")) def is_paid_status(value: Any) -> bool: text = _norm(value) return text in PAID_STATUSES or text.startswith("pag") def is_contested_status(value: Any) -> bool: text = _norm(value) return text in CONTESTED_STATUSES or "contest" in text def complete_invoices_context(payload: Any, invoice_id: Any = "") -> dict[str, Any]: source = payload if isinstance(payload, dict) else {} billing = source.get("billingProfile") or source.get("billing_profile") or {} billing = billing if isinstance(billing, dict) else {} raw_pt = billing.get("paymentTypeId") if billing.get("paymentTypeId") is not None else billing.get("payment_type_id") try: payment_type_id = int(raw_pt) if raw_pt is not None else None except (TypeError, ValueError): payment_type_id = None method = str( source.get("payment_method") or source.get("method_payment") or source.get("forma_pagamento") or billing.get("paymentMethod") or billing.get("methodPayment") or PAYMENT_TYPE_LABELS.get(payment_type_id, "") ).strip().lower() rows = payment_items(source) selected = select_invoice(source, invoice_id) statuses = [invoice_status(x) for x in rows if invoice_status(x)] selected_status = invoice_status(selected) if invoice_id: has_open = bool(selected and is_unpaid_status(selected_status)) else: has_open = any(is_unpaid_status(x) for x in statuses) return { "invoice": selected, "invoice_status": selected_status, "invoice_statuses": statuses, "has_open_bill": has_open, "payment_type_id": payment_type_id, "payment_method": method, "method_payment": method, "forma_pagamento": method, "requires_sms": payment_type_id in PAYMENT_TYPES_REQUIRE_SMS, "requer_sms": payment_type_id in PAYMENT_TYPES_REQUIRE_SMS, } def _is_dacc(method: Any) -> bool: text = _norm(method) return any(k in text for k in ("dacc", "debito automatico", "débito automático", "cartao", "cartão", "parcelamento")) def resolve_refund_option(*, payment_method: str, has_open_bill: bool, invoice_statuses: list[str]) -> dict[str, Any]: is_dacc = _is_dacc(payment_method) is_contested = any(is_contested_status(x) for x in invoice_statuses) is_paid = any(is_paid_status(x) for x in invoice_statuses) and not has_open_bill is_unpaid = bool(has_open_bill) or any(is_unpaid_status(x) for x in invoice_statuses) boleto = bool(is_unpaid and not is_dacc and not is_contested) reason = "credito_conta_futura" if boleto: reason = "fatura_nao_paga" elif is_dacc: reason = "forma_pagamento_dacc_cartao_parcelamento" elif is_contested: reason = "fatura_ja_contestada" elif is_paid: reason = "fatura_paga" return { "refund_option": "1" if boleto else "0", "format_text": "sms" if boleto else "conta_futura", "resolution_type": "new_boleto" if boleto else "credit_bill", "decision_reason": reason, "is_dacc": is_dacc, "is_paid": is_paid, "is_unpaid": is_unpaid, "is_contested": is_contested, } def parse_date(value: Any) -> datetime | None: text = str(value or "").strip() if not text: return None for fmt in ("%Y-%m-%d", "%Y%m%d", "%d/%m/%Y", "%d-%m-%Y", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M:%SZ"): try: return datetime.strptime(text[:20] if "T" in fmt else text, fmt) except ValueError: continue digits = re.sub(r"\D", "", text) if len(digits) >= 8: for fmt in ("%Y%m%d", "%d%m%Y"): try: return datetime.strptime(digits[:8], fmt) except ValueError: pass return None def due_day(value: Any) -> int: parsed = parse_date(value) if parsed: return parsed.day m = re.search(r"(?:^|[-/])(\d{1,2})$", str(value or "").strip()) return int(m.group(1)) if m else 0 def billing_cutoff_reference(cutoff_day: int, *, reference_datetime: datetime | None = None) -> tuple[bool, datetime | None]: if cutoff_day <= 0: return False, None ref = reference_datetime or datetime.now() day = min(cutoff_day, monthrange(ref.year, ref.month)[1]) cutoff = datetime(ref.year, ref.month, day) if ref < cutoff: month = ref.month - 1 year = ref.year if month == 0: month, year = 12, year - 1 cutoff = datetime(year, month, min(cutoff_day, monthrange(year, month)[1])) return ref >= cutoff, cutoff def next_cutoff_from_cut_date(cut_date: datetime, *, reference_datetime: datetime | None = None) -> tuple[bool, datetime]: ref = reference_datetime or datetime.now() month = cut_date.month + 1 year = cut_date.year if month == 13: month, year = 1, year + 1 nxt = datetime(year, month, min(cut_date.day, monthrange(year, month)[1])) return ref > nxt, nxt def manual_conta_certa_from_evidence(*, dependent_invoice_item: bool = False, invoice_item: dict[str, Any] | None = None, contract: Any = None, reference_datetime: datetime | None = None) -> bool: if dependent_invoice_item: return True row = invoice_item or {} for key in ("cutDate", "cutoffDate", "cut_date", "cutoff_date"): if row.get(key): parsed = parse_date(row.get(key)) if parsed: return next_cutoff_from_cut_date(parsed, reference_datetime=reference_datetime)[0] try: return billing_cutoff_reference(int(row.get(key)), reference_datetime=reference_datetime)[0] except (TypeError, ValueError): pass payload = contract if isinstance(contract, dict) else {} billing = payload.get("billing_profile") or payload.get("billingProfile") or {} billing = billing if isinstance(billing, dict) else {} date_info = billing.get("date") if isinstance(billing.get("date"), dict) else {} try: cutoff_day = int(date_info.get("cutoffDay") or 0) except (TypeError, ValueError): cutoff_day = 0 return billing_cutoff_reference(cutoff_day, reference_datetime=reference_datetime)[0] if cutoff_day else False