Coverage for src/lilbee/cli/tui/screens/analyze_report.py: 100%
203 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-11-06 08:17 +0000
1"""The /analyze task and the report it opens: findings, the recommended profile, Apply or save."""
3from __future__ import annotations
5from collections.abc import Callable, Mapping, Sequence
6from dataclasses import dataclass
7from enum import StrEnum
8from pathlib import Path
9from typing import TYPE_CHECKING, ClassVar, NoReturn
11from textual import on
12from textual.app import ComposeResult
13from textual.binding import Binding, BindingType
14from textual.containers import Horizontal, VerticalScroll
15from textual.content import Content
16from textual.screen import Screen
17from textual.widgets import DataTable, Footer, Static
19from lilbee.app import profiles
20from lilbee.app.analyze import (
21 PICK_REASON_KEY,
22 AnalyzeReport,
23 AnalyzeRequest,
24 LanguageRow,
25 Recommendation,
26 run_analysis,
27)
28from lilbee.cli.tui import messages as msg
29from lilbee.cli.tui.browse_bindings import browse_back_bindings
30from lilbee.cli.tui.screens.profile_dialogs import (
31 add_cost_column,
32 diff_cells,
33 run_profile_op,
34 shorten_middle,
35 start_profile_switch,
36 start_switch,
37 switch_plan,
38)
39from lilbee.cli.tui.task_queue import PERCENT, TaskType
40from lilbee.cli.tui.thread_safe import call_from_thread
41from lilbee.cli.tui.widgets.bottom_bars import BottomBars
42from lilbee.cli.tui.widgets.confirm_dialog import ConfirmPill
43from lilbee.cli.tui.widgets.status_bar import ViewTabs
44from lilbee.cli.tui.widgets.task_bar import TaskBar
45from lilbee.cli.tui.widgets.top_bars import TopBars
46from lilbee.core.profile_files import ProfileStore
47from lilbee.data.analyze import CorpusSignals
48from lilbee.runtime import asyncio_loop
49from lilbee.runtime.progress import AnalyzeEvent, EventType
51if TYPE_CHECKING:
52 from lilbee.cli.tui.app import LilbeeApp
53 from lilbee.cli.tui.widgets.task_bar_controller import ProgressReporter
54 from lilbee.runtime.progress import ProgressEvent
57class ReportAction(StrEnum):
58 """The report's action pills."""
60 APPLY = "apply"
61 SAVE = "save"
62 CLOSE = "close"
65_REASON_WIDTH = 48 # the longest reason a failed-file row shows
68@dataclass(frozen=True)
69class FinishedAnalysis:
70 """A finished analyze run: its report and the folder it read."""
72 report: AnalyzeReport
73 folder: str
76def _short_reason(error: str) -> str:
77 """The first line of a parser's *error*, cut to fit a table cell."""
78 lines = error.strip().splitlines()
79 return shorten_middle(lines[0].strip() if lines else "", _REASON_WIDTH)
82def start_analysis(app: LilbeeApp, directory: Path | None) -> None:
83 """Queue analyze of *directory*, or of the indexed corpus when None; a toast says when done."""
84 folder = str(directory) if directory is not None else msg.ANALYZE_YOUR_DOCUMENTS
86 def _target(reporter: ProgressReporter) -> None:
87 report = _analyze(app, directory, reporter)
88 call_from_thread(app, _finished, app, FinishedAnalysis(report, folder))
90 shown = (directory.name or folder) if directory is not None else folder
91 name = msg.TASK_NAME_ANALYZE.format(folder=shown)
92 app.task_bar.start_task(name, TaskType.ANALYZE, _target, indeterminate=True)
95def _finished(app: LilbeeApp, analysis: FinishedAnalysis) -> None:
96 """Keep the report for /analyze report; it never opens over what the user is typing."""
97 app.last_analysis = analysis
98 app.notify(msg.ANALYZE_DONE.format(folder=analysis.folder))
101def open_report(app: LilbeeApp) -> None:
102 """Open the last finished analyze report, or say there is none yet."""
103 analysis = app.last_analysis
104 if analysis is None:
105 app.notify(msg.ANALYZE_NO_REPORT, severity="warning")
106 return
107 app.push_screen(AnalyzeReportScreen(analysis.report, analysis.folder))
110def _analyze(app: LilbeeApp, directory: Path | None, reporter: ProgressReporter) -> AnalyzeReport:
111 """Run analyze on the task worker; a refusal toasts its reason and fails the task."""
112 reporter.update(0, msg.ANALYZE_STATUS_FINDING, indeterminate=True)
113 request = AnalyzeRequest(directory=directory)
115 def on_progress(event_type: EventType, data: ProgressEvent) -> None:
116 # the callback carries every event type; only analyze events move the bar
117 if event_type is EventType.ANALYZE and isinstance(data, AnalyzeEvent):
118 percent = data.done * PERCENT // data.total if data.total else 0
119 detail = msg.ANALYZE_STATUS_READING.format(
120 done=data.done, total=data.total, file=data.file
121 )
122 reporter.update(percent, detail, indeterminate=False)
124 run = run_analysis(ProfileStore(), request, on_progress=on_progress, cancel=reporter)
125 try:
126 return asyncio_loop.run(run)
127 except ValueError as exc:
128 _refused(app, str(exc), exc)
129 except OSError as exc:
130 _refused(app, profiles.file_failure_message(exc), exc)
133def _refused(app: LilbeeApp, text: str, exc: Exception) -> NoReturn:
134 """Toast why analyze stopped, then fail the task with the same reason."""
135 call_from_thread(app, app.notify, msg.ANALYZE_FAILED.format(error=text), severity="error")
136 raise RuntimeError(text) from exc
139def reading_text(signals: CorpusSignals) -> str:
140 """How many documents analyze read and counted, and whether it sampled them."""
141 template = msg.ANALYZE_READ_COUNTED if signals.files_counted else msg.ANALYZE_READ
142 text = template.format(
143 read=signals.files_read, total=signals.documents_total, counted=signals.files_counted
144 )
145 if signals.files_read + len(signals.failed) < signals.documents_total:
146 return f"{text} {msg.ANALYZE_SAMPLED.format(cap=signals.cap)}"
147 return text
150def _types_text(file_types: Mapping[str, int]) -> str:
151 ordered = sorted(file_types.items(), key=lambda item: (-item[1], item[0]))
152 counts = [msg.ANALYZE_TYPE_COUNT.format(count=count, kind=kind) for kind, count in ordered]
153 return ", ".join(counts) or msg.PROFILE_VALUE_NONE
156def _languages_text(rows: Sequence[LanguageRow]) -> str:
157 shares = [
158 msg.ANALYZE_LANGUAGE_SHARE.format(name=row.fts_language or row.code, share=row.share)
159 for row in rows
160 ]
161 if not shares:
162 return msg.ANALYZE_LANGUAGES_NONE
163 return msg.ANALYZE_LANGUAGES_TEXT.format(languages=", ".join(shares))
166def fact_rows(report: AnalyzeReport) -> list[tuple[str, str]]:
167 """The Found table: each signal analyze read, left out when there is nothing to show."""
168 signals = report.signals
169 pdf = signals.pdf
170 rows: list[tuple[str, str | None]] = [
171 (msg.ANALYZE_FACT_TYPES, _types_text(signals.file_types)),
172 (
173 msg.ANALYZE_FACT_CODE,
174 msg.ANALYZE_SHARE_OF_FILES.format(share=signals.code_share)
175 if signals.code_share
176 else None,
177 ),
178 (
179 msg.ANALYZE_FACT_SCANS,
180 msg.ANALYZE_SHARE_OF_PAGES.format(share=pdf.scanned_share)
181 if pdf.pages or signals.image_files
182 else None,
183 ),
184 (
185 msg.ANALYZE_FACT_TABLES,
186 msg.ANALYZE_TABLES_VALUE.format(tables=pdf.files_with_tables, files=pdf.files)
187 if pdf.files
188 else None,
189 ),
190 (msg.ANALYZE_FACT_LANGUAGES, _languages_text(report.languages)),
191 (
192 msg.ANALYZE_FACT_PDF_LENGTH,
193 None
194 if pdf.median_pages is None
195 else msg.ANALYZE_PAGES_VALUE.format(pages=pdf.median_pages),
196 ),
197 (
198 msg.ANALYZE_FACT_TEXT_LENGTH,
199 None
200 if signals.median_chars is None
201 else msg.ANALYZE_CHARS_VALUE.format(chars=signals.median_chars),
202 ),
203 ]
204 return [(label, value) for label, value in rows if value is not None]
207def _table(table_id: str, *columns: str) -> DataTable[str | Content]:
208 table: DataTable[str | Content] = DataTable(
209 id=table_id, cursor_type="none", zebra_stripes=False, show_cursor=False
210 )
211 table.add_columns(*columns)
212 # read-only tables: focus moves between the scroll area and the action pills
213 table.can_focus = False
214 return table
217def change_rows(recommendation: Recommendation) -> list[Sequence[str | Content]]:
218 """The changes table: each setting applying changes, its cost, and why analyze set it."""
219 why = {reason.key: reason.text for reason in recommendation.reasons}
220 # every adjustment carries a reason, so a change without one is the built-in's own value
221 builtin = msg.ANALYZE_WHY_BUILTIN.format(builtin=recommendation.builtin)
222 return [(*diff_cells(row), why.get(row.key, builtin)) for row in recommendation.changes]
225def _recommend_text(recommendation: Recommendation) -> str:
226 template = (
227 msg.ANALYZE_RECOMMEND_DERIVED if recommendation.name else msg.ANALYZE_RECOMMEND_BUILTIN
228 )
229 text = template.format(builtin=recommendation.builtin)
230 picked = [r.text for r in recommendation.reasons if r.key == PICK_REASON_KEY]
231 return " ".join([text, *(msg.ANALYZE_PICKED.format(reason=reason) for reason in picked)])
234class AnalyzeReportScreen(Screen[None]):
235 """What analyze found and the profile it recommends; nothing changes until Apply or Save."""
237 CSS_PATH = "analyze_report.tcss"
238 # the read-only scroll area: no key acts until the user moves to a pill
239 AUTO_FOCUS = "#analyze-report"
241 BINDINGS: ClassVar[list[BindingType]] = [
242 *browse_back_bindings(),
243 Binding("left", "app.focus_previous", "Previous", show=False),
244 Binding("right", "app.focus_next", "Next", show=False),
245 ]
247 app: LilbeeApp # type: ignore[assignment]
249 def __init__(self, report: AnalyzeReport, folder: str) -> None:
250 super().__init__()
251 self._report = report
252 self._folder = folder
253 self._actions: dict[ReportAction, Callable[[], None]] = {
254 ReportAction.APPLY: self.action_apply,
255 ReportAction.SAVE: self.action_save,
256 ReportAction.CLOSE: self.action_go_back,
257 }
259 def compose(self) -> ComposeResult:
260 with TopBars():
261 yield ViewTabs()
262 with VerticalScroll(id="analyze-report"):
263 yield Static(
264 msg.ANALYZE_TITLE.format(folder=self._folder), classes="analyze-heading first"
265 )
266 yield Static(reading_text(self._report.signals), classes="analyze-muted")
267 facts = _table("analyze-facts", msg.ANALYZE_FOUND, "")
268 facts.add_rows(fact_rows(self._report))
269 yield facts
270 yield from self._compose_failures()
271 yield from self._compose_recommendation()
272 yield from self._compose_notes()
273 with Horizontal(id="analyze-actions"):
274 yield ConfirmPill(
275 msg.PROFILE_APPLY_LABEL,
276 pill_id="analyze-apply",
277 answer=ReportAction.APPLY,
278 primary=True,
279 )
280 if self._report.recommendation.name is not None:
281 yield ConfirmPill(
282 msg.ANALYZE_SAVE_LABEL, pill_id="analyze-save", answer=ReportAction.SAVE
283 )
284 yield ConfirmPill(
285 msg.PROFILE_KEY_CLOSE, pill_id="analyze-close", answer=ReportAction.CLOSE
286 )
287 with BottomBars():
288 yield TaskBar()
289 yield Footer()
291 def _compose_failures(self) -> ComposeResult:
292 folders = self._report.signals.unreadable_folders
293 if folders:
294 yield Static(msg.ANALYZE_UNREADABLE_TITLE, classes="analyze-heading")
295 table = _table("analyze-unreadable", msg.ANALYZE_COL_FOLDER)
296 table.add_rows((folder,) for folder in folders)
297 yield table
298 failed = self._report.signals.failed
299 if not failed:
300 return
301 yield Static(msg.ANALYZE_FAILED_TITLE, classes="analyze-heading")
302 table = _table("analyze-failed", msg.ANALYZE_COL_FILE, msg.ANALYZE_COL_REASON)
303 table.add_rows((failure.file, _short_reason(failure.error)) for failure in failed)
304 yield table
306 def _compose_recommendation(self) -> ComposeResult:
307 rec = self._report.recommendation
308 title = msg.ANALYZE_RECOMMEND_TITLE.format(name=rec.name or rec.builtin)
309 yield Static(title, classes="analyze-heading", markup=False)
310 yield Static(_recommend_text(rec), classes="analyze-muted", markup=False)
311 if not rec.changes:
312 yield Static(msg.ANALYZE_NO_CHANGES, classes="analyze-muted")
313 else:
314 table = _table(
315 "analyze-changes",
316 msg.PROFILE_COL_SETTING,
317 msg.PROFILE_COL_NOW,
318 msg.PROFILE_COL_AFTER,
319 )
320 add_cost_column(table)
321 table.add_column(msg.ANALYZE_COL_WHY)
322 table.add_rows(change_rows(rec))
323 yield table
324 if rec.kept:
325 keys = ", ".join(rec.kept)
326 yield Static(msg.ANALYZE_KEEPS.format(keys=keys), classes="analyze-muted")
328 def _compose_notes(self) -> ComposeResult:
329 notes = self._report.recommendation.notes
330 if not notes:
331 return
332 yield Static(msg.ANALYZE_NOTES_TITLE, classes="analyze-heading")
333 for note in notes:
334 yield Static(note, classes="analyze-muted", markup=False)
336 @on(ConfirmPill.Picked)
337 def _on_picked(self, event: ConfirmPill.Picked) -> None:
338 event.stop()
339 self._actions[ReportAction(str(event.answer))]()
341 def action_apply(self) -> None:
342 """Ask with the Apply dialog, then save the recommended profile and switch to it."""
343 rec = self._report.recommendation
344 name = rec.name
345 if name is None:
346 start_profile_switch(self.app, self, rec.builtin, _ignore, self._applied)
347 return
348 values = rec.values
349 start_switch(
350 self.app,
351 self,
352 lambda: switch_plan(
353 profiles.recommended_file(name, values), profiles.preview(name, values)
354 ),
355 lambda: profiles.apply_recommended(
356 ProfileStore(), name, values, profiles.default_save_folder()
357 ),
358 on_close=_ignore,
359 on_applied=self._applied,
360 )
362 def action_save(self) -> None:
363 """Save the recommended profile without switching to it."""
364 rec = self._report.recommendation
365 name = rec.name
366 if name is None:
367 return
368 values = rec.values
369 run_profile_op(
370 self,
371 lambda: profiles.save_recommended(
372 ProfileStore(), name, values, profiles.default_save_folder(), switch=False
373 ),
374 self._saved,
375 )
377 def _saved(self, result: profiles.SaveResult) -> None:
378 location = result.location
379 self.notify(msg.PROFILE_SAVED.format(name=location.name, path=location.path))
381 def _applied(self) -> None:
382 self.dismiss()
384 def action_go_back(self) -> None:
385 self.dismiss()
388def _ignore() -> None:
389 """Keep the report open when the Apply dialog closes without applying."""