Coverage for presidio_analyzer / predefined_recognizers / third_party / ahds_recognizer.py: 59%

1import os 

2from typing import List, Optional 

3 

4try: 

5 from azure.health.deidentification import DeidentificationClient 

6 from azure.health.deidentification.models import ( 

7 DeidentificationContent, 

8 DeidentificationOperationType, 

9 PhiCategory, 

10 ) 

11except ImportError: 

12 DeidentificationClient = None 

13 DeidentificationContent = None 

14 DeidentificationOperationType = None 

15 PhiCategory = None 

16 

17try: 

18 from presidio_analyzer.llm_utils.azure_auth_helper import ( 

19 get_azure_credential, 

20 ) 

21 AZURE_AUTH_AVAILABLE = True 

22except ImportError: # pragma: no cover 

23 get_azure_credential = None 

24 AZURE_AUTH_AVAILABLE = False 

25 

26from presidio_analyzer import AnalysisExplanation, RecognizerResult, RemoteRecognizer 

27from presidio_analyzer.nlp_engine import NlpArtifacts 

28 

29 

30class AzureHealthDeidRecognizer(RemoteRecognizer): 

31 """Wrapper for PHI detection using Azure Health Data Services de-identification.""" 

32 

33 def __init__( 

34 self, 

35 supported_entities: Optional[List[str]] = None, 

36 supported_language: str = "en", 

37 client: Optional[DeidentificationClient] = None, 

38 name: Optional[str] = None, 

39 ): 

40 """ 

41 Wrap PHI detection using Azure Health Data Services de-identification. 

42 

43 :param supported_entities: List of supported entities for this recognizer. 

44 :param supported_language: Language code (not used, only 'en' supported). 

45 :param client: Optional DeidentificationClient instance. 

46 """ 

47 super().__init__( 

48 supported_entities=supported_entities, 

49 supported_language=supported_language, 

50 name=name if name else "Azure Health Data Services Deidentification", 

51 version="1.0.0", 

52 ) 

53 

54 

55 endpoint = os.getenv("AHDS_ENDPOINT", None) 

56 

57 if client is None: 

58 if endpoint is None: 

59 raise ValueError( 

60 "AHDS de-identification endpoint is required. " 

61 "Please provide an endpoint " 

62 "or set the AHDS_ENDPOINT environment variable." 

63 ) 

64 

65 if not DeidentificationClient: 

66 raise ImportError( 

67 "Azure Health Data Services Deidentification SDK is not available. " 

68 "Please install azure-health-deidentification and azure-identity." 

69 ) 

70 

71 # Use environment-aware credential (DefaultAzureCredential for dev, 

72 # ChainedTokenCredential for production) 

73 credential = get_azure_credential() 

74 client = DeidentificationClient(endpoint, credential) 

75 

76 self.deid_client = client 

77 

78 if not supported_entities: 

79 self.supported_entities = self._get_supported_entities() 

80 

81 @staticmethod 

82 def _get_supported_entities() -> List[str]: 

83 if PhiCategory: 

84 try: 

85 # PhiCategory is an enum, try to get the actual enum names 

86 return [category.name for category in PhiCategory] 

87 except Exception: 

88 return ImportError 

89 

90 def get_supported_entities(self) -> List[str]: 

91 """ 

92 Return the list of entities supported by this recognizer. 

93 

94 Returns 

95 List[str]: A list of supported entity names as strings. 

96 """ 

97 return self.supported_entities 

98 

99 def analyze( 

100 self, text: str, entities: List[str] = None, nlp_artifacts: NlpArtifacts = None 

101 ) -> List[RecognizerResult]: 

102 """ 

103 Analyze text using Azure Health Data Services Deidentification (TAG operation). 

104 

105 :param text: Text to analyze 

106 :param entities: List of entities to return (optional) 

107 :param nlp_artifacts: Not used 

108 :return: List of RecognizerResult for each PHI entity found 

109 """ 

110 if not entities: 

111 entities = self.supported_entities 

112 

113 body = DeidentificationContent( 

114 input_text=text, 

115 operation_type=DeidentificationOperationType.TAG 

116 ) 

117 result = self.deid_client.deidentify_text(body) 

118 

119 recognizer_results = [] 

120 if result.tagger_result and result.tagger_result.entities: 

121 for entity in result.tagger_result.entities: 

122 category = entity.category.upper() 

123 if category not in [e.upper() for e in entities]: 

124 continue 

125 analysis_explanation = AzureHealthDeidRecognizer._build_explanation( 

126 entity_type=category 

127 ) 

128 recognizer_results.append( 

129 RecognizerResult( 

130 entity_type=category, 

131 start=entity.offset.code_point, 

132 end=entity.offset.code_point + entity.length.code_point, 

133 score=round(entity.confidence_score, 2), 

134 analysis_explanation=analysis_explanation, 

135 ) 

136 ) 

137 return recognizer_results 

138 

139 @staticmethod 

140 def _build_explanation(entity_type: str) -> AnalysisExplanation: 

141 explanation = AnalysisExplanation( 

142 recognizer=AzureHealthDeidRecognizer.__class__.__name__, 

143 original_score=1.0, 

144 textual_explanation=( 

145 f"Identified as {entity_type} by Azure Health Data Services " 

146 "Deidentification" 

147 ), 

148 ) 

149 return explanation