from typing import List, Dict, Any, Optional from datetime import date from collections import Counter, defaultdict from src.infra.document_analysis.analyzers.document_analyst import DocumentAnalyst from src.infra.document_analysis.reports.company_consistency_report import ( build_company_consistency_report, ) from src.core.scoring.scoring_engine import ( calculate_credit_score, ScoringInput, ) from src.core.decision.decision_engine import make_credit_decision def _get_canonical_value_weighted(field: str, profiles: List[Dict], consistency: Dict[str, Any]) -> Any: """ Extrai o valor canónico para um campo a partir dos perfis (incluindo input), utilizando ponderação pela confiança de cada fonte. O relatório de consistência é usado para validar a homogeneidade. """ # Separa documentos (exclui input) doc_profiles = [p for p in profiles if p["source"] != "input"] # Se não houver documentos, usa input (se existir) if not doc_profiles: input_profile = next((p for p in profiles if p["source"] == "input"), None) return input_profile.get(field) if input_profile else None # Para cada valor, soma a confiança dos documentos que o apresentam confidence_sum = defaultdict(float) for p in doc_profiles: val = p.get(field) if val is not None: confidence_sum[val] += p.get("confidence", 1.0) # confiança padrão = 1 se não informada if not confidence_sum: return None # Escolhe o valor com maior soma de confiança best_val, _ = max(confidence_sum.items(), key=lambda x: x[1]) # Validações adicionais baseadas no relatório de consistência if field == "nuit" and not consistency.get("identity", {}).get("nuit_consistent"): # Se o NUIT for inconsistente, não podemos confiar; retorna None return None if field == "company_name": sim = consistency.get("identity", {}).get("company_name_similarity", 0) if sim < 0.8: # Nomes muito divergentes, retorna None para evitar erro return None # Para data de incorporação, não aplicamos validação extra aqui, # mas a consistência temporal já é usada no pipeline return best_val def _extract_bank_features(document_results: List[Any]) -> Dict[str, Any]: """ Extrai informações bancárias de documentos do tipo extrato. Retorna um dicionário com chaves padronizadas, incluindo flag de validade. """ bank_features = { "monthly_revenue": None, "average_balance": None, "total_credits": None, "days_analyzed": None, "records_count": None, "recommended_limits": None, "statement_score": None, "fluxo_liquido": None, "analysis_status": None, "has_valid_extrato": False, # flag adicionada } for result in document_results: if result.document_type_detected == "extrato_bancario": extrato_features = result.features.get("bank_statement_features", {}) if extrato_features: bank_features["monthly_revenue"] = extrato_features.get("media_entradas_mensal") bank_features["average_balance"] = extrato_features.get("saldo_medio") bank_features["total_credits"] = extrato_features.get("total_creditos") bank_features["days_analyzed"] = extrato_features.get("period_days") bank_features["records_count"] = extrato_features.get("records_count") bank_features["recommended_limits"] = extrato_features.get("recommended_max_by_term") bank_features["statement_score"] = extrato_features.get("statement_score") bank_features["fluxo_liquido"] = extrato_features.get("fluxo_liquido") bank_features["analysis_status"] = extrato_features.get("analysis_status") # Define a flag de validade: status OK e pelo menos um registro if (bank_features["analysis_status"] == "OK" and bank_features.get("records_count", 0) > 0): bank_features["has_valid_extrato"] = True break # considera apenas o primeiro extrato (pode ser melhorado no futuro) return bank_features def run_credit_pipeline( *, files: List[str], input_company_name: Optional[str] = None, input_nuit: Optional[str] = None, input_meses_atividade: Optional[int] = None, faturamento_mensal: Optional[float] = None, valor_credito: float, prazo_dias: int, ) -> Dict[str, Any]: """ Pipeline oficial end-to-end do Credit Scoring. Funciona exclusivamente com documentos, usando inputs manuais apenas como fallback. O relatório de consistência é gerado apenas a partir dos documentos. """ # ===================================================== # 1️⃣ ANÁLISE DOCUMENTAL # ===================================================== analyst = DocumentAnalyst() document_results = analyst.analyze_documents(files) # ===================================================== # 2️⃣ COLETA DE COMPANY PROFILES DOS DOCUMENTOS # ===================================================== document_profiles = [] # apenas dos documentos all_profiles = [] # inclui input (para fallback) for r in document_results: profile = r.features.get("company_profile") if profile: # Converte para dicionário e garante que 'confidence' exista profile_dict = profile.__dict__ if "confidence" not in profile_dict: profile_dict["confidence"] = 1.0 # fallback document_profiles.append(profile_dict) all_profiles.append(profile_dict) # ===================================================== # 3️⃣ PERFIL DE INPUT (APENAS PARA FALLBACK) # ===================================================== input_profile = { "source": "input", "company_name": input_company_name, "nuit": input_nuit, "entity_type": None, "incorporation_date": None, "company_age_months": input_meses_atividade, "issuer": "user_input", "confidence": 1.0 if (input_company_name or input_nuit) else 0.0, "raw": {}, } all_profiles.append(input_profile) # incluído apenas para fallback # ===================================================== # 4️⃣ CONSISTÊNCIA LEGAL DA EMPRESA (APENAS DOCUMENTOS) # ===================================================== company_consistency = build_company_consistency_report(profiles=document_profiles) # ===================================================== # 5️⃣ EXTRAI VALORES CANÓNICOS PARA O SCORING (COM FALLBACK PONDERADO) # ===================================================== canonical_name = _get_canonical_value_weighted("company_name", all_profiles, company_consistency) canonical_nuit = _get_canonical_value_weighted("nuit", all_profiles, company_consistency) # Para incorporação, escolhemos a data com maior confiança (já feito pela função) canonical_incorporation = _get_canonical_value_weighted("incorporation_date", all_profiles, company_consistency) # Meses de atividade: calcula a partir da data de constituição, se disponível if canonical_incorporation: today = date.today() delta_months = (today.year - canonical_incorporation.year) * 12 + (today.month - canonical_incorporation.month) canonical_months = max(0, delta_months) else: # Fallback: input ou valor de algum documento (com maior confiança) # Reutilizamos a lógica ponderada, mas o campo é company_age_months canonical_months = _get_canonical_value_weighted("company_age_months", all_profiles, company_consistency) if canonical_months is None: canonical_months = input_meses_atividade or 0 # ===================================================== # 6️⃣ EXTRAI INFORMAÇÕES BANCÁRIAS DOS DOCUMENTOS # ===================================================== bank_features = _extract_bank_features(document_results) # Se o usuário não forneceu faturamento, tenta usar o do extrato (apenas se válido) if faturamento_mensal is None and bank_features["has_valid_extrato"]: faturamento_mensal = bank_features.get("monthly_revenue") # ===================================================== # 7️⃣ SCORING NUMÉRICO # ===================================================== scoring_input = ScoringInput( company_name=canonical_name, nuit=canonical_nuit, meses_atividade=canonical_months or 0, faturamento_mensal=faturamento_mensal, valor_credito=valor_credito, prazo_dias=prazo_dias, documentos_submetidos=len(document_results), ) score_output = calculate_credit_score( scoring_input=scoring_input, document_results=document_results, company_consistency_report=company_consistency, bank_features=bank_features, # agora inclui has_valid_extrato e records_count ) # ===================================================== # 8️⃣ DECISÃO FINAL # ===================================================== decision = make_credit_decision( score=score_output.score, risk_class=score_output.risk_class, confidence=score_output.confidence, reasons=score_output.reasons, features_used={ **score_output.features_used, "company_consistency": company_consistency, "valor_credito": valor_credito, "prazo_dias": prazo_dias, "bank_features": bank_features, }, ) # ===================================================== # 9️⃣ PAYLOAD FINAL # ===================================================== return { "decision": decision, "risk": { "risk_class": score_output.risk_class, "recommended_max_value": score_output.recommended_max_value, "recommended_term_days": score_output.recommended_term_days, }, "score": score_output, "company_consistency": company_consistency, "documents": document_results, "bank_features": bank_features, }