Coverage for narrative_harm_classifier/classifier/rules/azure_nlp.py: 41%

1""" 

2classifier/rules/azure_nlp.py — Azure Text Analytics connector. 

3 

4Wraps Azure Cognitive Services Text Analytics for: 

5- Sentiment analysis (used as negative signal amplifier) 

6- Named Entity Recognition (detects group/identity mentions) 

7- Key phrase extraction (surface-level harm signal detection) 

8 

9Falls back gracefully when Azure credentials are not configured (dev mode). 

10""" 

11 

12import logging 

13from dataclasses import dataclass 

14from typing import Optional 

15 

16logger = logging.getLogger(__name__) 

17 

18 

19@dataclass 

20class AzureNLPResult: 

21 sentiment: str # positive | negative | neutral | mixed 

22 sentiment_score_negative: float # 0.0–1.0 

23 entities: list[dict] # [{text, category, confidence}] 

24 key_phrases: list[str] 

25 language: str 

26 is_fallback: bool = False # True when Azure not configured 

27 

28 

29class AzureNLPClient: 

30 """ 

31 Azure Text Analytics client with graceful dev-mode fallback. 

32 

33 In production: set AZURE_TEXT_ANALYTICS_ENDPOINT and AZURE_TEXT_ANALYTICS_KEY. 

34 In dev/test: client operates in fallback mode with neutral scores. 

35 """ 

36 

37 def __init__(self, endpoint: str = "", key: str = ""): 

38 self.endpoint = endpoint 

39 self.key = key 

40 self._client = None 

41 

42 if endpoint and key: 

43 try: 

44 from azure.ai.textanalytics import TextAnalyticsClient 

45 from azure.core.credentials import AzureKeyCredential 

46 self._client = TextAnalyticsClient( 

47 endpoint=endpoint, 

48 credential=AzureKeyCredential(key) 

49 ) 

50 logger.info("Azure Text Analytics client initialized") 

51 except ImportError: 

52 logger.warning("azure-ai-textanalytics not installed — running in fallback mode") 

53 except Exception as e: 

54 logger.warning(f"Azure Text Analytics init failed: {e} — running in fallback mode") 

55 else: 

56 logger.info("Azure credentials not configured — running in fallback/dev mode") 

57 

58 @property 

59 def is_configured(self) -> bool: 

60 return self._client is not None 

61 

62 def analyze(self, text: str, language: str = "en") -> AzureNLPResult: 

63 """Run sentiment + NER + key phrases in a single batched call.""" 

64 if not self.is_configured: 

65 return self._fallback_result(text) 

66 

67 try: 

68 from azure.ai.textanalytics import ( 

69 RecognizeEntitiesAction, 

70 AnalyzeSentimentAction, 

71 ExtractKeyPhrasesAction, 

72 ) 

73 

74 poller = self._client.begin_analyze_actions( 

75 documents=[{"id": "1", "text": text, "language": language}], 

76 actions=[ 

77 AnalyzeSentimentAction(), 

78 RecognizeEntitiesAction(), 

79 ExtractKeyPhrasesAction(), 

80 ], 

81 ) 

82 results = list(poller.result()) 

83 

84 sentiment_result = None 

85 entity_result = None 

86 keyphrase_result = None 

87 

88 for action_result in results[0]: 

89 if action_result.kind == "SentimentAnalysis" and not action_result.is_error: 

90 sentiment_result = action_result 

91 elif action_result.kind == "EntityRecognition" and not action_result.is_error: 

92 entity_result = action_result 

93 elif action_result.kind == "KeyPhraseExtraction" and not action_result.is_error: 

94 keyphrase_result = action_result 

95 

96 sentiment = "neutral" 

97 neg_score = 0.0 

98 if sentiment_result: 

99 sentiment = sentiment_result.sentiment 

100 neg_score = sentiment_result.confidence_scores.negative 

101 

102 entities = [] 

103 if entity_result: 

104 for ent in entity_result.entities: 

105 entities.append({ 

106 "text": ent.text, 

107 "category": ent.category, 

108 "confidence": ent.confidence_score, 

109 }) 

110 

111 key_phrases = [] 

112 if keyphrase_result: 

113 key_phrases = list(keyphrase_result.key_phrases) 

114 

115 return AzureNLPResult( 

116 sentiment=sentiment, 

117 sentiment_score_negative=neg_score, 

118 entities=entities, 

119 key_phrases=key_phrases, 

120 language=language, 

121 is_fallback=False, 

122 ) 

123 

124 except Exception as e: 

125 logger.error(f"Azure NLP analysis failed: {e}") 

126 return self._fallback_result(text) 

127 

128 def _fallback_result(self, text: str) -> AzureNLPResult: 

129 """Dev-mode fallback: neutral scores, no entities.""" 

130 return AzureNLPResult( 

131 sentiment="neutral", 

132 sentiment_score_negative=0.0, 

133 entities=[], 

134 key_phrases=[], 

135 language="en", 

136 is_fallback=True, 

137 )