from __future__ import annotations from dataclasses import dataclass from typing import Any, Dict, List, Optional, Tuple ALLOWED_TERMS = (15, 30, 45) @dataclass class RiskClassification: risk_class: str # A, B, C, D, E reasons: List[str] # reasons humanas, completas recommended_max_value: Optional[float] = None recommended_term_days: Optional[int] = None def _is_critical_document_problem(doc_results: List[Any]) -> Tuple[bool, bool, List[str]]: """ Retorna: - unreadable_critical (bool) - identity_unverifiable (bool) - flags (mensagens humanas) """ critical_types = { "alvara", "certidao_registo_comercial", "registo_comercial", "extrato_bancario", } flags: List[str] = [] unreadable_critical = False identity_unverifiable = False for r in doc_results: doc_type = (getattr(r, "document_type_detected", "") or "").lower() alerts = getattr(r, "alerts", []) or [] if any( "sem texto" in a.lower() or "baixa legibilidade" in a.lower() or "erro ao processar" in a.lower() for a in alerts ): if doc_type in critical_types or doc_type in {"desconhecido", "erro"}: unreadable_critical = True flags.append( f"Não foi possível ler corretamente um documento crítico ({doc_type or 'desconhecido'})." ) if doc_type in {"alvara", "certidao_registo_comercial", "registo_comercial"}: identity_unverifiable = True return unreadable_critical, identity_unverifiable, flags def _entity_identity_assessment( *, input_company_name: Optional[str], input_nuit: Optional[str], input_meses_atividade: Optional[int], entity_features: Dict[str, Any], doc_signals: Dict[str, float], ) -> Tuple[str, List[str], float]: """ Retorna: - status: "OK" | "NEEDS_REVIEW" | "FAIL" - reasons humanas - penalty (0..1) para score (opcional) entity_features esperado (exemplo): { "company_name_match": True/False, "nuit_match": True/False, "registration_age_months": 36, "activity_months_match": True/False, "issuer_verified": True/False, } """ reasons: List[str] = [] penalty = 0.0 cdc = float(doc_signals.get("cross_document_consistency", 0.0)) nuit_match = bool(entity_features.get("nuit_match", False)) name_match = bool(entity_features.get("company_name_match", False)) activity_match = bool(entity_features.get("activity_months_match", False)) issuer_verified = bool(entity_features.get("issuer_verified", False)) # Cross-doc if cdc >= 0.90: reasons.append("(Pos) Os documentos submetidos são consistentes e referem-se à mesma entidade.") elif cdc >= 0.60: reasons.append("(Neg) Existem diferenças entre documentos que sugerem possível inconsistência de entidade (recomenda-se revisão).") penalty = max(penalty, 0.25) else: reasons.append("(Neg) Inconsistência forte entre documentos: os dados não parecem pertencer à mesma entidade.") penalty = max(penalty, 0.60) # NUIT if input_nuit: if nuit_match: reasons.append("(Pos) O NUIT informado no sistema coincide com o NUIT encontrado nos documentos oficiais.") else: reasons.append("(Neg) O NUIT informado no sistema NÃO coincide com o NUIT encontrado nos documentos oficiais.") penalty = max(penalty, 0.60) else: reasons.append("(Info) O NUIT não foi informado no sistema; a verificação fica limitada.") penalty = max(penalty, 0.10) # Nome if input_company_name: if name_match: reasons.append("(Pos) O nome informado no sistema coincide (ou é compatível) com o nome nos documentos oficiais.") else: reasons.append("(Neg) O nome informado no sistema não coincide com o nome nos documentos oficiais.") penalty = max(penalty, 0.35) else: reasons.append("(Info) O nome da empresa não foi informado no sistema; a verificação fica limitada.") penalty = max(penalty, 0.10) # Tempo de atividade (input) × tempo de registo (docs) if input_meses_atividade is not None: if activity_match: reasons.append("(Pos) O tempo de atividade informado é compatível com o tempo de registo encontrado nos documentos.") else: reg_age = entity_features.get("registration_age_months") if reg_age is not None: reasons.append( f"(Neg) O tempo de atividade informado ({input_meses_atividade} meses) não é compatível com o tempo de registo encontrado ({reg_age} meses)." ) else: reasons.append("(Neg) Não foi possível confirmar o tempo de registo para validar o tempo de atividade informado.") penalty = max(penalty, 0.30) else: reasons.append("(Info) O tempo de atividade não foi informado no sistema; a validação fica limitada.") penalty = max(penalty, 0.10) # Emissor (quando existir regra) if issuer_verified: reasons.append("(Pos) O emissor do documento foi identificado como compatível com um emissor esperado (documento com aspeto oficial).") else: reasons.append("(Info) Não foi possível validar o emissor com alta confiança; pode exigir verificação manual dependendo do caso.") # Decide status if penalty >= 0.60: return "FAIL", reasons, penalty if penalty >= 0.25: return "NEEDS_REVIEW", reasons, penalty return "OK", reasons, penalty def _bank_affordability( *, requested_value: float, requested_term_days: int, bank_features: Dict[str, Any], ) -> Tuple[str, List[str], Optional[float], Optional[int]]: """ Retorna: - status: "OK" | "SUGGEST_VALUE" | "SUGGEST_TERM" | "NO_DATA" - reasons - recommended_max_value - recommended_term_days bank_features esperado: { "analysis_status": "OK" | "INSUFFICIENT_DATA", "confidence": 0..1, "capacidade_credito_mensal": float (se existir), "fluxo_liquido": float, "period_days": int (opcional), # opcionalmente: "recommended_max_by_term": {15: x, 30: y, 45: z} } """ reasons: List[str] = [] if requested_term_days not in ALLOWED_TERMS: reasons.append(f"(Info) Prazo {requested_term_days} dias não está no conjunto permitido {list(ALLOWED_TERMS)}.") return "NO_DATA", reasons, None, None status = str(bank_features.get("analysis_status", "INSUFFICIENT_DATA")).upper() conf = float(bank_features.get("confidence", 0.0)) if status != "OK": reasons.append("(Info) O extrato bancário não possui dados suficientes para estimar capacidade de pagamento com confiança.") return "NO_DATA", reasons, None, None # 1) Se existir tabela pronta por prazo, usa direto table = bank_features.get("recommended_max_by_term") if isinstance(table, dict): max_for_term = float(table.get(requested_term_days, 0.0)) if max_for_term <= 0: reasons.append("(Info) Não foi possível estimar valor máximo recomendado para o prazo escolhido.") return "NO_DATA", reasons, None, None if requested_value <= max_for_term: reasons.append("(Pos) Com base no extrato, o valor solicitado é compatível com o prazo escolhido.") return "OK", reasons, None, None # tenta achar prazo que comporte o valor (C) for term in ALLOWED_TERMS: if requested_value <= float(table.get(term, 0.0)): reasons.append("(Neg) O valor solicitado não é viável no prazo escolhido, mas pode ser viável com ajuste de prazo.") reasons.append(f"(Info) Sugestão: alterar o prazo para {term} dias mantendo o valor.") return "SUGGEST_TERM", reasons, None, term # se nenhum prazo comporta, sugere valor menor no mesmo prazo (B) reasons.append("(Neg) O valor solicitado excede a capacidade estimada para o prazo escolhido.") reasons.append(f"(Info) Valor máximo recomendado para {requested_term_days} dias: {int(max_for_term)}.") return "SUGGEST_VALUE", reasons, max_for_term, None # 2) Fallback: usa capacidade mensal se existir cap_mensal = bank_features.get("capacidade_credito_mensal") if isinstance(cap_mensal, (int, float)) and cap_mensal > 0: factor = requested_term_days / 30.0 max_for_term = float(cap_mensal) * factor if requested_value <= max_for_term: reasons.append("(Pos) O extrato indica capacidade financeira compatível com o valor e prazo solicitados.") return "OK", reasons, None, None # tenta sugerir prazo (C) for term in ALLOWED_TERMS: max_term = float(cap_mensal) * (term / 30.0) if requested_value <= max_term: reasons.append("(Neg) O valor solicitado não é viável no prazo escolhido, mas pode ser viável com ajuste de prazo.") reasons.append(f"(Info) Sugestão: alterar o prazo para {term} dias mantendo o valor.") return "SUGGEST_TERM", reasons, None, term # senão sugere valor no mesmo prazo (B) reasons.append("(Neg) O valor solicitado excede a capacidade estimada para o prazo escolhido.") reasons.append(f"(Info) Valor máximo recomendado para {requested_term_days} dias: {int(max_for_term)}.") return "SUGGEST_VALUE", reasons, max_for_term, None # 3) Último fallback: só fluxo_liquido fluxo = bank_features.get("fluxo_liquido") if isinstance(fluxo, (int, float)): if fluxo > 0: reasons.append("(Pos) O extrato apresenta fluxo de caixa líquido positivo no período analisado.") else: reasons.append("(Neg) O extrato apresenta fluxo de caixa líquido não positivo no período analisado.") reasons.append("(Info) Ainda não há dados suficientes para estimar valor máximo recomendado com precisão.") return "NO_DATA", reasons, None, None reasons.append("(Info) Métricas financeiras ausentes no extrato analisado.") return "NO_DATA", reasons, None, None def classify_risk( *, requested_value: float, requested_term_days: int, doc_results: List[Any], doc_signals: Dict[str, float], entity_features: Dict[str, Any], input_company_name: Optional[str] = None, input_nuit: Optional[str] = None, input_meses_atividade: Optional[int] = None, bank_features: Optional[Dict[str, Any]] = None, ) -> RiskClassification: """ Classificação A–E: A: tudo certo - identidade validada - documentos consistentes - valor e prazo compatíveis com o extrato B: valor alto para o fluxo - sugere valor menor - mantém o prazo C: prazo incompatível - sugere ajuste de prazo (15/30/45) - pode manter ou não o valor D: análise manual - documentos ilegíveis - dúvidas relevantes - dados financeiros inconclusivos E: recusa - falha grave de identidade - inconsistência documental forte - risco inaceitável """ reasons: List[str] = [] # ========================================================= # 1) Identidade da entidade (inclui tempo de atividade) # ========================================================= identity_status, identity_reasons, identity_penalty = _entity_identity_assessment( input_company_name=input_company_name, input_nuit=input_nuit, input_meses_atividade=input_meses_atividade, entity_features=entity_features, doc_signals=doc_signals, ) reasons.extend(identity_reasons) # Falha grave de identidade → E if identity_status == "FAIL": reasons.append( "(Neg) A identidade legal da empresa não pôde ser validada com segurança. " "O pedido é recusado." ) return RiskClassification( risk_class="E", reasons=reasons, ) # ========================================================= # 2) Documentos – legibilidade informa, crítico decide D/E # ========================================================= unreadable_critical, identity_unverifiable, doc_flags = _is_critical_document_problem( doc_results ) reasons.extend([f"(Info) {m}" for m in doc_flags]) # Documento crítico ilegível + identidade não confirmável → E if unreadable_critical and identity_unverifiable: reasons.append( "(Neg) Não foi possível confirmar a identidade da empresa devido à " "ilegibilidade de documentos legais críticos." ) return RiskClassification( risk_class="E", reasons=reasons, ) # Documento crítico ilegível (isolado) → D if unreadable_critical: reasons.append( "(Info) É necessária intervenção humana porque documentos críticos " "não foram lidos com qualidade suficiente." ) return RiskClassification( risk_class="D", reasons=reasons, ) # ========================================================= # 3) Extrato bancário – valor × prazo # ========================================================= bank_features = bank_features or {} aff_status, aff_reasons, rec_value, rec_term = _bank_affordability( requested_value=requested_value, requested_term_days=requested_term_days, bank_features=bank_features, ) reasons.extend(aff_reasons) # Identidade precisa revisão + extrato inconclusivo → D if identity_status == "NEEDS_REVIEW" and aff_status == "NO_DATA": reasons.append( "(Info) Existem dúvidas de validação da entidade e o extrato bancário " "não permite estimativa automática. Encaminhado para análise manual." ) return RiskClassification( risk_class="D", reasons=reasons, ) # ========================================================= # 4) Decisão final A / B / C / D / E # ========================================================= # A — tudo consistente if aff_status == "OK" and identity_status == "OK": reasons.append( "(Pos) A empresa apresenta identidade validada, documentação consistente " "e capacidade financeira compatível com o valor e prazo solicitados." ) return RiskClassification( risk_class="A", reasons=reasons, ) # C — ajuste de prazo (prioridade sobre B) if aff_status == "SUGGEST_TERM": reasons.append( "(Neg) O prazo selecionado não é compatível com o valor solicitado " "de acordo com o extrato bancário." ) reasons.append( "(Info) Recomenda-se ajustar o prazo para viabilizar o crédito." ) return RiskClassification( risk_class="C", reasons=reasons, recommended_term_days=rec_term, ) # B — ajuste de valor (mesmo prazo) if aff_status == "SUGGEST_VALUE": reasons.append( "(Neg) O valor solicitado excede a capacidade financeira estimada " "para o prazo selecionado." ) reasons.append( "(Info) Recomenda-se reduzir o valor mantendo o mesmo prazo." ) return RiskClassification( risk_class="B", reasons=reasons, recommended_max_value=rec_value, ) # Identidade precisa revisão → D if identity_status == "NEEDS_REVIEW": reasons.append( "(Info) Foram detectadas inconsistências que exigem verificação humana " "antes de prosseguir." ) return RiskClassification( risk_class="D", reasons=reasons, ) # Fallback defensivo → D reasons.append( "(Info) Não foi possível concluir a viabilidade financeira automaticamente. " "Encaminhado para análise manual." ) return RiskClassification( risk_class="D", reasons=reasons, )