Coverage for narrative_harm_classifier/classifier/tracking/models.py: 98%

1""" 

2classifier/tracking/models.py — Escalation-chain data model. 

3 

4Tracks a *source* (an account, outlet, or document stream identified by an 

5arbitrary caller-supplied source_id) across multiple classified observations, 

6and scores whether its rhetoric is climbing a harm-escalation ladder rather 

7than treating each text as an isolated event. 

8 

9The severity ladder below is a simplified, project-specific model inspired by 

10general narrative-escalation research (e.g. othering -> dehumanization -> 

11criminalization -> violence calls). It is not a validated academic scale — 

12it exists to give a consistent, explainable ordering across taxonomy rows so 

13trend direction can be computed deterministically. 

14""" 

15 

16from datetime import datetime 

17from enum import IntEnum 

18from typing import Optional 

19 

20from pydantic import BaseModel, Field 

21 

22 

23class SeverityLevel(IntEnum): 

24 NONE = 0 

25 NARRATIVE_DISTORTION = 1 

26 DEMONIZATION_OBJECTIFICATION = 2 

27 ANIMALIZATION_CRIMINALIZATION = 3 

28 DIRECT_VIOLENCE_CALL = 4 

29 

30 

31# harm_mechanism -> severity, per the ladder above 

32HARM_MECHANISM_SEVERITY: dict[str, SeverityLevel] = { 

33 "false_attribution": SeverityLevel.NARRATIVE_DISTORTION, 

34 "demonization": SeverityLevel.DEMONIZATION_OBJECTIFICATION, 

35 "objectification": SeverityLevel.DEMONIZATION_OBJECTIFICATION, 

36 "animalization": SeverityLevel.ANIMALIZATION_CRIMINALIZATION, 

37 "criminalization": SeverityLevel.ANIMALIZATION_CRIMINALIZATION, 

38 "direct_call_to_violence": SeverityLevel.DIRECT_VIOLENCE_CALL, 

39} 

40 

41 

42def severity_for_mechanism(harm_mechanism: Optional[str]) -> SeverityLevel: 

43 if not harm_mechanism: 

44 return SeverityLevel.NONE 

45 return HARM_MECHANISM_SEVERITY.get(harm_mechanism, SeverityLevel.NONE) 

46 

47 

48class Observation(BaseModel): 

49 id: Optional[int] = None 

50 source_id: str 

51 text_excerpt: str 

52 is_harmful: bool 

53 harm_category: str 

54 harm_mechanism: Optional[str] = None 

55 confidence: float 

56 severity: SeverityLevel 

57 observed_at: datetime = Field(default_factory=datetime.utcnow) 

58 content_hash: str = "" 

59 prev_hash: str = "" 

60 record_hash: str = "" 

61 

62 

63class SourceProfile(BaseModel): 

64 source_id: str 

65 observation_count: int 

66 current_severity: SeverityLevel 

67 rolling_avg_severity: float 

68 trend: str # "escalating" | "stable" | "de-escalating" | "insufficient_data" 

69 risk_level: str # "low" | "watch" | "elevated" | "critical" 

70 history: list[Observation] = [] 

71 

72 

73class ChainVerification(BaseModel): 

74 source_id: str 

75 observation_count: int 

76 intact: bool 

77 first_broken_id: Optional[int] = None 

78 

79 

80RISK_LEVELS = ("low", "watch", "elevated", "critical") 

81 

82 

83def risk_level_for(current_severity: SeverityLevel, trend: str) -> str: 

84 """ 

85 Deterministic risk mapping: base risk from current severity, bumped up one 

86 level when the trend is escalating. Kept simple and explainable rather than 

87 a learned model, consistent with the rest of this project's rationale-driven 

88 design. 

89 """ 

90 if current_severity == SeverityLevel.NONE: 

91 base = 0 

92 elif current_severity == SeverityLevel.NARRATIVE_DISTORTION: 

93 base = 1 

94 elif current_severity == SeverityLevel.DEMONIZATION_OBJECTIFICATION: 

95 base = 2 

96 else: # ANIMALIZATION_CRIMINALIZATION or DIRECT_VIOLENCE_CALL 

97 base = 3 

98 

99 if trend == "escalating" and base < 3: 

100 base += 1 

101 

102 return RISK_LEVELS[base]