import os from datetime import datetime from fastapi import FastAPI, HTTPException from groq import Groq from pydantic import BaseModel import pytz from app.config import config, require_env from app.core.gemini_client import get_gemini_model from modules.processing.cie10.RANGES_HTML import CIE10Retriever app = FastAPI() colombia_tz = pytz.timezone("America/Bogota") client_groq = Groq(api_key=require_env(config.GROQ_API_KEY, "GROQ_API_KEY")) # Instancia del RAG retriever de CIE‑10 retriever = CIE10Retriever( faiss_index_path=os.getenv("CIE10_FAISS_INDEX", "faiss_principal_jerarquico"), estructura_json_path=os.getenv("CIE10_ESTRUCTURA_JSON", "cie10_estructura_completa.json"), groq_api_key=config.GROQ_API_KEY, groq_model=config.GROQ_MODEL_CIE10 ) def _es_error_cuota_gemini(error: Exception) -> bool: texto = str(error or "") texto_upper = texto.upper() return "RESOURCE_EXHAUSTED" in texto_upper or "429" in texto_upper or "QUOTA EXCEEDED" in texto_upper def _generar_historia_con_groq(prompt: str, model_name: str) -> str: resp = client_groq.chat.completions.create( model=model_name, temperature=0.0, messages=[ {"role": "system", "content": "Eres un analista medico experto."}, {"role": "user", "content": prompt}, ], ) return resp.choices[0].message.content or "" class HistoriaClinicaRequest(BaseModel): descripcion: str @staticmethod def analizar_historiaclinica(descripcion: str) -> str: descripcion = (descripcion or "") resumen_previo = "" summary_threshold = config.HISTORIA_SUMMARY_THRESHOLD summary_chunk = max(500, config.HISTORIA_SUMMARY_CHUNK) max_chars = config.HISTORIA_MAX_CHARS def _chunk_text(text: str, size: int): return [text[i:i + size] for i in range(0, len(text), size)] def _resumir_largo(texto: str) -> str: instrucciones = ( "Resume la historia clinica manteniendo todos los datos clinicos criticos: " "diagnosticos, procedimientos, medicamentos (con dosis/via/frecuencia si aparece), " "fechas, numeros, hallazgos y contexto relevante. No inventes datos. " "Entrega el resumen en espanol, en texto plano con secciones claras." ) partes = _chunk_text(texto, summary_chunk) res_parcial = [] for idx, parte in enumerate(partes, start=1): prompt_resumen = ( f"{instrucciones}\n\n" f"Parte {idx} de {len(partes)}:\n{parte}" ) if config.LLM_PROVIDER == "gemini": try: model = get_gemini_model( api_key=require_env(config.GEMINI_API_KEY, "GEMINI_API_KEY"), model_name=config.GEMINI_MODEL_DEFAULT, ) resp = model.generate_content( prompt_resumen, generation_config={"temperature": 0.0}, ) res_text = getattr(resp, "text", "") or "" except Exception as gemini_error: if not _es_error_cuota_gemini(gemini_error): raise res_text = _generar_historia_con_groq( prompt_resumen, config.GROQ_MODEL_HISTORIA_LIGHT, ) else: res_text = _generar_historia_con_groq( prompt_resumen, config.GROQ_MODEL_HISTORIA_LIGHT, ) if res_text: res_parcial.append(res_text.strip()) if not res_parcial: return "" consolidado = "\n\n".join(res_parcial) prompt_final = ( "Consolida los siguientes resumenes parciales en un unico resumen clinico " "coherente y completo, preservando diagnosticos, procedimientos, medicamentos, " "fechas, cifras y hallazgos. No omitas datos relevantes.\n\n" f"{consolidado}" ) if config.LLM_PROVIDER == "gemini": try: model = get_gemini_model( api_key=require_env(config.GEMINI_API_KEY, "GEMINI_API_KEY"), model_name=config.GEMINI_MODEL_DEFAULT, ) resp = model.generate_content( prompt_final, generation_config={"temperature": 0.0}, ) return getattr(resp, "text", "") or "" except Exception as gemini_error: if not _es_error_cuota_gemini(gemini_error): raise return _generar_historia_con_groq( prompt_final, config.GROQ_MODEL_HISTORIA_LIGHT, ) return _generar_historia_con_groq( prompt_final, config.GROQ_MODEL_HISTORIA_LIGHT, ) if summary_threshold and len(descripcion) > summary_threshold: try: resumen_previo = _resumir_largo(descripcion) except Exception: resumen_previo = "" if resumen_previo: descripcion_para_prompt = ( "Resumen previo (generado automaticamente para reducir tokens):\n" f"{resumen_previo}" ) else: descripcion_para_prompt = descripcion if max_chars and len(descripcion_para_prompt) > max_chars: descripcion_para_prompt = descripcion_para_prompt[:max_chars] prompt = f""" Eres un asistente de procesamiento de texto médico. Se te proporcionará una historia clínica completa. Tu tarea es extraer TODA la información estructurada del documento y presentarla en HTML válido. **IMPORTANTE:** Extrae TODOS los metadatos del encabezado del documento: - Prestador de servicio / Institución médica - Número de caso - Datos de identificación del paciente (Tipo y número de documento) - Sexo - Edad - Fecha de ingreso - Fecha de nacimiento - Motivo de consulta **ESTRUCTURA DE SALIDA (HTML válido, SIN estilos inline):**

