Coverage for src/lilbee/app/analyze.py: 100%
236 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-11-06 08:17 +0000
1"""Corpus analysis use case: signals to a recommended profile, saving it, and the analyze tip."""
3from __future__ import annotations
5import asyncio
6import logging
7import os
8import re
9from collections.abc import Callable, Mapping
10from dataclasses import dataclass, replace
11from pathlib import Path
12from typing import Any
14from lilbee.app.profiles import (
15 DiffRow,
16 apply,
17 default_save_folder,
18 preview,
19 save_recommended,
20 show,
21)
22from lilbee.core.config import cfg
23from lilbee.core.config.enums import FtsLanguage, OcrMode
24from lilbee.core.config.resolve import read_layers, read_profile_table, resolve
25from lilbee.core.profile_files import (
26 DEFAULT_PROFILE_NAME,
27 ProfileFolder,
28 ProfileStore,
29 profile_key,
30)
31from lilbee.core.project_state import dismiss_tip, mark_analyzed, read_state
32from lilbee.core.system import LOCAL_ROOT_DIRNAME
33from lilbee.data.analyze import CorpusSignals, collect_signals, ocr_language_supported
34from lilbee.data.ingest.discovery import CorpusScan, discover_corpus, discover_dir
35from lilbee.runtime.cancellation import CancelSignal, TaskCancelledError
36from lilbee.runtime.progress import DetailedProgressCallback, noop_callback
38log = logging.getLogger(__name__)
40# Built-in pick thresholds, first match wins. Unmeasured: a run on real corpora tunes them.
41SCANNED_ARCHIVE_SHARE = 0.30
42CODE_REPOSITORY_SHARE = 0.50
43NOTES_SHARE = 0.50
44RESEARCH_TABLES_SHARE = 0.20
45# Languages below this share of text files are left out of ocr_language. Unmeasured.
46OCR_LANGUAGE_FLOOR = 0.10
47MAX_OCR_LANGUAGES = 3
49SCANNED_ARCHIVE = "Scanned archive"
50CODE_REPOSITORY = "Code repository"
51NOTES_AND_MARKDOWN = "Notes and markdown"
52RESEARCH_PAPERS = "Research papers"
53PICK_REASON_KEY = "profile"
54MAX_PROFILE_NAME_LENGTH = 40
55FALLBACK_PROJECT_NAME = "project"
57NOTE_CONTENT_TYPES = frozenset({"md", "markdown", "mdx", "txt"})
58OCR_LANGUAGE_REASON = "assumed from the text pages"
59OCR_MODEL_NOTE = "The catalog lists OCR models that fit this machine."
60DEFAULT_FITS_NOTE = "Default fits this corpus."
61IMAGE_SCAN_NOTE = (
62 "Image files are counted, not read: each adds one scanned page to the scanned share, "
63 "scaled by the share of documents sampled."
64)
65TARGET_WITHOUT_SAVE_MESSAGE = "target takes effect only with apply or save; nothing was read."
66RELATIVE_DIRECTORY_MESSAGE = "directory must be an absolute path on the lilbee server: {value!r}"
67DIRECTORY_NOT_READABLE_MESSAGE = "Path '{path}' is not readable."
68TIP_TEXT = (
69 "This project uses the Default profile. Run lilbee analyze to get a profile "
70 "recommendation, or lilbee analyze --off to hide this tip."
71)
73# ISO 639-3 codes xberg detects, mapped to the Snowball stemmers of full-text search.
74FTS_LANGUAGE_BY_CODE: Mapping[str, FtsLanguage] = {
75 "ara": FtsLanguage.ARABIC,
76 "dan": FtsLanguage.DANISH,
77 "nld": FtsLanguage.DUTCH,
78 "eng": FtsLanguage.ENGLISH,
79 "fin": FtsLanguage.FINNISH,
80 "fra": FtsLanguage.FRENCH,
81 "deu": FtsLanguage.GERMAN,
82 "ell": FtsLanguage.GREEK,
83 "hun": FtsLanguage.HUNGARIAN,
84 "ita": FtsLanguage.ITALIAN,
85 "nob": FtsLanguage.NORWEGIAN,
86 "nno": FtsLanguage.NORWEGIAN,
87 "nor": FtsLanguage.NORWEGIAN,
88 "por": FtsLanguage.PORTUGUESE,
89 "ron": FtsLanguage.ROMANIAN,
90 "rus": FtsLanguage.RUSSIAN,
91 "spa": FtsLanguage.SPANISH,
92 "swe": FtsLanguage.SWEDISH,
93 "tam": FtsLanguage.TAMIL,
94 "tur": FtsLanguage.TURKISH,
95}
97# ISO 639-3 codes xberg detects whose Tesseract language data goes by another name.
98TESSERACT_LANGUAGE_BY_CODE: Mapping[str, str] = {
99 "cmn": "chi_sim",
100 "nob": "nor",
101 "pes": "fas",
102}
104_NAME_UNSAFE = re.compile(r"[^A-Za-z0-9 _-]+")
107@dataclass(frozen=True)
108class LanguageRow:
109 """A detected language, its share, its stemmer, and its Tesseract name and support."""
111 code: str
112 share: float
113 fts_language: FtsLanguage | None
114 ocr_code: str
115 ocr_supported: bool
118@dataclass(frozen=True)
119class Reason:
120 """Why the recommendation sets *key*; ``profile`` explains the built-in pick."""
122 key: str
123 text: str
126@dataclass(frozen=True)
127class Recommendation:
128 """The picked built-in, the derived profile's name and values, and what applying it changes.
130 ``name`` is None when Default fits and there is nothing to save.
131 """
133 builtin: str
134 name: str | None
135 values: Mapping[str, Any]
136 changes: tuple[DiffRow, ...]
137 kept: tuple[str, ...]
138 reasons: tuple[Reason, ...]
139 notes: tuple[str, ...]
142@dataclass(frozen=True)
143class SavedProfile:
144 """The profile analyze saved or switched to, its file, whether the project now uses it,
145 and any notice switching to it raises."""
147 name: str
148 folder: ProfileFolder
149 path: Path
150 applied: bool
151 warnings: tuple[str, ...] = ()
154@dataclass(frozen=True)
155class AnalyzeReport:
156 """Everything one analyze run read and recommends."""
158 signals: CorpusSignals
159 languages: tuple[LanguageRow, ...]
160 recommendation: Recommendation
161 saved: SavedProfile | None
164@dataclass(frozen=True)
165class AnalyzeRequest:
166 """What to analyze and what to do with the recommendation.
168 No *directory* reads the corpus ingest reads. *save* names the saved profile; *apply* also
169 switches to it. *target* picks the folder and needs *apply* or *save*; without it the
170 profile goes to the project folder when there is one, else the global one.
171 """
173 directory: Path | None = None
174 apply: bool = False
175 save: str | None = None
176 target: ProfileFolder | None = None
179@dataclass(frozen=True)
180class _PickRule:
181 builtin: str
182 threshold: float
183 share: Callable[[CorpusSignals], float]
184 reason: str
187def _notes_share(signals: CorpusSignals) -> float:
188 notes = sum(n for kind, n in signals.file_types.items() if kind in NOTE_CONTENT_TYPES)
189 return notes / signals.files_total if signals.files_total else 0.0
192def _tables_share(signals: CorpusSignals) -> float:
193 pdf = signals.pdf
194 return pdf.files_with_tables / pdf.files if pdf.files else 0.0
197_PICK_RULES: tuple[_PickRule, ...] = (
198 _PickRule(
199 SCANNED_ARCHIVE, SCANNED_ARCHIVE_SHARE, lambda s: s.pdf.scanned_share, "of pages are scans"
200 ),
201 _PickRule(CODE_REPOSITORY, CODE_REPOSITORY_SHARE, lambda s: s.code_share, "of files are code"),
202 _PickRule(NOTES_AND_MARKDOWN, NOTES_SHARE, _notes_share, "of files are Markdown or text"),
203 _PickRule(RESEARCH_PAPERS, RESEARCH_TABLES_SHARE, _tables_share, "of PDFs have tables"),
204)
207def pick_builtin(signals: CorpusSignals) -> tuple[str, Reason | None]:
208 """The first built-in whose share reaches its threshold, else Default, with the reason."""
209 for rule in _PICK_RULES:
210 share = rule.share(signals)
211 if share >= rule.threshold:
212 return rule.builtin, Reason(PICK_REASON_KEY, f"{share:.0%} {rule.reason}")
213 return DEFAULT_PROFILE_NAME, None
216def _language_row(code: str, share: float) -> LanguageRow:
217 ocr_code = TESSERACT_LANGUAGE_BY_CODE.get(code, code)
218 return LanguageRow(
219 code, share, FTS_LANGUAGE_BY_CODE.get(code), ocr_code, ocr_language_supported(ocr_code)
220 )
223def language_rows(signals: CorpusSignals) -> tuple[LanguageRow, ...]:
224 """Each detected language with its stemmer and Tesseract support."""
225 return tuple(_language_row(lang.code, lang.share) for lang in signals.languages)
228def project_label(data_root: Path) -> str:
229 """The project's folder name: the parent of a ``.lilbee`` folder, else the data root's own."""
230 folder = data_root.parent if data_root.name == LOCAL_ROOT_DIRNAME else data_root
231 return folder.name
234def derived_name(builtin: str, project: str) -> str:
235 """The derived profile's name, "<built-in> (<project>)", cut to fit the profile name rules."""
236 room = MAX_PROFILE_NAME_LENGTH - len(builtin) - len(" ()")
237 safe = _NAME_UNSAFE.sub("-", project).strip(" -_")[:room].strip(" -_")
238 return f"{builtin} ({safe or FALLBACK_PROJECT_NAME[:room]})"
241class _Plan:
242 """The adjustments, reasons and notes a recommendation collects on top of a built-in."""
244 def __init__(self, builtin_values: Mapping[str, Any]) -> None:
245 layers = read_layers(cfg.data_root)
246 self._layers = replace(layers, profile=dict(builtin_values))
247 self.adjustments: dict[str, Any] = {}
248 self.reasons: list[Reason] = []
249 self.notes: list[str] = []
251 def baseline(self, key: str) -> Any:
252 """*key*'s value under the built-in alone, with user and env values on top."""
253 return resolve(key, self._layers).value
255 def adjust(self, key: str, value: Any, reason: str) -> None:
256 """Set *key* to *value* when the built-in alone gives something else."""
257 if value != self.baseline(key):
258 self.adjustments[key] = value
259 self.reasons.append(Reason(key, reason))
262def _plan_fts(plan: _Plan, rows: tuple[LanguageRow, ...]) -> None:
263 if not rows:
264 return
265 top = rows[0]
266 if top.fts_language is None:
267 plan.notes.append(
268 f"No stemmer for {top.code}; fts_language stays {plan.baseline('fts_language')}."
269 )
270 return
271 plan.adjust(
272 "fts_language", top.fts_language, f"{top.share:.0%} of text files are {top.fts_language}"
273 )
276def _plan_ocr_language(plan: _Plan, signals: CorpusSignals, rows: tuple[LanguageRow, ...]) -> None:
277 scans = signals.pdf.scanned_pages + signals.image_files
278 if scans == 0 or plan.baseline("ocr") == OcrMode.OFF:
279 return
280 common = [row for row in rows if row.share >= OCR_LANGUAGE_FLOOR][:MAX_OCR_LANGUAGES]
281 for row in common:
282 if not row.ocr_supported:
283 plan.notes.append(f"Tesseract has no language data for {row.ocr_code} on this machine.")
284 supported = [row.ocr_code for row in common if row.ocr_supported]
285 if supported:
286 plan.adjust("ocr_language", supported, OCR_LANGUAGE_REASON)
289def _plan_ocr_model(plan: _Plan, signals: CorpusSignals) -> None:
290 if signals.pdf.scanned_share >= SCANNED_ARCHIVE_SHARE and not cfg.vision_model:
291 plan.notes.append(OCR_MODEL_NOTE)
294def _builtin_values(store: ProfileStore, builtin: str) -> Mapping[str, Any]:
295 entry = store.scan().find(builtin)
296 if entry is None or entry.file is None:
297 raise ValueError(f"The built-in profile {builtin} is missing or broken")
298 return entry.file.values
301def recommend(
302 store: ProfileStore, signals: CorpusSignals, rows: tuple[LanguageRow, ...]
303) -> Recommendation:
304 """The built-in that fits *signals*, adjusted for the corpus, and what applying it changes."""
305 builtin, pick_reason = pick_builtin(signals)
306 builtin_values = _builtin_values(store, builtin)
307 plan = _Plan(builtin_values)
308 _plan_fts(plan, rows)
309 _plan_ocr_language(plan, signals, rows)
310 _plan_ocr_model(plan, signals)
311 if signals.image_files:
312 plan.notes.append(IMAGE_SCAN_NOTE)
313 values = {**builtin_values, **plan.adjustments}
314 fits_default = builtin == DEFAULT_PROFILE_NAME and not plan.adjustments
315 name = None if fits_default else derived_name(builtin, project_label(cfg.data_root))
316 if fits_default:
317 plan.notes.append(DEFAULT_FITS_NOTE)
318 diff = preview(name or builtin, values)
319 reasons = ([pick_reason] if pick_reason else []) + plan.reasons
320 return Recommendation(
321 builtin, name, values, diff.changes, diff.kept, tuple(reasons), tuple(plan.notes)
322 )
325def _save(
326 store: ProfileStore, recommendation: Recommendation, request: AnalyzeRequest
327) -> SavedProfile | None:
328 """Save the recommendation when the request asks; applying a fitting Default switches to it."""
329 if not request.apply and request.save is None:
330 return None
331 name = request.save if request.save is not None else recommendation.name
332 if name is None:
333 applied = apply(store, recommendation.builtin)
334 entry = show(store, applied.name)
335 return SavedProfile(
336 entry.name, entry.folder, entry.path, applied=True, warnings=applied.warnings
337 )
338 target = request.target or default_save_folder()
339 result = save_recommended(store, name, recommendation.values, target, switch=request.apply)
340 location = result.location
341 return SavedProfile(
342 location.name, location.folder, location.path, request.apply, result.warnings
343 )
346def validate_request(request: AnalyzeRequest) -> None:
347 """Refuse a *target* with no *apply* or *save*, and a *directory* that is not readable."""
348 if request.target is not None and not request.apply and request.save is None:
349 raise ValueError(TARGET_WITHOUT_SAVE_MESSAGE)
350 if request.directory is not None and not request.directory.is_dir():
351 raise ValueError(f"{request.directory} is not a folder")
352 if request.directory is not None and not os.access(request.directory, os.R_OK | os.X_OK):
353 raise ValueError(DIRECTORY_NOT_READABLE_MESSAGE.format(path=request.directory))
356def server_directory(value: str | None) -> Path | None:
357 """A directory named by a remote caller: absolute on the server, or None for the corpus."""
358 if value is None:
359 return None
360 directory = Path(value)
361 if not directory.is_absolute():
362 raise ValueError(RELATIVE_DIRECTORY_MESSAGE.format(value=value))
363 return directory
366def _scan(directory: Path | None) -> CorpusScan:
367 return discover_corpus() if directory is None else discover_dir(directory)
370def _record_run() -> None:
371 """Record the completed run; a failed write must not fail a run that already saved."""
372 try:
373 mark_analyzed(cfg.data_root)
374 except OSError as exc:
375 log.warning("Could not record that analyze ran: %s", exc)
378def _finish(
379 store: ProfileStore,
380 signals: CorpusSignals,
381 request: AnalyzeRequest,
382 cancel: CancelSignal | None,
383) -> AnalyzeReport:
384 rows = language_rows(signals)
385 recommendation = recommend(store, signals, rows)
386 if cancel is not None and cancel.is_set():
387 raise TaskCancelledError
388 saved = _save(store, recommendation, request)
389 _record_run()
390 return AnalyzeReport(signals, rows, recommendation, saved)
393async def run_analysis(
394 store: ProfileStore,
395 request: AnalyzeRequest,
396 *,
397 on_progress: DetailedProgressCallback = noop_callback,
398 cancel: CancelSignal | None = None,
399) -> AnalyzeReport:
400 """Read the corpus, recommend a profile, save it when asked, and record the run.
402 Raises ``TaskCancelledError`` on cancel, before anything is saved or recorded;
403 raises ``ValueError`` when *directory* is not a readable folder, a *target* comes without
404 *apply* or *save*, or the save is refused.
405 """
406 await asyncio.to_thread(validate_request, request)
407 scan = await asyncio.to_thread(_scan, request.directory)
408 signals = await collect_signals(scan.files, on_progress=on_progress, cancel=cancel)
409 signals = replace(signals, unreadable_folders=scan.unreadable)
410 return await asyncio.to_thread(_finish, store, signals, request, cancel)
413@dataclass(frozen=True)
414class TipState:
415 """Whether the project was analyzed, hid the tip, and would see the tip now."""
417 analyzed: bool
418 tip_dismissed: bool
419 tip_shows: bool
422def tip_state(root: Path) -> TipState:
423 """The analyze tip state of the project at *root*."""
424 state = read_state(root)
425 analyzed = state.analyzed_at is not None
426 name = read_profile_table(root).name or DEFAULT_PROFILE_NAME
427 on_default = profile_key(name) == profile_key(DEFAULT_PROFILE_NAME)
428 fresh = on_default and not analyzed and not state.tip_dismissed
429 return TipState(analyzed, state.tip_dismissed, fresh)
432def tip_shows(root: Path) -> bool:
433 """True when the project at *root* is on Default, unanalyzed, and not hiding the tip."""
434 return tip_state(root).tip_shows
437def hide_tip(root: Path) -> None:
438 """Stop showing the analyze tip for the project at *root*."""
439 dismiss_tip(root)