Coverage for narrative_harm_classifier/cli.py: 84%

1""" 

2narrative_harm_classifier/cli.py — `nhc` command-line interface. 

3 

4Ties together classification, the API server, escalation-chain tracking, 

5and the templated benchmark suite for local/scriptable use without needing 

6to write any Python or run curl against the API. 

7""" 

8 

9import json 

10import sys 

11from typing import Optional 

12 

13import typer 

14 

15# Rationale/report text uses non-ASCII characters (e.g. "≥"). On Windows, 

16# stdout defaults to the legacy console codepage (cp1252) rather than UTF-8, 

17# which crashes on encode. Force UTF-8 so `nhc` works in a default terminal. 

18for _stream in (sys.stdout, sys.stderr): 

19 if hasattr(_stream, "reconfigure"): 

20 _stream.reconfigure(encoding="utf-8", errors="replace") 

21 

22from narrative_harm_classifier.core.config import get_settings 

23from narrative_harm_classifier.core.models import ClassifyRequest 

24from narrative_harm_classifier.classifier.factory import build_engine, build_tracker, build_benchmark_runner 

25from narrative_harm_classifier.classifier.validators.i18n_smoke import run_i18n_smoke 

26 

27app = typer.Typer(name="nhc", help="Narrative Harm Classifier — classify, track, and benchmark.") 

28track_app = typer.Typer(help="Escalation-chain tracking across a source's observation history.") 

29benchmark_app = typer.Typer(help="Templated functional-test benchmark suite.") 

30app.add_typer(track_app, name="track") 

31app.add_typer(benchmark_app, name="benchmark") 

32 

33 

34@app.command() 

35def serve( 

36 host: str = typer.Option("0.0.0.0", help="Bind host"), 

37 port: int = typer.Option(8000, help="Bind port"), 

38 reload: bool = typer.Option(False, help="Auto-reload on code changes (development only)"), 

39): 

40 """Start the FastAPI server (docs at http://<host>:<port>/docs).""" 

41 import uvicorn 

42 

43 uvicorn.run("narrative_harm_classifier.api.main:app", host=host, port=port, reload=reload) 

44 

45 

46@app.command() 

47def classify( 

48 text: str = typer.Argument(..., help="Text to classify"), 

49 context: Optional[str] = typer.Option(None, help="Optional surrounding context"), 

50 language: str = typer.Option("en", "--language", "-l", help="ISO 639-1 language code (e.g. en, es, fr, ru, ar, ig, yo, ha)"), 

51): 

52 """Classify a single text item and print the full result as JSON.""" 

53 engine = build_engine(get_settings()) 

54 result = engine.classify(ClassifyRequest(text=text, context=context, language=language)) 

55 typer.echo(result.model_dump_json(indent=2)) 

56 

57 

58@track_app.command("observe") 

59def track_observe( 

60 source_id: str = typer.Argument(..., help="Identifier for the tracked source (URL, handle, doc id...)"), 

61 text: str = typer.Argument(..., help="Text to classify and append to this source's history"), 

62 language: str = typer.Option("en", "--language", "-l", help="ISO 639-1 language code"), 

63): 

64 """Classify a text and append it to a source's escalation history.""" 

65 tracker = build_tracker(get_settings()) 

66 obs = tracker.observe(source_id, ClassifyRequest(text=text, language=language)) 

67 typer.echo(obs.model_dump_json(indent=2)) 

68 

69 

70@track_app.command("show") 

71def track_show( 

72 source_id: str = typer.Argument(..., help="Source to show"), 

73 window: int = typer.Option(20, help="Number of most recent observations to consider"), 

74): 

75 """Show a source's current severity, trend, and risk level.""" 

76 tracker = build_tracker(get_settings()) 

77 profile = tracker.profile(source_id, window=window) 

78 if profile.observation_count == 0: 

79 typer.echo(f"No observations recorded for source '{source_id}'", err=True) 

80 raise typer.Exit(code=1) 

81 

82 typer.echo(f"Source: {profile.source_id}") 

83 typer.echo(f"Observations: {profile.observation_count}") 

84 typer.echo(f"Current severity: {profile.current_severity.name} ({int(profile.current_severity)})") 

85 typer.echo(f"Rolling avg severity:{profile.rolling_avg_severity:.2f}") 

86 typer.echo(f"Trend: {profile.trend}") 

87 typer.echo(f"Risk level: {profile.risk_level.upper()}") 

88 

89 

90@track_app.command("list") 