Nombre del paciente

[Nombre completo EXACTO del paciente extraído del documento]

Prestador de servicio

[Nombre de la institución médica]

Número de caso

[Número de caso del documento]

Datos de identificación del paciente

[Tipo y número de documento, ej: CC - 29465123]

Sexo

[Sexo del paciente]

Edad

[Edad del paciente con formato "XX AÑOS"]

Fecha de nacimiento

[Fecha de nacimiento en formato DD/MM/YYYY]

Fecha de ingreso

[Fecha de ingreso en formato DD/MM/YYYY]

Motivo de consulta

[Motivo de consulta extraído]

Resumen

[Resumen clínico completo en un párrafo]

Diagnósticos

  1. [Diagnóstico 1 con código si está disponible]
  2. [Diagnóstico 2 con código si está disponible]

Procedimientos

  1. [Procedimiento 1 - Código de 6 dígitos si está en "Ordenes Generadas"]
  2. [Procedimiento 2 - Código de 6 dígitos si está en "Ordenes Generadas"]

Medicamentos administrados

  1. [Código 4 dígitos] - [Nombre medicamento + dosis + vía + frecuencia]
  2. [Código 4 dígitos] - [Nombre medicamento + dosis + vía + frecuencia]
**NOTAS IMPORTANTES:** 1. Los códigos de medicamentos son de 4 dígitos (ejemplo: 5656, 6491) 2. Los códigos de procedimientos son de 6 dígitos (ejemplo: 890201) 3. Extrae TODOS los medicamentos listados en el documento 4. Si algún campo no está disponible, coloca "No especificado" 5. Mantén el formato EXACTO de las fechas como aparecen en el documento Historia clínica: {descripcion_para_prompt} """ try: if config.LLM_PROVIDER == "gemini": if not config.GEMINI_API_KEY: raise HTTPException(status_code=500, detail="GEMINI_API_KEY no configurada.") try: model = get_gemini_model( api_key=require_env(config.GEMINI_API_KEY, "GEMINI_API_KEY"), model_name=config.GEMINI_MODEL_HISTORIA, ) response = model.generate_content( prompt, generation_config={"temperature": 0.0}, ) return getattr(response, "text", "") or "" except Exception as gemini_error: if not _es_error_cuota_gemini(gemini_error): raise return _generar_historia_con_groq( prompt, config.GROQ_MODEL_HISTORIA, ) modelo_groq = config.GROQ_MODEL_HISTORIA if config.LLM_PROVIDER != "gemini" and config.HISTORIA_LARGE_THRESHOLD: if len(descripcion) > config.HISTORIA_LARGE_THRESHOLD: modelo_groq = config.GROQ_MODEL_HISTORIA_LIGHT resp = client_groq.chat.completions.create( model=modelo_groq, temperature=0.0, messages=[ {"role": "system", "content": "Eres un asistente de procesamiento de texto médico. Usa HTML válido."}, {"role": "user", "content": prompt}, ], ) return resp.choices[0].message.content except Exception as e: proveedor = "gemini" if config.LLM_PROVIDER == "gemini" else "groq" raise HTTPException(status_code=500, detail=f"Error usando {proveedor}: {e}") @staticmethod def guardar_historia(descripcion: str, analisis: str) -> str: # Desactivado: esta colección no se usa en flujos posteriores. return ""