Coverage for src/lilbee/cli/commands/analyze.py: 100%

1"""The analyze command: read the corpus, recommend a profile, and the analyze tip.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6from collections.abc import Iterator 

7from contextlib import contextmanager 

8from pathlib import Path 

9from typing import TYPE_CHECKING 

10 

11import typer 

12from rich.table import Table 

13from rich.text import Text 

14 

15from lilbee.app.analyze import hide_tip, tip_state 

16from lilbee.app.profiles import file_failure_message 

17from lilbee.cli import theme 

18from lilbee.cli.app import apply_overrides, console, data_dir_option, global_option 

19from lilbee.cli.commands._shared import MAX_FAILURES_SHOWN, fail, line 

20from lilbee.cli.commands.profile import render_changes 

21from lilbee.cli.helpers import json_output 

22from lilbee.core.config import cfg 

23from lilbee.core.profile_files import ProfileFolder, ProfileStore 

24from lilbee.runtime.cancellation import sigint_cancel 

25from lilbee.runtime.console import PlainConsole 

26from lilbee.runtime.progress import AnalyzeEvent, EventType 

27 

28if TYPE_CHECKING: 

29 from lilbee.app.analyze import AnalyzeRequest 

30 from lilbee.runtime.progress import DetailedProgressCallback, ProgressEvent 

31 from lilbee.server.models import AnalyzeResponse 

32 

33READING_LABEL = "Reading files" 

34CANCELLED_MESSAGE = "Analyze was cancelled; nothing was saved." 

35OFF_ALONE_MESSAGE = "--off hides the tip and takes no folder, --apply, --save or --target." 

36TIP_HIDDEN_MESSAGE = "The analyze tip is hidden for this project." 

37 

38_directory_argument = typer.Argument( 

39 None, 

40 metavar="[DIR]", 

41 help="A folder to analyze; omit it to analyze the files lilbee indexes.", 

42 show_default=False, 

43) 

44_apply_option = typer.Option( 

45 False, "--apply", help="Save the recommended profile and switch this project to it." 

46) 

47_save_option = typer.Option( 

48 None, 

49 "--save", 

50 metavar="NAME", 

51 help="Save the recommended profile as NAME without switching to it.", 

52) 

53_target_option = typer.Option( 

54 None, 

55 "--target", 

56 help="Where to save with --apply or --save: project (the default when there is one) or global.", 

57 show_default=False, 

58) 

59_off_option = typer.Option(False, "--off", help="Hide the analyze tip; reads no files.") 

60 

61 

62def print_tip_if_shown(root: Path) -> None: 

63 """Print the analyze tip in text mode when the project at *root* should see it.""" 

64 from lilbee.app.analyze import TIP_TEXT, tip_shows 

65 

66 if not cfg.json_mode and tip_shows(root): 

67 console.print(Text(TIP_TEXT, style=theme.MUTED), soft_wrap=True) 

68 

69 

70@contextmanager 

71def _progress() -> Iterator[DetailedProgressCallback]: 

72 """A progress bar over the files analyze extracts; off in JSON mode.""" 

73 from rich.progress import BarColumn, MofNCompleteColumn, Progress 

74 

75 from lilbee.runtime.progress.columns import literal_text_column 

76 

77 with Progress( 

78 literal_text_column("{task.description}"), 

79 BarColumn(), 

80 MofNCompleteColumn(), 

81 transient=True, 

82 console=PlainConsole(stderr=True), 

83 disable=cfg.json_mode, 

84 ) as progress: 

85 task = progress.add_task(READING_LABEL, total=None) 

86 

87 def on_progress(event_type: EventType, data: ProgressEvent) -> None: 

88 # the callback carries every event type; only analyze events move this bar 

89 if event_type is EventType.ANALYZE and isinstance(data, AnalyzeEvent): 

90 progress.update(task, completed=data.done, total=data.total) 

91 

92 yield on_progress 

93 

94 

95def _reading_line(report: AnalyzeResponse) -> str: 

96 read = f"Read {report.files_read} of {report.documents_total} documents" 

97 if report.files_counted: 

98 return f"{read}; counted {report.files_counted} code, image and archive files." 

99 return f"{read}." 

100 

101 

102def _render_reading(report: AnalyzeResponse) -> None: 

103 line(_reading_line(report), theme.ACCENT) 

104 if report.files_read + len(report.failed) < report.documents_total: 

105 line(f"The documents are sampled evenly; analyze_max_files is {report.cap}.", theme.MUTED) 

106 types = Table("File type", "Files") 

107 for kind, count in report.file_types.items(): 

108 types.add_row(Text(kind), str(count)) 

109 console.print(types) 

110 pdf = report.pdf 

111 line(f"Code files: {report.code_share:.0%}") 

112 line(f"Scanned pages: {pdf.scanned_pages} of {pdf.pages} PDF pages") 

113 line(f"Scanned share, image files included: {pdf.scanned_share:.0%}") 

114 line(f"PDFs with tables: {pdf.files_with_tables} of {pdf.files}") 

115 if pdf.median_pages is not None: 

116 line(f"Median PDF length: {pdf.median_pages:g} pages") 

117 if report.median_chars is not None: 

118 line(f"Median length of other files: {report.median_chars:g} characters") 

119 

120 

121def _render_languages(report: AnalyzeResponse) -> None: 

122 if not report.languages: 

123 line("No language detected: no file has enough text.") 

124 return 

125 table = Table("Language", "Share", "Search stemmer", "Tesseract") 

126 for lang in report.languages: 

127 stemmer = lang.fts_language.value if lang.fts_language else "none" 

128 tesseract = "installed" if lang.ocr_supported else "not installed" 

129 table.add_row(lang.code, f"{lang.share:.0%}", stemmer, tesseract) 

130 console.print(table) 

131 

132 

133def _render_failures(report: AnalyzeResponse) -> None: 

134 for folder in report.unreadable_folders: 

135 line(f"Could not open folder: {folder}", theme.WARNING) 

136 if not report.failed: 

137 return 

138 line(f"{len(report.failed)} files could not be read:", theme.WARNING) 

139 for failure in report.failed[:MAX_FAILURES_SHOWN]: 

140 line(f" {failure.file}: {failure.error}") 

141 hidden = len(report.failed) - MAX_FAILURES_SHOWN 

142 if hidden > 0: 

143 line(f" and {hidden} more; --json lists them all") 

144 

145 

146def _render_recommendation(report: AnalyzeResponse) -> None: 

147 rec = report.recommendation 

148 line(f"Recommended: {rec.name or rec.builtin}", theme.ACCENT) 

149 for reason in rec.reasons: 

150 line(f" {reason.key}: {reason.text}") 

151 render_changes(rec.changes) 

152 if rec.kept: 

153 line(f"Keeps your values of: {', '.join(rec.kept)}") 

154 for note in rec.notes: 

155 line(note, theme.MUTED) 

156 

157 

158def _render_saved(report: AnalyzeResponse) -> None: 

159 saved = report.saved 

160 if saved is not None and saved.applied: 

161 line(f"This project now uses {saved.name}: {saved.path}", theme.ACCENT) 

162 elif saved is not None: 

163 line(f"Saved {saved.name}: {saved.path}") 

164 elif report.recommendation.name is not None: 

165 line("Run lilbee analyze --apply to save it and switch to it.", theme.MUTED) 

166 if saved is not None: 

167 for warning in saved.warnings: 

168 line(warning, theme.WARNING) 

169 

170 

171def _render(report: AnalyzeResponse) -> None: 

172 _render_reading(report) 

173 _render_languages(report) 

174 _render_failures(report) 

175 _render_recommendation(report) 

176 _render_saved(report) 

177 

178 

179def _analyze(request: AnalyzeRequest) -> AnalyzeResponse: 

180 """Run analyze with a progress bar and Ctrl+C as a clean cancel; a refusal exits 1.""" 

181 from lilbee.app.analyze import run_analysis 

182 from lilbee.runtime.cancellation import TaskCancelledError 

183 from lilbee.server.models import AnalyzeResponse 

184 

185 try: 

186 with _progress() as on_progress, sigint_cancel() as cancel: 

187 report = asyncio.run( 

188 run_analysis(ProfileStore(), request, on_progress=on_progress, cancel=cancel) 

189 ) 

190 except TaskCancelledError: 

191 fail(CANCELLED_MESSAGE) 

192 except ValueError as exc: 

193 fail(str(exc)) 

194 except OSError as exc: 

195 fail(file_failure_message(exc)) 

196 return AnalyzeResponse.from_report(report) 

197 

198 

199def _hide_tip() -> None: 

200 from lilbee.server.models import AnalyzeStateResponse # CLI startup: 35 ms to import 

201 

202 try: 

203 hide_tip(cfg.data_root) 

204 except OSError as exc: 

205 fail(file_failure_message(exc)) 

206 if cfg.json_mode: 

207 json_output( 

208 AnalyzeStateResponse.from_state(tip_state(cfg.data_root)).model_dump(mode="json") 

209 ) 

210 else: 

211 line(TIP_HIDDEN_MESSAGE) 

212 

213 

214def analyze_cmd( 

215 directory: Path | None = _directory_argument, 

216 apply: bool = _apply_option, 

217 save: str | None = _save_option, 

218 target: ProfileFolder | None = _target_option, 

219 off: bool = _off_option, 

220 data_dir: Path | None = data_dir_option, 

221 use_global: bool = global_option, 

222) -> None: 

223 """Read your documents and recommend a profile; --apply saves it and switches to it.""" 

224 from lilbee.app.analyze import AnalyzeRequest 

225 

226 apply_overrides(data_dir=data_dir, use_global=use_global) 

227 if off: 

228 if directory is not None or apply or save is not None or target is not None: 

229 fail(OFF_ALONE_MESSAGE) 

230 _hide_tip() 

231 return 

232 request = AnalyzeRequest(directory=directory, apply=apply, save=save, target=target) 

233 response = _analyze(request) 

234 if cfg.json_mode: 

235 json_output(response.model_dump(mode="json")) 

236 return 

237 _render(response)