"""Generación de workbooks Excel para epicrisis consolidadas por lote.""" from __future__ import annotations import math import re from collections.abc import Iterable from dataclasses import dataclass from io import BytesIO from typing import Any, Final from openpyxl import Workbook from openpyxl.styles import Alignment, Border, Font, PatternFill, Side from openpyxl.utils import get_column_letter from openpyxl.worksheet.properties import PageSetupProperties from openpyxl.worksheet.worksheet import Worksheet from app.case_epicrisis.application.utils import ( RECOMMENDATION_CATEGORY_LABELS, build_selected_pdf_procedimientos, format_ayuda_diagnostica_interpretacion_exportable, format_ayuda_diagnostica_nombre_exportable, ) @dataclass(frozen=True) class RowHeightProfile: chars_per_width: float line_points: float padding: float max_height: float uppercase_ratio_threshold: float = 1.0 uppercase_width_factor: float = 1.0 long_paragraph_threshold: int = 0 long_paragraph_bonus_lines: int = 0 @dataclass(frozen=True) class HeightEntry: value: str start_col: int end_col: int profile: str _INVALID_SHEET_CHARS = re.compile(r"[:\\/?*\[\]]") _HTML_BREAK_PATTERN = re.compile(r"<\s*br\s*/?\s*>", flags=re.IGNORECASE) _HTML_TAG_PATTERN = re.compile(r"<[^>]+>") _CIE10_PATTERN = re.compile(r"^[A-TV-Z]\d{2}[0-9A-Z]?(?:\.[0-9A-Z]{1,2})?$", flags=re.IGNORECASE) _DECORATIVE_LINE_PATTERN = re.compile(r"^[\s\-_=╔╗╚╝║═•*]+$") _MARKDOWN_BULLET_PATTERN = re.compile(r"^\s*[•*-]\s+") _MARKDOWN_EMPHASIS_PATTERN = re.compile(r"\*\*(.*?)\*\*") _THIN_SIDE = Side(style="thin", color="D8E1EB") _BORDER = Border(left=_THIN_SIDE, right=_THIN_SIDE, top=_THIN_SIDE, bottom=_THIN_SIDE) _TITLE_FILL = PatternFill("solid", fgColor="1F4E78") _SECTION_FILL = PatternFill("solid", fgColor="D9EAF7") _HEADER_FILL = PatternFill("solid", fgColor="EAF3FA") _CARD_FILL = PatternFill("solid", fgColor="F8FBFE") _TEXT_FORMAT = "@" _CASE_SHEET_WIDTHS = { "A": 16, "B": 28, "C": 28, "D": 28, "E": 12, "F": 12, "G": 12, "H": 16, } _SECTION_GAP = 1 _ROW_HEIGHT_BASE = 15.0 _ROW_HEIGHT_SECTION = 18.0 _ROW_HEIGHT_TITLE = 20.0 _HEIGHT_PROFILE_SHORT_VALUE = "short_value" _HEIGHT_PROFILE_TABLE_VALUE = "table_value" _HEIGHT_PROFILE_RICH_VALUE = "rich_value" _HEIGHT_PROFILE_NARRATIVE_BLOCK = "narrative_block" _HEIGHT_PROFILE_ANALYSIS_BLOCK = "analysis_block" _ROW_HEIGHT_PROFILES = { _HEIGHT_PROFILE_SHORT_VALUE: RowHeightProfile( chars_per_width=1.55, line_points=13.0, padding=2.0, max_height=60.0, ), _HEIGHT_PROFILE_TABLE_VALUE: RowHeightProfile( chars_per_width=1.35, line_points=13.0, padding=4.0, max_height=90.0, uppercase_ratio_threshold=0.55, uppercase_width_factor=0.92, long_paragraph_threshold=120, long_paragraph_bonus_lines=1, ), _HEIGHT_PROFILE_RICH_VALUE: RowHeightProfile( chars_per_width=1.02, line_points=13.0, padding=6.0, max_height=320.0, uppercase_ratio_threshold=0.55, uppercase_width_factor=0.88, long_paragraph_threshold=140, long_paragraph_bonus_lines=1, ), _HEIGHT_PROFILE_NARRATIVE_BLOCK: RowHeightProfile( chars_per_width=1.0, line_points=13.0, padding=8.0, max_height=340.0, uppercase_ratio_threshold=0.55, uppercase_width_factor=0.84, long_paragraph_threshold=160, long_paragraph_bonus_lines=1, ), _HEIGHT_PROFILE_ANALYSIS_BLOCK: RowHeightProfile( chars_per_width=0.9, line_points=13.0, padding=10.0, max_height=540.0, uppercase_ratio_threshold=0.5, uppercase_width_factor=0.84, long_paragraph_threshold=160, long_paragraph_bonus_lines=2, ), } _WIDE_VALUE_KEYS = { "descripcion", "procedimiento", "analisis_clinico", "hallazgos_quirurgicos", "descripcion_procedimiento", "glosa_analisis", "resumen", } _LABELS = { "case_key": "Case key", "nombre_paciente": "Nombre paciente", "numero_caso": "Número de caso", "identificacion_paciente": "Identificación paciente", "procedimiento": "Procedimiento", "fuente": "Fuente", "documento_id": "Documento ID", "documento_nombre": "Documento", "fecha_analisis": "Fecha análisis", "tipo": "Tipo", "nombre": "Ayuda", "concepto": "Concepto", "ordenado": "Ordenado", "interpretado": "Interpretado", "facturado": "Facturado", "glosado": "Glosado", "estado_interpretacion": "Estado interpretación", "alertas": "Alertas", "exportable": "Exportable", "trazabilidad": "Trazabilidad", "codigo_soat": "Código SOAT", "codigo_referencia": "CUM/IUM", "codigo_facturacion": "Código interno", "categoria": "Categoría", "indicacion": "Indicación", "duracion": "Duración", "fecha": "Fecha", "cie10_principal": "CIE10 principal", "cie10_secundarios": "CIE10 secundarios", "cups_principal": "CUPS principal", "cups_alternativos": "CUPS alternativos", "analisis_clinico": "Análisis clínico", } _LABEL_WORD_REPLACEMENTS = { "analisis": "análisis", "area": "área", "atencion": "atención", "cirugia": "cirugía", "clinica": "clínica", "clinico": "clínico", "codigo": "código", "decripcion": "descripción", "descripcion": "descripción", "diagnostico": "diagnóstico", "diagnosticos": "diagnósticos", "dias": "días", "evolucion": "evolución", "farmacologica": "farmacológica", "farmacologico": "farmacológico", "identificacion": "identificación", "informacion": "información", "interaccion": "interacción", "justificacion": "justificación", "medica": "médica", "medicacion": "medicación", "medico": "médico", "numero": "número", "patologia": "patología", "posologia": "posología", "quirurgica": "quirúrgica", "quirurgico": "quirúrgico", "quirurgicas": "quirúrgicas", "quirurgicos": "quirúrgicos", "tecnica": "técnica", "tecnico": "técnico", "via": "vía", } _NARRATIVE_SECTION_TITLES = { "Resumen", "Hallazgos quirúrgicos", "Descripción del procedimiento", } _ANALYSIS_SECTION_TITLES = {"Análisis clínico", "Glosa analítica"} _RICH_VALUE_LABELS = { "CIE10 principal", "CIE10 secundarios", "CUPS principal", "CUPS alternativos", } _COMPACT_VALUE_LABELS = { "Case key", "Nombre paciente", "Número de caso", "Identificación paciente", "Procedimiento", "Código SOAT", "Descripción", "Fuente", "Origen", } NO_INFORMATION_TEXT: Final = "" @dataclass(frozen=True) class BatchEpicrisisExcelCaseRow: case_key: str patient_name: str patient_id: str case_number: str procedure_description: str epicrisis_status: str included_in_excel: bool omission_reason: str context: dict[str, Any] | None = None class BatchEpicrisisExcelWorkbookBuilder: """Construye un workbook con un resumen del lote y una hoja por caso.""" def build( self, *, batch_id: str, rows: list[BatchEpicrisisExcelCaseRow], ) -> bytes: workbook = Workbook() used_titles = {"Resumen"} case_sheet_map: dict[str, str] = {} for row in rows: if not row.included_in_excel or not row.context: continue sheet_title = self._unique_sheet_title( row.case_key or row.case_number or "Caso", used_titles, ) used_titles.add(sheet_title) case_sheet_map[row.case_key] = sheet_title summary_sheet = workbook.active summary_sheet.title = "Resumen" self._build_summary_sheet( summary_sheet, batch_id=batch_id, rows=rows, case_sheet_map=case_sheet_map, ) for row in rows: if not row.included_in_excel or not row.context: continue sheet_title = case_sheet_map[row.case_key] sheet = workbook.create_sheet(title=sheet_title) self._build_case_sheet(sheet, batch_id=batch_id, row=row) buffer = BytesIO() workbook.save(buffer) return buffer.getvalue() def _build_summary_sheet( self, sheet: Worksheet, *, batch_id: str, rows: list[BatchEpicrisisExcelCaseRow], case_sheet_map: dict[str, str], ) -> None: headers = self._summary_headers() start_row = 4 self._configure_summary_columns(sheet) self._write_summary_titles(sheet, batch_id=batch_id) self._write_summary_headers(sheet, row_index=start_row, headers=headers) for row_index, item in enumerate(rows, start=start_row + 1): self._write_summary_data_row( sheet, row_index=row_index, item=item, case_sheet_map=case_sheet_map, ) sheet.freeze_panes = "A5" sheet.sheet_view.showGridLines = False self._configure_print(sheet, end_col=8, orientation=sheet.ORIENTATION_LANDSCAPE) def _configure_summary_columns(self, sheet: Worksheet) -> None: widths = [34, 28, 24, 18, 38, 20, 16, 42] for index, width in enumerate(widths, start=1): sheet.column_dimensions[get_column_letter(index)].width = width def _write_summary_titles(self, sheet: Worksheet, *, batch_id: str) -> None: self._merge_row( sheet, 1, f"Reporte de epicrisis por lote - {batch_id}", fill=_TITLE_FILL, font=Font(bold=True, color="FFFFFF", size=14), end_col=8, minimum=_ROW_HEIGHT_TITLE, ) self._merge_row( sheet, 2, "Resumen consolidado por case_key", fill=_HEADER_FILL, font=Font(bold=True, color="1F1F1F"), end_col=8, minimum=_ROW_HEIGHT_SECTION, ) def _summary_headers(self) -> list[str]: return [ "case_key", "Paciente", "Identificación paciente", "Número de caso", "Procedimiento", "Estado epicrisis", "Incluido en Excel", "Motivo omisión", ] def _write_summary_headers(self, sheet: Worksheet, *, row_index: int, headers: list[str]) -> None: for index, header in enumerate(headers, start=1): cell = sheet.cell(row=row_index, column=index, value=header) cell.font = Font(bold=True) cell.fill = _SECTION_FILL cell.border = _BORDER cell.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True) self._set_row_height( sheet, row_index, [ self._height_entry(header, index, index, profile=_HEIGHT_PROFILE_SHORT_VALUE) for index, header in enumerate(headers, start=1) ], ) def _summary_row_values(self, item: BatchEpicrisisExcelCaseRow) -> list[str]: return [ item.case_key or NO_INFORMATION_TEXT, item.patient_name or NO_INFORMATION_TEXT, self._patient_identification(item) or NO_INFORMATION_TEXT, item.case_number or NO_INFORMATION_TEXT, item.procedure_description or NO_INFORMATION_TEXT, item.epicrisis_status or NO_INFORMATION_TEXT, "Sí" if item.included_in_excel else "No", item.omission_reason or "", ] def _write_summary_data_row( self, sheet: Worksheet, *, row_index: int, item: BatchEpicrisisExcelCaseRow, case_sheet_map: dict[str, str], ) -> None: row_candidates: list[HeightEntry] = [] for column_index, value in enumerate(self._summary_row_values(item), start=1): cell = sheet.cell(row=row_index, column=column_index, value=value) cell.border = _BORDER cell.alignment = Alignment(vertical="top", wrap_text=True) row_candidates.append( self._height_entry( str(value or ""), column_index, column_index, profile=_HEIGHT_PROFILE_SHORT_VALUE, ) ) self._apply_summary_hyperlink(cell, column_index=column_index, item=item, case_sheet_map=case_sheet_map) self._set_row_height(sheet, row_index, row_candidates) def _apply_summary_hyperlink( self, cell, *, column_index: int, item: BatchEpicrisisExcelCaseRow, case_sheet_map: dict[str, str], ) -> None: if column_index != 1: return target_sheet = case_sheet_map.get(item.case_key) if not target_sheet: return cell.hyperlink = f"#'{target_sheet}'!A1" cell.style = "Hyperlink" def _build_case_sheet( self, sheet: Worksheet, *, batch_id: str, row: BatchEpicrisisExcelCaseRow, ) -> None: context = row.context or {} for column, width in _CASE_SHEET_WIDTHS.items(): sheet.column_dimensions[column].width = width current_row = 1 self._merge_row( sheet, current_row, f"Epicrisis del caso {row.case_key}", fill=_TITLE_FILL, font=Font(bold=True, color="FFFFFF", size=14), minimum=_ROW_HEIGHT_TITLE, ) current_row += 1 self._merge_row( sheet, current_row, f"Lote: {batch_id} | Número de caso: {row.case_number or ''} | Paciente: {row.patient_name or ''}", fill=_HEADER_FILL, font=Font(bold=True), minimum=_ROW_HEIGHT_SECTION, ) current_row += 1 identification = { "case_key": row.case_key or NO_INFORMATION_TEXT, "nombre_paciente": context.get("nombre_paciente") or row.patient_name or NO_INFORMATION_TEXT, "numero_caso": context.get("case_number") or row.case_number or NO_INFORMATION_TEXT, "identificacion_paciente": self._patient_identification(row) or NO_INFORMATION_TEXT, "procedimiento": row.procedure_description or NO_INFORMATION_TEXT, } current_row = self._write_key_value_section(sheet, current_row, "Identificación", identification) current_row = self._write_value_section( sheet, current_row, "Metadatos HC", context.get("metadatos_hc") ) recommendations = self._serialize_recommendations(context.get("recomendaciones_medicas")) if recommendations: current_row = self._write_value_section( sheet, current_row, "Recomendaciones médicas", recommendations ) current_row = self._write_value_section( sheet, current_row, "Antecedentes HC", context.get("antecedentes_hc") or context.get("antecedentes") ) if context.get("curation_schema_version"): publishable_statuses = {"validado_automaticamente", "aprobado", "corregido"} diagnoses = [ { "codigo_cie10": item.get("code") or "", "descripcion": item.get("description") or "", } for item in context.get("diagnosticos_curados") or [] if isinstance(item, dict) and item.get("status") in publishable_statuses ] current_row = self._write_value_section( sheet, current_row, "Diagnósticos CIE-10", diagnoses, ) procedures = build_selected_pdf_procedimientos(context) for classification, title in ( ("quirurgico", "Procedimientos - Quirúrgicos"), ("no_quirurgico", "Procedimientos - No quirúrgicos"), ): selected = [ { "codigo_cups": item.get("codigo_cups") or "", "descripcion": item.get("descripcion") or "", } for item in procedures if isinstance(item, dict) and item.get("clasificacion") == classification ] current_row = self._write_value_section(sheet, current_row, title, selected) current_row = self._write_value_section( sheet, current_row, "Medicamentos", self._serialize_medications(context), ) surgical_detail = { "hallazgos_quirurgicos": context.get("hallazgos_quirurgicos") or "", "descripcion_procedimiento": context.get("descripcion_procedimiento") or "", } current_row = self._write_value_section(sheet, current_row, "Detalle quirúrgico", surgical_detail) current_row = self._write_value_section( sheet, current_row, "Ayudas diagnósticas", self._serialize_ayudas_diagnosticas(context.get("ayudas_diagnosticas")), ) sheet.freeze_panes = "A4" sheet.sheet_view.showGridLines = False self._configure_print( sheet, end_col=max(4, sheet.max_column), orientation=sheet.ORIENTATION_PORTRAIT, left_margin=0.25, right_margin=0.25, ) def _write_documents_section( self, sheet: Worksheet, row_index: int, context: dict[str, Any], ) -> int: sources = [] for key, label in ( ("historia", "Historia clínica"), ("quirurgico", "Documento quirúrgico"), ("factura", "Factura"), ): serialized = self._serialize_document_summary(context.get(key), label) if serialized: sources.append(serialized) for key, label in (("radiologia", "Radiología"), ("laboratorio", "Laboratorio"), ("generico", "Genérico")): for item in context.get(key) or []: serialized = self._serialize_document_summary(item, label) if serialized: sources.append(serialized) return self._write_value_section(sheet, row_index, "Fuentes documentales", sources) def _serialize_document_summary(self, item: Any, label: str) -> dict[str, str] | None: if not isinstance(item, dict): return None return { "fuente": label, "documento_id": str(item.get("_id") or "").strip() or NO_INFORMATION_TEXT, "fecha_analisis": str(item.get("fecha_analisis") or "").strip() or NO_INFORMATION_TEXT, } def _serialize_ayudas_diagnosticas(self, items: Any) -> list[dict[str, Any]]: result: list[dict[str, Any]] = [] for item in items or []: if not isinstance(item, dict): continue if item.get("exportable") is False: continue result.append( { "tipo": item.get("tipo") or "", "nombre": format_ayuda_diagnostica_nombre_exportable(item), "interpretacion": format_ayuda_diagnostica_interpretacion_exportable(item), } ) return result def _serialize_medications(self, context: dict[str, Any]) -> Any: items = context.get("medicamentos_caso") or context.get("medicamentos_hc") if not isinstance(items, list) or not items: return context.get("medicamentos_hc_display") or [] result: list[dict[str, Any]] = [] for item in items: if not isinstance(item, dict): continue statuses = { str(status or "").strip().casefold() for status in (item.get("estados") or [item.get("tipo_uso")]) } if "administrado" not in statuses: continue name = str(item.get("nombre") or item.get("medicamento") or "").strip() presentation = str(item.get("presentacion") or "").strip() medication = " · ".join(value for value in (name, presentation) if value) row = { "codigo_referencia": item.get("codigo_referencia") or "", "medicamento": medication, "posologia": item.get("posologia") or "", "dosis": item.get("dosis") or "", "cantidad": item.get("cantidad") if item.get("cantidad") is not None else "", } if any(value not in (None, "") for value in row.values()): result.append(row) return result def _serialize_recommendations(self, items: Any) -> list[dict[str, str]]: result: list[dict[str, str]] = [] raw_items = [item for item in items if isinstance(item, dict)] if isinstance(items, list) else [] for category, label in RECOMMENDATION_CATEGORY_LABELS.items(): for item in raw_items: if str(item.get("categoria") or "") != category: continue row = { "categoria": label, "indicacion": str(item.get("indicacion") or "").strip(), "duracion": str(item.get("duracion") or "").strip(), "fecha": str(item.get("fecha") or "").strip(), } if any(row[key] for key in ("indicacion", "duracion", "fecha")): result.append(row) return result def _format_boolean_state(self, value: Any) -> str: return "Sí" if bool(value) else "No" def _format_interpretation_state(self, value: Any) -> str: normalized = str(value or "").strip().lower() if not normalized: return NO_INFORMATION_TEXT labels = { "no_interpretado": "No interpretado", "pendiente_revision": "Pendiente revisión", "interpretado": "Interpretado", } return labels.get(normalized, normalized.replace("_", " ").capitalize()) def _write_key_value_section( self, sheet: Worksheet, row_index: int, title: str, data: dict[str, Any], ) -> int: self._merge_row( sheet, row_index, title, fill=_SECTION_FILL, font=Font(bold=True), minimum=_ROW_HEIGHT_SECTION, ) row_index += 1 row_index = self._write_key_value_rows(sheet, row_index, data) return self._next_section_row(row_index) def _write_key_value_rows( self, sheet: Worksheet, row_index: int, data: dict[str, Any], ) -> int: for key, value in data.items(): label = self._humanize_key(key) serialized = self._stringify(value) self._write_labeled_value_row( sheet, row_index, label, serialized, profile=self._profile_for_labeled_value(label, serialized), ) row_index += 1 return row_index def _write_value_section( self, sheet: Worksheet, row_index: int, title: str, value: Any, ) -> int: normalized = self._normalize_value(value) section_end_col = 4 if self._is_dict_table(normalized): section_end_col = max(section_end_col, len(self._collect_columns(normalized))) self._merge_row( sheet, row_index, title, fill=_SECTION_FILL, font=Font(bold=True), end_col=section_end_col, minimum=_ROW_HEIGHT_SECTION, ) row_index += 1 if not normalized: row_index = self._write_empty_value_row(sheet, row_index) return self._next_section_row(row_index) if self._is_dict_table(normalized): row_index = self._write_tabular_value_rows(sheet, row_index, normalized) return self._next_section_row(row_index) if isinstance(normalized, dict): row_index = self._write_key_value_rows(sheet, row_index, normalized) return self._next_section_row(row_index) if isinstance(normalized, list): row_index = self._write_list_value_rows(sheet, row_index, normalized) return self._next_section_row(row_index) row_index = self._write_scalar_value_row( sheet, row_index, normalized, profile=self._profile_for_section_value(title), ) return self._next_section_row(row_index) def _write_empty_value_row(self, sheet: Worksheet, row_index: int) -> int: self._merge_range(sheet, row_index, 1, row_index, 4, NO_INFORMATION_TEXT) self._set_row_height( sheet, row_index, [self._height_entry(NO_INFORMATION_TEXT, 1, 4, profile=_HEIGHT_PROFILE_SHORT_VALUE)], ) return row_index + 1 def _is_dict_table(self, normalized: Any) -> bool: return ( isinstance(normalized, list) and bool(normalized) and all(isinstance(item, dict) for item in normalized) ) def _write_tabular_value_rows( self, sheet: Worksheet, row_index: int, normalized: list[dict[str, Any]], ) -> int: columns = self._collect_columns(normalized) column_layout = self._build_table_layout(columns) row_candidates = self._write_tabular_header(sheet, row_index, columns, column_layout) self._set_row_height(sheet, row_index, row_candidates, minimum=_ROW_HEIGHT_SECTION) row_index += 1 for item in normalized: row_index = self._write_tabular_data_row( sheet, row_index, item, columns, column_layout, ) return row_index def _write_tabular_header( self, sheet: Worksheet, row_index: int, columns: list[str], column_layout: list[tuple[int, int]], ) -> list[HeightEntry]: row_candidates: list[HeightEntry] = [] for (start_col, end_col), column_name in zip(column_layout, columns, strict=False): header = self._humanize_key(column_name) self._merge_range( sheet, row_index, start_col, row_index, end_col, header, fill=_HEADER_FILL, font=Font(bold=True), ) cell = sheet.cell(row=row_index, column=start_col) cell.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True) row_candidates.append( self._height_entry(header, start_col, end_col, profile=_HEIGHT_PROFILE_SHORT_VALUE) ) return row_candidates def _write_tabular_data_row( self, sheet: Worksheet, row_index: int, item: dict[str, Any], columns: list[str], column_layout: list[tuple[int, int]], ) -> int: data_candidates = [] for (start_col, end_col), column_name in zip(column_layout, columns, strict=False): serialized = self._stringify(item.get(column_name)) self._merge_range( sheet, row_index, start_col, row_index, end_col, serialized, ) cell = sheet.cell(row=row_index, column=start_col) cell.border = _BORDER cell.alignment = Alignment(vertical="top", wrap_text=True) if column_name in { "codigo", "codigo_soat", "codigo_referencia", "codigo_facturacion", }: cell.number_format = _TEXT_FORMAT data_candidates.append( self._height_entry( serialized, start_col, end_col, profile=self._profile_for_table_value(column_name, serialized), ) ) self._set_row_height(sheet, row_index, data_candidates) return row_index + 1 def _write_list_value_rows( self, sheet: Worksheet, row_index: int, normalized: list[Any], ) -> int: for item in normalized: serialized = self._stringify(item) self._merge_range(sheet, row_index, 1, row_index, 4, serialized) self._set_row_height( sheet, row_index, [self._height_entry(serialized, 1, 4, profile=self._profile_for_full_width_value(serialized))], ) row_index += 1 return row_index def _write_scalar_value_row( self, sheet: Worksheet, row_index: int, normalized: Any, *, profile: str, ) -> int: serialized = self._stringify(normalized) self._merge_range(sheet, row_index, 1, row_index, 4, serialized) self._set_row_height(sheet, row_index, [self._height_entry(serialized, 1, 4, profile=profile)]) return row_index + 1 def _write_soat_sections(self, sheet: Worksheet, row_index: int, items: Any) -> int: soat_rows, qx_rows = self._split_soat_rows(items) row_index = self._write_soat_card_section( sheet, row_index, title="Códigos desde SOAT", items=soat_rows, qx_mode=False, ) row_index = self._write_soat_card_section( sheet, row_index, title="Hallazgos quirúrgicos codificados", items=qx_rows, qx_mode=True, ) return row_index def _write_soat_card_section( self, sheet: Worksheet, row_index: int, *, title: str, items: list[dict[str, Any]], qx_mode: bool, ) -> int: self._merge_row( sheet, row_index, title, fill=_SECTION_FILL, font=Font(bold=True), minimum=_ROW_HEIGHT_SECTION, ) row_index += 1 if not items: self._merge_range(sheet, row_index, 1, row_index, 4, NO_INFORMATION_TEXT) self._set_row_height( sheet, row_index, [self._height_entry(NO_INFORMATION_TEXT, 1, 4, profile=_HEIGHT_PROFILE_SHORT_VALUE)], ) return self._next_section_row(row_index + 1) for item in items: row_index = self._write_soat_card(sheet, row_index, item=item, qx_mode=qx_mode) return self._next_section_row(row_index) def _write_soat_card( self, sheet: Worksheet, row_index: int, *, item: dict[str, Any], qx_mode: bool, ) -> int: start_row = row_index card_title = "Registro de hallazgos QX" if qx_mode else "Registro SOAT" self._merge_range( sheet, row_index, 1, row_index, 4, card_title, fill=_HEADER_FILL, font=Font(bold=True), ) self._set_row_height( sheet, row_index, [self._height_entry(card_title, 1, 4, profile=_HEIGHT_PROFILE_SHORT_VALUE)], minimum=_ROW_HEIGHT_SECTION, ) row_index += 1 if qx_mode: row_index = self._write_labeled_value_row( sheet, row_index, "Origen", "Hallazgos quirúrgicos codificados", profile=_HEIGHT_PROFILE_SHORT_VALUE, value_fill=_CARD_FILL, ) else: row_index = self._write_labeled_value_row( sheet, row_index, "Código SOAT", self._stringify(self._soat_code(item)), as_text=True, profile=_HEIGHT_PROFILE_SHORT_VALUE, value_fill=_CARD_FILL, ) description = self._display_description(item) row_index = self._write_labeled_value_row( sheet, row_index, "Descripción", description, profile=self._profile_for_labeled_value("Descripción", description), value_fill=_CARD_FILL, ) source = self._display_source(item, qx_mode=qx_mode) row_index = self._write_labeled_value_row( sheet, row_index, "Fuente", source, profile=_HEIGHT_PROFILE_SHORT_VALUE, value_fill=_CARD_FILL, ) analysis_title = self._humanize_key("analisis_clinico") self._merge_range( sheet, row_index, 1, row_index, 4, analysis_title, fill=_HEADER_FILL, font=Font(bold=True), ) self._set_row_height( sheet, row_index, [self._height_entry(analysis_title, 1, 4, profile=_HEIGHT_PROFILE_SHORT_VALUE)], minimum=_ROW_HEIGHT_SECTION, ) row_index += 1 analysis = self._clean_llm_text(item.get("analisis_clinico")) self._merge_range( sheet, row_index, 1, row_index, 4, analysis or NO_INFORMATION_TEXT, fill=_CARD_FILL, ) self._set_row_height( sheet, row_index, [ self._height_entry( analysis or NO_INFORMATION_TEXT, 1, 4, profile=_HEIGHT_PROFILE_ANALYSIS_BLOCK, ) ], ) row_index += 1 row_index = self._write_labeled_value_row( sheet, row_index, "CIE10 principal", self._format_primary_cie10(item), profile=_HEIGHT_PROFILE_RICH_VALUE, value_fill=_CARD_FILL, ) row_index = self._write_labeled_value_row( sheet, row_index, "CIE10 secundarios", self._format_code_entries(item.get("cie10_secundarios")), profile=_HEIGHT_PROFILE_RICH_VALUE, value_fill=_CARD_FILL, ) row_index = self._write_labeled_value_row( sheet, row_index, "CUPS principal", self._format_code_entry(item.get("cups_principal")), profile=_HEIGHT_PROFILE_RICH_VALUE, value_fill=_CARD_FILL, ) row_index = self._write_labeled_value_row( sheet, row_index, "CUPS alternativos", self._format_code_entries(item.get("cups_alternativos")), profile=_HEIGHT_PROFILE_RICH_VALUE, value_fill=_CARD_FILL, ) for current in range(start_row, row_index): for column in range(1, 5): cell = sheet.cell(current, column) cell.border = _BORDER return row_index def _write_labeled_value_row( self, sheet: Worksheet, row_index: int, label: str, value: str, *, as_text: bool = False, profile: str | None = None, value_fill: PatternFill | None = None, ) -> int: label_cell = sheet.cell(row=row_index, column=1, value=label) label_cell.font = Font(bold=True) label_cell.fill = _HEADER_FILL label_cell.border = _BORDER label_cell.alignment = Alignment(vertical="top", wrap_text=True) self._merge_range( sheet, row_index, 2, row_index, 4, value, fill=value_fill, ) value_cell = sheet.cell(row=row_index, column=2) value_cell.alignment = Alignment(vertical="top", wrap_text=True) if as_text: value_cell.number_format = _TEXT_FORMAT resolved_profile = profile or self._profile_for_labeled_value(label, value) self._set_row_height( sheet, row_index, [ self._height_entry(label, 1, 1, profile=_HEIGHT_PROFILE_SHORT_VALUE), self._height_entry(value, 2, 4, profile=resolved_profile), ], ) return row_index + 1 def _split_soat_rows(self, items: Any) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: soat_rows: list[dict[str, Any]] = [] qx_rows: list[dict[str, Any]] = [] for item in items or []: if not isinstance(item, dict): continue codigo_soat = str(item.get("codigo_soat") or "").strip().upper() fuente = str(item.get("fuente") or "").strip().lower() if codigo_soat == "QX-HALLAZGOS" or fuente == "agente_hallazgos_qx": qx_rows.append(dict(item)) continue soat_rows.append(dict(item)) return soat_rows, qx_rows def _normalize_value(self, value: Any) -> Any: if value in (None, "", [], {}): return None if isinstance(value, dict): normalized = {} for key, item in value.items(): if key in {"analisis_html", "regen_url", "respuesta_completa"}: continue normalized[key] = self._normalize_value(item) return { key: item for key, item in normalized.items() if item not in (None, "", [], {}) or key == "interpretacion" } or None if isinstance(value, list): items = [self._normalize_value(item) for item in value] return [item for item in items if item not in (None, "", [], {})] or None return self._strip_html(str(value)) def _stringify(self, value: Any) -> str: normalized = self._normalize_value(value) if normalized is None: return NO_INFORMATION_TEXT if isinstance(normalized, dict): return ( "\n".join( f"{self._humanize_key(key)}: {self._stringify(item)}" for key, item in normalized.items() ) or NO_INFORMATION_TEXT ) if isinstance(normalized, list): if all(not isinstance(item, (dict, list)) for item in normalized): return "\n".join(f"- {item}" for item in normalized) or NO_INFORMATION_TEXT return "\n".join(self._stringify(item) for item in normalized) or NO_INFORMATION_TEXT text = str(normalized).strip() return text or NO_INFORMATION_TEXT def _collect_columns(self, rows: Iterable[dict[str, Any]]) -> list[str]: columns: list[str] = [] preferred = [ "fuente", "documento_id", "tipo_documento", "fecha_analisis", "nombre_paciente", "numero_caso", "codigo", "codigo_soat", "codigo_referencia", "codigo_facturacion", "diagnostico", "descripcion", "medicamento", "posologia", "dosis", "cantidad", ] seen = set() for item in preferred: if any(item in row for row in rows): columns.append(item) seen.add(item) for row in rows: for key in row: if key in {"analisis_html", "respuesta_completa"} or key in seen: continue columns.append(key) seen.add(key) return columns or ["valor"] def _build_table_layout(self, columns: list[str]) -> list[tuple[int, int]]: column_count = len(columns) if column_count <= 1: return [(1, 4)] if column_count == 2: wide_index = self._preferred_wide_column_index(columns) if wide_index == 0: return [(1, 3), (4, 4)] return [(1, 1), (2, 4)] if column_count == 3: wide_index = self._preferred_wide_column_index(columns) if wide_index == 0: return [(1, 2), (3, 3), (4, 4)] if wide_index == 1: return [(1, 1), (2, 3), (4, 4)] return [(1, 1), (2, 2), (3, 4)] return [(index, index) for index in range(1, column_count + 1)] def _next_section_row(self, row_index: int) -> int: return row_index + _SECTION_GAP def _preferred_wide_column_index(self, columns: list[str]) -> int: for index, column_name in enumerate(columns): if str(column_name or "").strip().lower() in _WIDE_VALUE_KEYS: return index return max(0, len(columns) - 1) def _merge_row( self, sheet: Worksheet, row_index: int, value: str, *, fill: PatternFill, font: Font, end_col: int = 4, minimum: float = _ROW_HEIGHT_SECTION, ) -> None: self._merge_range(sheet, row_index, 1, row_index, end_col, value, fill=fill, font=font) self._set_row_height( sheet, row_index, [self._height_entry(value, 1, end_col, profile=_HEIGHT_PROFILE_SHORT_VALUE)], minimum=minimum, ) def _merge_range( self, sheet: Worksheet, start_row: int, start_col: int, end_row: int, end_col: int, value: str, *, fill: PatternFill | None = None, font: Font | None = None, ) -> None: sheet.merge_cells( start_row=start_row, start_column=start_col, end_row=end_row, end_column=end_col, ) cell = sheet.cell(row=start_row, column=start_col, value=value) cell.alignment = Alignment(vertical="top", wrap_text=True) cell.border = _BORDER if fill is not None: cell.fill = fill if font is not None: cell.font = font for row in sheet.iter_rows(min_row=start_row, max_row=end_row, min_col=start_col, max_col=end_col): for item in row: item.border = _BORDER if fill is not None: item.fill = fill def _configure_print( self, sheet: Worksheet, *, end_col: int, orientation: str, left_margin: float = 0.25, right_margin: float = 0.25, ) -> None: sheet.sheet_properties.pageSetUpPr = PageSetupProperties(fitToPage=True) sheet.page_setup.paperSize = sheet.PAPERSIZE_LETTER sheet.page_setup.orientation = orientation sheet.page_setup.fitToWidth = 1 sheet.page_setup.fitToHeight = 0 sheet.page_margins.left = left_margin sheet.page_margins.right = right_margin sheet.page_margins.top = 0.5 sheet.page_margins.bottom = 0.5 sheet.page_margins.header = 0.3 sheet.page_margins.footer = 0.3 sheet.print_area = f"A1:{get_column_letter(end_col)}{sheet.max_row}" def _set_row_height( self, sheet: Worksheet, row_index: int, entries: list[HeightEntry], *, minimum: float = _ROW_HEIGHT_BASE, ) -> None: height = minimum max_height = max( (self._height_profile(entry.profile).max_height for entry in entries), default=_ROW_HEIGHT_PROFILES[_HEIGHT_PROFILE_SHORT_VALUE].max_height, ) for entry in entries: height = max(height, self._estimate_height(sheet, entry)) sheet.row_dimensions[row_index].height = min(height, max_height) def _estimate_height(self, sheet: Worksheet, entry: HeightEntry) -> float: text = self._normalize_text_for_height(entry.value) or " " width = sum(self._column_width(sheet, column) for column in range(entry.start_col, entry.end_col + 1)) profile = self._height_profile(entry.profile) chars_per_width = profile.chars_per_width if self._uppercase_ratio(text) >= profile.uppercase_ratio_threshold: chars_per_width *= profile.uppercase_width_factor approx_chars = max(int(width * chars_per_width), 10) lines = 0 for line in text.splitlines() or [" "]: current = re.sub(r"\s+", " ", line).strip() if not current: lines += 1 continue lines += max(1, math.ceil(len(current) / approx_chars)) if ( profile.long_paragraph_threshold and "\n" not in text and len(text) >= profile.long_paragraph_threshold ): lines += profile.long_paragraph_bonus_lines return max(_ROW_HEIGHT_BASE, (lines * profile.line_points) + profile.padding) def _height_entry(self, value: str, start_col: int, end_col: int, *, profile: str) -> HeightEntry: return HeightEntry(value=value, start_col=start_col, end_col=end_col, profile=profile) def _height_profile(self, profile: str) -> RowHeightProfile: return _ROW_HEIGHT_PROFILES[profile] def _profile_for_section_value(self, title: str) -> str: if title in _ANALYSIS_SECTION_TITLES: return _HEIGHT_PROFILE_ANALYSIS_BLOCK if title in _NARRATIVE_SECTION_TITLES: return _HEIGHT_PROFILE_NARRATIVE_BLOCK return _HEIGHT_PROFILE_RICH_VALUE def _profile_for_labeled_value(self, label: str, value: str) -> str: normalized_label = str(label or "").strip() if normalized_label in _ANALYSIS_SECTION_TITLES: return _HEIGHT_PROFILE_ANALYSIS_BLOCK if normalized_label in _NARRATIVE_SECTION_TITLES: return _HEIGHT_PROFILE_NARRATIVE_BLOCK if normalized_label in _RICH_VALUE_LABELS: return _HEIGHT_PROFILE_RICH_VALUE if normalized_label in _COMPACT_VALUE_LABELS: return self._profile_for_compact_value(value) return self._profile_for_compact_value(value, fallback=_HEIGHT_PROFILE_RICH_VALUE) def _profile_for_table_value(self, column_name: str, value: str) -> str: if str(column_name or "").strip().lower() in _WIDE_VALUE_KEYS: return _HEIGHT_PROFILE_RICH_VALUE return self._profile_for_compact_value(value, fallback=_HEIGHT_PROFILE_TABLE_VALUE) def _profile_for_full_width_value(self, value: str) -> str: if self._value_is_analysis_like(value): return _HEIGHT_PROFILE_ANALYSIS_BLOCK if self._value_is_narrative(value): return _HEIGHT_PROFILE_NARRATIVE_BLOCK return self._profile_for_compact_value(value, fallback=_HEIGHT_PROFILE_TABLE_VALUE) def _profile_for_compact_value(self, value: str, *, fallback: str = _HEIGHT_PROFILE_SHORT_VALUE) -> str: normalized = self._normalize_text_for_height(value) if not normalized: return _HEIGHT_PROFILE_SHORT_VALUE if self._value_is_analysis_like(normalized): return _HEIGHT_PROFILE_ANALYSIS_BLOCK if self._value_is_rich(normalized): return fallback return _HEIGHT_PROFILE_SHORT_VALUE def _value_is_rich(self, value: str) -> bool: normalized = self._normalize_text_for_height(value) return "\n" in normalized or len(normalized) >= 140 def _value_is_narrative(self, value: str) -> bool: normalized = self._normalize_text_for_height(value) return len(normalized) >= 160 def _value_is_analysis_like(self, value: str) -> bool: normalized = self._normalize_text_for_height(value) lowered = normalized.lower() return "coherencia diagnóstico" in lowered or "análisis de riesgo de glosa" in lowered def _uppercase_ratio(self, text: str) -> float: letters = [char for char in text if char.isalpha()] if not letters: return 0.0 uppercase_letters = [char for char in letters if char.isupper()] return len(uppercase_letters) / len(letters) def _column_width(self, sheet: Worksheet, column_index: int) -> float: column_letter = get_column_letter(column_index) width = sheet.column_dimensions[column_letter].width return float(width or 10.0) def _soat_code(self, item: dict[str, Any]) -> str: return str(item.get("codigo_soat") or "").strip() or NO_INFORMATION_TEXT def _display_source(self, item: dict[str, Any], *, qx_mode: bool) -> str: source = str(item.get("fuente") or "").strip() if qx_mode: return "Agente hallazgos QX" return self._humanize_key(source) if source else "SOAT" def _display_description(self, item: dict[str, Any]) -> str: return self._clean_display_text(item.get("descripcion")) or NO_INFORMATION_TEXT def _format_primary_cie10(self, item: dict[str, Any]) -> str: fallback = str(item.get("codigo") or "").strip().upper() fallback_code = fallback if _CIE10_PATTERN.match(fallback) else None return self._format_code_entry(item.get("cie10_principal"), fallback_code=fallback_code) def _format_code_entries(self, items: Any) -> str: entries = [self._format_code_entry(item) for item in items or []] entries = [entry for entry in entries if entry and entry != NO_INFORMATION_TEXT] return "\n\n".join(entries) or NO_INFORMATION_TEXT def _format_code_entry(self, item: Any, *, fallback_code: str | None = None) -> str: if isinstance(item, dict): details = self._code_entry_details(item, fallback_code=fallback_code) if details: return "\n".join(details) return NO_INFORMATION_TEXT code = self._extract_code(item) or (fallback_code or "").strip().upper() return f"Código: {code}" if code else NO_INFORMATION_TEXT def _code_entry_details(self, item: dict[str, Any], *, fallback_code: str | None = None) -> list[str]: details: list[str] = [] code = self._extract_code(item) or (fallback_code or "").strip().upper() if code: details.append(f"Código: {code}") description = self._clean_display_text(item.get("descripcion")) if description: details.append(f"Descripción: {description}") justification = self._clean_llm_text(item.get("justificacion")) if justification: details.append(f"Justificación: {justification}") for key, value in item.items(): if key in {"codigo", "descripcion", "justificacion", "analisis_html", "respuesta_completa"}: continue serialized = self._clean_display_text(value) if serialized: details.append(f"{self._humanize_key(key)}: {serialized}") return details def _extract_code(self, item: Any) -> str: if isinstance(item, dict): return str(item.get("codigo") or "").strip().upper() return str(item or "").strip().upper() def _clean_display_text(self, value: Any) -> str: text = self._strip_html(str(value or "")) text = _MARKDOWN_EMPHASIS_PATTERN.sub(r"\1", text) text = text.replace("\u202f", " ") return re.sub(r"\n{3,}", "\n\n", text).strip() def _clean_llm_text(self, value: Any) -> str: cleaned_lines: list[str] = [] for raw_line in self._clean_display_text(value).splitlines(): line = raw_line.strip() if not line: if cleaned_lines and cleaned_lines[-1] != "": cleaned_lines.append("") continue if _DECORATIVE_LINE_PATTERN.match(line): continue line = _MARKDOWN_BULLET_PATTERN.sub("", line) line = _MARKDOWN_EMPHASIS_PATTERN.sub(r"\1", line) cleaned_lines.append(line) while cleaned_lines and cleaned_lines[-1] == "": cleaned_lines.pop() return "\n".join(cleaned_lines).strip() def _unique_sheet_title(self, raw_title: str, used_titles: set[str]) -> str: base = _INVALID_SHEET_CHARS.sub("-", str(raw_title or "Caso")).strip().strip("'") or "Caso" base = base[:31] candidate = base counter = 1 while candidate in used_titles: suffix = f"_{counter}" candidate = f"{base[: 31 - len(suffix)]}{suffix}" counter += 1 return candidate def _humanize_key(self, value: str) -> str: key = str(value or "").strip() if not key: return "Valor" if key in _LABELS: return _LABELS[key] normalized = key.replace("_", " ").strip() normalized = " ".join(_LABEL_WORD_REPLACEMENTS.get(part.lower(), part.lower()) for part in normalized.split()) return normalized[:1].upper() + normalized[1:] def _strip_html(self, value: str) -> str: cleaned = _HTML_BREAK_PATTERN.sub("\n", value or "") cleaned = _HTML_TAG_PATTERN.sub("", cleaned) cleaned = cleaned.replace(" ", " ") return re.sub(r"\n{3,}", "\n\n", cleaned).strip() def _normalize_text_for_height(self, value: Any) -> str: text = self._strip_html(str(value or "")) text = text.replace("\r\n", "\n").replace("\r", "\n").replace("\u202f", " ") text = re.sub(r"[ \t]+\n", "\n", text) return re.sub(r"\n{3,}", "\n\n", text).strip() def _patient_identification(self, row: BatchEpicrisisExcelCaseRow) -> str: context_id = ((row.context or {}).get("metadatos_hc") or {}).get("identificacion_paciente") return str(context_id or row.patient_id or "").strip()