Coverage for slack_bot / obsidian / cli.py: 0%

1import cmd 

2import os 

3import sys 

4from slack_bot.obsidian.indexer import ObsidianIndexer 

5from slack_bot.obsidian.indexer import ObsidianIndexer 

6from slack_bot.obsidian.generators import WritingAssistant, ReplyGenerator, DecisionSupport, SearchAnalyzer 

7from health.utils.logging_config import setup_logger 

8 

9logger = setup_logger(__name__) 

10 

11class ObsidianShell(cmd.Cmd): 

12 intro = 'Welcome to the Obsidian Assistant Shell. Type help or ? to list commands.\n' 

13 prompt = '(obsidian) ' 

14 

15 # Modes 

16 MODE_NONE = "none" 

17 MODE_WRITE = "write" 

18 MODE_REPLY = "reply" 

19 MODE_DECIDE = "decide" 

20 MODE_SEARCH = "search" 

21 

22 def __init__(self, vault_path: str): 

23 super().__init__() 

24 self.vault_path = vault_path 

25 self.indexer = ObsidianIndexer(vault_path) 

26 self.indexer.scan_vault() 

27 

28 self.generators = { 

29 self.MODE_WRITE: WritingAssistant(self.indexer), 

30 self.MODE_REPLY: ReplyGenerator(self.indexer), 

31 self.MODE_DECIDE: DecisionSupport(self.indexer), 

32 self.MODE_SEARCH: SearchAnalyzer(self.indexer) 

33 } 

34 

35 self.current_mode = self.MODE_NONE 

36 self.history = [] # Stores chat history for current session 

37 self._set_prompt() 

38 

39 def _set_prompt(self): 

40 if self.current_mode == self.MODE_NONE: 

41 self.prompt = '(obsidian) ' 

42 else: 

43 turn_indicator = f" [{len(self.history)//2} turns]" if self.history else "" 

44 self.prompt = f'(obsidian:{self.current_mode}{turn_indicator}) ' 

45 

46 def do_mode(self, arg): 

47 """Switch mode: mode [write|reply|decide|search]""" 

48 if arg in [self.MODE_WRITE, self.MODE_REPLY, self.MODE_DECIDE, self.MODE_SEARCH]: 

49 self.current_mode = arg 

50 self.history = [] # Clear history on mode switch 

51 self._set_prompt() 

52 print(f"Switched to mode: {arg} (History cleared)") 

53 else: 

54 print(f"Invalid mode. Available modes: {self.MODE_WRITE}, {self.MODE_REPLY}, {self.MODE_DECIDE}, {self.MODE_SEARCH}") 

55 self.current_mode = self.MODE_NONE 

56 self.history = [] 

57 self._set_prompt() 

58 

59 def do_reload(self, arg): 

60 """Reloads the vault index.""" 

61 print("Reloading index...") 

62 self.indexer.scan_vault() 

63 print("Index reloaded.") 

64 

65 def do_clear(self, arg): 

66 """Clears the current conversation history.""" 

67 self.history = [] 

68 self._set_prompt() 

69 print("Context cleared.") 

70 

71 def do_context(self, arg): 

72 """Show current context samples.""" 

73 if self.current_mode == self.MODE_NONE: 

74 print("No mode selected.") 

75 return 

76 

77 print(f"--- Context for {self.current_mode} ---") 

78 if self.current_mode == self.MODE_WRITE: 

79 samples = self.indexer.writing_samples 

80 print(f"Total Writing Samples (#writing_sample): {len(samples)}") 

81 for s in samples: 

82 print(f" - {os.path.basename(s)}") 

83 

84 elif self.current_mode == self.MODE_REPLY: 

85 samples = self.indexer.reply_samples 

86 print(f"Total Reply Samples (#reply_sample): {len(samples)}") 

87 for s in samples: 

88 print(f" - {os.path.basename(s)}") 

89 print(" + REPLY-SAMPLE.md (Local)") 

90 

91 elif self.current_mode == self.MODE_DECIDE: 

92 print("Referencing: decision.md, methodology.md") 

93 

94 elif self.current_mode == self.MODE_SEARCH: 

95 print("Hybrid Search: Obsidian (Local) + DuckDuckGo (Web)") 

96 

97 def default(self, line): 

98 """Handle input based on current mode.""" 

99 if self.current_mode == self.MODE_NONE: 

100 print("Please select a mode first using `mode [write|reply|decide]`") 

101 return 

102 

103 user_input = line.strip() 

104 if not user_input: 

105 return 

106 

107 print("\nThinking...\n") 

108 generator = self.generators[self.current_mode] 

109 

110 try: 

111 # All generators now use chat interface 

112 response, updated_history = generator.chat(user_input, self.history) 

113 

114 self.history = updated_history 

115 self._set_prompt() 

116 

117 print(f"\n{response}\n") 

118 

119 except Exception as e: 

120 logger.error(f"Generation failed: {e}") 

121 print(f"Error: {e}") 

122 

123 def do_quit(self, arg): 

124 """Quit the shell.""" 

125 return True 

126 

127 def do_exit(self, arg): 

128 """Exit the shell.""" 

129 return True 

130 

131 def emptyline(self): 

132 pass