Coverage for narrative_harm_classifier/classifier/taxonomy/loader.py: 92%

1""" 

2classifier/taxonomy/loader.py — Versioned taxonomy config loader. 

3 

4Loads taxonomy_v1.yaml (or any versioned config) and provides structured access 

5to taxonomy rows, thresholds, and ambiguity resolution rules. 

6Supports M1 baseline reproducibility via version pinning. 

7""" 

8 

9from typing import Optional 

10from dataclasses import dataclass, field 

11from functools import lru_cache 

12 

13from narrative_harm_classifier.core.yaml_loader import load_yaml_file 

14 

15 

16@dataclass 

17class TaxonomyRow: 

18 row_id: str 

19 target_type: str 

20 harm_mechanism: str 

21 identity_axis: str 

22 signal_weight: float 

23 decision_threshold: float 

24 positive_examples: list[str] = field(default_factory=list) 

25 negative_examples: list[str] = field(default_factory=list) 

26 

27 

28@dataclass 

29class CategorySpec: 

30 id: str 

31 name: str 

32 description: str 

33 priority: int 

34 precision_min: float 

35 recall_min: float 

36 fpr_max: float 

37 rows: list[TaxonomyRow] = field(default_factory=list) 

38 

39 

40@dataclass 

41class AmbiguityRules: 

42 multi_signal_conflict: str 

43 threshold_tie: str 

44 context_window_tokens: int 

45 require_target_present: bool 

46 min_signal_count: int 

47 

48 

49@dataclass 

50class TaxonomyConfig: 

51 version: str 

52 effective_date: str 

53 baseline_tag: str 

54 priority_category: str 

55 categories: list[CategorySpec] 

56 ambiguity_rules: AmbiguityRules 

57 

58 def get_category(self, name: str) -> Optional[CategorySpec]: 

59 return next((c for c in self.categories if c.name == name), None) 

60 

61 def get_row(self, row_id: str) -> Optional[TaxonomyRow]: 

62 for cat in self.categories: 

63 for row in cat.rows: 

64 if row.row_id == row_id: 

65 return row 

66 return None 

67 

68 def all_rows(self) -> list[tuple[str, TaxonomyRow]]: 

69 """Returns (category_name, row) pairs for all taxonomy rows.""" 

70 result = [] 

71 for cat in self.categories: 

72 for row in cat.rows: 

73 result.append((cat.name, row)) 

74 return result 

75 

76 

77@lru_cache(maxsize=8) 

78def load_taxonomy(config_path: str) -> TaxonomyConfig: 

79 """Load and parse taxonomy YAML. Cached per path for performance.""" 

80 raw = load_yaml_file(config_path) 

81 

82 categories = [] 

83 for cat_raw in raw.get("categories", []): 

84 rows = [] 

85 for row_raw in cat_raw.get("taxonomy_rows", []): 

86 rows.append(TaxonomyRow( 

87 row_id=row_raw["row_id"], 

88 target_type=row_raw["target_type"], 

89 harm_mechanism=row_raw["harm_mechanism"], 

90 identity_axis=row_raw["identity_axis"], 

91 signal_weight=row_raw["signal_weight"], 

92 decision_threshold=row_raw["decision_threshold"], 

93 positive_examples=row_raw.get("examples", {}).get("positive", []), 

94 negative_examples=row_raw.get("examples", {}).get("negative", []), 

95 )) 

96 

97 thresholds = cat_raw.get("thresholds", {}) 

98 categories.append(CategorySpec( 

99 id=cat_raw["id"], 

100 name=cat_raw["name"], 

101 description=cat_raw["description"], 

102 priority=cat_raw["priority"], 

103 precision_min=thresholds.get("precision_min", 0.70), 

104 recall_min=thresholds.get("recall_min", 0.65), 

105 fpr_max=thresholds.get("fpr_max", 0.20), 

106 rows=rows, 

107 )) 

108 

109 ar = raw.get("ambiguity_rules", {}) 

110 ambiguity_rules = AmbiguityRules( 

111 multi_signal_conflict=ar.get("multi_signal_conflict", "highest_weight_wins"), 

112 threshold_tie=ar.get("threshold_tie", "conservative"), 

113 context_window_tokens=ar.get("context_window_tokens", 256), 

114 require_target_present=ar.get("require_target_present", True), 

115 min_signal_count=ar.get("min_signal_count", 1), 

116 ) 

117 

118 return TaxonomyConfig( 

119 version=raw["version"], 

120 effective_date=raw["effective_date"], 

121 baseline_tag=raw["baseline_tag"], 

122 priority_category=raw["priority_category"], 

123 categories=categories, 

124 ambiguity_rules=ambiguity_rules, 

125 )