Coverage for narrative_harm_classifier/classifier/rules/engine.py: 96%
133 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-20 13:25 +0000
1"""
2classifier/rules/engine.py — Multi-dimensional, multi-language classification engine.
4Implements D2.4a specification: operationalizes relationships between
5targets, identities, harm mechanisms, and decision thresholds.
7Architecture:
8 1. Signal detection — pattern matching (per-language) + dog-whistle lexicon
9 + Azure NLP entity/sentiment signals
10 2. Suppression — negation / counter-speech / benign-context heuristics
11 3. Signal scoring — weighted aggregation per taxonomy row
12 4. Multi-signal resolution — ambiguity rules from taxonomy config
13 5. Decision — threshold gating with rationale + provenance hash
14"""
16import re
17import logging
18from pathlib import Path
19from typing import Optional
21from narrative_harm_classifier.core.models import (
22 ClassifyRequest,
23 ClassificationResult,
24 SignalMatch,
25 HarmCategory,
26 TargetType,
27 IdentityAxis,
28)
29from narrative_harm_classifier.classifier.taxonomy.loader import TaxonomyConfig, TaxonomyRow, CategorySpec
30from narrative_harm_classifier.classifier.rules.azure_nlp import AzureNLPClient, AzureNLPResult
31from narrative_harm_classifier.classifier.rules.patterns_loader import (
32 LanguagePatterns,
33 load_language_patterns,
34 DEFAULT_LANGUAGE,
35)
36from narrative_harm_classifier.classifier.rules.dogwhistles import DogwhistleLexicon, DogwhistleEntry
37from narrative_harm_classifier.classifier.counter_narrative import guidance_for
38from narrative_harm_classifier.classifier.provenance import content_hash
40logger = logging.getLogger(__name__)
42DEFAULT_PATTERNS_DIR = str(Path(__file__).parent.parent.parent / "data" / "patterns")
44# Negation cues are checked in a window immediately before a matched harm
45# pattern ("these people are NOT vermin"), rather than across the whole
46# text, so an unrelated un-negated claim later in the same text isn't
47# suppressed by a negation word that has nothing to do with it.
48_NEGATION_WINDOW_CHARS = 60
51# ---------------------------------------------------------------------------
52# Signal detection
53# ---------------------------------------------------------------------------
55def _detect_identity_anchor(text: str, patterns: LanguagePatterns) -> Optional[str]:
56 """Return the identity axis if a group mention is found in the text."""
57 text_lower = text.lower()
58 deobf_lower = patterns.deobfuscate(text_lower)
59 for axis, compiled_patterns in patterns.identity_anchors.items():
60 for compiled in compiled_patterns:
61 if compiled.search(text_lower) or compiled.search(deobf_lower):
62 return axis
63 return None
66def _detect_harm_signals(
67 text: str, row: TaxonomyRow, patterns: LanguagePatterns
68) -> tuple[bool, Optional[str], Optional[re.Match], str]:
69 """
70 Check text against harm patterns for a given taxonomy row, against both
71 the literal text and a deobfuscated ("v3rmin" -> "vermin") version.
73 Returns (matched, matched_pattern, match_object, matched_text_lower) — the
74 match object and the text version it was found in are returned so callers
75 can run negation-window / counter-speech checks against the same text
76 the match actually came from.
77 """
78 text_lower = text.lower()
79 deobf_lower = patterns.deobfuscate(text_lower)
80 compiled_patterns = patterns.harm_patterns.get(row.harm_mechanism, [])
81 for compiled in compiled_patterns:
82 match = compiled.search(text_lower)
83 if match:
84 return True, compiled.pattern, match, text_lower
85 match = compiled.search(deobf_lower)
86 if match:
87 return True, compiled.pattern, match, deobf_lower
88 return False, None, None, text_lower
91def _azure_negative_amplifier(azure_result: Optional[AzureNLPResult]) -> float:
92 """
93 Use Azure sentiment negativity as a signal amplifier (0.0 → 1.0).
94 High negative sentiment boosts confidence; neutral is neutral.
95 Falls back to 1.0 (no penalty) when Azure is unavailable.
96 """
97 if azure_result is None or azure_result.is_fallback:
98 return 1.0 # no penalty when Azure not configured
99 # Map: 0.0 negative → 0.2, 1.0 negative → 1.0
100 return 0.2 + (azure_result.sentiment_score_negative * 0.8)
103# ---------------------------------------------------------------------------
104# Suppression pipeline — negation / counter-speech / benign-context
105# ---------------------------------------------------------------------------
106# Each rule answers "should this matched signal be discarded?" A match is
107# suppressed if ANY rule fires. Extracted into a uniform list (rather than
108# three separate inline `if` statements) so a new heuristic — or a new
109# language's cues — plugs in without restructuring classify() again.
111def _rule_negation(matched_text: str, match_obj: Optional[re.Match], patterns: LanguagePatterns) -> bool:
112 if patterns.negation_cues is None or match_obj is None:
113 return False
114 window = matched_text[max(0, match_obj.start() - _NEGATION_WINDOW_CHARS) : match_obj.start()]
115 return bool(patterns.negation_cues.search(window))
118def _rule_counter_speech(matched_text: str, match_obj: Optional[re.Match], patterns: LanguagePatterns) -> bool:
119 if patterns.reporting_cues is None or patterns.condemnation_cues is None:
120 return False
121 return bool(patterns.reporting_cues.search(matched_text)) and bool(
122 patterns.condemnation_cues.search(matched_text)
123 )
126def _rule_benign_context(matched_text: str, match_obj: Optional[re.Match], patterns: LanguagePatterns) -> bool:
127 if patterns.benign_context_cues is None:
128 return False
129 return bool(patterns.benign_context_cues.search(matched_text))
132_SUPPRESSION_RULES = (_rule_counter_speech, _rule_negation, _rule_benign_context)
135def _is_suppressed(matched_text: str, match_obj: Optional[re.Match], patterns: LanguagePatterns) -> bool:
136 return any(rule(matched_text, match_obj, patterns) for rule in _SUPPRESSION_RULES)
139# identity_axis -> TargetType, shared by taxonomy rows and dog-whistle entries
140_AXIS_TO_TARGET_TYPE = {
141 "race_ethnicity": TargetType.ETHNIC_GROUP,
142 "religion": TargetType.RELIGIOUS_GROUP,
143 "gender": TargetType.GENDER_GROUP,
144 "national_origin": TargetType.NATIONAL_ORIGIN_GROUP,
145 "political_affiliation": TargetType.POLITICAL_GROUP,
146}
149# ---------------------------------------------------------------------------
150# Engine
151# ---------------------------------------------------------------------------
153class ClassificationEngine:
154 """
155 Multi-signal, multi-language classification engine implementing the
156 D2.4a spec plus an optional dog-whistle lexicon.
158 Signal resolution order:
159 1. Load the request's language vocabulary (falls back to English)
160 2. Identity anchor check (require_target_present rule)
161 3. Per-row harm pattern matching + dog-whistle lexicon matching
162 4. Suppression (negation / counter-speech / benign-context)
163 5. Azure NLP sentiment amplification
164 6. Weighted score aggregation
165 7. Ambiguity resolution (highest_weight_wins / conservative tie-break)
166 8. Threshold decision + provenance hash + counter-narrative guidance
167 """
169 def __init__(
170 self,
171 taxonomy: TaxonomyConfig,
172 azure_client: Optional[AzureNLPClient] = None,
173 patterns_dir: Optional[str] = None,
174 dogwhistles: Optional[DogwhistleLexicon] = None,
175 ):
176 self.taxonomy = taxonomy
177 self.azure = azure_client or AzureNLPClient() # fallback mode if no creds
178 self.patterns_dir = patterns_dir or DEFAULT_PATTERNS_DIR
179 self.dogwhistles = dogwhistles
180 # harm_mechanism -> category name, derived once from the taxonomy,
181 # so dog-whistle matches (which name a harm_mechanism directly, not
182 # a category) can be scored through the same pipeline as taxonomy rows.
183 self._mechanism_to_category = {row.harm_mechanism: category for category, row in taxonomy.all_rows()}
185 def _load_patterns(self, requested_language: str) -> tuple[LanguagePatterns, Optional[str]]:
186 try:
187 return load_language_patterns(self.patterns_dir, requested_language), None
188 except FileNotFoundError:
189 fallback = load_language_patterns(self.patterns_dir, DEFAULT_LANGUAGE)
190 return fallback, f"Language '{requested_language}' not available — fell back to '{DEFAULT_LANGUAGE}'."
192 def classify(self, request: ClassifyRequest) -> ClassificationResult:
193 text = request.text
194 full_text = f"{request.context or ''} {text}".strip()
195 patterns, language_fallback_note = self._load_patterns(request.language or DEFAULT_LANGUAGE)
196 hash_value = content_hash(text, request.context, self.taxonomy.version)
198 # Step 1: Run Azure NLP (non-blocking — fallback if unavailable)
199 azure_result = self.azure.analyze(full_text)
201 # Step 2: Check for identity anchor (target group presence)
202 identity_axis_detected = _detect_identity_anchor(full_text, patterns)
204 rules = self.taxonomy.ambiguity_rules
205 if rules.require_target_present and not identity_axis_detected:
206 rationale = "No target group identity detected — require_target_present=True"
207 if language_fallback_note:
208 rationale += f" {language_fallback_note}"
209 return self._no_harm_result(request, rationale, patterns, hash_value)
211 # Step 3: Score each taxonomy row + dog-whistle lexicon entries.
212 # Each signal carries its own decision_threshold directly (rather
213 # than being re-derived later via a row_id lookup) so a dog-whistle
214 # match — which has no corresponding taxonomy row — can't fall
215 # through to an unrelated config value as a threshold.
216 # Each entry: (score, SignalMatch, CategorySpec, dogwhistle_term_or_None, decision_threshold)
217 signal_matches: list[tuple[float, SignalMatch, CategorySpec, Optional[str], float]] = []
218 azure_amp = _azure_negative_amplifier(azure_result)
220 for category, row in self.taxonomy.all_rows():
221 harm_matched, pattern, match_obj, matched_text = _detect_harm_signals(full_text, row, patterns)
222 if not harm_matched:
223 continue
224 if _is_suppressed(matched_text, match_obj, patterns):
225 continue
227 cat_spec = self.taxonomy.get_category(category)
228 score = row.signal_weight * azure_amp
229 match = SignalMatch(
230 row_id=row.row_id,
231 harm_mechanism=row.harm_mechanism,
232 target_type=TargetType(row.target_type) if row.target_type in TargetType._value2member_map_ else TargetType.UNKNOWN,
233 identity_axis=IdentityAxis(row.identity_axis) if row.identity_axis in IdentityAxis._value2member_map_ else IdentityAxis.UNKNOWN,
234 signal_weight=row.signal_weight,
235 matched_pattern=pattern,
236 azure_sentiment_score=azure_result.sentiment_score_negative if not azure_result.is_fallback else None,
237 azure_entity_detected=None,
238 )
239 signal_matches.append((score, match, cat_spec, None, row.decision_threshold))
241 if self.dogwhistles is not None:
242 for entry in self.dogwhistles.detect(full_text):
243 category = self._mechanism_to_category.get(entry.harm_mechanism)
244 cat_spec = self.taxonomy.get_category(category) if category else None
245 if cat_spec is None:
246 continue # dog-whistle maps to a mechanism this taxonomy doesn't define
247 score = entry.signal_weight * azure_amp
248 match = SignalMatch(
249 row_id=f"DOGWHISTLE:{entry.term}",
250 harm_mechanism=entry.harm_mechanism,
251 target_type=_AXIS_TO_TARGET_TYPE.get(entry.identity_axis, TargetType.UNKNOWN),
252 identity_axis=IdentityAxis(entry.identity_axis) if entry.identity_axis in IdentityAxis._value2member_map_ else IdentityAxis.UNKNOWN,
253 signal_weight=entry.signal_weight,
254 matched_pattern=entry.term,
255 azure_sentiment_score=azure_result.sentiment_score_negative if not azure_result.is_fallback else None,
256 azure_entity_detected=None,
257 )
258 signal_matches.append((score, match, cat_spec, entry.term, entry.decision_threshold))
260 # Step 4: Check min_signal_count
261 if len(signal_matches) < rules.min_signal_count:
262 rationale = f"Signal count {len(signal_matches)} below minimum {rules.min_signal_count}"
263 return self._no_harm_result(request, rationale, patterns, hash_value)
265 # Step 5: Ambiguity resolution — highest_weight_wins
266 if rules.multi_signal_conflict == "highest_weight_wins":
267 signal_matches.sort(key=lambda x: x[0], reverse=True)
269 best_score, best_match, best_cat, best_dogwhistle_term, threshold = signal_matches[0]
271 # Step 6: Threshold decision
272 # Conservative tie-break
273 if best_score == threshold and rules.threshold_tie == "conservative":
274 is_harmful = True
275 else:
276 is_harmful = best_score >= threshold
278 all_matches = [m for _, m, _, _, _ in signal_matches]
280 rationale = self._build_rationale(
281 is_harmful, best_score, threshold, best_match, azure_result, identity_axis_detected
282 )
283 if language_fallback_note:
284 rationale += f" {language_fallback_note}"
286 category_map = {
287 "dehumanization": HarmCategory.DEHUMANIZATION,
288 "incitement": HarmCategory.INCITEMENT,
289 "narrative_distortion": HarmCategory.NARRATIVE_DISTORTION,
290 }
292 return ClassificationResult(
293 request_id=request.request_id,
294 text_excerpt=text[:200],
295 is_harmful=is_harmful,
296 harm_category=category_map.get(best_cat.name, HarmCategory.NONE) if is_harmful else HarmCategory.NONE,
297 confidence=round(min(best_score, 1.0), 4),
298 target_type=best_match.target_type if is_harmful else None,
299 identity_axis=best_match.identity_axis if is_harmful else None,
300 harm_mechanism=best_match.harm_mechanism if is_harmful else None,
301 signals_matched=all_matches,
302 decision_rationale=rationale,
303 taxonomy_version=self.taxonomy.version,
304 language=patterns.language,
305 language_confidence=patterns.confidence,
306 dogwhistle_matched=best_dogwhistle_term if is_harmful else None,
307 counter_narrative_guidance=guidance_for(best_match.harm_mechanism) if is_harmful else None,
308 content_hash=hash_value,
309 )
311 def _no_harm_result(
312 self, request: ClassifyRequest, rationale: str, patterns: LanguagePatterns, hash_value: str
313 ) -> ClassificationResult:
314 return ClassificationResult(
315 request_id=request.request_id,
316 text_excerpt=request.text[:200],
317 is_harmful=False,
318 harm_category=HarmCategory.NONE,
319 confidence=0.0,
320 signals_matched=[],
321 decision_rationale=rationale,
322 taxonomy_version=self.taxonomy.version,
323 language=patterns.language,
324 language_confidence=patterns.confidence,
325 content_hash=hash_value,
326 )
328 def _build_rationale(
329 self,
330 is_harmful: bool,
331 score: float,
332 threshold: float,
333 match: SignalMatch,
334 azure: AzureNLPResult,
335 identity_axis: Optional[str],
336 ) -> str:
337 parts = []
338 if is_harmful:
339 # Use the axis actually detected in the text rather than the
340 # matched row's static tag when both are available — the two
341 # can differ (a row tagged race_ethnicity can still fire on text
342 # whose only identity anchor is national_origin), and reporting
343 # both a static "targeting X" and a detected "Identity axis: Y"
344 # in the same rationale read as self-contradictory.
345 axis_for_rationale = identity_axis or match.identity_axis.value
346 parts.append(f"HARM DETECTED: {match.harm_mechanism} targeting {axis_for_rationale}.")
347 parts.append(f"Confidence {score:.3f} ≥ threshold {threshold:.3f}.")
348 parts.append(f"Matched row {match.row_id}.")
349 else:
350 parts.append(f"NO HARM: score {score:.3f} below threshold {threshold:.3f}.")
352 if not azure.is_fallback:
353 parts.append(f"Azure sentiment: {azure.sentiment} (neg={azure.sentiment_score_negative:.2f}).")
354 else:
355 parts.append("Azure NLP: fallback mode (no credentials).")
357 if identity_axis:
358 parts.append(f"Identity axis detected: {identity_axis}.")
360 return " ".join(parts)