import os from pathlib import Path from typing import Optional, List, Dict, Any from slack_sdk import WebClient from health.utils.logging_config import setup_logger from slack_bot.obsidian.indexer import ObsidianIndexer from slack_bot.obsidian.generators import WritingAssistant, ReplyGenerator, DecisionSupport, SearchAnalyzer, DeAIReviser, ZhihuGenerator from slack_bot.obsidian.info_manager import InfoManager from slack_bot.context.storage import ContextStorage as BaseStorage logger = setup_logger(__name__) class ObsidianContextStorage(BaseStorage): """Specific storage for Obsidian bot to avoid mixing with Health bot.""" def __init__(self, channel_id: str): super().__init__(channel_id) from health import config self.context_dir = config.DATA_DIR / "obsidian_context" self.context_dir.mkdir(parents=True, exist_ok=True) self.file_path = self.context_dir / f"{channel_id}.json" self.state_file = self.context_dir / f"{channel_id}_state.json" def save_state(self, state: Dict[str, Any]): import json with open(self.state_file, "w") as f: json.dump(state, f) def load_state(self) -> Dict[str, Any]: import json if not self.state_file.exists(): return {"mode": "none"} try: with open(self.state_file, "r") as f: return json.load(f) except: return {"mode": "none"} class ObsidianDispatcher: """Routes Obsidian Slack Bot messages.""" MODE_WRITE = "write" MODE_REPLY = "reply" MODE_DECIDE = "decide" MODE_SEARCH = "search" MODE_DEAI = "deai" MODE_ZHIHU = "zhihu" MODE_NOTE = "note" MODE_ZHIHU_HUNTER = "zhihu-hunter" MODE_INFO = "info" VALID_MODES = {MODE_WRITE, MODE_REPLY, MODE_DECIDE, MODE_SEARCH, MODE_DEAI, MODE_ZHIHU, MODE_NOTE, MODE_ZHIHU_HUNTER, MODE_INFO} def __init__(self, vault_path: str): self.indexer = ObsidianIndexer(vault_path) self.indexer.scan_vault() # Initialize RAG vector store and note ingester from slack_bot.obsidian.embeddings import get_embedding_provider from slack_bot.obsidian.vector_store import ChromaVectorStore from slack_bot.obsidian.note_ingester import NoteIngester self.vector_store = ChromaVectorStore(embedding_provider=get_embedding_provider()) self.note_ingester = NoteIngester( vault_path=Path(vault_path), vector_store=self.vector_store, ) self.generators = { self.MODE_WRITE: WritingAssistant(self.indexer, self.vector_store), self.MODE_REPLY: ReplyGenerator(self.indexer, self.vector_store), self.MODE_DECIDE: DecisionSupport(self.indexer), self.MODE_SEARCH: SearchAnalyzer(self.indexer), self.MODE_DEAI: DeAIReviser(self.indexer), self.MODE_ZHIHU: ZhihuGenerator(self.indexer, self.vector_store), } self.info_manager = InfoManager(vault_path=Path(vault_path)) token = os.environ.get("OBSIDIAN_SLACK_BOT_TOKEN") self.client = WebClient(token=token) # Initialize Zhihu Hunter components from slack_bot.zhihu.zhihu_hunter import ZhihuHunter from slack_bot.zhihu.zhihu_playwright_engine import ZhihuPlaywrightEngine from slack_bot.zhihu.slack_interactive_gateway import SlackInteractiveGateway zhihu_notify_channel = os.environ.get("ZHIHU_NOTIFY_CHANNEL", "") self.zhihu_hunter = ZhihuHunter( vault_path=Path(vault_path), vector_store=self.vector_store, indexer=self.indexer, ) _zhihu_engine = ZhihuPlaywrightEngine( slack_client=self.client, notify_channel=zhihu_notify_channel, ) self.zhihu_gateway = SlackInteractiveGateway( slack_client=self.client, hunter=self.zhihu_hunter, engine=_zhihu_engine, notify_channel=zhihu_notify_channel, ) logger.info("ObsidianDispatcher initialized") def dispatch(self, text: str, channel_id: str, user_id: str, response_ts: Optional[str] = None): storage = ObsidianContextStorage(channel_id) state = storage.load_state() current_mode = state.get("mode", "none") cmd = text.strip().lower() # 1. Handle Commands if cmd.startswith("mode "): new_mode = cmd.split(" ", 1)[1].strip() if new_mode in self.VALID_MODES: storage.clear() storage.save_state({"mode": new_mode}) msg = f"✅ Switched to mode: *{new_mode}* (History cleared)" if new_mode == self.MODE_INFO: msg += ( "\n\n📌 *使用说明:*\n" "• `project <名称>` — 切换/创建项目\n" "• 直接发送文字 — 新增条目，末尾加 `(From 来源)` 标注出处\n" "• `update` — 查看最近条目，选序号追加更新\n" "• `ask <问题>` — 基于项目信息回答问题（附带日期和来源）\n" "• `summary` — 汇总项目进展\n" "• `summary + ` — 汇总 + 融合外部笔记\n" "• `search <关键词>` — 搜索当前项目\n" "• `list` — 查看所有项目" ) self._reply(channel_id, msg, response_ts) return else: self._reply(channel_id, f"❌ Invalid mode. Available: {', '.join(sorted(self.VALID_MODES))}", response_ts) return if cmd == "clear": storage.clear() self._reply(channel_id, "🧹 Context cleared.", response_ts) return if cmd == "reload": self.indexer.scan_vault() self._reply(channel_id, "🔄 Vault reloaded.", response_ts) return # 2. Handle Content based on mode if current_mode == "none": self._reply(channel_id, "⚠️ No mode selected. Use `mode write/reply/decide/search/deai/zhihu/note/info/zhihu-hunter` to start.", response_ts) return # MODE_NOTE: ingest text or URL into vault + vector store if current_mode == self.MODE_NOTE: try: result = self.note_ingester.ingest(text) self._reply(channel_id, result, response_ts) except Exception as e: logger.error(f"Note ingestion failed: {e}", exc_info=True) self._reply(channel_id, f"🐛 Ingestion error: {str(e)}", response_ts) return # MODE_INFO: project info fragment tracker if current_mode == self.MODE_INFO: try: self._handle_info_mode(text, channel_id, storage, state, response_ts) except Exception as e: logger.error(f"Info mode failed: {e}", exc_info=True) self._reply(channel_id, f"🐛 Error: {str(e)}", response_ts) return # MODE_ZHIHU_HUNTER: scan vault for keywords, surface Zhihu question candidates # # Command format: # 开始 → full scan, all dirs, auto keywords # article → scan Article only (Revised first, no time filter) # article agent开发 → scan Article, focus on "agent开发" # notes agent开发 → scan notes, focus on "agent开发" # 关键词1 关键词2 → skip scan, search directly with keywords # Article/8-xxx-Revised.md → single file: grounded in that article # Article/*openclaw*-Revised.md → glob: combine matching files # https://www.zhihu.com/question/... → direct URL: answer this specific question # # Dir shortcuts: notes / article / clippings / claw / deepfact / deep-dive if current_mode == self.MODE_ZHIHU_HUNTER: from slack_bot.zhihu.zhihu_hunter import DIR_SHORTCUTS try: trigger_words = {"开始", "hunt", "扫描", "start", "go"} words = [w for w in text.strip().split() if w.lower() not in trigger_words] if not words: # Bare trigger → full auto scan self._reply(channel_id, "🔍 正在扫描知乎候选问题（所有目录）...", response_ts) questions = self.zhihu_hunter.scan_and_hunt(since_days=7) elif words and 'zhihu.com/question/' in words[0]: # Direct URL paste — user was invited or found a question to answer. # Any text after the URL is treated as an optional answer outline. # Strip trailing Slack/HTML junk chars (>, ", etc.) from the URL. import re as _re url = _re.sub(r'[>\'")\].,;]+$', '', words[0]).lstrip("<") outline = " ".join(words[1:])[:500] if len(words) > 1 else "" hint = f"（已记录回答思路）" if outline else "" self._reply(channel_id, f"🔍 获取问题标题，准备生成草稿{hint}...", response_ts) questions = self.zhihu_hunter.hunt_direct_url(url) if outline and questions: questions[0].outline = outline logger.info(f"ZhihuHunter: direct URL {url!r} → {questions[0].title!r}" + (f" outline={outline[:40]!r}" if outline else "")) elif words and ('*' in words[0] or '?' in words[0]): # Glob pattern → combine all matched files for keyword extraction pattern = words[0] self._reply( channel_id, f"🔍 匹配 `{pattern}` 中的文件，提取观点搜索知乎问题...", response_ts, ) questions = self.zhihu_hunter.scan_glob_files(pattern) logger.info(f"ZhihuHunter: glob '{pattern}' → {len(questions)} questions") elif words and ('/' in words[0] or words[0].endswith('.md')): # Single-file mode: extract questions from one specific file, # and pass source_file so answers stay grounded in that article. rel_path = words[0] self._reply( channel_id, f"🔍 从 `{rel_path}` 提取观点，搜索匹配知乎问题...", response_ts, ) questions = self.zhihu_hunter.scan_single_file(rel_path) logger.info(f"ZhihuHunter: single-file scan '{rel_path}' → {len(questions)} questions") elif words[0].lower() in DIR_SHORTCUTS: # Dir-scoped scan with optional topic hint rel_dir = DIR_SHORTCUTS[words[0].lower()] topic_hint = " ".join(words[1:]) hint_desc = f"，聚焦：{topic_hint}" if topic_hint else "" self._reply( channel_id, f"🔍 扫描 {rel_dir} 目录{hint_desc}，请稍候...", response_ts, ) questions = self.zhihu_hunter.scan_and_hunt( since_days=7, dirs=[rel_dir], topic_hint=topic_hint ) else: # Direct keywords → skip vault scan entirely self._reply(channel_id, f"🔍 使用指定关键词搜索：{' · '.join(words)}", response_ts) questions = self.zhihu_hunter._find_questions(words) logger.info(f"ZhihuHunter: manual keywords {words} → {len(questions)} questions") self.zhihu_gateway.post_question_cards(questions, channel=channel_id) except Exception as e: logger.error(f"Zhihu hunt failed: {e}", exc_info=True) self._reply(channel_id, f"❌ 扫描失败：{e}", response_ts) return # Get history history = storage.get_context() generator = self.generators[current_mode] try: logger.info(f"Generating Obsidian response in mode: {current_mode}") # The chat method handles first turn logic and returns updated history # But wait, our generators' chat method appends to history internally. # In our Slack setup, we want to append the result to ContextStorage. # Note: generator.chat(text, history) returns (response, new_history_list) # new_history_list contains [...history, {user: rich_prompt}, {assistant: response}] response_text, updated_history = generator.chat(text, history) # Since generator.chat already did the LLM call and returned history, # we just need to save the ONLY NEW messages to our storage. # However, during FIRST TURN, text is replaced by rich_prompt. # For simplicity: completely overwrite storage with updated_history # ContextStorage doesn't support overwrite easily, but we can clear and re-add. # Better: Modify ObsidianContextStorage to support set_context. # Actually, let's just use the updated_history and manually add to storage # Warning: first turn 'user' message in storage will be the RICH PROMPT # This is actually better for context continuity in Slack too. storage.clear() for msg in updated_history: storage.add_message(msg["role"], msg["content"]) self._reply(channel_id, response_text, response_ts) except Exception as e: logger.error(f"Obsidian generation failed: {e}", exc_info=True) self._reply(channel_id, f"🐛 Error: {str(e)}", response_ts) def _handle_info_mode( self, text: str, channel_id: str, storage: ObsidianContextStorage, state: dict, response_ts: Optional[str], ) -> None: """Route commands within MODE_INFO.""" cmd = text.strip() cmd_lower = cmd.lower() active_project = state.get("active_project", "") pending_idx = state.get("pending_update_entry_index") # --- Command: project --- if cmd_lower.startswith("project "): project_name = cmd[8:].strip() if not project_name or "/" in project_name or ".." in project_name: self._reply(channel_id, "⚠️ 无效项目名。", response_ts) return state["active_project"] = project_name state["pending_update_entry_index"] = None storage.save_state(state) self._reply(channel_id, f"📂 已切换到项目: *{project_name}*", response_ts) return # --- Command: list --- if cmd_lower == "list": result = self.info_manager.list_projects() self._reply(channel_id, result, response_ts) return # Guard: require active project for all other commands if not active_project: self._reply( channel_id, "⚠️ 请先选择项目。用 `project <名称>` 切换或创建项目。\n用 `list` 查看已有项目。", response_ts, ) return # --- Command: update --- if cmd_lower == "update": msg, entries = self.info_manager.list_recent(active_project) if entries: # Store entries in state for later reference state["pending_update_entries"] = [e.model_dump() for e in entries] state["pending_update_entry_index"] = -1 # waiting for index storage.save_state(state) self._reply(channel_id, msg, response_ts) return # --- State: waiting for update index selection --- if pending_idx == -1: try: idx = int(cmd.strip()) - 1 # user sees 1-based entries_data = state.get("pending_update_entries", []) if 0 <= idx < len(entries_data): state["pending_update_entry_index"] = idx storage.save_state(state) self._reply( channel_id, f"✏️ 已选择第 {idx + 1} 条。请发送更新内容：", response_ts, ) return else: state["pending_update_entry_index"] = None storage.save_state(state) self._reply(channel_id, "⚠️ 无效序号，已退出更新模式。", response_ts) return except ValueError: # Not a number — treat as cancel, fall through to add_entry state["pending_update_entry_index"] = None state.pop("pending_update_entries", None) storage.save_state(state) # --- State: waiting for update content --- if isinstance(pending_idx, int) and pending_idx >= 0: from slack_bot.obsidian.info_manager import InfoEntry entries_data = state.get("pending_update_entries", []) entries = [InfoEntry(**d) for d in entries_data] result = self.info_manager.update_entry( active_project, entries, pending_idx, cmd ) state["pending_update_entry_index"] = None state.pop("pending_update_entries", None) storage.save_state(state) self._reply(channel_id, result, response_ts) return # --- Command: summary --- if cmd_lower.startswith("summary"): extra_paths = None if "+" in cmd: _, paths_str = cmd.split("+", 1) extra_paths = [p.strip() for p in paths_str.split(",") if p.strip()] result = self.info_manager.summarize(active_project, extra_paths) self._reply(channel_id, result, response_ts) return # --- Command: ask --- if cmd_lower.startswith("ask "): question = cmd[4:].strip() if question: extra_paths = None if "+" in question: question, paths_str = question.split("+", 1) extra_paths = [p.strip() for p in paths_str.split(",") if p.strip()] question = question.strip() result = self.info_manager.ask(active_project, question, extra_paths) self._reply(channel_id, result, response_ts) return # --- Command: search --- if cmd_lower.startswith("search "): keyword = cmd[7:].strip() if keyword: result = self.info_manager.search_entries(active_project, keyword) self._reply(channel_id, result, response_ts) return # --- Default: add new entry --- result = self.info_manager.add_entry(active_project, cmd) self._reply(channel_id, result, response_ts) def _reply(self, channel_id: str, text: str, ts: Optional[str] = None): from slack_bot.utils.mrkdwn import SlackFormatter # Slack API limits (official documentation): # - chat.postMessage: 40,000 chars # - chat.update: 4,000 chars # Slack markdown formatting can expand text by 50-100% (bold, links, code blocks, etc.) TRUNCATE_WARNING = "\n\n⚠️ (Response truncated, full text uploaded as file)" FILE_UPLOAD_HINT = "📄 Full response uploaded as file above ☝️" # Calculate safe pre-formatting limit with aggressive overhead buffer if ts: # For updates: 4000 limit - 60% overhead for safety = ~1600 chars max_raw_length = 1500 else: # For new messages: 40000 limit - 40% overhead = ~24000 chars max_raw_length = 24000 original_length = len(text) should_upload_file = original_length > max_raw_length if should_upload_file: # Upload full text as file first self._upload_as_file(channel_id, text, ts) # Then send a truncated preview truncate_at = text.rfind('\n\n', 0, 800) if truncate_at == -1 or truncate_at < 400: truncate_at = text.rfind('\n', 0, 800) if truncate_at == -1 or truncate_at < 400: truncate_at = 800 preview_text = text[:truncate_at] + f"\n\n...\n\n{FILE_UPLOAD_HINT}" formatted_text = SlackFormatter.convert(preview_text) logger.info(f"Uploaded full response ({original_length} chars) as file, sending preview ({len(preview_text)} chars)") else: # Text is short enough, just format and send formatted_text = SlackFormatter.convert(text) # Final safety check after formatting final_limit = 3900 if ts else 39900 if len(formatted_text) > final_limit: logger.warning(f"Formatted text ({len(formatted_text)} chars) exceeds limit after formatting. Uploading as file.") self._upload_as_file(channel_id, text, ts) # Send simple notification formatted_text = FILE_UPLOAD_HINT # Send the message try: if ts: self.client.chat_update(channel=channel_id, ts=ts, text=formatted_text) else: self.client.chat_postMessage(channel=channel_id, text=formatted_text) except Exception as e: if "msg_too_long" in str(e): logger.error(f"Still got msg_too_long error even after truncation. Text length: {len(formatted_text)}") # Ultimate fallback: send error message only try: error_msg = "❌ Response formatting error. Please check the uploaded file above." if ts: self.client.chat_update(channel=channel_id, ts=ts, text=error_msg) else: self.client.chat_postMessage(channel=channel_id, text=error_msg) except: logger.error("Failed to send even error message", exc_info=True) else: raise def _upload_as_file(self, channel_id: str, content: str, ts: Optional[str] = None): """Fallback: upload long responses as text files.""" try: import tempfile from datetime import datetime filename = f"response_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt" with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False, encoding='utf-8') as tmp: tmp.write(content) tmp_path = tmp.name self.client.files_upload_v2( channel=channel_id, file=tmp_path, filename=filename, initial_comment="📄 Response too long for Slack message, uploaded as file:", thread_ts=ts ) os.unlink(tmp_path) logger.info(f"Uploaded long response as file: {filename}") except Exception as e: logger.error(f"Failed to upload file: {e}", exc_info=True) # Ultimate fallback: send error message error_msg = "❌ Response too long to display. Please try a more specific query." if ts: self.client.chat_update(channel=channel_id, ts=ts, text=error_msg) else: self.client.chat_postMessage(channel=channel_id, text=error_msg)