Coverage for src/lilbee/core/profile_files.py: 100%
371 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-11-06 08:17 +0000
1"""Profile files: the file format, its validation, discovery across the folders, and writes."""
3from __future__ import annotations
5import re
6import tomllib
7from collections.abc import Callable, Mapping, Sequence
8from dataclasses import dataclass, replace
9from enum import StrEnum
10from functools import cache, partial
11from importlib.metadata import version as installed_version
12from pathlib import Path
13from typing import Any, TypeVar
15import tomli_w
16from packaging.version import InvalidVersion, Version
17from pydantic import ValidationError
19from lilbee.core.config import Config, cfg
20from lilbee.core.config.enums import ProfileScope
21from lilbee.core.config.resolve import PROFILE_FIELDS, builtin_value
22from lilbee.core.security import write_text_atomically
23from lilbee.core.system import canonical_data_root, default_data_dir
25T = TypeVar("T")
27PROFILES_DIRNAME = "profiles"
28BUILTIN_DIRNAME = "builtin"
29PROFILE_SUFFIX = ".toml"
30PACKAGE_PROFILES_DIR = Path(__file__).resolve().parent.parent / PROFILES_DIRNAME
31MAX_PROFILE_BYTES = 256 * 1024
33DEFAULT_PROFILE_NAME = "Default"
34PROFILE_FORMAT = 1
36META_TABLE = "profile"
37VALUES_TABLE = "values"
38_TOP_LEVEL_KEYS = frozenset({META_TABLE, VALUES_TABLE})
39_STRING_META_KEYS = ("name", "description", "tested_on", "min_lilbee", "evidence")
40_META_KEYS = frozenset({*_STRING_META_KEYS, "authors", "format"})
41_AUTHOR_KEYS = frozenset({"name", "github"})
43MAX_NAME_LENGTH = 40
44_NAME_PATTERN = re.compile(rf"[A-Za-z0-9 _()-]{{1,{MAX_NAME_LENGTH}}}")
45# words a profile name cannot be: fixed route words and the file names Windows keeps for devices
46RESERVED_NAME_KEYS = frozenset(
47 {
48 "active",
49 "new",
50 "discard",
51 "import",
52 "validate",
53 "con",
54 "prn",
55 "aux",
56 "nul",
57 *(f"{port}{digit}" for port in ("com", "lpt") for digit in range(10)),
58 }
59)
60_KEY_DROPPED = re.compile(r"[()]")
61_KEY_SEPARATORS = re.compile(r"[\s_-]+")
64class ProfileFolder(StrEnum):
65 """A folder lilbee reads profiles from, highest precedence first."""
67 PROJECT = "project"
68 GLOBAL = "global"
69 BUILTIN = "builtin"
72class NameProblem(StrEnum):
73 """Why a name cannot go on a new profile file in a folder."""
75 EMPTY = "empty"
76 INVALID = "invalid"
77 RESERVED = "reserved"
78 RESERVED_WORD = "reserved_word"
79 TAKEN = "taken"
82class ProfileFileError(ValueError):
83 """A profile file lilbee cannot use, with the reason as its message."""
86class ProfileNameClashError(ProfileFileError):
87 """A write's profile name already names another file in the same folder."""
90@dataclass(frozen=True)
91class ProfileAuthor:
92 """One credited author of a profile."""
94 name: str
95 github: str | None
98@dataclass(frozen=True)
99class ProfileFile:
100 """A valid profile file: its metadata and the settings it holds."""
102 name: str
103 description: str | None
104 authors: tuple[ProfileAuthor, ...]
105 tested_on: str | None
106 format: int
107 min_lilbee: str | None
108 evidence: str | None
109 values: Mapping[str, Any]
112@dataclass(frozen=True)
113class ProfileEntry:
114 """One profile file found in a folder; broken files carry the reason in ``error``."""
116 name: str
117 folder: ProfileFolder
118 path: Path
119 file: ProfileFile | None
120 error: str | None
121 shadowed_by: ProfileFolder | None = None
124@dataclass(frozen=True)
125class ProfileCatalog:
126 """Every profile file found: Default, then the other built-ins, then the rest."""
128 entries: tuple[ProfileEntry, ...]
130 def usable_names(self) -> list[str]:
131 """Names a pick can apply: valid profiles that no higher folder hides."""
132 return [e.name for e in self.entries if e.file is not None and e.shadowed_by is None]
134 def find(self, name: str) -> ProfileEntry | None:
135 """The entry a name picks: the highest-precedence one with that lookup key."""
136 key = profile_key(name)
137 return next(
138 (e for e in self.entries if e.shadowed_by is None and profile_key(e.name) == key),
139 None,
140 )
143def profile_key(name: str) -> str:
144 """The lookup key of a name and the stem of a file lilbee writes for it: a lowercase slug."""
145 words = _KEY_DROPPED.sub("", name.casefold())
146 return _KEY_SEPARATORS.sub("-", words).strip("-")
149def profile_folders(data_root: Path) -> list[tuple[ProfileFolder, Path]]:
150 """The profile folders for *data_root*, highest precedence first."""
151 global_root = default_data_dir()
152 folders = [
153 (ProfileFolder.GLOBAL, global_root / PROFILES_DIRNAME),
154 (ProfileFolder.BUILTIN, PACKAGE_PROFILES_DIR / BUILTIN_DIRNAME),
155 ]
156 if canonical_data_root(data_root) != canonical_data_root(global_root):
157 folders.insert(0, (ProfileFolder.PROJECT, data_root / PROFILES_DIRNAME))
158 return folders
161def _needs_newer(min_lilbee: str | None) -> str | None:
162 """The "needs a newer lilbee" message when *min_lilbee* is above the running version."""
163 if min_lilbee is None or Version(min_lilbee) <= Version(installed_version("lilbee")):
164 return None
165 return f"Needs lilbee {min_lilbee} or newer"
168def _optional_str(meta: Mapping[str, Any], key: str) -> str | None:
169 value = meta.get(key)
170 # untyped TOML: a metadata field can hold any TOML type
171 if value is not None and not isinstance(value, str):
172 raise ProfileFileError(f"{key} must be text")
173 return value
176def _parse_author(raw: Any) -> ProfileAuthor:
177 # untyped TOML: each author must be a table with a text name
178 if not isinstance(raw, dict) or not isinstance(raw.get("name"), str):
179 raise ProfileFileError("Each author needs a name")
180 unknown = sorted(set(raw) - _AUTHOR_KEYS)
181 if unknown:
182 raise ProfileFileError(f"Unknown author field: {unknown[0]}")
183 return ProfileAuthor(name=raw["name"], github=_optional_str(raw, "github"))
186def _parse_authors(meta: Mapping[str, Any]) -> tuple[ProfileAuthor, ...]:
187 raw = meta.get("authors", [])
188 # untyped TOML: authors must be an array of tables
189 if not isinstance(raw, list):
190 raise ProfileFileError("authors must be a list of { name, github } tables")
191 return tuple(_parse_author(item) for item in raw)
194def _parse_format(meta: Mapping[str, Any]) -> int:
195 value = meta.get("format", PROFILE_FORMAT)
196 if value != PROFILE_FORMAT or isinstance(value, bool):
197 raise ProfileFileError(
198 f"Unknown profile format {value!r}; this lilbee reads format {PROFILE_FORMAT}"
199 )
200 return PROFILE_FORMAT
203def _parse_min_lilbee(meta: Mapping[str, Any]) -> str | None:
204 value = _optional_str(meta, "min_lilbee")
205 if value is None:
206 return None
207 try:
208 Version(value)
209 except InvalidVersion:
210 raise ProfileFileError(f"min_lilbee is not a version: {value!r}") from None
211 return value
214def _bad_name(name: str) -> bool:
215 return _NAME_PATTERN.fullmatch(name) is None or not profile_key(name)
218def _check_name(meta: Mapping[str, Any], stem: str) -> str:
219 raw = _optional_str(meta, "name")
220 name = stem if raw is None else raw
221 if _bad_name(name):
222 raise ProfileFileError(
223 f"Bad name {name!r}: use 1 to {MAX_NAME_LENGTH} ASCII letters, digits, spaces, "
224 "hyphens, underscores or parentheses"
225 )
226 if profile_key(name) in RESERVED_NAME_KEYS:
227 raise ProfileFileError(f"Reserved name: {name} cannot name a profile")
228 return name
231def _assign(trial: Config, key: str, value: Any) -> Any:
232 """*value* as *trial* holds it after its field validator runs; raises on a bad value."""
233 try:
234 setattr(trial, key, value)
235 except ValidationError as exc:
236 raise ProfileFileError(f"Bad value for {key}: {exc.errors()[0]['msg']}") from None
237 # A validator's TypeError is a bad value
238 except TypeError as exc:
239 raise ProfileFileError(f"Bad value for {key}: {exc}") from None
240 return getattr(trial, key)
243def normalized_values(values: Mapping[str, Any]) -> dict[str, Any]:
244 """Each value as Config holds it once its field validator runs; raises on a bad value."""
245 trial = cfg.model_copy()
246 return {key: _assign(trial, key, value) for key, value in values.items()}
249def _normalized(trial: Config, key: str, value: Any, min_lilbee: str | None) -> Any:
250 """*value* as Config holds it; raises when *key* is not a profile setting or *value* is bad."""
251 if key not in Config.model_fields:
252 raise ProfileFileError(_needs_newer(min_lilbee) or f"Unknown setting: {key}")
253 if key not in PROFILE_FIELDS:
254 raise ProfileFileError(f"Profiles cannot set {key}")
255 try:
256 return _assign(trial, key, value)
257 except ProfileFileError as exc:
258 raise ProfileFileError(_needs_newer(min_lilbee) or str(exc)) from None
261def _overlap_problems(values: Mapping[str, Any]) -> list[str]:
262 size, overlap = values.get("chunk_size"), values.get("chunk_overlap")
263 if size is not None and overlap is not None and overlap >= size:
264 return [f"chunk_overlap ({overlap}) must be < chunk_size ({size})"]
265 return []
268def _evidence_problems(values: Mapping[str, Any], evidence: str | None) -> list[str]:
269 """A built-in profile sets a retrieval value unequal to the default only with evidence."""
270 if evidence is not None:
271 return []
272 return [
273 f"Sets retrieval setting {key} without evidence"
274 for key, value in values.items()
275 if PROFILE_FIELDS[key] is ProfileScope.RETRIEVAL and value != builtin_value(key)
276 ]
279def _value_problems(
280 raw: Mapping[str, Any], min_lilbee: str | None, folder: ProfileFolder, evidence: str | None
281) -> list[str]:
282 problems: list[str] = []
283 normalized: dict[str, Any] = {}
284 trial = cfg.model_copy()
285 for key, value in raw.items():
286 try:
287 normalized[key] = _normalized(trial, key, value, min_lilbee)
288 except ProfileFileError as exc:
289 problems.append(str(exc))
290 problems += _overlap_problems(normalized)
291 if folder is ProfileFolder.BUILTIN:
292 problems += _evidence_problems(normalized, evidence)
293 return problems
296def _checked(problems: list[str], check: Callable[[], T]) -> T | None:
297 """The result of *check*, or None with its reason added to *problems*."""
298 try:
299 return check()
300 except ProfileFileError as exc:
301 problems.append(str(exc))
302 return None
305def _meta_problems(meta: Mapping[str, Any], stem: str) -> tuple[list[str], str | None]:
306 """Every problem in the ``[profile]`` table, and its ``min_lilbee`` when that is valid."""
307 problems = [f"Unknown profile field: {key}" for key in sorted(set(meta) - _META_KEYS)]
308 _checked(problems, partial(_check_name, meta, stem))
309 for key in ("description", "tested_on", "evidence"):
310 _checked(problems, partial(_optional_str, meta, key))
311 min_lilbee = _checked(problems, partial(_parse_min_lilbee, meta))
312 _checked(problems, partial(_parse_authors, meta))
313 _checked(problems, partial(_parse_format, meta))
314 return problems, min_lilbee
317def profile_problems(data: Mapping[str, Any], stem: str, folder: ProfileFolder) -> list[str]:
318 """Every reason parsed TOML is not a valid profile file; empty when it is one."""
319 problems = [f"Unknown table: {key}" for key in sorted(set(data) - _TOP_LEVEL_KEYS)]
320 meta = data.get(META_TABLE, {})
321 # untyped TOML: [profile] must be a table
322 if not isinstance(meta, dict):
323 problems.append("[profile] must be a table")
324 meta = {}
325 meta_problems, min_lilbee = _meta_problems(meta, stem)
326 problems += meta_problems
327 raw = data.get(VALUES_TABLE)
328 # untyped TOML: [values] must be a table
329 if not isinstance(raw, dict):
330 problems.append("Missing [values] table")
331 else:
332 evidence = meta.get("evidence")
333 # untyped TOML: a non-text evidence is already a problem and counts as none
334 evidence = evidence if isinstance(evidence, str) else None
335 problems += _value_problems(raw, min_lilbee, folder, evidence)
336 return list(dict.fromkeys(problems))
339def parse_profile(data: Mapping[str, Any], stem: str, folder: ProfileFolder) -> ProfileFile:
340 """Validate parsed TOML as a profile file; raises ``ProfileFileError`` with the first reason."""
341 problems = profile_problems(data, stem, folder)
342 if problems:
343 raise ProfileFileError(problems[0])
344 meta = data.get(META_TABLE, {})
345 return ProfileFile(
346 name=_check_name(meta, stem),
347 description=meta.get("description"),
348 authors=_parse_authors(meta),
349 tested_on=meta.get("tested_on"),
350 format=PROFILE_FORMAT,
351 min_lilbee=meta.get("min_lilbee"),
352 evidence=meta.get("evidence"),
353 values=dict(data[VALUES_TABLE]),
354 )
357def _check_size(size: int) -> None:
358 """Raise ``ProfileFileError`` when *size* bytes is over the profile file cap."""
359 if size > MAX_PROFILE_BYTES:
360 raise ProfileFileError(
361 f"The file is over {MAX_PROFILE_BYTES // 1024} KB, too large for a profile file"
362 )
365def _parse_toml(text: str) -> dict[str, Any]:
366 """The parsed TOML in *text*; raises ``ProfileFileError`` when it is too large or not TOML."""
367 _check_size(len(text.encode("utf-8")))
368 try:
369 return tomllib.loads(text)
370 # tomllib raises RecursionError on arrays nested a few hundred deep
371 except (tomllib.TOMLDecodeError, RecursionError) as exc:
372 raise ProfileFileError(f"Not valid TOML: {exc}") from None
375def read_profile_text(path: Path) -> str:
376 """The text of the file at *path*; raises ``ProfileFileError`` when it cannot be read."""
377 try:
378 with path.open("rb") as f:
379 data = f.read(MAX_PROFILE_BYTES + 1)
380 except FileNotFoundError:
381 raise ProfileFileError(f"No file at {path}") from None
382 except OSError as exc:
383 raise ProfileFileError(f"Cannot read the file: {exc}") from None
384 _check_size(len(data))
385 try:
386 return data.decode("utf-8")
387 except UnicodeDecodeError:
388 raise ProfileFileError("Not UTF-8 text") from None
391def _load_toml(path: Path) -> dict[str, Any]:
392 """The parsed TOML at *path*; raises ``ProfileFileError`` when it cannot be read or parsed."""
393 return _parse_toml(read_profile_text(path))
396def read_entry(path: Path, folder: ProfileFolder) -> ProfileEntry:
397 """Read and validate one profile file; a broken file is returned with its reason."""
398 try:
399 data = _load_toml(path)
400 except ProfileFileError as exc:
401 return ProfileEntry(path.stem, folder, path, None, str(exc))
402 try:
403 profile = parse_profile(data, path.stem, folder)
404 except ProfileFileError as exc:
405 return ProfileEntry(_entry_name(data, path.stem), folder, path, None, str(exc))
406 return ProfileEntry(profile.name, folder, path, profile, None)
409def _entry_name(data: Mapping[str, Any], stem: str) -> str:
410 """The name a broken file shows under: its own name when readable, else the file stem."""
411 meta = data.get(META_TABLE)
412 # untyped TOML: a broken file's [profile] table may hold anything
413 name = meta.get("name") if isinstance(meta, dict) else None
414 return name if isinstance(name, str) else stem
417def _read_folder(folder: ProfileFolder, path: Path) -> list[ProfileEntry]:
418 """Every profile file in *path*, sorted by file name; a missing folder is empty."""
419 if not path.is_dir():
420 return []
421 files = sorted(p for p in path.glob(f"*{PROFILE_SUFFIX}") if p.is_file())
422 return _mark_duplicates([read_entry(p, folder) for p in files])
425def _broken(entry: ProfileEntry, reason: str) -> ProfileEntry:
426 return replace(entry, file=None, error=reason)
429def _mark_duplicates(entries: list[ProfileEntry]) -> list[ProfileEntry]:
430 """Mark every entry whose lookup key another entry in the same folder shares."""
431 keys = [profile_key(e.name) for e in entries]
432 return [
433 _broken(e, f"Duplicate name in the {e.folder.value} folder") if keys.count(key) > 1 else e
434 for e, key in zip(entries, keys, strict=True)
435 ]
438def _reserved_reason(name: str) -> str:
439 return f"Reserved name: {name} is a built-in profile"
442def _settle_names(entries: list[ProfileEntry]) -> list[ProfileEntry]:
443 """Break non-builtin entries that take a built-in name; shadow each lower entry by name."""
444 reserved = {profile_key(e.name) for e in entries if e.folder is ProfileFolder.BUILTIN}
445 owner: dict[str, ProfileFolder] = {}
446 settled: list[ProfileEntry] = []
447 for entry in entries:
448 key = profile_key(entry.name)
449 if entry.folder is not ProfileFolder.BUILTIN and key in reserved:
450 entry = _broken(entry, _reserved_reason(entry.name))
451 settled.append(replace(entry, shadowed_by=ProfileFolder.BUILTIN))
452 continue
453 winner = owner.setdefault(key, entry.folder)
454 settled.append(entry if winner is entry.folder else replace(entry, shadowed_by=winner))
455 return settled
458def scan(folders: Sequence[tuple[ProfileFolder, Path]]) -> ProfileCatalog:
459 """Read every profile file in *folders*, given highest precedence first.
461 The catalog lists Default, then the other built-ins, then the rest.
462 """
463 entries = [entry for folder, path in folders for entry in _read_folder(folder, path)]
464 return ProfileCatalog(entries=tuple(sorted(_settle_names(entries), key=_listing_rank)))
467def _listing_rank(entry: ProfileEntry) -> int:
468 """Sort key that lists Default first, then the other built-ins, then every other profile."""
469 if entry.folder is not ProfileFolder.BUILTIN:
470 return 2
471 return 0 if profile_key(entry.name) == profile_key(DEFAULT_PROFILE_NAME) else 1
474@dataclass(frozen=True)
475class ProfileStore:
476 """Scans the profile folders of the current data root on each call."""
478 def scan(self) -> ProfileCatalog:
479 """Every profile visible from ``cfg.data_root`` now."""
480 return scan(profile_folders(cfg.data_root))
483@dataclass(frozen=True)
484class ProfileValidation:
485 """Every problem found in one profile file; empty ``problems`` means it is valid."""
487 path: Path
488 name: str
489 problems: tuple[str, ...]
491 @property
492 def valid(self) -> bool:
493 """True when the file has no problems."""
494 return not self.problems
497@dataclass(frozen=True)
498class PlannedWrite:
499 """A validated profile file and the free path it goes to; nothing is written until ``write``."""
501 path: Path
502 folder: ProfileFolder
503 profile: ProfileFile
504 text: str
505 replacing: Path | None
507 def write(self) -> Path:
508 """Write the file atomically, then remove the file it replaces when that is another file."""
509 stale = (
510 None
511 if self.replacing is None or _same_file(self.replacing, self.path)
512 else self.replacing
513 )
514 write_text_atomically(self.path, self.text)
515 if stale is not None:
516 stale.unlink(missing_ok=True)
517 return self.path
520def _same_file(first: Path, second: Path) -> bool:
521 """True when both paths exist and name one file, whatever their spelling or case."""
522 try:
523 return first.samefile(second)
524 except OSError:
525 return False
528@cache
529def _keys_in(folder: Path) -> frozenset[str]:
530 """The lookup keys of the built-in files in *folder*, which ship with the package."""
531 return frozenset(profile_key(e.name) for e in _read_folder(ProfileFolder.BUILTIN, folder))
534def builtin_keys() -> frozenset[str]:
535 """The lookup keys of the built-in profiles, which no other profile may take."""
536 return _keys_in(PACKAGE_PROFILES_DIR / BUILTIN_DIRNAME)
539def validate_file(path: Path, folder: ProfileFolder) -> ProfileValidation:
540 """Every problem with *path* as a profile file in *folder*."""
541 try:
542 text = read_profile_text(path)
543 except ProfileFileError as exc:
544 return ProfileValidation(path, path.stem, (str(exc),))
545 return validate_text(text, path, folder)
548def validate_text(text: str, path: Path, folder: ProfileFolder) -> ProfileValidation:
549 """Every problem with *text*, the content of the file at *path*, as a profile in *folder*."""
550 try:
551 data = _parse_toml(text)
552 except ProfileFileError as exc:
553 return ProfileValidation(path, path.stem, (str(exc),))
554 name = _entry_name(data, path.stem)
555 problems = profile_problems(data, path.stem, folder)
556 if folder is not ProfileFolder.BUILTIN and profile_key(name) in builtin_keys():
557 problems.append(_reserved_reason(name))
558 return ProfileValidation(path, name, tuple(problems))
561def profile_text(profile: ProfileFile) -> str:
562 """*profile* as the text of a clean profile file."""
563 authors = [
564 {k: v for k, v in (("name", a.name), ("github", a.github)) if v is not None}
565 for a in profile.authors
566 ]
567 meta = {
568 "name": profile.name,
569 "description": profile.description,
570 "authors": authors or None,
571 "tested_on": profile.tested_on,
572 "format": profile.format,
573 "min_lilbee": profile.min_lilbee,
574 "evidence": profile.evidence,
575 }
576 table = {key: value for key, value in meta.items() if value is not None}
577 return tomli_w.dumps({META_TABLE: table, VALUES_TABLE: dict(profile.values)})
580def parse_text(text: str, stem: str, folder: ProfileFolder) -> ProfileFile:
581 """Validate *text* as a profile file named *stem* when it names none; raises on a problem."""
582 return parse_profile(_parse_toml(text), stem, folder)
585def _clash(
586 directory: Path, folder: ProfileFolder, planned: Path, replacing: Path | None
587) -> Path | None:
588 """A file in *directory*, other than *replacing*, at *planned* or holding the same name."""
589 if not directory.is_dir():
590 return None
591 key = planned.stem.casefold()
592 for path in sorted(directory.glob(f"*{PROFILE_SUFFIX}")):
593 if path != replacing and key in _name_keys(read_entry(path, folder)):
594 return path
595 return None
598def _name_keys(entry: ProfileEntry) -> tuple[str, str]:
599 """The keys *entry* takes: its name and its file stem, since a file system can fold case."""
600 return profile_key(entry.name), entry.path.stem.casefold()
603def name_problem(name: str, folder: ProfileFolder, catalog: ProfileCatalog) -> NameProblem | None:
604 """Why ``plan_write`` would refuse *name* in *folder*, judged from *catalog*, or None."""
605 if not name.strip():
606 return NameProblem.EMPTY
607 if _bad_name(name):
608 return NameProblem.INVALID
609 key = profile_key(name)
610 if key in RESERVED_NAME_KEYS:
611 return NameProblem.RESERVED_WORD
612 taken = {(e.folder, k) for e in catalog.entries for k in _name_keys(e)}
613 if (ProfileFolder.BUILTIN, key) in taken:
614 return NameProblem.RESERVED
615 return NameProblem.TAKEN if (folder, key) in taken else None
618def plan_write(
619 directory: Path, folder: ProfileFolder, text: str, *, stem: str, replacing: Path | None = None
620) -> PlannedWrite:
621 """Validate *text* and pick its file in *directory*, named by the profile's slug.
623 Raises ``ProfileFileError`` when the text is invalid or another file there has the name;
624 *replacing* is the file this write supersedes.
625 """
626 profile = parse_text(text, stem, folder)
627 key = profile_key(profile.name)
628 if key in builtin_keys():
629 raise ProfileFileError(_reserved_reason(profile.name))
630 path = directory / f"{key}{PROFILE_SUFFIX}"
631 clash = _clash(directory, folder, path, replacing)
632 if clash is not None:
633 raise ProfileNameClashError(f"A profile named {profile.name} already exists: {clash}")
634 return PlannedWrite(path, folder, profile, text, replacing)