from __future__ import annotations import re import unicodedata as ud from datetime import date from decimal import Decimal, InvalidOperation, ROUND_HALF_UP from typing import Any _CENT = Decimal('0.01') def decimal_from_any(value: Any) -> Decimal | None: if value is None or isinstance(value, bool): return None if isinstance(value, Decimal): return value if isinstance(value, (int, float)): return Decimal(str(value)) text = str(value or '').strip().replace('R$', '').replace(' ', '') if not text: return None if ',' in text and '.' in text: text = text.replace('.', '').replace(',', '.') elif ',' in text: text = text.replace(',', '.') try: return Decimal(text) except (InvalidOperation, ValueError): return None def money(value: Decimal) -> Decimal: return value.quantize(_CENT, rounding=ROUND_HALF_UP) def amount_text(value: Decimal) -> str: return f'{money(value):.2f}' def normalize_match_text(value: Any) -> str: text = re.sub(r'\s*\([^)]*\)', '', str(value or '')).strip() text = ud.normalize('NFKD', text) text = ''.join(ch for ch in text if not ud.combining(ch)).casefold() return re.sub(r'\s+', ' ', re.sub(r'[^a-z0-9]+', ' ', text)).strip() def same_plan_name(left: Any, right: Any) -> bool: a, b = normalize_match_text(left), normalize_match_text(right) return bool(a and b and (a == b or a in b or b in a)) def resolve_plano_controle(planos: list[dict[str, Any]]) -> tuple[dict[str, Any], dict[str, Any]] | None: controls = [p for p in planos if isinstance(p, dict) and bool(p.get('is_controle'))] if len(controls) != 1: return None control = controls[0] other = next((p for p in planos if p is not control), None) return (control, other) if isinstance(other, dict) else None def resolve_liquid_value(plano: dict[str, Any]) -> Decimal | None: for key in ('valor_final','valorFinal','valor_liquido','valorLiquido','net_value','netValue','subtotal','value_final'): val = decimal_from_any(plano.get(key)) if val is not None: return val gross = next((decimal_from_any(plano.get(k)) for k in ('valor_bruto','valorBruto','gross_value','grossValue','valor_bruto_plano','valorBrutoPlano','preco_unit') if decimal_from_any(plano.get(k)) is not None), None) discounts = next((decimal_from_any(plano.get(k)) for k in ('total_descontos','totalDescontos','discount_total') if decimal_from_any(plano.get(k)) is not None), None) return gross + discounts if gross is not None and discounts is not None else None def _parse_emission_year(invoice: dict[str, Any], emission: str='') -> int | None: for text in [emission] + [str(x.get('emissao') or '') for x in invoice.get('Fatura Resumo', []) if isinstance(x, dict)]: m = re.search(r'\b\d{2}/\d{2}/(?P\d{4})\b', text) if m: return int(m.group('y')) return None def _period_text(invoice: dict[str, Any], period: str='') -> str: if period: return period for item in invoice.get('Fatura Resumo', []) or []: if isinstance(item, dict) and normalize_match_text(item.get('desc')) == 'periodo': return str(item.get('period') or '').strip() return '' def _parse_period(text: str, year: int) -> tuple[date,date] | None: m = re.search(r'(?P\d{2})/(?P\d{2})\s+a\s+(?P\d{2})/(?P\d{2})', text) if not m: return None sd,sm,ed,em = map(int, (m['sd'],m['sm'],m['ed'],m['em'])) sy = year - 1 if sm > em else year try: start,end = date(sy,sm,sd), date(year,em,ed) except ValueError: return None return (start,end) if start <= end else None def resolve_period_days(other: dict[str, Any], invoice: dict[str, Any], *, period: str='', emission: str='') -> tuple[int,int] | None: year = _parse_emission_year(invoice, emission) ptxt = _period_text(invoice, period) if year is None or not ptxt: return None pr = _parse_period(ptxt, year) if pr is None: return None cycle = (pr[1]-pr[0]).days + 1 raw = other.get('days') if other.get('days') is not None else other.get('dias') try: days_other = int(Decimal(str(raw)).to_integral_value(rounding=ROUND_HALF_UP)) except Exception: return None if days_other <= 0 or cycle <= 0: return None if days_other >= cycle: return cycle, 1 return cycle, max(1, cycle-days_other) def find_danfe_plan_items(danfe: dict[str, Any], control: dict[str, Any]) -> list[dict[str, Any]]: plans = danfe.get('Planos') if not isinstance(plans, dict): return [] desc = control.get('desc') for name, raw in plans.items(): if same_plan_name(name, desc) and isinstance(raw, list): return [x for x in raw if isinstance(x, dict)] return [] def build_contestation_items(danfe: dict[str, Any], control: dict[str, Any], refund: Decimal) -> tuple[list[dict[str, Any]], Decimal]: remaining = money(refund) out: list[dict[str, Any]] = [] for item in find_danfe_plan_items(danfe, control): name = str(item.get('desc') or '').strip() claimed = decimal_from_any(item.get('valor_final') if item.get('valor_final') is not None else item.get('valorFinal')) if not name or claimed is None or claimed <= 0: continue claimed = money(claimed) validated = min(remaining, claimed) if validated <= 0: continue out.append({'itemName': name, 'itemType':'PRO_RATA', 'claimedAmount':float(claimed), 'validatedAmount':float(validated)}) remaining = money(remaining-validated) if remaining <= 0: break return out, remaining def human_validation_text(control: dict[str, Any], liquid: Decimal, cycle: int, control_days: int, used: Decimal, refund: Decimal, items: list[dict[str,Any]]) -> str: name = re.sub(r'\s*\([^)]*\)', '', str(control.get('desc') or '')).strip() or 'Plano Controle' parts=[] for idx,item in enumerate(items,1): claimed=decimal_from_any(item.get('claimedAmount')) or Decimal('0') validated=decimal_from_any(item.get('validatedAmount')) or Decimal('0') parts.append(f"{idx}. {item.get('itemName','')}: valor DANFE R$ {money(claimed):.2f}; valor a abater R$ {money(validated):.2f}") return ( f'Validacao humana pro-rata:Plano Controle identificado: {name}. ' f'Base liquida do plano: R$ {money(liquid):.2f}. Calculo: ciclo de {cycle} dias, uso considerado de {control_days} dias; ' f'valor usado R$ {money(used):.2f}; valor a devolver R$ {money(refund):.2f}. ' f"Itens selecionados no DANFE, na ordem de abatimento: {'; '.join(parts) if parts else 'nenhum item gerado'}. " ) def calculate_refund(*, planos: list[dict[str,Any]], invoice_detail: dict[str,Any], invoice_period: str='', invoice_emissao: str='') -> dict[str,Any]: if len(planos) != 2: raise ValueError('pro_rata exige exatamente dois planos para calcular devolucao.') resolved = resolve_plano_controle(planos) if resolved is None: raise ValueError('Nao foi possivel identificar exatamente um Plano Controle.') control, other = resolved liquid = resolve_liquid_value(control) if liquid is None or liquid <= 0: raise ValueError('Valor liquido do Plano Controle ausente ou invalido.') period_days = resolve_period_days(other, invoice_detail, period=invoice_period, emission=invoice_emissao) if period_days is None: raise ValueError('Periodo da fatura ou dos planos ausente ou invalido para calcular pro-rata.') cycle, control_days = period_days liquid = money(liquid) used = (liquid / Decimal(cycle)) * Decimal(control_days) refund = money(liquid-used) danfe = invoice_detail.get('DANFE-COM') if not isinstance(danfe, dict) or not danfe: raise ValueError('DANFE-COM nao encontrado na fatura recuperada.') items, remaining = build_contestation_items(danfe, control, refund) if remaining > 0: raise ValueError(f'Itens do DANFE insuficientes para cobrir devolucao de R$ {money(remaining):.2f}.') total = money(sum((decimal_from_any(x.get('validatedAmount')) or Decimal('0') for x in items), Decimal('0'))) return { 'items': items, 'invoice_amount_open': amount_text(total), 'invoice_amount': amount_text(total), 'texto_validacao_humana': human_validation_text(control, liquid, cycle, control_days, used, refund, items), 'valor_liquido': amount_text(liquid), 'dias_ciclo': cycle, 'dias_controle': control_days, 'valor_usado': amount_text(money(used)), 'valor_devolver': amount_text(refund), } def payment_message(devolucao: dict[str,Any]) -> str: items = devolucao.get('items') if isinstance(devolucao.get('items'), list) else [] total=Decimal('0'); plan='' for item in items: if not isinstance(item,dict): continue if not plan: plan=str(item.get('itemName') or item.get('item_name') or '').strip() total += decimal_from_any(item.get('validatedAmount') if item.get('validatedAmount') is not None else item.get('validated_amount')) or Decimal('0') amount=f'{money(total):.2f}'.replace('.',',') ptxt=f' {plan}' if plan else '' proto=str(devolucao.get('protocolo_id') or '').strip() proto_txt=f' Seu numero de protocolo e {proto}.' if proto else '' due=str(devolucao.get('data_credito_proxima_fatura') or '').strip() due_txt=f' na fatura com vencimento em {due}, considerando o seu ciclo de faturamento' if due else ' em uma proxima fatura' if str(devolucao.get('format_text') or '').strip()=='sms': barcode_txt='com o codigo de barras atualizado' if str(devolucao.get('barcode') or '').strip() else 'com as orientacoes para pagamento' return f'Realizei a contestacao da fatura considerando o valor proporcional do Plano Controle{ptxt}. O valor contestado, de R$ {amount}, foi retirado da sua fatura. Enviamos uma mensagem {barcode_txt}, com prazo de 4 dias para pagamento.{proto_txt}'.strip() return f'Realizei a contestacao considerando o valor proporcional do Plano Controle{ptxt}. O valor contestado, de R$ {amount}, ficou registrado como credito{due_txt}.{proto_txt}'.strip()