Coverage for src/lilbee/app/reset.py: 100%

1"""Knowledge base reset (delete all documents and data).""" 

2 

3from __future__ import annotations 

4 

5import logging 

6import shutil 

7from pathlib import Path 

8 

9from pydantic import BaseModel 

10 

11from lilbee.core import settings 

12from lilbee.core.config import cfg 

13from lilbee.data.ingest.skip_marker import clear_skip_markers 

14 

15 

16class ResetResult(BaseModel): 

17 """Result of a full knowledge base reset.""" 

18 

19 command: str = "reset" 

20 deleted_docs: int 

21 deleted_data: int 

22 skipped: list[str] = [] 

23 documents_dir: str 

24 data_dir: str 

25 

26 

27def _clear_dir(base_dir: Path, skipped: list[str]) -> int: 

28 """Delete all items in *base_dir*, appending undeletable paths to *skipped*.""" 

29 log = logging.getLogger(__name__) 

30 deleted = 0 

31 if not base_dir.exists(): 

32 return deleted 

33 for item in list(base_dir.iterdir()): 

34 try: 

35 # iterdir yields direct children only, so the entry is within 

36 # base_dir by construction. Registered source roots live outside this 

37 # dir (only their config entry is here), so reset never reaches a 

38 # user's corpus: it deletes owned files, and perform_reset 

39 # un-registers the roots from config.toml separately. 

40 if item.is_dir(): 

41 shutil.rmtree(item) 

42 else: 

43 item.unlink() 

44 except OSError as exc: 

45 # best-effort: reset is "delete as much as you can", and the 

46 # caller surfaces the skipped list to the user verbatim. 

47 log.warning("Could not delete %s: %s", item, exc) 

48 skipped.append(str(item)) 

49 continue 

50 deleted += 1 

51 return deleted 

52 

53 

54def perform_reset() -> ResetResult: 

55 """Delete all documents and data and un-register every linked source.""" 

56 skipped: list[str] = [] 

57 deleted_docs = _clear_dir(cfg.documents_dir, skipped) 

58 deleted_data = _clear_dir(cfg.data_dir, skipped) 

59 

60 # config.toml and the skip sidecars live at data_root, next to (not inside) 

61 # the two cleared dirs. Without un-registering the roots here, in the file 

62 # AND in this process, the next sync walks every linked root and re-indexes 

63 # it, so content reappears after a "factory reset". Other settings survive: 

64 # reset deletes data, not configuration. 

65 settings.delete_value(cfg.data_root, "linked_roots") 

66 cfg.linked_roots = {} 

67 clear_skip_markers(cfg.data_root) 

68 

69 return ResetResult( 

70 deleted_docs=deleted_docs, 

71 deleted_data=deleted_data, 

72 skipped=skipped, 

73 documents_dir=str(cfg.documents_dir), 

74 data_dir=str(cfg.data_dir), 

75 )