202 lines
7.9 KiB
Python
202 lines
7.9 KiB
Python
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
|