"""Generación de workbooks Excel para epicrisis consolidadas por lote.""" from __future__ import annotations from dataclasses import dataclass from io import BytesIO import re from typing import Any, Iterable from openpyxl import Workbook from openpyxl.styles import Alignment, Border, Font, PatternFill, Side from openpyxl.utils import get_column_letter from openpyxl.worksheet.worksheet import Worksheet _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"<[^>]+>") _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") @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: self._merge_row(sheet, 1, f"Reporte de epicrisis por lote - {batch_id}", fill=_TITLE_FILL, font=Font(bold=True, color="FFFFFF", size=14)) self._merge_row( sheet, 2, "Resumen consolidado por case_key", fill=_HEADER_FILL, font=Font(bold=True, color="1F1F1F"), ) headers = [ "case_key", "Paciente", "Identificación paciente", "Número de caso", "Procedimiento", "Estado epicrisis", "Incluido en Excel", "Motivo omisión", ] start_row = 4 for index, header in enumerate(headers, start=1): cell = sheet.cell(row=start_row, 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) for row_index, item in enumerate(rows, start=start_row + 1): values = [ item.case_key or "Sin información", item.patient_name or "Sin información", self._patient_identification(item) or "Sin información", item.case_number or "Sin información", item.procedure_description or "Sin información", item.epicrisis_status or "Sin información", "Sí" if item.included_in_excel else "No", item.omission_reason or "", ] for column_index, value in enumerate(values, start=1): cell = sheet.cell(row=row_index, column=column_index, value=value) cell.border = _BORDER cell.alignment = Alignment(vertical="top", wrap_text=True) if column_index == 1: target_sheet = case_sheet_map.get(item.case_key) if target_sheet: cell.hyperlink = f"#'{target_sheet}'!A1" cell.style = "Hyperlink" 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 sheet.freeze_panes = "A5" sheet.sheet_view.showGridLines = False sheet.page_setup.orientation = "landscape" sheet.page_setup.fitToWidth = 1 def _build_case_sheet( self, sheet: Worksheet, *, batch_id: str, row: BatchEpicrisisExcelCaseRow, ) -> None: context = row.context or {} 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), ) current_row += 1 self._merge_row( sheet, current_row, f"Lote: {batch_id} | Número de caso: {row.case_number or 'Sin información'} | Paciente: {row.patient_name or 'Sin información'}", fill=_HEADER_FILL, font=Font(bold=True), ) current_row += 2 identification = { "case_key": row.case_key or "Sin información", "nombre_paciente": context.get("nombre_paciente") or row.patient_name or "Sin información", "numero_caso": context.get("case_number") or row.case_number or "Sin información", "identificacion_paciente": self._patient_identification(row) or "Sin información", "procedimiento": row.procedure_description or "Sin información", } current_row = self._write_key_value_section(sheet, current_row, "Identificación", identification) current_row = self._write_documents_section(sheet, current_row, context) current_row = self._write_value_section(sheet, current_row, "Metadatos HC", context.get("metadatos_hc")) current_row = self._write_value_section(sheet, current_row, "Procedimientos HC", context.get("procedimientos_hc")) current_row = self._write_value_section( sheet, current_row, "Medicamentos", context.get("medicamentos_hc_display") or context.get("medicamentos_hc"), ) current_row = self._write_value_section(sheet, current_row, "Hallazgos quirúrgicos", context.get("hallazgos_quirurgicos")) current_row = self._write_value_section(sheet, current_row, "Descripción del procedimiento", context.get("descripcion_procedimiento")) current_row = self._write_value_section(sheet, current_row, "Procedimientos de factura", context.get("procedimientos_factura")) current_row = self._write_value_section(sheet, current_row, "Resultados SOAT", context.get("soat_resultados")) current_row = self._write_value_section(sheet, current_row, "Códigos desde SOAT", context.get("codigos_desde_soat")) self._write_value_section(sheet, current_row, "Glosa analítica", context.get("glosa_analisis")) for column, width in {"A": 26, "B": 30, "C": 28, "D": 44, "E": 24, "F": 20}.items(): sheet.column_dimensions[column].width = width sheet.freeze_panes = "A4" sheet.sheet_view.showGridLines = False sheet.page_setup.orientation = "portrait" sheet.page_setup.fitToWidth = 1 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) 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 "Sin información", "fecha_analisis": str(item.get("fecha_analisis") or "").strip() or "Sin información", } 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)) row_index += 1 for key, value in data.items(): label = sheet.cell(row=row_index, column=1, value=self._humanize_key(key)) label.font = Font(bold=True) label.fill = _HEADER_FILL label.border = _BORDER label.alignment = Alignment(vertical="top", wrap_text=True) self._merge_range(sheet, row_index, 2, row_index, 4, self._stringify(value)) row_index += 1 return row_index + 1 def _write_value_section( self, sheet: Worksheet, row_index: int, title: str, value: Any, ) -> int: self._merge_row(sheet, row_index, title, fill=_SECTION_FILL, font=Font(bold=True)) row_index += 1 normalized = self._normalize_value(value) if not normalized: self._merge_range(sheet, row_index, 1, row_index, 4, "Sin información") return row_index + 2 if isinstance(normalized, list) and normalized and all(isinstance(item, dict) for item in normalized): columns = self._collect_columns(normalized) column_layout = self._build_table_layout(len(columns)) for (start_col, end_col), column_name in zip(column_layout, columns): self._merge_range( sheet, row_index, start_col, row_index, end_col, self._humanize_key(column_name), 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_index += 1 for item in normalized: for (start_col, end_col), column_name in zip(column_layout, columns): self._merge_range( sheet, row_index, start_col, row_index, end_col, self._stringify(item.get(column_name)), ) cell = sheet.cell(row=row_index, column=start_col) cell.border = _BORDER cell.alignment = Alignment(vertical="top", wrap_text=True) row_index += 1 return row_index + 1 if isinstance(normalized, dict): return self._write_key_value_section(sheet, row_index - 1, title, normalized) if isinstance(normalized, list): for item in normalized: self._merge_range(sheet, row_index, 1, row_index, 4, self._stringify(item)) row_index += 1 return row_index + 1 self._merge_range(sheet, row_index, 1, row_index, 4, self._stringify(normalized)) return row_index + 2 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"}: continue normalized[key] = self._normalize_value(item) return {key: item for key, item in normalized.items() if item not in (None, "", [], {})} 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 "Sin información" if isinstance(normalized, dict): return "\n".join( f"{self._humanize_key(key)}: {self._stringify(item)}" for key, item in normalized.items() ) or "Sin información" 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 "Sin información" return "\n".join(self._stringify(item) for item in normalized) or "Sin información" text = str(normalized).strip() return text or "Sin información" 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", "diagnostico", "descripcion", "medicamento", "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.keys(): if key in {"analisis_html"} or key in seen: continue columns.append(key) seen.add(key) return columns or ["valor"] def _build_table_layout(self, column_count: int) -> list[tuple[int, int]]: if column_count <= 1: return [(1, 4)] if column_count == 2: return [(1, 2), (3, 4)] if column_count == 3: return [(1, 1), (2, 2), (3, 4)] return [(index, index) for index in range(1, column_count + 1)] def _merge_row( self, sheet: Worksheet, row_index: int, value: str, *, fill: PatternFill, font: Font, ) -> None: self._merge_range(sheet, row_index, 1, row_index, 4, value, fill=fill, font=font) 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 _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: return str(value or "").replace("_", " ").strip().capitalize() or "Valor" 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 _patient_identification(self, row: BatchEpicrisisExcelCaseRow) -> str: context_id = ( ((row.context or {}).get("metadatos_hc") or {}).get("datos_identificacion_paciente") ) return str(context_id or row.patient_id or "").strip()