Coverage for src/lilbee/cli/commands/ingest_sync.py: 100%
416 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-11-06 08:17 +0000
1"""Sync, rebuild, add, chunks, and remove commands."""
3from __future__ import annotations
5import asyncio
6import functools
7import threading
8from contextlib import contextmanager
9from dataclasses import dataclass, field
10from pathlib import Path
11from typing import TYPE_CHECKING, Any, NoReturn
13import typer
14from rich.text import Text
16if TYPE_CHECKING:
17 from collections.abc import Awaitable, Callable, Iterator
19 from lilbee.data.ingest import SyncResult
20 from lilbee.data.store.types import RemoveResult
21 from lilbee.runtime.progress import DetailedProgressCallback
23from lilbee.app.ingest import (
24 LIBRARY_REFUSALS,
25 AddRollback,
26 RegisterResult,
27 expand_remove_targets,
28 leave_as_cancel,
29 register_sources,
30 removable_names,
31 remove_documents_durably,
32)
33from lilbee.app.search import clean_results
34from lilbee.app.services import get_services
35from lilbee.cli import theme
36from lilbee.cli.app import (
37 apply_overrides,
38 console,
39 data_dir_option,
40 global_option,
41)
42from lilbee.cli.commands._shared import CHUNK_PREVIEW_LEN, MAX_FAILURES_SHOWN
43from lilbee.cli.commands.analyze import print_tip_if_shown
44from lilbee.cli.commands.crawl_login import crawl_login
45from lilbee.cli.helpers import (
46 add_paths,
47 json_output,
48 print_prefixed,
49 sync_result_to_json,
50)
51from lilbee.core.config import cfg
52from lilbee.core.config.defaults import MAX_CRAWL_DEPTH, MAX_CRAWL_PAGES
53from lilbee.core.config.enums import OcrMode
54from lilbee.core.text import strip_control_characters
55from lilbee.crawler import is_url
56from lilbee.crawler.models import NO_LOGIN, CrawlLogin
57from lilbee.crawler.url_filter import without_userinfo
59_SYNC_CANCELLED_MESSAGE = "Sync cancelled."
60# The shell's exit status for a command stopped by Ctrl+C (128 + SIGINT).
61_EXIT_INTERRUPTED = 130
63_ocr_option = typer.Option(
64 None,
65 "--ocr",
66 help=(
67 "Scanned pages for this run: auto reads each page without usable text and keeps "
68 "the text of every other page; all reads every page of every file; off skips them. "
69 "Leave it out to use the ocr setting."
70 ),
71)
72_retry_skipped_option = typer.Option(
73 False,
74 "--retry-skipped",
75 help="Retry files that were skipped on a previous sync (clears the failed-file markers).",
76)
77_prune_ignored_option = typer.Option(
78 False,
79 "--prune-ignored",
80 help="Also drop indexed documents a .lilbeeignore now excludes. Source files are kept.",
81)
82_ocr_timeout_option = typer.Option(
83 None,
84 "--ocr-timeout",
85 help="Per-page timeout in seconds for vision OCR (default: 300, 0 = no limit).",
86)
89def _apply_ocr_overrides(ocr: OcrMode | None, ocr_timeout: float | None) -> None:
90 """Apply --ocr and --ocr-timeout CLI overrides to config.
92 The CLI is a single-shot, single-process invocation, so mutating the global
93 cfg here is safe (it mirrors ``apply_overrides`` for the data dir). The
94 daemon-shared per-request OCR override uses a ContextVar instead; see
95 ``temporary_ocr_config``.
96 """
97 if ocr is not None:
98 cfg.ocr = ocr
99 if ocr_timeout is not None:
100 cfg.ocr_timeout = ocr_timeout
103_paths_argument = typer.Argument(
104 ...,
105 help="Files, directories, or URLs to add to the knowledge base.",
106)
108_force_option = typer.Option(False, "--force", "-f", help="Overwrite existing files.")
109_max_cpus_option = typer.Option(
110 None,
111 "--max-cpus",
112 min=1,
113 help="Cap the workers used to discover and hash files. Unset = auto (all available cores).",
114)
115_processes_option = typer.Option(
116 None,
117 "--processes",
118 min=0,
119 help=(
120 "Ingest worker processes, one GPU each: N explicit, 0 = auto (one per card),"
121 " 1 = this process only."
122 ),
123)
124_crawl_option = typer.Option(
125 False,
126 "--crawl",
127 help="Recursively crawl URLs (whole site by default; see --depth and --max-pages).",
128)
129_depth_option = typer.Option(
130 None,
131 "--depth",
132 min=0,
133 max=MAX_CRAWL_DEPTH,
134 help=(
135 "Cap link-follow depth for --crawl. Unset = unbounded; 0 = single URL only; "
136 f"at most {MAX_CRAWL_DEPTH}."
137 ),
138)
139_max_pages_option = typer.Option(
140 None,
141 "--max-pages",
142 min=0,
143 max=MAX_CRAWL_PAGES,
144 help=(
145 "Cap pages for --crawl. Unset = protective default; 0 = unlimited; "
146 f"N = hard cap, at most {MAX_CRAWL_PAGES}."
147 ),
148)
149_include_subdomains_option = typer.Option(
150 False,
151 "--include-subdomains",
152 help=(
153 "Allow --crawl to follow links into sibling subdomains of the start "
154 "host (e.g. en.wikipedia.org plus af.wikipedia.org). Default scopes "
155 "the crawl to the exact start host only."
156 ),
157)
159_authorization_option = typer.Option(
160 "",
161 "--authorization",
162 help="Authorization header value sent to the one host of the URLs, e.g. 'Bearer <token>'.",
163)
164_username_option = typer.Option(
165 "", "--username", help="Basic-auth username sent to the one host of the URLs."
166)
167_password_option = typer.Option("", "--password", help="Basic-auth password sent with --username.")
168_cookie_option = typer.Option(
169 "",
170 "--cookie",
171 help="Cookie header value sent to the one host of the URLs, e.g. 'session=<id>'.",
172)
175def _partition_inputs(inputs: list[str]) -> tuple[list[Path], list[str]]:
176 """Split inputs into file paths and URLs."""
177 paths: list[Path] = []
178 urls: list[str] = []
179 for inp in inputs:
180 if is_url(inp):
181 urls.append(inp)
182 else:
183 paths.append(Path(inp))
184 return paths, urls
187@dataclass(frozen=True)
188class _FailedSource:
189 """A requested URL that added nothing, and why."""
191 url: str
192 reason: str
195@dataclass
196class _CrawlOutcome:
197 """The pages the crawls of one add saved, and the URLs that failed."""
199 saved: list[Path] = field(default_factory=list)
200 failed: list[_FailedSource] = field(default_factory=list)
202 def record(
203 self,
204 url: str,
205 crawl_one: Callable[[], list[Path]],
206 page_failures: list[str],
207 crawled: dict[str, int],
208 ) -> list[Path]:
209 """Crawl the seed shown as *url* with *crawl_one*; keep its pages, or its failure.
211 *url* failed when its address is refused (ValueError), when its crawl cannot start
212 (RuntimeError), or when it saved no page and every page its crawl counted failed.
213 """
214 try:
215 paths = crawl_one()
216 except (ValueError, RuntimeError) as exc:
217 self.failed.append(_FailedSource(url, f"{url}: {exc}"))
218 return []
219 self.saved.extend(paths)
220 if page_failures and not paths and len(page_failures) >= crawled.get("n", 0):
221 self.failed.append(_FailedSource(url, "\n".join(page_failures)))
222 return paths
225def _crawl_urls_blocking(
226 urls: list[str],
227 *,
228 crawl: bool,
229 depth: int | None,
230 max_pages: int | None,
231 cancel_event: threading.Event,
232 include_subdomains: bool = False,
233 login: CrawlLogin = NO_LOGIN,
234) -> _CrawlOutcome:
235 """Crawl each URL in turn, sending *login* to the one host of *urls*; a set *cancel_event*
236 raises a cancel.
238 A URL whose address is refused, or whose crawl saved nothing and failed on every page,
239 is a failed source; the next URL is still crawled.
240 """
241 from rich.progress import Progress, SpinnerColumn, TaskID
243 from lilbee.crawler import crawl_and_save
244 from lilbee.runtime.progress import (
245 CrawlDoneEvent,
246 CrawlPageEvent,
247 CrawlPageFailedEvent,
248 EventType,
249 ProgressEvent,
250 )
251 from lilbee.runtime.progress.columns import literal_text_column
253 effective_depth, effective_pages = (depth, max_pages) if crawl else (0, None)
255 from lilbee.runtime.console import PlainConsole
257 err_console = PlainConsole(stderr=True)
258 outcome = _CrawlOutcome()
259 with Progress(
260 SpinnerColumn(),
261 literal_text_column("{task.description}"),
262 transient=True,
263 console=err_console,
264 disable=cfg.json_mode,
265 ) as progress:
266 for url in urls:
267 shown = without_userinfo(url)
268 ptask = progress.add_task(f"Crawling {shown}...", total=None)
269 crawled: dict[str, int] = {}
270 failures: list[str] = []
272 def _make_callback(
273 _t: TaskID = ptask,
274 _crawled: dict[str, int] = crawled,
275 _failures: list[str] = failures,
276 ) -> DetailedProgressCallback:
277 def on_progress(event_type: EventType, data: ProgressEvent) -> None:
278 if event_type == EventType.CRAWL_PAGE:
279 if not isinstance(data, CrawlPageEvent):
280 raise TypeError(f"Expected CrawlPageEvent, got {type(data).__name__}")
281 total_str = str(data.total) if data.total > 0 else "?"
282 progress.update(
283 _t,
284 description=f"Crawled {data.current}/{total_str}: {data.url}",
285 )
286 elif event_type == EventType.CRAWL_DONE and isinstance(data, CrawlDoneEvent):
287 _crawled["n"] = data.pages_crawled
288 elif event_type == EventType.CRAWL_PAGE_FAILED and isinstance(
289 data, CrawlPageFailedEvent
290 ):
291 _failures.append(f"{data.url}: {data.reason}")
293 return on_progress
295 paths = outcome.record(
296 shown,
297 functools.partial(
298 _run_crawl_with_signal_cancel,
299 url,
300 depth=effective_depth,
301 max_pages=effective_pages,
302 on_progress=_make_callback(),
303 cancel_event=cancel_event,
304 crawl_and_save=crawl_and_save,
305 include_subdomains=include_subdomains,
306 login=login,
307 ),
308 failures,
309 crawled,
310 )
311 progress.update(ptask, description=f"Done: {shown} ({len(paths)} pages)")
312 # No explicit cap given and the crawl filled the protective default:
313 # tell the user how to go unlimited without editing settings.
314 default_cap = cfg.crawl_max_pages or cfg.crawl_safety_max_pages
315 if crawl and max_pages is None and crawled.get("n", 0) >= default_cap:
316 err_console.print(
317 f"Stopped at the default {default_cap}-page limit; "
318 f"pass --max-pages 0 to crawl unlimited (or --max-pages N for a higher cap).",
319 )
320 return outcome
323def _run_crawl_with_signal_cancel(
324 url: str,
325 *,
326 depth: int | None,
327 max_pages: int | None,
328 on_progress: DetailedProgressCallback,
329 cancel_event: threading.Event,
330 crawl_and_save: Callable[..., Awaitable[list[Path]]],
331 include_subdomains: bool = False,
332 login: CrawlLogin = NO_LOGIN,
333) -> list[Path]:
334 """Crawl one URL on its own event loop; a crawl that ends under a set cancel raises it."""
335 # Manage the event loop explicitly. In the CLI this runs once per process,
336 # but under pytest-xdist the same worker thread runs many tests; leaving a
337 # closed loop set as the "current" loop for the thread poisons every later
338 # asyncio.get_event_loop() call and hangs macOS 3.12/3.13 unit-test CI.
339 # Always clear the thread-current loop in finally.
340 loop = asyncio.new_event_loop()
341 try:
342 asyncio.set_event_loop(loop)
343 coro = crawl_and_save(
344 url,
345 depth=depth,
346 max_pages=max_pages,
347 on_progress=on_progress,
348 cancel=cancel_event,
349 include_subdomains=include_subdomains,
350 login=login,
351 )
352 result: list[Path] = loop.run_until_complete(coro)
353 # A cancelled crawl returns the pages it saved; the command still stops here.
354 if cancel_event.is_set():
355 raise asyncio.CancelledError
356 return result
357 finally:
358 loop.close()
359 asyncio.set_event_loop(None)
362def _cancellable_progress(
363 cancel_event: threading.Event, chain: DetailedProgressCallback
364) -> DetailedProgressCallback:
365 """Wrap *chain* so a set *cancel_event* aborts the in-flight file cooperatively.
367 The ingest pipeline and the per-page vision OCR loop both call the progress
368 callback between units of work; raising :class:`TaskCancelledError` there is
369 the established cooperative-cancel signal, so a Ctrl+C stops a long OCR
370 between pages instead of after the whole document.
371 """
372 from lilbee.runtime.cancellation import TaskCancelledError
374 def _callback(event_type: object, data: object) -> None:
375 if cancel_event.is_set():
376 raise TaskCancelledError
377 chain(event_type, data) # type: ignore[arg-type]
379 return _callback
382def _exit_cancelled(
383 rollback: AddRollback, nothing_dropped: str = _SYNC_CANCELLED_MESSAGE
384) -> NoReturn:
385 """Report a cancelled command and exit with the Ctrl+C status."""
386 message = rollback.message(nothing_dropped)
387 if cfg.json_mode:
388 payload: dict[str, object] = {"error": message}
389 if rollback.not_added:
390 payload["not_added"] = rollback.not_added
391 if rollback.absorbed_into:
392 payload["copied"] = list(rollback.absorbed_into)
393 payload["absorbed"] = [
394 label for labels in rollback.absorbed_into.values() for label in labels
395 ]
396 json_output(payload)
397 else:
398 console.print(Text(message, style=theme.WARNING))
399 raise SystemExit(_EXIT_INTERRUPTED) from None
402@contextmanager
403def _ctrl_c_sets(cancel_event: threading.Event) -> Iterator[None]:
404 """Make Ctrl+C set *cancel_event* instead of raising KeyboardInterrupt, for the block."""
405 import signal
407 # signal.signal raises ValueError off the main thread (e.g. under pytest-xdist
408 # workers); the cancel_event can still be driven externally there.
409 if threading.current_thread() is not threading.main_thread():
410 yield
411 return
413 def _on_sigint(_signum: int, _frame: object) -> None:
414 cancel_event.set()
416 previous_handler = signal.signal(signal.SIGINT, _on_sigint)
417 try:
418 yield
419 finally:
420 signal.signal(signal.SIGINT, previous_handler)
423@contextmanager
424def _ctrl_c_stops(cancel_event: threading.Event, rollback: AddRollback) -> Iterator[None]:
425 """Run a command under Ctrl+C; one it stops exits 130 after *rollback*."""
426 try:
427 # Only the user's Ctrl+C sets this cancel.
428 with _ctrl_c_sets(cancel_event), leave_as_cancel(rollback, cancel_event, lambda: True):
429 yield
430 except asyncio.CancelledError:
431 _exit_cancelled(rollback)
434def run_sync_with_signal_cancel(
435 *,
436 force_rebuild: bool = False,
437 retry_skipped: bool = False,
438 prune_ignored: bool = False,
439 on_progress: DetailedProgressCallback | None = None,
440 started_by: str = "lilbee sync",
441) -> SyncResult:
442 """Run ``sync`` under Ctrl+C; a sync it stops ends the command with exit status 130."""
443 cancel_event = threading.Event()
444 with _ctrl_c_stops(cancel_event, AddRollback()):
445 return _run_sync(
446 cancel_event,
447 force_rebuild=force_rebuild,
448 retry_skipped=retry_skipped,
449 prune_ignored=prune_ignored,
450 on_progress=on_progress,
451 started_by=started_by,
452 )
455def _wait_shown_on_stderr(chain: DetailedProgressCallback) -> DetailedProgressCallback:
456 """Wrap *chain* so the notice of a sync that waits for another prints to stderr.
458 stderr keeps stdout to the one JSON document in JSON mode.
459 """
460 from lilbee.runtime.console import PlainConsole
461 from lilbee.runtime.progress import EventType, ProgressEvent, SyncWaitEvent
463 err = PlainConsole(stderr=True)
465 def _callback(event_type: EventType, data: ProgressEvent) -> None:
466 if event_type == EventType.SYNC_WAIT and isinstance(data, SyncWaitEvent):
467 err.print(data.message, style=theme.MUTED, soft_wrap=True)
468 chain(event_type, data)
470 return _callback
473def _default_progress() -> DetailedProgressCallback:
474 """Print a line per file when the live bar cannot draw (piped output); else stay silent.
476 JSON mode keeps stdout to the one document.
477 """
478 from lilbee.cli.sync import file_start_printer
479 from lilbee.runtime.progress import noop_callback
481 if cfg.json_mode or console.is_terminal:
482 return noop_callback
483 return file_start_printer(console)
486def _run_sync(
487 cancel_event: threading.Event,
488 *,
489 force_rebuild: bool = False,
490 retry_skipped: bool = False,
491 prune_ignored: bool = False,
492 on_progress: DetailedProgressCallback | None = None,
493 before_sync: Callable[[], None] | None = None,
494 started_by: str,
495) -> SyncResult:
496 """Run ``sync`` on its own event loop; it polls *cancel_event* between files and pages.
498 *before_sync* runs once the eager warm is off, before the sync starts. *started_by*
499 names the command for the notice of a sync that waits for this one.
500 """
501 from lilbee.data.ingest import sync
503 # Batch ingest is a headless one-shot: skip the eager warm so services init
504 # doesn't spawn every role. With lazy per-role spawn, the sync brings up only
505 # the embed server (plus vision/chat if those steps actually run), instead of
506 # holding an idle chat server's VRAM for the whole build.
507 cfg.worker_pool_eager_start = False
508 if before_sync is not None:
509 before_sync()
511 callback = _cancellable_progress(
512 cancel_event, _wait_shown_on_stderr(on_progress or _default_progress())
513 )
514 loop = asyncio.new_event_loop()
515 try:
516 asyncio.set_event_loop(loop)
517 return loop.run_until_complete(
518 sync(
519 force_rebuild=force_rebuild,
520 quiet=cfg.json_mode,
521 on_progress=callback,
522 cancel=cancel_event,
523 retry_skipped=retry_skipped,
524 prune_ignored=prune_ignored,
525 started_by=started_by,
526 )
527 )
528 finally:
529 loop.close()
530 asyncio.set_event_loop(None)
533def sync_cmd(
534 data_dir: Path | None = data_dir_option,
535 use_global: bool = global_option,
536 ocr: OcrMode | None = _ocr_option,
537 ocr_timeout: float | None = _ocr_timeout_option,
538 retry_skipped: bool = _retry_skipped_option,
539 prune_ignored: bool = _prune_ignored_option,
540 max_cpus: int | None = _max_cpus_option,
541 processes: int | None = _processes_option,
542) -> None:
543 """Manually trigger document sync."""
544 apply_overrides(data_dir=data_dir, use_global=use_global)
545 _apply_ocr_overrides(ocr, ocr_timeout)
546 if max_cpus is not None:
547 cfg.ingest_workers = max_cpus
548 if processes is not None:
549 cfg.ingest_processes = processes
551 try:
552 result = run_sync_with_signal_cancel(
553 retry_skipped=retry_skipped, prune_ignored=prune_ignored, started_by="lilbee sync"
554 )
555 except RuntimeError as exc:
556 if cfg.json_mode:
557 json_output({"error": str(exc)})
558 raise SystemExit(1) from None
559 print_prefixed(console, "Error: ", exc, style=theme.ERROR)
560 raise SystemExit(1) from None
561 if cfg.json_mode:
562 json_output(sync_result_to_json(result))
563 return
564 console.print(result)
565 if _changed_content(result):
566 print_tip_if_shown(cfg.data_root)
569def _changed_content(result: SyncResult) -> bool:
570 """True when the sync added, updated or removed a file: the tip is about new content."""
571 return bool(result.added or result.updated or result.removed)
574def rebuild_failure_message(failed: list[str]) -> str:
575 """The failed-file report of a rebuild: how many, which, and what to do next."""
576 shown = ", ".join(failed[:MAX_FAILURES_SHOWN])
577 if len(failed) > MAX_FAILURES_SHOWN:
578 shown += f" and {len(failed) - MAX_FAILURES_SHOWN} more"
579 noun = "document" if len(failed) == 1 else "documents"
580 return (
581 f"{len(failed)} {noun} failed to ingest: {shown}. "
582 "They keep their earlier chunks. Fix the cause, then run `lilbee rebuild` again."
583 )
586def rebuild_or_raise() -> SyncResult:
587 """Re-ingest every document in place; raises RuntimeError on a refused rebuild."""
588 from lilbee.data.ingest import SyncResult
590 result = run_sync_with_signal_cancel(force_rebuild=True, started_by="lilbee rebuild")
591 # untyped return: run_sync_with_signal_cancel returns object to avoid a heavy top-level import
592 if not isinstance(result, SyncResult):
593 raise TypeError(f"Expected SyncResult, got {type(result).__name__}")
594 return result
597def run_rebuild() -> SyncResult:
598 """Re-ingest every document in place; a refused rebuild exits 1 with its reason."""
599 try:
600 return rebuild_or_raise()
601 except RuntimeError as exc:
602 if cfg.json_mode:
603 json_output({"error": str(exc)})
604 raise SystemExit(1) from None
605 print_prefixed(console, "Error: ", exc, style=theme.ERROR)
606 raise SystemExit(1) from None
609def rebuild(
610 data_dir: Path | None = data_dir_option,
611 use_global: bool = global_option,
612 ocr: OcrMode | None = _ocr_option,
613 ocr_timeout: float | None = _ocr_timeout_option,
614 max_cpus: int | None = _max_cpus_option,
615 processes: int | None = _processes_option,
616) -> None:
617 """Re-ingest everything from documents/ over the index; a failed file exits 1."""
618 apply_overrides(data_dir=data_dir, use_global=use_global)
619 _apply_ocr_overrides(ocr, ocr_timeout)
620 if max_cpus is not None:
621 cfg.ingest_workers = max_cpus
622 if processes is not None:
623 cfg.ingest_processes = processes
624 result = run_rebuild()
625 if cfg.json_mode:
626 payload: dict[str, Any] = {
627 "command": "rebuild",
628 "ingested": len(result.added),
629 "failed": len(result.failed),
630 }
631 if result.failed:
632 payload["error"] = rebuild_failure_message(result.failed)
633 json_output(payload)
634 else:
635 console.print(f"Rebuilt: {len(result.added)} documents ingested")
636 if result.failed:
637 print_prefixed(
638 console, "Error: ", rebuild_failure_message(result.failed), style=theme.ERROR
639 )
640 if result.failed:
641 raise SystemExit(1)
644def index(
645 data_dir: Path | None = data_dir_option,
646 use_global: bool = global_option,
647) -> None:
648 """Build the search indexes now (vector ANN + full-text).
650 Useful before publishing a large index so downloaders get fast search
651 without waiting for it to build on first query. Forces the vector index
652 even below the auto-build threshold.
653 """
654 apply_overrides(data_dir=data_dir, use_global=use_global)
655 store = get_services().store
656 store.ensure_fts_index()
657 store.ensure_scalar_indexes()
658 built = store.ensure_vector_index(force=True)
659 if cfg.json_mode:
660 json_output({"command": "index", "vector_index": built})
661 return
662 if built:
663 console.print("Search indexes built (vector ANN + full-text).")
664 else:
665 console.print("Full-text index built; vector index needs more chunks.")
668def _validate_file_paths(file_paths: list[Path]) -> None:
669 """Exit on the first missing path; respects ``cfg.json_mode``."""
670 for fp in file_paths:
671 if fp.exists():
672 continue
673 if cfg.json_mode:
674 json_output({"error": f"Path not found: {fp}"})
675 raise SystemExit(1)
676 print_prefixed(console, "Error: ", f"Path not found: {fp}", style=theme.ERROR)
677 raise SystemExit(1)
680def _crawl_urls_step(
681 urls: list[str],
682 *,
683 crawl: bool,
684 depth: int | None,
685 max_pages: int | None,
686 include_subdomains: bool,
687 cancel_event: threading.Event,
688 typed_login: CrawlLogin,
689) -> _CrawlOutcome:
690 """Crawl URLs; without the crawler extra every URL fails with the install message."""
691 if not urls:
692 return _CrawlOutcome()
693 from lilbee.crawler import CRAWLER_MISSING_MESSAGE, crawler_available
695 if not crawler_available():
696 return _CrawlOutcome(
697 failed=[_FailedSource(without_userinfo(url), CRAWLER_MISSING_MESSAGE) for url in urls]
698 )
699 try:
700 login = crawl_login(urls, typed_login)
701 except ValueError as exc:
702 raise typer.BadParameter(str(exc)) from None
703 outcome = _crawl_urls_blocking(
704 urls,
705 crawl=crawl,
706 depth=depth,
707 max_pages=max_pages,
708 cancel_event=cancel_event,
709 include_subdomains=include_subdomains,
710 login=login,
711 )
712 if not cfg.json_mode:
713 console.print(
714 f"Crawled {len(outcome.saved)} page(s) from {len(urls)} URL(s)",
715 style=theme.MUTED,
716 )
717 return outcome
720def _failed_reasons(failed: list[_FailedSource]) -> str:
721 """The distinct failure reasons of *failed*, one per line."""
722 return "\n".join(dict.fromkeys(source.reason for source in failed))
725def _failed_json(failed: list[_FailedSource]) -> list[dict[str, str]]:
726 return [{"url": s.url, "reason": s.reason} for s in failed]
729def _report_failed_sources(payload: dict | None, failed: list[_FailedSource]) -> None:
730 """Print the add result with each failed source's reason; exit 1 when a source failed."""
731 if payload is not None:
732 if failed:
733 payload["error"] = _failed_reasons(failed)
734 payload["failed"] = _failed_json(failed)
735 json_output(payload)
736 elif failed:
737 console.print(_failed_reasons(failed), style=theme.ERROR, soft_wrap=True)
738 if failed:
739 raise SystemExit(1)
742def _report_add_error(exc: RuntimeError, failed: list[_FailedSource]) -> NoReturn:
743 """Report the failed sources, then the error that stopped the add; exit 1."""
744 if cfg.json_mode:
745 payload: dict = {"error": str(exc)}
746 if failed:
747 payload["failed"] = _failed_json(failed)
748 json_output(payload)
749 raise SystemExit(1) from None
750 if failed:
751 console.print(_failed_reasons(failed), style=theme.ERROR, soft_wrap=True)
752 print_prefixed(console, "Error: ", exc, style=theme.ERROR)
753 raise SystemExit(1) from None
756def _add_json_mode(
757 file_paths: list[Path],
758 crawled_paths: list[Path],
759 *,
760 force: bool,
761 run_sync: Callable[[RegisterResult], object],
762) -> dict:
763 """Run the JSON-mode finish: register roots, sync, return the one structured result."""
764 reg_result = RegisterResult()
765 if file_paths:
766 reg_result = register_sources(file_paths, force=force)
767 # A sync is a whole-vault pass; run it only when something named reached the corpus.
768 result = run_sync(reg_result) if reg_result.reached_corpus or crawled_paths else None
769 return {
770 "command": "add",
771 "copied": reg_result.registered,
772 "name_taken": reg_result.name_taken,
773 "overlapping": reg_result.overlapping,
774 "absorbed": reg_result.absorbed,
775 "tracked": reg_result.tracked,
776 "refused": reg_result.refused,
777 "crawled": len(crawled_paths),
778 "sync": None if result is None else sync_result_to_json(result),
779 }
782def _register_and_sync(
783 file_paths: list[Path],
784 crawled_paths: list[Path],
785 *,
786 force: bool,
787 cancel_event: threading.Event,
788 rollback: AddRollback,
789) -> dict | None:
790 """Register the files and sync; returns the JSON result, or None after human output."""
792 synced = False
794 def _sync(registration: RegisterResult) -> object:
795 nonlocal synced
796 rollback.absorbed_into = registration.absorbed_into
797 result = _run_sync(
798 cancel_event,
799 before_sync=lambda: rollback.registered(registration.revocable, cancel_event),
800 started_by="lilbee add",
801 )
802 synced = _changed_content(result)
803 return result
805 if cfg.json_mode:
806 return _add_json_mode(file_paths, crawled_paths, force=force, run_sync=_sync)
807 if file_paths:
808 # Crawled pages are new content even when no file reached the corpus.
809 add_paths(
810 file_paths,
811 console,
812 force=force,
813 run_sync=_sync,
814 sync_anyway=bool(crawled_paths),
815 )
816 elif crawled_paths:
817 console.print(_sync(RegisterResult()))
818 # After the result: a refused, cancelled or failed sync raises before this line.
819 if synced:
820 print_tip_if_shown(cfg.data_root)
821 return None
824def add(
825 paths: list[str] = _paths_argument,
826 data_dir: Path | None = data_dir_option,
827 use_global: bool = global_option,
828 force: bool = _force_option,
829 ocr: OcrMode | None = _ocr_option,
830 ocr_timeout: float | None = _ocr_timeout_option,
831 crawl: bool = _crawl_option,
832 depth: int | None = _depth_option,
833 max_pages: int | None = _max_pages_option,
834 include_subdomains: bool = _include_subdomains_option,
835 authorization: str = _authorization_option,
836 username: str = _username_option,
837 password: str = _password_option,
838 cookie: str = _cookie_option,
839 max_cpus: int | None = _max_cpus_option,
840 processes: int | None = _processes_option,
841) -> None:
842 """Link files or crawl URLs into the knowledge base and ingest them."""
843 apply_overrides(data_dir=data_dir, use_global=use_global)
844 # Registration reads the store before the sync starts. The first services
845 # init must not warm a chat model that an add never uses.
846 cfg.worker_pool_eager_start = False
847 _apply_ocr_overrides(ocr, ocr_timeout)
848 try:
849 typed_login = CrawlLogin(authorization, username, password, cookie)
850 except ValueError as exc:
851 raise typer.BadParameter(str(exc)) from None
852 if max_cpus is not None:
853 cfg.ingest_workers = max_cpus
854 if processes is not None:
855 cfg.ingest_processes = processes
857 file_paths, urls = _partition_inputs(paths)
858 _validate_file_paths(file_paths)
860 cancel_event = threading.Event()
861 rollback = AddRollback(paths=file_paths, at_sync=False)
862 outcome = _CrawlOutcome()
863 try:
864 with _ctrl_c_stops(cancel_event, rollback):
865 outcome = _crawl_urls_step(
866 urls,
867 crawl=crawl,
868 depth=depth,
869 max_pages=max_pages,
870 include_subdomains=include_subdomains,
871 cancel_event=cancel_event,
872 typed_login=typed_login,
873 )
874 payload = _register_and_sync(
875 file_paths,
876 outcome.saved,
877 force=force,
878 cancel_event=cancel_event,
879 rollback=rollback,
880 )
881 # An add that ends before its sync under a Ctrl+C still stops; a sync
882 # that returned has finished, and its result stands.
883 if cancel_event.is_set() and not rollback.at_sync:
884 raise asyncio.CancelledError
885 except RuntimeError as exc:
886 _report_add_error(exc, outcome.failed)
887 _report_failed_sources(payload, outcome.failed)
890_chunks_source_argument = typer.Argument(..., help="Source name to inspect chunks for.")
893def chunks(
894 source: str = _chunks_source_argument,
895 data_dir: Path | None = data_dir_option,
896 use_global: bool = global_option,
897) -> None:
898 """Show chunks a document was split into (useful for debugging retrieval)."""
899 apply_overrides(data_dir=data_dir, use_global=use_global)
901 store = get_services().store
902 known = {s["filename"] for s in store.get_sources()}
903 if source not in known:
904 if cfg.json_mode:
905 json_output({"error": f"Source not found: {source}"})
906 raise SystemExit(1)
907 print_prefixed(console, "Source not found: ", source, style=theme.ERROR)
908 raise SystemExit(1)
910 raw_chunks = store.get_chunks_by_source(source)
911 cleaned = sorted(
912 clean_results(raw_chunks),
913 key=lambda c: c.get("chunk_index", 0),
914 )
916 if cfg.json_mode:
917 json_output({"command": "chunks", "source": source, "chunks": cleaned})
918 return
920 console.print(
921 Text.assemble(
922 (str(len(cleaned)), theme.LABEL),
923 " chunks from ",
924 (strip_control_characters(source), theme.ACCENT),
925 "\n",
926 ),
927 soft_wrap=True,
928 )
929 for c in cleaned:
930 idx = c.get("chunk_index", "?")
931 chunk_text = strip_control_characters(c.get("chunk", ""), keep_lines=True)
932 preview = chunk_text[:CHUNK_PREVIEW_LEN]
933 if len(chunk_text) > CHUNK_PREVIEW_LEN:
934 preview += "..."
935 console.print(Text.assemble(f" [{idx}] ", preview), soft_wrap=True)
938_remove_names_argument = typer.Argument(
939 ..., help="Source name(s), folder(s), or glob pattern(s) to remove from the knowledge base."
940)
942_remove_yes_option = typer.Option(
943 False, "--yes", "-y", help="Skip the confirmation prompt when a name expands to many documents."
944)
947def _remove_or_exit(names: list[str], targets: list[str]) -> RemoveResult:
948 """Remove *targets*; a library that refuses the removal exits 1, nothing removed."""
949 try:
950 return remove_documents_durably(names, targets=targets)
951 except LIBRARY_REFUSALS as exc:
952 if cfg.json_mode:
953 json_output({"error": str(exc)})
954 else:
955 print_prefixed(console, "Error: ", exc, style=theme.ERROR)
956 raise SystemExit(1) from None
959def remove(
960 names: list[str] = _remove_names_argument,
961 data_dir: Path | None = data_dir_option,
962 use_global: bool = global_option,
963 yes: bool = _remove_yes_option,
964) -> None:
965 """Remove documents from the knowledge base by source name, folder, or glob pattern.
967 A folder name removes every document indexed beneath it; a glob pattern
968 (containing ``*``, ``?``, or ``[]``) removes every source it matches. A file
969 held out because its ingestion failed can be named too; it stays out of later
970 syncs. Source files on disk are never deleted.
971 """
972 apply_overrides(data_dir=data_dir, use_global=use_global)
973 # Remove only touches the store, never the engine; skip the eager fleet warm.
974 cfg.worker_pool_eager_start = False
976 known = removable_names()
977 targets = expand_remove_targets(names, known=known)
978 expanded = sorted(set(targets)) != sorted(set(names))
979 if expanded and not yes and not cfg.json_mode:
980 # Count only what actually exists; not-found names are kept in targets.
981 removable = sum(1 for t in targets if t in set(known))
982 typer.confirm(f"Remove {removable} document(s)? Source files on disk are kept.", abort=True)
984 result = _remove_or_exit(names, targets)
986 if cfg.json_mode:
987 payload: dict = {"command": "remove", "removed": result.removed}
988 if result.not_found:
989 payload["not_found"] = result.not_found
990 json_output(payload)
991 if not result.removed and result.not_found:
992 raise SystemExit(1)
993 return
995 for name in result.removed:
996 console.print(
997 Text.assemble("Removed ", (strip_control_characters(name), theme.ACCENT)),
998 soft_wrap=True,
999 )
1000 for name in result.not_found:
1001 print_prefixed(console, "Not found: ", name, style=theme.ERROR)
1002 if not result.removed and result.not_found:
1003 raise SystemExit(1)