from __future__ import annotations import re from collections import defaultdict from datetime import UTC, date, datetime from typing import TYPE_CHECKING, Annotated, Any from urllib.parse import quote from zoneinfo import ZoneInfo from fastapi import APIRouter, Depends, HTTPException, Query, Request from fastapi.responses import HTMLResponse, JSONResponse from starlette import status from app.auth import get_current_user from app.core.dependencies import get_services from app.core.operational_queue import ( BatchQueueSource, EpicrisisQueueSource, ManualSessionQueueSource, OperationalQueueFacade, ) from app.models import UserInDB from app.routes.history import build_user_histories_snapshot from app.routes.shell_navigation import build_shell_context from app.services.case_deletion_service import CaseDeletionError from app.services.demo_identity_service import get_demo_identity_service if TYPE_CHECKING: from app.core.services import AppServices else: AppServices = Any router = APIRouter() CurrentUser = Annotated[UserInDB, Depends(get_current_user)] ServicesDep = Annotated[AppServices, Depends(get_services)] _DIGITS_PATTERN = re.compile(r"\D+") _CASE_FILTER_STATUS = {"completado", "pendiente", "procesando", "en_cola"} _RIPS_FILTER_STATUS = {"sin_rips", "rips_listo", "rips_con_observaciones"} _DOC_COUNT_RANGES = {"1-2", "3-5", "6+"} _EPICRISIS_PDF_FILTERS = {"yes", "no"} _BOGOTA_TZ = ZoneInfo("America/Bogota") def _normalize_digits(value: Any) -> str: return _DIGITS_PATTERN.sub("", str(value or "")) def _safe_iso_to_datetime(value: Any) -> datetime: if not value: return datetime.min.replace(tzinfo=UTC) try: parsed = datetime.fromisoformat(str(value)) except ValueError: return datetime.min.replace(tzinfo=UTC) if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=_BOGOTA_TZ) return parsed.astimezone(UTC) def _safe_iso_to_date(value: str | None) -> date | None: if not value: return None try: return date.fromisoformat(str(value)) except ValueError: return None def _normalize_case_key(value: str) -> str: normalized = str(value or "").strip() if not normalized: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="El parámetro 'case_key' es obligatorio.", ) return normalized def _humanize_document_type(document_type: str) -> str: normalized = str(document_type or "sin_tipo").replace("_", " ").strip() if not normalized: return "Sin tipo" return normalized[:1].upper() + normalized[1:] def _build_case_history_href(case_group: dict[str, Any], *, fallback_document_id: str = "") -> str: case_key = str(case_group.get("case_key") or "").strip() document_id = str(fallback_document_id or "").strip() if case_key: href = f"/casos?case_key={quote(case_key)}" if document_id: href = f"{href}&documento_id={quote(document_id)}" return href return "/casos" def _build_case_actions(case_group: dict[str, Any], *, document_id: str = "") -> list[dict[str, str]]: case_key = str(case_group.get("case_key") or "").strip() epicrisis_status = str(case_group.get("epicrisis_status") or "").strip().lower() epicrisis_label = "Ver epicrisis" if epicrisis_status == "completado" else "Generar epicrisis" actions = [ { "label": "Ver detalle", "href": _build_case_history_href(case_group, fallback_document_id=document_id), "variant": "secondary", } ] if not case_key: return actions actions.extend( [ { "label": epicrisis_label, "href": str(case_group.get("epicrisis_url") or f"/epicrisis?case_key={quote(case_key)}"), "variant": "primary", }, { "label": "RIPS", "href": f"/rips/operativo?case_key={quote(case_key)}", "variant": "secondary", }, { "label": "PDF", "href": f"/epicrisis/pdf_cache/ultimo?case_key={quote(case_key)}", "variant": "ghost", }, ] ) return actions def _load_latest_rips_lookup( *, username: str, services: AppServices, case_keys: list[str], ) -> dict[str, dict[str, Any]]: normalized_case_keys = [str(case_key or "").strip() for case_key in case_keys if str(case_key or "").strip()] if not normalized_case_keys: return {} results = services.mongo_analyses.collection.find( { "usuario": username, "tipo_documento": "rips_case_payload", "case_key": {"$in": normalized_case_keys}, } ).sort([("case_key", 1), ("version", -1), ("fecha_analisis", -1)]) lookup: dict[str, dict[str, Any]] = {} for item in results: case_key = str(item.get("case_key") or "").strip() if not case_key or case_key in lookup: continue lookup[case_key] = dict(item) return lookup def _derive_rips_status(rips_payload: dict[str, Any] | None) -> str: if not rips_payload: return "sin_rips" if bool(rips_payload.get("ready_for_cuv")): return "rips_listo" return "rips_con_observaciones" def _build_rips_meta(rips_payload: dict[str, Any] | None) -> dict[str, Any]: status = _derive_rips_status(rips_payload) updated_at = (rips_payload or {}).get("fecha_analisis") label_map = { "sin_rips": "Sin RIPS", "rips_listo": "RIPS listo", "rips_con_observaciones": "RIPS con observaciones", } return { "status": status, "label": label_map[status], "ready_for_cuv": bool((rips_payload or {}).get("ready_for_cuv")), "version": int((rips_payload or {}).get("version") or 0) or None, "reject_count": int((rips_payload or {}).get("reject_count") or 0), "notify_count": int((rips_payload or {}).get("notify_count") or 0), "updated_at": ( updated_at.isoformat() if isinstance(updated_at, datetime) else str(updated_at or "") ), } def _normalize_case_groups(snapshot: dict[str, Any]) -> tuple[list[dict[str, Any]], int]: cases: list[dict[str, Any]] = [] legacy_excluded_count = 0 for case_group in snapshot.get("historias_por_caso", {}).values(): case_key = str(case_group.get("case_key") or "").strip() if not case_key: legacy_excluded_count += 1 continue cases.append(case_group) return cases, legacy_excluded_count def _build_case_summary(case_group: dict[str, Any], *, rips_payload: dict[str, Any] | None) -> dict[str, Any]: documents = list(case_group.get("documentos") or []) latest_doc = documents[0] if documents else {} latest_at = str(latest_doc.get("fecha_analisis") or "") epicrisis_status = str(case_group.get("epicrisis_status") or "pendiente") rips = _build_rips_meta(rips_payload) return { "case_key": str(case_group.get("case_key") or "").strip(), "case_number": str(case_group.get("case_number") or "").strip() or "Sin número", "patient_name": str(case_group.get("patient_name") or "Paciente sin nombre"), "patient_id": str(case_group.get("patient_id") or "").strip(), "document_count": len(documents), "document_types": list(case_group.get("tipos_documento") or []), "updated_at": latest_at, "epicrisis_status": epicrisis_status, "ready_for_epicrisis": bool(case_group.get("ready_for_epicrisis", False)), "epicrisis_rule_status": str(case_group.get("epicrisis_rule_status") or ""), "epicrisis_blocking_reason": str(case_group.get("epicrisis_blocking_reason") or ""), "epicrisis_missing_documents": list(case_group.get("epicrisis_missing_documents") or []), "epicrisis_url": str(case_group.get("epicrisis_url") or f"/epicrisis?case_key={quote(str(case_group.get('case_key') or '').strip())}"), "has_epicrisis_pdf": epicrisis_status == "completado", "rips": rips, "actions": _build_case_actions(case_group, document_id=str(latest_doc.get("_id") or "").strip()), } def _project_case_summary( *, services: AppServices, username: str, case_group: dict[str, Any], summary: dict[str, Any], ) -> dict[str, Any]: demo_service = get_demo_identity_service(services) visible_identity = demo_service.project_case_identity( username=username, case_key=str(case_group.get("case_key") or "").strip(), case_number=str(case_group.get("case_number") or "").strip(), patient_id=str(case_group.get("patient_id") or "").strip(), patient_name=str(case_group.get("patient_name") or ""), ) visible_case_group = { **case_group, "case_key": visible_identity["case_key"], "patient_name": visible_identity["patient_name"], "patient_id": visible_identity["patient_id"], } return { **summary, "case_key": visible_identity["case_key"], "patient_name": visible_identity["patient_name"], "patient_id": visible_identity["patient_id"], "epicrisis_url": str( visible_case_group.get("epicrisis_url") or f"/epicrisis?case_key={quote(visible_identity['case_key'])}" ), "actions": _build_case_actions( visible_case_group, document_id=str(((case_group.get("documentos") or [{}])[0].get("_id") if case_group.get("documentos") else "") or ""), ), } def _matches_document_count_range(document_count: int, value: str) -> bool: if value == "1-2": return 1 <= document_count <= 2 if value == "3-5": return 3 <= document_count <= 5 if value == "6+": return document_count >= 6 return True def _filter_case_summaries( case_summaries: list[dict[str, Any]], *, q: str, epicrisis_status: str, document_type: str, rips_status: str, document_count_range: str, has_epicrisis_pdf: str, updated_from: str, updated_to: str, ) -> list[dict[str, Any]]: normalized_query = str(q or "").strip().lower() normalized_epicrisis_status = str(epicrisis_status or "").strip().lower() normalized_document_type = str(document_type or "").strip().lower() normalized_rips_status = str(rips_status or "").strip().lower() normalized_document_count_range = str(document_count_range or "").strip() normalized_pdf_filter = str(has_epicrisis_pdf or "").strip().lower() from_date = _safe_iso_to_date(updated_from) to_date = _safe_iso_to_date(updated_to) query_digits = _normalize_digits(normalized_query) filtered: list[dict[str, Any]] = [] for item in case_summaries: patient_id_digits = _normalize_digits(item.get("patient_id")) search_blob = " ".join( [ str(item.get("case_number") or "").lower(), str(item.get("case_key") or "").lower(), str(item.get("patient_name") or "").lower(), ] ) if normalized_query and normalized_query not in search_blob and ( not query_digits or query_digits not in patient_id_digits ): continue if ( normalized_epicrisis_status and normalized_epicrisis_status in _CASE_FILTER_STATUS and str(item.get("epicrisis_status") or "").strip().lower() != normalized_epicrisis_status ): continue if normalized_document_type: available_document_types = {str(doc_type or "").strip().lower() for doc_type in item.get("document_types") or []} if normalized_document_type not in available_document_types: continue if ( normalized_rips_status and normalized_rips_status in _RIPS_FILTER_STATUS and str(((item.get("rips") or {}).get("status")) or "").strip().lower() != normalized_rips_status ): continue if ( normalized_document_count_range and normalized_document_count_range in _DOC_COUNT_RANGES and not _matches_document_count_range(int(item.get("document_count") or 0), normalized_document_count_range) ): continue has_pdf = bool(item.get("has_epicrisis_pdf")) if ( normalized_pdf_filter and normalized_pdf_filter in _EPICRISIS_PDF_FILTERS and ( (normalized_pdf_filter == "yes" and not has_pdf) or (normalized_pdf_filter == "no" and has_pdf) ) ): continue updated_at = _safe_iso_to_datetime(item.get("updated_at")) updated_date = updated_at.astimezone(_BOGOTA_TZ).date() if updated_at != datetime.min else None if from_date and (updated_date is None or updated_date < from_date): continue if to_date and (updated_date is None or updated_date > to_date): continue filtered.append(item) return sorted(filtered, key=lambda current: _safe_iso_to_datetime(current.get("updated_at")), reverse=True) def _build_case_list_payload( *, username: str, services: AppServices, q: str = "", epicrisis_status: str = "", document_type: str = "", rips_status: str = "", document_count_range: str = "", has_epicrisis_pdf: str = "", updated_from: str = "", updated_to: str = "", ) -> dict[str, Any]: snapshot = build_user_histories_snapshot(username=username, services=services) normalized_cases, legacy_excluded_count = _normalize_case_groups(snapshot) case_keys = [str(case_group.get("case_key") or "").strip() for case_group in normalized_cases] rips_lookup = _load_latest_rips_lookup(username=username, services=services, case_keys=case_keys) all_case_summaries = [ _project_case_summary( services=services, username=username, case_group=case_group, summary=_build_case_summary( case_group, rips_payload=rips_lookup.get(str(case_group.get("case_key") or "").strip()), ), ) for case_group in normalized_cases ] filtered_cases = _filter_case_summaries( all_case_summaries, q=q, epicrisis_status=epicrisis_status, document_type=document_type, rips_status=rips_status, document_count_range=document_count_range, has_epicrisis_pdf=has_epicrisis_pdf, updated_from=updated_from, updated_to=updated_to, ) available_document_types = sorted( { str(document_type or "").strip() for case_item in all_case_summaries for document_type in case_item.get("document_types") or [] if str(document_type or "").strip() } ) return { "summary": { "total_cases": len(all_case_summaries), "filtered_cases": len(filtered_cases), "total_documents": sum(int(item.get("document_count") or 0) for item in all_case_summaries), "filtered_documents": sum(int(item.get("document_count") or 0) for item in filtered_cases), "completed_epicrisis": sum( 1 for item in filtered_cases if str(item.get("epicrisis_status") or "").strip().lower() == "completado" ), "rips_ready": sum( 1 for item in filtered_cases if str(((item.get("rips") or {}).get("status")) or "") == "rips_listo" ), "legacy_excluded_count": legacy_excluded_count, "available_document_types": available_document_types, }, "filters": { "q": str(q or ""), "epicrisis_status": str(epicrisis_status or ""), "document_type": str(document_type or ""), "rips_status": str(rips_status or ""), "document_count_range": str(document_count_range or ""), "has_epicrisis_pdf": str(has_epicrisis_pdf or ""), "updated_from": str(updated_from or ""), "updated_to": str(updated_to or ""), }, "cases": filtered_cases, "empty": not filtered_cases, } def _build_case_detail_payload(*, username: str, services: AppServices, case_key: str) -> dict[str, Any]: demo_service = get_demo_identity_service(services) snapshot = build_user_histories_snapshot(username=username, services=services) normalized_case_key = _normalize_case_key( demo_service.resolve_case_key(username=username, visible_case_key=case_key) ) case_group = next( ( item for item in snapshot.get("historias_por_caso", {}).values() if str(item.get("case_key") or "").strip() == normalized_case_key ), None, ) if case_group is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Caso no encontrado o no disponible en la bandeja actual.", ) rips_lookup = _load_latest_rips_lookup(username=username, services=services, case_keys=[normalized_case_key]) case_summary = _project_case_summary( services=services, username=username, case_group=case_group, summary=_build_case_summary(case_group, rips_payload=rips_lookup.get(normalized_case_key)), ) documents_by_type: dict[str, list[dict[str, Any]]] = defaultdict(list) visible_case_group = {**case_group, "case_key": case_summary["case_key"]} for document in case_group.get("documentos") or []: current_type = str(document.get("tipo_documento") or "sin_tipo") projected_document = demo_service.project_document(username=username, document=document) visible_document = projected_document if isinstance(projected_document, dict) else dict(document) documents_by_type[current_type].append( { "document_id": str(visible_document.get("_id") or "").strip(), "name": str(visible_document.get("nombre_archivo") or "Documento"), "type": current_type, "type_label": _humanize_document_type(current_type), "document_title": str(visible_document.get("document_title") or "").strip(), "updated_at": str(visible_document.get("fecha_analisis") or ""), "href": _build_case_history_href( visible_case_group, fallback_document_id=str(visible_document.get("_id") or "").strip(), ), "analysis_available": bool( visible_document.get("analysis_structured") or visible_document.get("analisis_html") or visible_document.get("analisis") ), } ) grouped_documents = [ { "type": document_type, "type_label": _humanize_document_type(document_type), "count": len(items), "documents": sorted(items, key=lambda item: _safe_iso_to_datetime(item.get("updated_at")), reverse=True), } for document_type, items in sorted(documents_by_type.items(), key=lambda current: current[0]) ] return { "case": case_summary, "document_groups": grouped_documents, } def _build_operational_queue_payload( *, username: str, services: AppServices, include_terminal: bool = False, ) -> dict[str, Any]: facade = OperationalQueueFacade( sources=[ ManualSessionQueueSource( runtime=services.individual_ingestion_runtime, demo_identity_service=get_demo_identity_service(services), ), BatchQueueSource(runtime=services.batch_runtime), EpicrisisQueueSource( case_repository=services.case_epicrisis_runtime_service.case_repository, demo_identity_service=get_demo_identity_service(services), ), ] ) return facade.build(username=username, include_terminal=include_terminal) @router.get("/casos", response_class=HTMLResponse) async def cases_page( request: Request, user: CurrentUser, services: ServicesDep, case_key: Annotated[str | None, Query()] = None, documento_id: Annotated[str | None, Query()] = None, ): return services.templates.TemplateResponse( request, "cases.html", { "request": request, "user": user, "initial_case_key": str(case_key or "").strip(), "initial_documento_id": str(documento_id or "").strip(), **build_shell_context(request=request, user=user), }, ) @router.get("/api/casos") async def list_cases( user: CurrentUser, services: ServicesDep, q: Annotated[str, Query(max_length=120)] = "", epicrisis_status: Annotated[str, Query(max_length=32)] = "", document_type: Annotated[str, Query(max_length=64)] = "", rips_status: Annotated[str, Query(max_length=64)] = "", document_count_range: Annotated[str, Query(max_length=16)] = "", has_epicrisis_pdf: Annotated[str, Query(max_length=16)] = "", updated_from: Annotated[str, Query(max_length=10)] = "", updated_to: Annotated[str, Query(max_length=10)] = "", ) -> JSONResponse: payload = _build_case_list_payload( username=user.username, services=services, q=q, epicrisis_status=epicrisis_status, document_type=document_type, rips_status=rips_status, document_count_range=document_count_range, has_epicrisis_pdf=has_epicrisis_pdf, updated_from=updated_from, updated_to=updated_to, ) return JSONResponse(content=payload) @router.get("/api/cola-operativa") async def operational_queue( user: CurrentUser, services: ServicesDep, include_terminal: bool = False, ) -> JSONResponse: payload = _build_operational_queue_payload( username=user.username, services=services, include_terminal=include_terminal, ) return JSONResponse(content=payload) @router.get("/api/casos/{case_key}") async def case_detail( case_key: str, user: CurrentUser, services: ServicesDep, ) -> JSONResponse: payload = _build_case_detail_payload(username=user.username, services=services, case_key=case_key) return JSONResponse(content=payload) @router.delete("/api/casos/documentos/{document_id}") async def delete_case_document( document_id: str, user: CurrentUser, services: ServicesDep, ) -> JSONResponse: try: payload = services.case_deletion_service.delete_document( username=user.username, document_id=document_id, ) except CaseDeletionError as exc: raise HTTPException(status_code=exc.status_code, detail=str(exc)) from exc return JSONResponse(content=payload) @router.delete("/api/casos/{case_key}") async def delete_case( case_key: str, user: CurrentUser, services: ServicesDep, ) -> JSONResponse: real_case_key = get_demo_identity_service(services).resolve_case_key( username=user.username, visible_case_key=case_key, ) try: payload = services.case_deletion_service.delete_case(username=user.username, case_key=real_case_key) except CaseDeletionError as exc: raise HTTPException(status_code=exc.status_code, detail=str(exc)) from exc if isinstance(payload, dict): payload = { **payload, "case_key": get_demo_identity_service(services).project_case_identity( username=user.username, case_key=real_case_key, ).get("case_key", real_case_key), } return JSONResponse(content=payload)