Projeto do Agent Contas ORACLE
This commit is contained in:
201
app/domain/contas/contestation_rules.py
Normal file
201
app/domain/contas/contestation_rules.py
Normal file
@@ -0,0 +1,201 @@
|
||||
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
|
||||
Reference in New Issue
Block a user