Coverage for src / lilbee / cli / helpers.py: 100%

1"""CLI-specific helpers: JSON formatter, Rich rendering, and CLI workflows.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6import json 

7from collections.abc import Generator 

8from pathlib import Path 

9from typing import TYPE_CHECKING 

10 

11from rich.console import Console, RenderableType 

12from rich.table import Table 

13 

14from lilbee.app.ingest import copy_files 

15from lilbee.app.status import StatusResult 

16from lilbee.cli import theme 

17from lilbee.core.config import cfg 

18 

19if TYPE_CHECKING: 

20 from lilbee.cli.sync import SyncStatus 

21 

22 

23def json_output(data: dict) -> None: 

24 """Print a JSON object to stdout.""" 

25 print(json.dumps(data)) 

26 

27 

28def render_status_result(status: StatusResult) -> Generator[RenderableType, None, None]: 

29 """Yield Rich renderables for a :class:`StatusResult`.""" 

30 yield f"[{theme.LABEL}]Documents:[/{theme.LABEL}] {status.config.documents_dir}" 

31 yield f"[{theme.LABEL}]Database:[/{theme.LABEL}] {status.config.data_dir}" 

32 yield f"[{theme.LABEL}]Chat model:[/{theme.LABEL}] {status.config.chat_model}" 

33 yield f"[{theme.LABEL}]Embeddings:[/{theme.LABEL}] {status.config.embedding_model}" 

34 vision = status.config.vision_model or "(disabled)" 

35 reranker = status.config.reranker_model or "(disabled)" 

36 yield f"[{theme.LABEL}]Vision:[/{theme.LABEL}] {vision}" 

37 yield f"[{theme.LABEL}]Reranker:[/{theme.LABEL}] {reranker}" 

38 if status.config.enable_ocr is not None: 

39 ocr_label = "enabled" if status.config.enable_ocr else "disabled" 

40 yield f"[{theme.LABEL}]Vision OCR:[/{theme.LABEL}] {ocr_label}" 

41 yield "" 

42 

43 if not status.sources: 

44 yield ( 

45 "No documents indexed. Drop files into the documents directory and run 'lilbee sync'." 

46 ) 

47 return 

48 

49 table = Table(title="Indexed Documents") 

50 table.add_column("File", style=theme.ACCENT) 

51 table.add_column("Hash", style=theme.MUTED, max_width=12) 

52 table.add_column("Chunks", justify="right") 

53 table.add_column("Ingested", style=theme.MUTED) 

54 for s in status.sources: 

55 table.add_row(s.filename, s.file_hash, str(s.chunk_count), s.ingested_at) 

56 yield table 

57 b = theme.LABEL 

58 yield f"\n[{b}]{len(status.sources)}[/{b}] documents, [{b}]{status.total_chunks}[/{b}] chunks" 

59 

60 

61def render_status(con: Console) -> None: 

62 """Print status info (documents, paths, chunk counts).""" 

63 from lilbee.app.status import gather_status 

64 

65 for renderable in render_status_result(gather_status()): 

66 con.print(renderable) 

67 

68 

69def copy_paths(paths: list[Path], con: Console, *, force: bool = False) -> list[str]: 

70 """Copy *paths* into the documents directory. Returns list of copied names.""" 

71 result = copy_files(paths, force=force) 

72 for name in result.skipped: 

73 con.print( 

74 f"[{theme.WARNING}]Warning:[/{theme.WARNING}] {name} already exists in knowledge base " 

75 f"(use --force to overwrite)" 

76 ) 

77 return result.copied 

78 

79 

80def add_paths( 

81 paths: list[Path], 

82 con: Console, 

83 *, 

84 force: bool = False, 

85 background: bool = False, 

86 chat_mode: bool = False, 

87 sync_status: SyncStatus | None = None, 

88) -> None: 

89 """Copy *paths* into the knowledge base and sync (human output). 

90 When *background* is True (chat ``/add``), sync runs in a background thread 

91 and this function returns immediately after copying files. 

92 """ 

93 from lilbee.data.ingest import sync 

94 

95 copied = copy_paths(paths, con, force=force) 

96 if chat_mode: 

97 print(f"Copied {len(copied)} path(s) to {cfg.documents_dir}") 

98 else: 

99 con.print( 

100 f"[{theme.MUTED}]Copied {len(copied)} path(s) to {cfg.documents_dir}[/{theme.MUTED}]" 

101 ) 

102 

103 if background: 

104 from lilbee.cli.sync import run_sync_background 

105 

106 run_sync_background(con, chat_mode=chat_mode, sync_status=sync_status) 

107 return 

108 

109 result = asyncio.run(sync()) 

110 con.print(result) 

111 

112 

113def sync_result_to_json(result: object) -> dict: 

114 """Convert a SyncResult to the JSON output envelope.""" 

115 from lilbee.data.ingest import SyncResult 

116 

117 if not isinstance(result, SyncResult): 

118 raise TypeError(f"Expected SyncResult, got {type(result).__name__}") 

119 return {"command": "sync", **result.model_dump()} 

120 

121 

122def auto_sync(con: Console, *, background: bool = False) -> None: 

123 """Run document sync before queries. 

124 When *background* is True, sync runs in a background thread and this 

125 function returns immediately (for chat/REPL). When False (default), 

126 sync blocks until complete (for ``lilbee ask``). 

127 """ 

128 if background: 

129 from lilbee.cli.sync import run_sync_background 

130 

131 run_sync_background(con) 

132 return 

133 

134 from lilbee.data.ingest import sync 

135 

136 try: 

137 result = asyncio.run(sync()) 

138 except RuntimeError as exc: 

139 con.print(f"[{theme.ERROR}]Error:[/{theme.ERROR}] {exc}") 

140 raise SystemExit(1) from None 

141 total = ( 

142 len(result.added) 

143 + len(result.updated) 

144 + len(result.removed) 

145 + len(result.failed) 

146 + len(result.skipped) 

147 ) 

148 if total: 

149 con.print( 

150 f"[{theme.MUTED}]Synced: {len(result.added)} added, " 

151 f"{len(result.updated)} updated, " 

152 f"{len(result.removed)} removed, " 

153 f"{len(result.skipped)} skipped, " 

154 f"{len(result.failed)} failed[/{theme.MUTED}]" 

155 )