Coverage for src/lilbee/app/ingest.py: 100%
402 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-11-06 08:17 +0000
1"""Register external source roots, and remove indexed documents durably."""
3from __future__ import annotations
5import asyncio
6import fnmatch
7import logging
8from collections import Counter
9from collections.abc import Callable, Collection, Generator, Iterable
10from contextlib import contextmanager
11from dataclasses import dataclass, field
12from pathlib import Path
14from lilbee.app.services import get_services
15from lilbee.core import settings
16from lilbee.core.config import Config, active_config
17from lilbee.core.config.enums import OcrMode
18from lilbee.core.system import DiskSpelling, disk_path
19from lilbee.data.ingest.discovery import (
20 excluded_extension_reasons,
21 file_hash,
22 resolve_source_path,
23 walk_reaches,
24)
25from lilbee.data.ingest.ignore import IgnoreRules
26from lilbee.data.store import REMOVED_REASON, HeldOut, HeldRecord
27from lilbee.data.store.locations import SourceLocations, root_location
28from lilbee.data.store.types import RemoveResult
29from lilbee.data.types import is_under
30from lilbee.runtime.cancellation import CancelSignal, TaskCancelledError
31from lilbee.runtime.lock import (
32 LibraryUnlockableError,
33 LockingUnsupportedError,
34 refuse_where_a_sync_cannot_lock,
35)
37_ADD_CANCELLED = "Add cancelled."
38_ADD_CANCELLED_ONE = "Add cancelled. {name} was not added."
39_ADD_CANCELLED_MANY = "Add cancelled. {names} were not added."
40_ALSO_HIT_ERROR = " It also hit an error: {error}."
41_CANCEL_ERRORS = (asyncio.CancelledError, TaskCancelledError, KeyboardInterrupt)
42# What an add, a remove or a sync raises when another lilbee process or a data
43# root it cannot lock keeps it from starting.
44LIBRARY_REFUSALS = (
45 LibraryUnlockableError,
46 LockingUnsupportedError,
47)
50@dataclass
51class RegisterResult:
52 """Result of registering source roots into the knowledge base."""
54 registered: list[str] = field(default_factory=list) # labels newly registered
55 name_taken: list[str] = field(default_factory=list)
56 """Labels held by a different live source or an owned entry; ``--force`` overwrites."""
57 overlapping: list[str] = field(default_factory=list)
58 """Paths nesting under or over ``documents_dir`` or a live root; none is registered."""
59 containing: list[str] = field(default_factory=list)
60 """The overlapping paths that contain the documents directory; none is registered."""
61 absorbed_into: dict[str, list[str]] = field(default_factory=dict)
62 """Each newly registered label that took in registered sources, with their labels."""
63 refused: list[str] = field(default_factory=list)
64 """Files whose format lilbee does not index, as ``name: reason``."""
65 outside_corpus: list[str] = field(default_factory=list)
66 """One name per distinct path that is refused, name-taken or around the documents directory."""
67 tracked: list[str] = field(default_factory=list)
68 """Named sources the knowledge base already tracks, so nothing was registered.
70 Either the path already lives under ``documents_dir`` or this exact source is
71 already registered under that label. Nothing is wrong and ``--force`` would
72 change nothing: the sync that follows covers them.
73 """
75 @property
76 def reached_corpus(self) -> bool:
77 """Whether a named path is in the corpus, so a sync has something to index for it.
79 A refused or missing path is not, and neither is one whose label another
80 source holds; a sync after one is a whole-vault pass whose summary would
81 read as the outcome of the add.
82 """
83 return bool(self.registered or self.tracked or self.overlapping)
85 @property
86 def absorbed(self) -> list[str]:
87 """The labels of the sources a newly registered parent took in."""
88 return [label for labels in self.absorbed_into.values() for label in labels]
90 @property
91 def revocable(self) -> list[str]:
92 """The registered labels a cancelled add un-registers: each that took in no source."""
93 return [label for label in self.registered if label not in self.absorbed_into]
95 @property
96 def overlapping_inside(self) -> list[str]:
97 """The overlapping paths a registered root already walks."""
98 return list((Counter(self.overlapping) - Counter(self.containing)).elements())
101def _resolve_label(
102 base: str, roots: dict[str, str], docs_resolved: Path, *, force: bool
103) -> str | None:
104 """Choose the source-key label for a new root, or None when the name is taken.
106 An owned ``documents_dir`` top-level entry of the same name always wins and is
107 never shadowed, even under ``force`` -- a label that shadows it would make
108 resolve_source_path disagree with how discovery keyed the owned file. Reuses
109 the label when a root of that name was registered before but its path has since
110 vanished (the source moved: re-register in place, no ``--force`` needed) or
111 when ``force`` overwrites a live registered root of the same name.
112 """
113 if (docs_resolved / base).exists():
114 return None # an owned entry holds this name; never shadow it
115 existing = roots.get(base)
116 if existing is not None and not Path(existing).exists():
117 return base # dangling root; the source moved, re-point it to the new path
118 if force:
119 return base
120 if base in roots:
121 return None
122 return base
125def _live_roots(spelled: dict[str, Path]) -> list[Path]:
126 """Each of the disk paths in *spelled* that exists; a vanished root overlaps nothing."""
127 return [path for path in spelled.values() if path.exists()]
130def _inside_a_root(src: Path, live_roots: list[Path]) -> bool:
131 """Whether a live root already walks *src*; a second root there would index it twice."""
132 return any(src.is_relative_to(root) for root in live_roots)
135def _reached_below(src: Path, spelled: dict[str, Path]) -> dict[str, Path]:
136 """Each root of *spelled* below *src* that the walk of *src* reaches, by label.
138 *spelled* holds the disk path of each registered root. A root whose folder is
139 gone counts by its stored path.
140 """
141 below = {
142 label: path for label, path in spelled.items() if path != src and path.is_relative_to(src)
143 }
144 if not below:
145 return below
146 rules = IgnoreRules.for_corpus(active_config().data_root)
147 return {label: path for label, path in below.items() if walk_reaches(src, path, rules)}
150def _classify(
151 paths: list[Path], roots: dict[str, str], docs_resolved: Path, *, force: bool
152) -> RegisterResult:
153 """Sort *paths* by what registering each one does, leaving *roots* as the registry after it.
155 A path of this call that a later path of it contains was never a source: it
156 leaves the registry and is not reported as taken in.
157 """
158 result = RegisterResult()
159 sources = frozenset(roots)
160 spelling = DiskSpelling()
161 # The disk is asked one time for each root; every comparison below reads this.
162 spelled = {label: spelling.of(Path(target)) for label, target in roots.items()}
163 by_target = {path: label for label, path in spelled.items()}
164 refused = excluded_extension_reasons()
165 outside: dict[Path, str] = {}
166 for p in paths:
167 src = spelling.of(p)
168 reason = refused.get(src.suffix.lower()) if src.is_file() else None
169 if reason is not None:
170 result.refused.append(f"{p.name}: {reason}")
171 outside.setdefault(src, p.name)
172 continue
173 if src == docs_resolved or docs_resolved in src.parents:
174 result.tracked.append(p.name) # already owned by the knowledge base
175 continue
176 already = by_target.get(src)
177 if already is not None:
178 result.tracked.append(already) # this exact source is already registered
179 continue
180 if _inside_a_root(src, _live_roots(spelled)):
181 result.overlapping.append(p.name) # would walk the same files twice
182 continue
183 reached = _reached_below(src, spelled)
184 if docs_resolved.is_relative_to(src):
185 result.overlapping.append(p.name)
186 result.containing.append(p.name)
187 outside.setdefault(src, p.name)
188 continue
189 others = {name: target for name, target in roots.items() if name not in reached}
190 label = _resolve_label(src.name, others, docs_resolved, force=force)
191 if label is None:
192 result.name_taken.append(src.name)
193 outside.setdefault(src, src.name)
194 continue
195 for child in reached:
196 del roots[child]
197 by_target.pop(spelled.pop(child), None)
198 taken = sorted(child for child in reached if child in sources)
199 roots[label] = str(src)
200 spelled[label] = src
201 by_target[src] = label
202 result.registered = [name for name in result.registered if name not in reached] + [label]
203 if taken:
204 result.absorbed_into[label] = taken
205 result.outside_corpus = list(outside.values())
206 return result
209def names_outside_corpus(paths: list[Path]) -> list[str]:
210 """The name of each of *paths* the corpus does not hold, by the rule registration applies."""
211 if not paths:
212 return [] # a stopped sync has no paths, and reads no registry to say so
213 config = active_config()
214 roots = dict(settings.load(config.data_root).get("linked_roots") or {})
215 plan = _classify(paths, roots, disk_path(config.documents_dir), force=False)
216 return [*plan.outside_corpus, *plan.registered]
219def source_label_taken(name: str, target: Path | None = None) -> bool:
220 """Whether registering *target* under *name* would collide with a different source.
222 The confirm-before-overwrite affordance in the TUI reads this. The label
223 rule is delegated to :func:`_resolve_label` (the authority register_sources
224 itself applies) rather than mirrored, and a *target* whose exact path is
225 already registered under *name* is not a collision: re-adding the same
226 source is idempotent, matching register_sources' by-target no-op.
227 """
228 config = active_config()
229 roots = dict(config.linked_roots)
230 held = roots.get(name)
231 if target is not None and held is not None and disk_path(Path(held)) == disk_path(target):
232 return False
233 return _resolve_label(name, roots, config.documents_dir.resolve(), force=False) is None
236def register_sources(paths: list[Path], *, force: bool = False) -> RegisterResult:
237 """Register each path as a root lilbee indexes where it already lives.
239 A prepared corpus is already on local disk, so ``add`` records where it is
240 rather than copying or linking it: discovery walks the registered root and
241 keys its files under the root's label (its basename). A path already inside
242 ``documents_dir`` is left to the owned-files walk; a path already registered
243 under the same target is a no-op; a label already taken by a different live
244 root or an owned entry is skipped unless ``force``. The registry is persisted
245 so later processes index the same roots.
247 A directory that contains registered sources takes each one its walk reaches:
248 the source leaves the registry, and its files, held-out records and wiki
249 citations show below the new label, with nothing moved, extracted or embedded
250 again. A label that takes a new folder takes the sources of its old one along;
251 an add that stops before that leaves them to the next sync, which pairs them
252 with their files by content.
253 A data root whose filesystem cannot lock files raises ``LockingUnsupportedError``
254 for every add, and a sync lock that cannot be opened raises
255 ``LibraryUnlockableError``, because the sync after the add refuses there.
256 """
257 config = active_config()
258 if paths:
259 refuse_where_a_sync_cannot_lock(config.data_root)
260 documents_dir = config.documents_dir
261 documents_dir.mkdir(parents=True, exist_ok=True)
262 docs_resolved = disk_path(documents_dir)
263 if not paths:
264 return RegisterResult()
266 def _mutate(
267 persisted: dict[str, str] | None,
268 ) -> tuple[dict[str, str], tuple[RegisterResult, dict[str, str]]]:
269 # Read the registry from config.toml INSIDE the lock (not the possibly
270 # stale in-memory copy) so two processes registering roots concurrently
271 # cannot lose each other's entry.
272 before = dict(persisted or {})
273 roots = dict(before)
274 result = _classify(paths, roots, docs_resolved, force=force)
275 return roots, (result, before)
277 result, before = settings.mutate_value(config.data_root, "linked_roots", _mutate)
278 settings.sync_from_resolver(config, ["linked_roots"])
279 if result.registered:
280 _follow_registered(before, result)
281 names = SourceLocations(config.linked_roots)
282 taken = [names.name(root_location(before[label])) for label in result.absorbed]
283 if taken:
284 show_current_paths_in_wiki()
285 unmark_sources_under(paths, spared=[*taken, *_kept_sources(result)])
286 return result
289def _follow_registered(before: dict[str, str], result: RegisterResult) -> None:
290 """Move the sources of each label *result* gave a new folder there, then name by location.
292 A source whose label is now registered is stored where it is. A parent that
293 takes in the source of its own label leaves that source's files where they are.
294 """
295 store = get_services().store
296 for label in result.registered:
297 if label in before and label not in result.absorbed_into.get(label, []):
298 store.relocate_root(before[label], active_config().linked_roots[label])
299 store.fill_locations()
302def _kept_sources(result: RegisterResult) -> list[str]:
303 """The label of each registered source the add did not name, whose records stay."""
304 named = {*result.registered, *result.tracked}
305 return [label for label in active_config().linked_roots if label not in named]
308def unmark_sources_under(paths: list[Path], *, spared: Iterable[str] = ()) -> None:
309 """Hold out of a sync no more any source *paths* covers.
311 A record at or below a key in *spared* stays, unless a path names a file
312 inside that source: the add asks for that file back. A parent that takes in a
313 source keeps what the user removed from it and what failed in it.
315 A record exists to stop *discovery* from resurrecting a source the user
316 removed, or from re-paying the extract cost on a file that yielded nothing.
317 Naming the path outranks it: ``add`` is the user asking for that source
318 back, so the record goes and the sync that follows ingests the file again.
319 Without this a removal would be permanent, undoable only by ``rebuild``,
320 which the user has no reason to reach for after typing the path they want.
322 Each root in *paths* must be registered when this runs: held keys resolve
323 to files through the live registry.
324 """
325 kept = list(spared)
326 store = get_services().store
327 covered, inside = _held_covering(store.held_records(), paths)
328 store.release_held(
329 name for name in covered if name in inside or not any(is_under(name, key) for key in kept)
330 )
333def _held_covering(held: Collection[str], paths: list[Path]) -> tuple[set[str], set[str]]:
334 """Held keys whose file is one of *paths* or lives beneath one, and those a path below
335 their own root covers.
337 Each key is resolved back to the file it tracks -- the same mapping
338 discovery keyed it by -- so an owned ``documents_dir`` entry, a file under a
339 registered root, and a single-file root are all matched by the one rule
340 instead of three shape-specific ones.
341 """
342 if not held:
343 return set(), set() # nothing to match, so the disk is not asked
344 config = active_config()
345 with settings.fresh_registry_pinned(config):
346 return _covering_one_registry(config, held, paths)
349def _covering_one_registry(
350 config: Config, held: Collection[str], paths: list[Path]
351) -> tuple[set[str], set[str]]:
352 """Match *held* against *paths* while the registry in *config* stays fixed."""
353 spelling = DiskSpelling()
354 named = [spelling.of(p) for p in paths]
355 bases = (config.documents_dir, *map(Path, config.linked_roots.values()))
356 covered: set[str] = set()
357 inside: set[str] = set()
358 for name in held:
359 stored = resolve_source_path(name)
360 base = next(base for base in bases if stored.is_relative_to(base))
361 # The disk spells the root, one time; the rest of a key is the walk's own spelling.
362 root = spelling.of(base)
363 tracked = (root / stored.relative_to(base)).resolve(strict=False)
364 hits = [path for path in named if tracked == path or path in tracked.parents]
365 if hits:
366 covered.add(name)
367 if any(path != root and path.is_relative_to(root) for path in hits):
368 inside.add(name)
369 return covered, inside
372_GLOB_CHARS = frozenset("*?[")
375def _is_glob(name: str) -> bool:
376 """Whether *name* should be matched as a glob rather than a literal source."""
377 return any(char in _GLOB_CHARS for char in name)
380def folder_members(name: str, known: Iterable[str]) -> list[str]:
381 """Known sources under folder *name*, matched on whole path segments.
383 ``myrepo`` covers ``myrepo/a.py`` but never ``myrepo-2/x``. Empty when *name*
384 is not a parent directory of any known source.
385 """
386 prefix = name.rstrip("/") + "/"
387 return [source for source in known if source.startswith(prefix)]
390def held_failures() -> dict[str, HeldRecord]:
391 """The record of each file an ingestion failure holds out, by path, sorted."""
392 held = get_services().store.held_records()
393 return {name: held[name] for name in sorted(held) if held[name].kind is HeldOut.FAILED}
396def removable_names(indexed: list[str] | None = None) -> list[str]:
397 """What a remove can name: *indexed* (the store's sources when not given), then failures."""
398 if indexed is None:
399 indexed = [s["filename"] for s in get_services().store.get_sources()]
400 seen = set(indexed)
401 return indexed + [name for name in held_failures() if name not in seen]
404def expand_remove_targets(names: list[str], known: list[str]) -> list[str]:
405 """Expand folder names and glob patterns to the known sources they cover.
407 An exact source name is kept. A folder name (a parent directory of known
408 sources) expands to every source beneath it. A glob (a name containing
409 ``* ? [``) expands to every source it fnmatches. A name matching none of
410 these is kept unchanged so the caller reports it not-found. Order and
411 de-duplication are preserved. *known* is what a remove can name,
412 ``removable_names()``.
413 """
414 known_set = set(known)
415 expanded: list[str] = []
416 seen: set[str] = set()
418 def _add(candidate: str) -> None:
419 if candidate not in seen:
420 seen.add(candidate)
421 expanded.append(candidate)
423 for name in names:
424 if name in known_set:
425 _add(name)
426 continue
427 if _is_glob(name):
428 matches = [source for source in known if fnmatch.fnmatchcase(source, name)]
429 else:
430 matches = folder_members(name, known)
431 if matches:
432 for match in matches:
433 _add(match)
434 else:
435 _add(name) # not-found; reported by the store
436 return expanded
439def unregister_roots(names: Iterable[str]) -> list[str]:
440 """Un-register any top-level source root named in *names*. Returns removed labels.
442 ``add`` registers a source root; removing it by its label drops the registry
443 entry so discovery stops finding its files, which then need no held-out record.
444 The source bytes on disk are never touched. Nested names (``corpus/a.txt``)
445 are not roots and are left alone.
446 """
447 config = active_config()
448 labels = list(names)
449 removed: list[str] = []
450 if not labels:
451 return removed
453 def _mutate(persisted: dict[str, str] | None) -> tuple[dict[str, str], list[str]]:
454 roots = dict(persisted or {})
455 for name in labels:
456 label = name.strip("/")
457 if "/" in label or label not in roots:
458 continue
459 del roots[label]
460 removed.append(label)
461 return roots, removed
463 result = settings.mutate_value(config.data_root, "linked_roots", _mutate)
464 settings.sync_from_resolver(config, ["linked_roots"])
465 return result
468log = logging.getLogger(__name__)
471def remove_documents_durably(
472 names: list[str],
473 targets: list[str] | None = None,
474) -> RemoveResult:
475 """Remove documents from the index (folders and globs expand) and make it stick.
477 Never deletes source bytes. A name is an indexed source, a file an ingestion
478 failure holds out, a folder or glob covering either, or a registered root.
479 Each removed file is held out of every later sync as a removal, at its
480 current hash, so a held-out failure stays out instead of being retried.
481 Removing a registered root un-registers it and drops the held-out records under
482 it: discovery can no longer find its files. Editing the source (new hash),
483 ``rebuild`` or adding the path again restores it; ``retry-skipped`` does not.
484 *targets* (the expanded names) is computed when not supplied; a caller that
485 already expanded for a confirmation prompt passes it to avoid re-expanding.
487 The removal and the records it holds out are one write of the index.
488 """
489 store = get_services().store
490 known = removable_names()
491 if targets is None:
492 targets = expand_remove_targets(names, known)
493 failed = held_failures()
494 settings.refresh_registry(active_config())
495 linked = active_config().linked_roots
496 going = [label for label in (name.strip("/") for name in names) if label in linked]
497 removable = set(known)
498 holds = _removal_holds([name for name in targets if name in removable], going, failed)
499 result = store.remove_documents(targets, holds)
500 held = [name for name in result.not_found if name in failed]
501 roots = forget_roots(names)
502 forget_removed_from_wiki_index(list(result.removed))
503 missing = [name for name in result.not_found if name not in failed]
504 emptied = [name for name in missing if name.strip("/") in roots]
505 return RemoveResult(
506 removed=[*result.removed, *held, *emptied],
507 not_found=[name for name in missing if name not in emptied],
508 )
511def forget_roots(names: list[str]) -> list[str]:
512 """Un-register every root named in *names* and drop the held-out records under it.
514 A source row a root keeps for a citation is stored by its shown name first.
515 The roots are un-registered even when the records cannot be changed; the
516 error is raised after.
517 """
518 settings.refresh_registry(active_config())
519 roots = active_config().linked_roots
520 labels = [label for label in (name.strip("/") for name in names) if label in roots]
521 named = [Path(roots[label]) for label in labels]
522 try:
523 # Records resolve through the registry, so they go before un-registering.
524 # A root that stays keeps its records, also when it names the folder of one that goes.
525 unmark_sources_under(named, spared=roots.keys() - set(labels))
526 get_services().store.name_by_label(labels)
527 finally:
528 unregistered = unregister_roots(names)
529 return unregistered
532@dataclass
533class AddRollback:
534 """What an interrupted add did not add, and any error it hit once stopped."""
536 not_added: list[str] = field(default_factory=list)
537 error: str | None = None
538 paths: list[Path] = field(default_factory=list)
539 """The paths the add was given."""
540 at_sync: bool = True
541 """Whether the add reached its sync."""
542 absorbed_into: dict[str, list[str]] = field(default_factory=dict)
543 """Each label the add registered that took in sources, with their labels; it stays."""
544 _roots: list[str] = field(default_factory=list)
545 _before: dict[str, str] = field(default_factory=dict)
547 def message(self, nothing_dropped: str) -> str:
548 """Why the add stopped, naming what it did not add and any error it also hit."""
549 stopped = self._stopped(nothing_dropped)
550 return stopped if self.error is None else stopped + _ALSO_HIT_ERROR.format(error=self.error)
552 def note_error(self, exc: BaseException) -> None:
553 """Log and name *exc*, raised once the add was stopped; a cancel names its cause."""
554 if isinstance(exc, _CANCEL_ERRORS):
555 # A cancel that took a failed write carries that error as its cause.
556 carried = exc.__cause__
557 if carried is not None and not isinstance(carried, _CANCEL_ERRORS):
558 self.error = str(carried) or type(carried).__name__
559 return
560 log.warning("A stopped sync also hit an error", exc_info=exc)
561 self.error = str(exc) or type(exc).__name__
563 def registered(self, labels: list[str], cancel: CancelSignal) -> None:
564 """Take *labels* as the roots the add registered; its sync starts here."""
565 self._roots, self.at_sync = labels, True
566 try:
567 self._before = indexed_stamps(labels)
568 except Exception as exc:
569 if not cancel.is_set():
570 raise
571 self.note_error(exc)
572 # A read that fails once stopped keeps every root with an indexed file.
573 self._before = {}
575 def forget_unfinished(self) -> None:
576 """Un-register each root the sync indexed nothing under, then name what the corpus lacks.
578 An error a step raises is named with the cancel, and the other step still runs.
579 """
580 for step in (self._forget_roots, self._name_missing):
581 try:
582 step()
583 except Exception as exc:
584 self.note_error(exc)
586 def _forget_roots(self) -> None:
587 """Un-register each root with no file indexed since the sync started."""
588 forget_unfinished_roots(self._roots, self._before)
590 def _name_missing(self) -> None:
591 """Take the name of each given path the corpus does not hold."""
592 self.not_added = names_outside_corpus(self.paths)
594 def _stopped(self, nothing_dropped: str) -> str:
595 match self.not_added:
596 case [] if self.at_sync:
597 return nothing_dropped
598 case []:
599 return _ADD_CANCELLED
600 case [name]:
601 return _ADD_CANCELLED_ONE.format(name=name)
602 case _:
603 return _ADD_CANCELLED_MANY.format(names=", ".join(self.not_added))
606def indexed_stamps(labels: list[str]) -> dict[str, str]:
607 """The ``ingested_at`` stamp of every indexed source under the roots *labels*."""
608 if not labels:
609 return {}
610 return {
611 source["filename"]: source["ingested_at"]
612 for source in get_services().store.get_sources()
613 if any(is_under(source["filename"], label) for label in labels)
614 }
617def forget_unfinished_roots(labels: list[str], before: dict[str, str]) -> list[str]:
618 """Un-register each root in *labels* with no file indexed since *before*; returns them.
620 A file is indexed since *before* when its source is new or its stamp changed.
621 """
622 finished = [name for name, stamp in indexed_stamps(labels).items() if before.get(name) != stamp]
623 unfinished = [label for label in labels if not any(is_under(name, label) for name in finished)]
624 return forget_roots(unfinished) if unfinished else []
627@contextmanager
628def leave_as_cancel(
629 rollback: AddRollback, cancel: CancelSignal, user_cancelled: Callable[[], bool]
630) -> Generator[None, None, None]:
631 """Wrap an add; anything raised while *cancel* is set leaves as a cancel after *rollback*."""
632 try:
633 yield
634 except BaseException as exc:
635 if not cancel.is_set():
636 raise
637 rollback.note_error(exc)
638 if user_cancelled():
639 rollback.forget_unfinished()
640 raise asyncio.CancelledError from exc
643@contextmanager
644def forget_unfinished_on_cancel(
645 paths: list[Path], labels: list[str], cancel: CancelSignal, user_cancelled: Callable[[], bool]
646) -> Generator[AddRollback, None, None]:
647 """Wrap an add of *paths* after registration; a raise under a set *cancel* is a cancel."""
648 rollback = AddRollback(paths=paths)
649 rollback.registered(labels, cancel)
650 with leave_as_cancel(rollback, cancel, user_cancelled):
651 yield rollback
654def _removal_holds(
655 names: list[str], roots: list[str], failed: dict[str, HeldRecord]
656) -> dict[str, HeldRecord]:
657 """The removal record each of *names* takes, except under a root that is un-registered.
659 The record takes the file's current hash. A source with no file, an imported
660 one or an archive member, takes none; a held-out failure that is not
661 reachable keeps the hash it failed at.
662 """
663 holds: dict[str, HeldRecord] = {}
664 for name in names:
665 if any(is_under(name, root) for root in roots):
666 continue # the root is gone; discovery won't resurrect these
667 path = resolve_source_path(name)
668 if path.is_file():
669 holds[name] = HeldRecord(HeldOut.REMOVED, file_hash(path), REMOVED_REASON)
670 elif name in failed:
671 holds[name] = HeldRecord(HeldOut.REMOVED, failed[name].file_hash, REMOVED_REASON)
672 return holds
675def remove_sources_and_forget_in_wiki(names: list[str]) -> RemoveResult:
676 """Remove indexed sources by name and drop them from the wiki's browse index."""
677 result = get_services().store.remove_documents(names)
678 forget_removed_from_wiki_index(list(result.removed))
679 return result
682def forget_removed_from_wiki_index(removed: list[str]) -> None:
683 """Drop removed documents from the wiki's browse index.
685 Their removal records keep them out of later syncs, so no refresh would ever
686 revisit their entries and the tree would keep offering pages the library
687 can no longer support. Best effort: the removal itself already succeeded.
688 """
689 if removed:
690 forget_missing_from_wiki_index()
693def forget_missing_from_wiki_index() -> None:
694 """Drop from the wiki's browse index each source it lists that holds no source row.
696 For the parent of a fan-out that stopped: its workers remove sources and
697 report nothing. Best effort, as the removal already succeeded.
698 """
699 config = active_config()
700 if not config.wiki:
701 return
702 from lilbee.wiki.stubs import drop_removed_sources
704 try:
705 drop_removed_sources(get_services().store, config)
706 except Exception:
707 log.warning("Failed to drop missing documents from the wiki index", exc_info=True)
710def show_current_paths_in_wiki() -> None:
711 """Render again each wiki page that shows a path its cited source no longer has.
713 Best effort, as the sync itself already succeeded.
714 """
715 store = get_services().store
716 try:
717 citations = store.wiki_citations_by_page()
718 if not citations:
719 return
720 # heavy: the wiki package loads spaCy
721 from lilbee.wiki.rerender import rerender_stale_pages
723 rerender_stale_pages(store, active_config(), citations)
724 except Exception:
725 log.warning("Failed to show the current document paths in the wiki pages", exc_info=True)
728@contextmanager
729def temporary_ocr_config(
730 ocr: OcrMode | None = None,
731 ocr_timeout: float | None = None,
732) -> Generator[None, None, None]:
733 """Override OCR config for the duration of the block, per request.
735 Backed by a ContextVar rather than a global ``cfg`` mutation, so concurrent
736 ingests on the shared HTTP daemon do not clobber one another's OCR settings.
737 """
738 from lilbee.data.extract.document import ocr_override
740 with ocr_override(ocr, ocr_timeout):
741 yield