91def track_list(window: int = typer.Option(20, help="Number of most recent observations to consider per source")): 

92 """List all tracked sources, sorted by risk (highest first).""" 

93 tracker = build_tracker(get_settings()) 

94 profiles = tracker.list_profiles(window=window) 

95 if not profiles: 

96 typer.echo("No tracked sources yet.") 

97 return 

98 

99 typer.echo(f"{'SOURCE':<30}{'RISK':<10}{'TREND':<20}{'SEVERITY':<24}{'OBS':<5}") 

100 for p in profiles: 

101 typer.echo( 

102 f"{p.source_id:<30}{p.risk_level.upper():<10}{p.trend:<20}" 

103 f"{p.current_severity.name:<24}{p.observation_count:<5}" 

104 ) 

105 

106 

107@track_app.command("verify") 

108def track_verify(source_id: str = typer.Argument(..., help="Source whose observation history to verify")): 

109 """ 

110 Recompute the tamper-evident hash chain for a source's observation 

111 history and report whether it's intact. 

112 """ 

113 tracker = build_tracker(get_settings()) 

114 result = tracker.verify_chain(source_id) 

115 if result.observation_count == 0: 

116 typer.echo(f"No observations recorded for source '{source_id}'", err=True) 

117 raise typer.Exit(code=1) 

118 

119 typer.echo(f"Source: {result.source_id}") 

120 typer.echo(f"Observations: {result.observation_count}") 

121 typer.echo(f"Chain intact: {'YES' if result.intact else 'NO — TAMPERING DETECTED'}") 

122 if not result.intact: 

123 typer.echo(f"First broken link at observation id: {result.first_broken_id}", err=True) 

124 raise typer.Exit(code=1) 

125 

126 

127@benchmark_app.command("run") 

128def benchmark_run(): 

129 """Run the templated benchmark suite and print aggregate + per-test-type results.""" 

130 runner = build_benchmark_runner(get_settings()) 

131 report = runner.run() 

132 

133 typer.echo(f"Taxonomy version: {report.taxonomy_version}") 

134 typer.echo(f"Total cases: {report.sample_count}\n") 

135 

136 typer.echo(f"{'TEST TYPE':<22}{'N':<6}{'PRECISION':<11}{'RECALL':<9}{'FPR':<8}{'F1':<8}") 

137 typer.echo( 

138 f"{'overall':<22}{report.overall.sample_count:<6}{report.overall.precision:<11}" 

139 f"{report.overall.recall:<9}{report.overall.fpr:<8}{report.overall.f1:<8}" 

140 ) 

141 for t in report.by_test_type: 

142 typer.echo(f"{t.test_type:<22}{t.sample_count:<6}{t.precision:<11}{t.recall:<9}{t.fpr:<8}{t.f1:<8}") 

143 

144 inconsistent = [g for g in report.group_consistency if not g.consistent] 

145 typer.echo(f"\nCross-group consistency: {len(report.group_consistency) - len(inconsistent)}/" 

146 f"{len(report.group_consistency)} templates consistent across groups") 

147 if inconsistent: 

148 typer.echo("Inconsistent templates (engine's verdict changed depending on which group was named):") 

149 for g in inconsistent: 

150 typer.echo(f" {g.template_id} ({g.test_type}): {json.dumps(g.verdicts)}") 

151 

152 

153@benchmark_app.command("i18n") 

154def benchmark_i18n(): 

155 """ 

156 Run the smaller per-language smoke test suite (see data/i18n_smoke_tests.yaml — 

157 NOT a full replication of the English benchmark; see README Limitations 

158 for why the experimental-tier languages get lighter coverage). 

159 """ 

160 settings = get_settings() 

161 engine = build_engine(settings) 

162 report = run_i18n_smoke(engine, settings.i18n_smoke_tests_path) 

163 

164 typer.echo(f"Total cases: {report.total} Passed: {report.passed}\n") 

165 typer.echo(f"{'LANGUAGE':<10}{'PASSED':<10}{'TOTAL':<10}") 

166 for lang, (passed, total) in sorted(report.by_language.items()): 

167 typer.echo(f"{lang:<10}{passed:<10}{total:<10}") 

168 

169 if report.failed_cases: 

170 typer.echo("\nFailed cases:") 

171 for c in report.failed_cases: 

172 typer.echo(f" [{c.language}] expected={c.expected_is_harmful} got={c.actual_is_harmful}: {c.text!r}") 

173 

174 

175def main(): 

176 app() 

177 

178 

179if __name__ == "__main__": 

180 main()