from __future__ import annotations from dataclasses import dataclass from app.rda.application.fhir_projection import build_rda_fhir_projection from app.rda.domain.fhir_projection import RdaFhirProjection from app.rda.domain.fhir_validation import ( FhirLikeResourceType, RdaValidationFinding, RdaValidationResult, ValidationSeverity, ) from app.rda.domain.models import RdaArtifact, RdaArtifactType @dataclass class ValidateRdaFhirProjectionUseCase: def execute(self, artifact: RdaArtifact) -> RdaValidationResult: projection = build_rda_fhir_projection(artifact) findings: list[RdaValidationFinding] = [] findings.extend(_validate_bundle(projection)) findings.extend(_validate_composition(projection)) findings.extend(_validate_patient(projection)) findings.extend(_validate_encounter(projection)) findings.extend(_validate_expected_organizations(artifact, projection)) findings.extend(_warnings_from_projection_gaps(projection)) return RdaValidationResult.from_findings( artifact_type=artifact.artifact_type, findings=findings, ) def _validate_bundle(projection: RdaFhirProjection) -> list[RdaValidationFinding]: findings: list[RdaValidationFinding] = [] references = {entry.reference.reference for entry in projection.bundle.entries} expected = { f"Composition/{projection.composition.id}", f"Patient/{projection.patient.id}", } missing = sorted(expected - references) if missing: findings.append( _error( resource_type=FhirLikeResourceType.BUNDLE, resource_path="bundle.entries", rule_code="RDA-FHIR-ERR-001", message="Bundle debe contener al menos Composition y Patient.", observed_value=missing, ) ) return findings def _validate_composition(projection: RdaFhirProjection) -> list[RdaValidationFinding]: findings: list[RdaValidationFinding] = [] expected_patient_reference = f"Patient/{projection.patient.id}" if projection.composition.subject.reference != expected_patient_reference: findings.append( _error( resource_type=FhirLikeResourceType.COMPOSITION, resource_path="composition.subject.reference", rule_code="RDA-FHIR-ERR-002", message="Composition debe referenciar al Patient proyectado.", observed_value=projection.composition.subject.reference, ) ) if projection.artifact_type == RdaArtifactType.EMERGENCY: expected_encounter_reference = ( f"Encounter/{projection.encounter.id}" if projection.encounter is not None else None ) actual_encounter_reference = ( projection.composition.encounter.reference if projection.composition.encounter is not None else None ) if expected_encounter_reference is None or actual_encounter_reference != expected_encounter_reference: findings.append( _error( resource_type=FhirLikeResourceType.COMPOSITION, resource_path="composition.encounter.reference", rule_code="RDA-FHIR-ERR-004", message="Composition de emergency debe referenciar al Encounter proyectado.", observed_value=actual_encounter_reference, ) ) expected_organization_references = { f"Organization/{organization.id}" for organization in projection.organizations } actual_organization_references = { reference.reference for reference in projection.composition.organization_references } missing_organizations = sorted(expected_organization_references - actual_organization_references) if missing_organizations: findings.append( _error( resource_type=FhirLikeResourceType.COMPOSITION, resource_path="composition.organization_references", rule_code="RDA-FHIR-ERR-005", message="Composition debe referenciar todas las Organization proyectadas.", observed_value=missing_organizations, ) ) if projection.practitioner is not None: expected_practitioner_reference = f"Practitioner/{projection.practitioner.id}" actual_practitioner_reference = ( projection.composition.practitioner_reference.reference if projection.composition.practitioner_reference is not None else None ) if actual_practitioner_reference != expected_practitioner_reference: findings.append( _error( resource_type=FhirLikeResourceType.COMPOSITION, resource_path="composition.practitioner_reference.reference", rule_code="RDA-FHIR-ERR-006", message="Composition debe referenciar al Practitioner proyectado cuando exista.", observed_value=actual_practitioner_reference, ) ) return findings def _validate_patient(projection: RdaFhirProjection) -> list[RdaValidationFinding]: if projection.patient.identifier_type and projection.patient.identifier_value: return [] return [ _error( resource_type=FhirLikeResourceType.PATIENT, resource_path="patient.identifier", rule_code="RDA-FHIR-ERR-003", message="Patient debe tener tipo y número de documento para identidad mínima usable.", observed_value={ "identifier_type": projection.patient.identifier_type, "identifier_value": projection.patient.identifier_value, }, ) ] def _validate_encounter(projection: RdaFhirProjection) -> list[RdaValidationFinding]: if projection.artifact_type != RdaArtifactType.EMERGENCY: return [] if projection.encounter is None: return [ _error( resource_type=FhirLikeResourceType.ENCOUNTER, resource_path="encounter", rule_code="RDA-FHIR-ERR-004", message="Emergency debe proyectar un Encounter.", observed_value=None, ) ] expected_patient_reference = f"Patient/{projection.patient.id}" if projection.encounter.subject.reference == expected_patient_reference: return [] return [ _error( resource_type=FhirLikeResourceType.ENCOUNTER, resource_path="encounter.subject.reference", rule_code="RDA-FHIR-ERR-004", message="Encounter debe referenciar al Patient proyectado.", observed_value=projection.encounter.subject.reference, ) ] def _validate_expected_organizations( artifact: RdaArtifact, projection: RdaFhirProjection, ) -> list[RdaValidationFinding]: findings: list[RdaValidationFinding] = [] expected_roles = _expected_organization_roles(artifact) projected_roles = {organization.role for organization in projection.organizations} missing_roles = sorted(expected_roles - projected_roles) if missing_roles: findings.append( _error( resource_type=FhirLikeResourceType.ORGANIZATION, resource_path="organizations", rule_code="RDA-FHIR-ERR-005", message="Payload con provider/payer debe proyectar Organization para cada rol presente.", observed_value=missing_roles, ) ) return findings def _expected_organization_roles(artifact: RdaArtifact) -> set[str]: payload = artifact.payload if isinstance(artifact.payload, dict) else {} if artifact.artifact_type == RdaArtifactType.PATIENT: organization = payload.get("organization") if not isinstance(organization, dict): return set() provider = organization.get("provider") payer = organization.get("payer") else: encounter = payload.get("encounter") if not isinstance(encounter, dict): return set() provider = encounter.get("provider") payer = encounter.get("payer") roles: set[str] = set() if _has_organization_data(provider): roles.add("provider") if _has_organization_data(payer): roles.add("payer") return roles def _has_organization_data(data: object) -> bool: if not isinstance(data, dict): return False return any(str(value).strip() for value in data.values() if value is not None) def _warnings_from_projection_gaps(projection: RdaFhirProjection) -> list[RdaValidationFinding]: findings: list[RdaValidationFinding] = [] for index, gap in enumerate(projection.projection_gaps, start=1): resource_type = _resource_type_for_gap(gap) findings.append( _warning( resource_type=resource_type, resource_path=f"projection_gaps[{index - 1}]", rule_code=f"RDA-FHIR-WARN-{index:03d}", message=f"Projection gap detected: {gap}.", observed_value=gap, ) ) return findings def _resource_type_for_gap(gap: str) -> FhirLikeResourceType: if gap.startswith("document_reference:"): return FhirLikeResourceType.DOCUMENT_REFERENCE if gap.startswith("practitioner:"): return FhirLikeResourceType.PRACTITIONER if gap.startswith("diagnoses:") or gap.startswith("clinical_background:"): return FhirLikeResourceType.COMPOSITION if gap.startswith("procedures:") or gap.startswith("medications:") or gap.startswith("diagnostic_support:"): return FhirLikeResourceType.ENCOUNTER return FhirLikeResourceType.COMPOSITION def _error( *, resource_type: FhirLikeResourceType, resource_path: str, rule_code: str, message: str, observed_value: object, ) -> RdaValidationFinding: return RdaValidationFinding( severity=ValidationSeverity.ERROR, resource_type=resource_type, resource_path=resource_path, rule_code=rule_code, message=message, observed_value=observed_value, ) def _warning( *, resource_type: FhirLikeResourceType, resource_path: str, rule_code: str, message: str, observed_value: object, ) -> RdaValidationFinding: return RdaValidationFinding( severity=ValidationSeverity.WARNING, resource_type=resource_type, resource_path=resource_path, rule_code=rule_code, message=message, observed_value=observed_value, )