From dfd8421dc6ed7a2f7f973a822a1d5b5a3ef770a4 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Fri, 26 Jun 2026 11:32:43 +0800 Subject: [PATCH] =?UTF-8?q?feat(=C2=A703):=20=E5=AE=9E=E7=8E=B0=20MCP=20Se?= =?UTF-8?q?rver=20=E6=A0=B8=E5=BF=83=20-=20Storage/Service/Tool/MCP=20Prot?= =?UTF-8?q?ocol=20Layers?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit **Storage Layer:** - Database 类（aiosqlite + WAL + 并发保护） - WikiPage 数据模型 - fix_dirty_states 恢复机制 - FTS5 全文搜索支持 - 完整的 SQLite 表结构 **Service Layer:** - CacheService（LRU 缓存 + TTL + 大小限制） - QueryService（查询服务） - ParserService（Markdown 解析） - IndexerService（索引服务） - GraphService（链接图服务） **Tool Layer (8 个 MCP Tools):** - wiki_query - FTS5 全文搜索 + 标签过滤 - memory_bridge - 按工具来源浏览 - wiki_status - 索引状态 - wiki_lint - 健康审计 - cross_linker - 缺失链接发现 - tag_taxonomy - 标签一致性 - wiki_synthesize - 跨概念综合分析 - daily_update - 日常维护 + hot.md 生成 **MCP Protocol Layer:** - MCP 协议解析和封装 - 工具注册和路由 - 错误处理和日志 - stdio 模式支持 **配置和部署:** - requirements.txt - config.example.yaml - ecosystem.config.cjs (PM2) Co-Authored-By: Claude Opus 4.8 (1M context) --- config.example.yaml | 22 ++ ecosystem.config.cjs | 18 + mcp_server/__init__.py | 7 + mcp_server/main.py | 218 ++++++++++++ mcp_server/services/__init__.py | 19 + mcp_server/services/cache.py | 91 +++++ mcp_server/services/graph.py | 137 +++++++ mcp_server/services/indexer.py | 226 ++++++++++++ mcp_server/services/parser.py | 239 +++++++++++++ mcp_server/services/query.py | 120 +++++++ mcp_server/storage/__init__.py | 14 + mcp_server/storage/database.py | 531 ++++++++++++++++++++++++++++ mcp_server/tools/__init__.py | 25 ++ mcp_server/tools/cross_linker.py | 87 +++++ mcp_server/tools/daily_update.py | 105 ++++++ mcp_server/tools/memory_bridge.py | 112 ++++++ mcp_server/tools/tag_taxonomy.py | 117 ++++++ mcp_server/tools/wiki_lint.py | 164 +++++++++ mcp_server/tools/wiki_query.py | 116 ++++++ mcp_server/tools/wiki_status.py | 75 ++++ mcp_server/tools/wiki_synthesize.py | 105 ++++++ requirements.txt | 13 + 22 files changed, 2561 insertions(+) create mode 100644 config.example.yaml create mode 100644 ecosystem.config.cjs create mode 100644 mcp_server/__init__.py create mode 100644 mcp_server/main.py create mode 100644 mcp_server/services/__init__.py create mode 100644 mcp_server/services/cache.py create mode 100644 mcp_server/services/graph.py create mode 100644 mcp_server/services/indexer.py create mode 100644 mcp_server/services/parser.py create mode 100644 mcp_server/services/query.py create mode 100644 mcp_server/storage/__init__.py create mode 100644 mcp_server/storage/database.py create mode 100644 mcp_server/tools/__init__.py create mode 100644 mcp_server/tools/cross_linker.py create mode 100644 mcp_server/tools/daily_update.py create mode 100644 mcp_server/tools/memory_bridge.py create mode 100644 mcp_server/tools/tag_taxonomy.py create mode 100644 mcp_server/tools/wiki_lint.py create mode 100644 mcp_server/tools/wiki_query.py create mode 100644 mcp_server/tools/wiki_status.py create mode 100644 mcp_server/tools/wiki_synthesize.py create mode 100644 requirements.txt diff --git a/config.example.yaml b/config.example.yaml new file mode 100644 index 0000000..eeed01c --- /dev/null +++ b/config.example.yaml @@ -0,0 +1,22 @@ +# Wiki MCP Server 配置文件示例 +# 保存到 ~/.sanguo-llmwiki/config.yaml + +wiki: + vault_path: "/Volumes/KnowledgeBase/wiki-vault" + index_path: "~/.sanguo-llmwiki/index.db" + max_page_size_kb: 500 + +mcp: + mode: "stdio" # stdio 或 SSE + host: "localhost" + port: 8080 + +performance: + query_timeout_ms: 5000 + cache_ttl_seconds: 3600 + fts_cache_size_mb: 100 + max_concurrent_indexing: 5 + +logging: + level: "INFO" + file: "~/.sanguo-llmwiki/wiki-mcp.log" diff --git a/ecosystem.config.cjs b/ecosystem.config.cjs new file mode 100644 index 0000000..aece6bf --- /dev/null +++ b/ecosystem.config.cjs @@ -0,0 +1,18 @@ +module.exports = { + apps: [{ + name: 'wiki-mcp', + script: 'python', + args: '-m mcp_server.main', + cwd: '/Users/chufeng/.openclaw/sanguo_projects/sanguo_llmwiki', + instances: 1, + autorestart: true, + watch: false, + max_memory_restart: '500M', + env: { + PYTHONUNBUFFERED: '1', + WIKI_VAULT_PATH: '/Volumes/KnowledgeBase/wiki-vault', + WIKI_INDEX_PATH: '~/.sanguo-llmwiki/index.db', + LOG_LEVEL: 'INFO' + } + }] +}; diff --git a/mcp_server/__init__.py b/mcp_server/__init__.py new file mode 100644 index 0000000..d31bd4e --- /dev/null +++ b/mcp_server/__init__.py @@ -0,0 +1,7 @@ +""" +MCP Server - Wiki MCP Server + +混合架构 Wiki 系统的 MCP Server 实现。 +""" + +__version__ = "1.0.0" diff --git a/mcp_server/main.py b/mcp_server/main.py new file mode 100644 index 0000000..d1bc380 --- /dev/null +++ b/mcp_server/main.py @@ -0,0 +1,218 @@ +""" +MCP Server - Wiki MCP Server 主入口 + +实现 MCP 协议解析、工具注册和路由。 + +参考设计文档：第 2.1 节 +""" + +import asyncio +import json +import logging +import sys +from typing import Any, Callable, Dict, Optional +from pathlib import Path + +# MCP SDK +try: + from mcp.server.models import InitializationOptions + from mcp.server import Server, NotificationOptions + from mcp.types import Tool, TextContent + MCP_AVAILABLE = True +except ImportError: + MCP_AVAILABLE = False + logger = logging.getLogger(__name__) + logger.warning("MCP SDK not available, using mock implementation") + +# 本地模块 +from .storage import Database, fix_dirty_states +from .services import CacheService, QueryService, ParserService, IndexerService, GraphService +from .tools import ( + WikiQueryTool, MemoryBridgeTool, WikiStatusTool, + WikiLintTool, CrossLinkerTool, TagTaxonomyTool, + WikiSynthesizeTool, DailyUpdateTool +) + +# 配置 +DEFAULT_WIKI_VAULT = "/Volumes/KnowledgeBase/wiki-vault" +DEFAULT_INDEX_PATH = "~/.sanguo-llmwiki/index.db" +DEFAULT_LOG_LEVEL = "INFO" + +# 配置日志 +logging.basicConfig( + level=getattr(logging, DEFAULT_LOG_LEVEL), + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' +) +logger = logging.getLogger(__name__) + + +class MCPServer: + """MCP 服务器封装类""" + + def __init__(self, wiki_vault_path: str, index_path: str): + self.wiki_vault_path = wiki_vault_path + self.index_path = index_path + self.db: Optional[Database] = None + self.tools: Dict[str, Callable] = {} + self.tool_schemas: Dict[str, dict] = {} + + # 如果 MCP SDK 可用，创建 Server 实例 + if MCP_AVAILABLE: + self.server = Server("wiki-mcp") + self._setup_handlers() + else: + self.server = None + logger.warning("Running in mock mode (MCP SDK not available)") + + async def initialize(self) -> None: + """初始化数据库和服务""" + logger.info("Initializing Wiki MCP Server...") + + # 初始化数据库 + self.db = Database(self.index_path) + await self.db.connect() + + # 恢复脏状态 + await fix_dirty_states(self.db) + + # 初始化服务 + self.cache = CacheService(max_size=1000) + self.query_service = QueryService(self.db, self.cache) + self.parser_service = ParserService() + self.indexer_service = IndexerService(self.db, self.parser_service, self.wiki_vault_path) + self.graph_service = GraphService(self.db) + + # 初始化工具 + self._init_tools() + + logger.info("Wiki MCP Server initialized") + + def _init_tools(self) -> None: + """初始化所有工具""" + wiki_query = WikiQueryTool(self.query_service) + memory_bridge = MemoryBridgeTool(self.query_service) + wiki_status = WikiStatusTool(self.query_service, self.indexer_service, self.graph_service) + wiki_lint = WikiLintTool(self.parser_service, self.query_service, self.graph_service) + cross_linker = CrossLinkerTool(self.graph_service) + tag_taxonomy = TagTaxonomyTool(self.query_service) + wiki_synthesize = WikiSynthesizeTool(self.query_service, self.graph_service) + daily_update = DailyUpdateTool(self.query_service, self.indexer_service) + + # 注册工具 + self.tools = { + "wiki_query": wiki_query.handle, + "memory_bridge": memory_bridge.handle, + "wiki_status": wiki_status.handle, + "wiki_lint": wiki_lint.handle, + "cross_linker": cross_linker.handle, + "tag_taxonomy": tag_taxonomy.handle, + "wiki_synthesize": wiki_synthesize.handle, + "daily_update": daily_update.handle + } + + # 注册工具 schema + self.tool_schemas = { + "wiki_query": wiki_query.get_schema(), + "memory_bridge": memory_bridge.get_schema(), + "wiki_status": wiki_status.get_schema(), + "wiki_lint": wiki_lint.get_schema(), + "cross_linker": cross_linker.get_schema(), + "tag_taxonomy": tag_taxonomy.get_schema(), + "wiki_synthesize": wiki_synthesize.get_schema(), + "daily_update": daily_update.get_schema() + } + + logger.info(f"Registered {len(self.tools)} tools") + + def _setup_handlers(self) -> None: + """设置 MCP 处理器""" + if not self.server: + return + + @self.server.list_tools() + async def list_tools() -> list[Tool]: + """列出所有可用工具""" + return [ + Tool( + name=schema["name"], + description=schema["description"], + inputSchema=schema["inputSchema"] + ) + for schema in self.tool_schemas.values() + ] + + @self.server.call_tool() + async def call_tool(name: str, arguments: dict) -> list[TextContent]: + """调用工具""" + if name not in self.tools: + raise ValueError(f"Unknown tool: {name}") + + try: + result = await self.tools[name](**arguments) + return [TextContent(type="text", text=json.dumps(result, ensure_ascii=False, indent=2))] + except Exception as e: + logger.error(f"Error calling tool {name}: {e}") + return [TextContent(type="text", text=json.dumps({ + "success": False, + "error": str(e) + }, ensure_ascii=False))] + + async def run(self) -> None: + """运行 MCP 服务器""" + if not self.server: + logger.error("MCP Server not available (MCP SDK not installed)") + return + + # 运行服务器 + async with self.server.stdio_stdio() as (read_stream, write_stream): + await self.server.run( + read_stream, + write_stream, + InitializationOptions( + server_name="wiki-mcp", + server_version="1.0.0", + capabilities=self.server.get_capabilities( + notification_options=NotificationOptions(), + experimental_capabilities={} + ) + ) + ) + + async def close(self) -> None: + """关闭服务器""" + if self.db: + await self.db.close() + logger.info("Database connection closed") + + +async def main(): + """主入口""" + import os + + # 从环境变量读取配置 + wiki_vault_path = os.environ.get("WIKI_VAULT_PATH", DEFAULT_WIKI_VAULT) + index_path = os.path.expanduser(os.environ.get("WIKI_INDEX_PATH", DEFAULT_INDEX_PATH)) + log_level = os.environ.get("LOG_LEVEL", DEFAULT_LOG_LEVEL) + + # 设置日志级别 + logging.getLogger().setLevel(getattr(logging, log_level)) + + # 检查 wiki vault 路径 + if not Path(wiki_vault_path).exists(): + logger.error(f"Wiki vault path does not exist: {wiki_vault_path}") + sys.exit(1) + + # 创建并运行服务器 + server = MCPServer(wiki_vault_path, index_path) + + try: + await server.initialize() + await server.run() + except KeyboardInterrupt: + logger.info("Interrupted by user") + finally: + await server.close() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/mcp_server/services/__init__.py b/mcp_server/services/__init__.py new file mode 100644 index 0000000..dbba744 --- /dev/null +++ b/mcp_server/services/__init__.py @@ -0,0 +1,19 @@ +""" +Service Layer - 服务层 + +提供各种服务：缓存、查询、索引、解析、图服务。 +""" + +from .cache import CacheService +from .query import QueryService +from .parser import ParserService +from .indexer import IndexerService +from .graph import GraphService + +__all__ = [ + "CacheService", + "QueryService", + "ParserService", + "IndexerService", + "GraphService" +] diff --git a/mcp_server/services/cache.py b/mcp_server/services/cache.py new file mode 100644 index 0000000..c303c69 --- /dev/null +++ b/mcp_server/services/cache.py @@ -0,0 +1,91 @@ +""" +Service Layer - CacheService（缓存服务） + +使用 LRU 缓存 + TTL 过期策略，防止内存泄漏。 + +参考设计文档：第 2.3 节 +""" + +import asyncio +import time +from collections import OrderedDict +from typing import Optional, Any +import fnmatch +import logging + +logger = logging.getLogger(__name__) + + +class CacheService: + """缓存服务 - LRU 缓存 + TTL 过期""" + + def __init__(self, max_size: int = 1000): + self.cache: OrderedDict[str, tuple] = OrderedDict() # key -> (value, expire_time) + self.max_size = max_size + self.lock = asyncio.Lock() + + async def get(self, key: str) -> Optional[Any]: + """获取缓存值（异步，带锁）""" + async with self.lock: + if key not in self.cache: + return None + + value, expire_time = self.cache[key] + + # 检查是否过期 + if expire_time and time.time() > expire_time: + del self.cache[key] + logger.debug(f"Cache expired: {key}") + return None + + # LRU: 移到末尾 + self.cache.move_to_end(key) + return value + + async def set(self, key: str, value: Any, ttl: int = 3600) -> None: + """设置缓存值（异步，带锁）""" + async with self.lock: + expire_time = time.time() + ttl if ttl else None + + # 如果缓存已满，删除最旧的条目 + if len(self.cache) >= self.max_size and key not in self.cache: + self.cache.popitem(last=False) # FIFO 删除 + logger.debug(f"Cache full, evicted oldest entry") + + self.cache[key] = (value, expire_time) + self.cache.move_to_end(key) + logger.debug(f"Cache set: {key} (TTL={ttl}s)") + + async def invalidate(self, pattern: str) -> int: + """按模式清除缓存（支持 * 通配符）""" + async with self.lock: + if pattern == "*": + count = len(self.cache) + self.cache.clear() + logger.info(f"Cache cleared: {count} entries") + return count + + keys_to_delete = [k for k in self.cache.keys() if fnmatch.fnmatch(k, pattern)] + for key in keys_to_delete: + del self.cache[key] + logger.info(f"Cache invalidated: {len(keys_to_delete)} entries matching '{pattern}'") + return len(keys_to_delete) + + async def get_stats(self) -> dict: + """获取缓存统计信息""" + async with self.lock: + now = time.time() + expired_count = sum(1 for _, expire_time in self.cache.values() if expire_time and now > expire_time) + + return { + "size": len(self.cache), + "max_size": self.max_size, + "expired_count": expired_count, + "utilization": len(self.cache) / self.max_size if self.max_size > 0 else 0 + } + + async def clear(self) -> None: + """清空缓存""" + async with self.lock: + self.cache.clear() + logger.info("Cache cleared") diff --git a/mcp_server/services/graph.py b/mcp_server/services/graph.py new file mode 100644 index 0000000..5c8efb9 --- /dev/null +++ b/mcp_server/services/graph.py @@ -0,0 +1,137 @@ +""" +Service Layer - GraphService（图服务） + +管理页面间的链接关系，查找孤立页面和缺失链接。 + +参考设计文档：第 2.3 节 +""" + +import logging +from typing import Set, List, Tuple +from ..storage import Database + +logger = logging.getLogger(__name__) + + +class GraphService: + """图服务 - 管理链接关系""" + + def __init__(self, db: Database): + self.db = db + + async def get_links(self, path: str) -> Set[str]: + """获取页面的出链""" + return await self.db.get_links(path) + + async def get_backlinks(self, path: str) -> Set[str]: + """获取页面的反向链接""" + return await self.db.get_backlinks(path) + + async def find_orphans(self) -> Set[str]: + """查找孤立页面（无反向链接）""" + all_pages = await self.db.get_all_pages() + orphans = set() + + for page in all_pages: + backlinks = await self.db.get_backlinks(page.path) + if not backlinks and page.path != "index.md": + orphans.add(page.path) + + return orphans + + async def find_missing_links(self) -> List[Tuple[str, str]]: + """查找缺失的交叉引用""" + # 获取所有页面内容 + all_pages = await self.db.get_all_pages() + page_paths = {page.path for page in all_pages} + + # 获取所有链接关系 + links_rows = await self.db.fetch_all("SELECT source, target FROM wiki_links") + + missing = [] + for source, target in links_rows: + if target not in page_paths: + missing.append((source, target)) + + return missing + + async def get_link_graph(self) -> dict: + """获取完整的链接图""" + pages = await self.db.get_all_pages() + graph = {} + + for page in pages: + links = await self.db.get_links(page.path) + backlinks = await self.db.get_backlinks(page.path) + graph[page.path] = { + "title": page.title, + "category": page.category, + "links": sorted(links), + "backlinks": sorted(backlinks), + "links_count": len(links), + "backlinks_count": len(backlinks) + } + + return graph + + async def get_highly_connected_pages(self, threshold: int = 5) -> List[dict]: + """获取高度连接的页面（入链 + 出链 >= threshold）""" + pages = await self.db.get_all_pages() + highly_connected = [] + + for page in pages: + links = await self.db.get_links(page.path) + backlinks = await self.db.get_backlinks(page.path) + total = len(links) + len(backlinks) + + if total >= threshold: + highly_connected.append({ + "path": page.path, + "title": page.title, + "links_count": len(links), + "backlinks_count": len(backlinks), + "total_connections": total + }) + + return sorted(highly_connected, key=lambda x: x["total_connections"], reverse=True) + + async def get_disconnected_components(self) -> List[Set[str]]: + """获取不连通的图分量（使用 BFS）""" + pages = await self.db.get_all_pages() + page_paths = {page.path for page in pages} + + if not page_paths: + return [] + + visited = set() + components = [] + + for start_path in page_paths: + if start_path in visited: + continue + + # BFS 构建连通分量 + component = set() + queue = [start_path] + + while queue: + current = queue.pop(0) + if current in visited: + continue + + visited.add(current) + component.add(current) + + # 添加出链和入链 + for link in await self.db.get_links(current): + if link in page_paths and link not in visited: + queue.append(link) + for backlink in await self.db.get_backlinks(current): + if backlink in page_paths and backlink not in visited: + queue.append(backlink) + + components.append(component) + + # 返回最大的分量以外的所有分量（即孤立的子图） + main_component = max(components, key=len) + return [c for c in components if c != main_component] diff --git a/mcp_server/services/indexer.py b/mcp_server/services/indexer.py new file mode 100644 index 0000000..631fbde --- /dev/null +++ b/mcp_server/services/indexer.py @@ -0,0 +1,226 @@ +""" +Service Layer - IndexerService（索引服务） + +管理 wiki 页面索引，支持增量更新和全量重建。 + +参考设计文档：第 2.3 节 +""" + +import logging +import asyncio +from pathlib import Path +from typing import List, Set, Optional +from datetime import datetime +from ..storage import Database, WikiPage, compute_content_hash +from .parser import ParserService + +logger = logging.getLogger(__name__) + + +class IndexerService: + """索引服务 - 管理页面索引""" + + def __init__(self, db: Database, parser: ParserService, wiki_vault_path: str): + self.db = db + self.parser = parser + self.wiki_vault_path = Path(wiki_vault_path) + self._lock = asyncio.Lock() + + async def index_page(self, page_path: str) -> Optional[WikiPage]: + """索引单个页面""" + try: + full_path = self.wiki_vault_path / page_path + + # 读取文件内容 + with open(full_path, 'r', encoding='utf-8') as f: + content = f.read() + + # 解析页面 + parsed = self.parser.parse_wiki_page(page_path, content) + + # 计算内容哈希 + content_hash = compute_content_hash(content) + + # 检查是否需要更新 + stored_hash = await self.db.get_page_hash(page_path) + if stored_hash == content_hash: + logger.debug(f"Page unchanged, skipping: {page_path}") + return None + + # 创建 WikiPage 对象 + page = WikiPage( + path=parsed["path"], + title=parsed["title"], + category=parsed["category"], + tags=parsed["tags"], + summary=parsed["summary"], + content_hash=content_hash, + lifecycle="draft", # 默认为 draft，后续可通过 wiki_lint 提升 + source_tool=parsed["source_tool"], + created_at=parsed["created_at"], + updated_at=parsed["updated_at"], + indexed_at=parsed["indexed_at"] + ) + + # 更新数据库 + await self.db.upsert_page(page) + + # 更新 FTS5 索引 + await self.db.update_fts_content( + page_path, + page.title, + content, + page.summary + ) + + # 更新链接关系 + await self.db.delete_links(page_path) + for link in parsed.get("links", []): + await self.db.upsert_link(page_path, link) + + # 更新标签 + for tag in page.tags: + await self.db.upsert_tag(tag) + + logger.info(f"Indexed page: {page_path}") + return page + + except Exception as e: + logger.error(f"Failed to index page {page_path}: {e}") + return None + + async def index_batch(self, page_paths: List[str]) -> dict: + """批量索引页面""" + results = { + "success": 0, + "failed": 0, + "skipped": 0, + "pages": [] + } + + for page_path in page_paths: + try: + page = await self.index_page(page_path) + if page: + results["success"] += 1 + results["pages"].append(page_path) + else: + results["skipped"] += 1 + except Exception as e: + logger.error(f"Failed to index {page_path}: {e}") + results["failed"] += 1 + + return results + + async def get_dirty_pages(self) -> List[str]: + """获取需要重新索引的页面（增量更新）""" + all_pages = self.scan_wiki_vault() + dirty_pages = [] + + for page_path in all_pages: + full_path = self.wiki_vault_path / page_path + try: + with open(full_path, 'r', encoding='utf-8') as f: + content = f.read() + current_hash = compute_content_hash(content) + stored_hash = await self.db.get_page_hash(page_path) + if stored_hash != current_hash: + dirty_pages.append(page_path) + except Exception as e: + logger.warning(f"Failed to check {page_path}: {e}") + + return dirty_pages + + async def incremental_update(self) -> dict: + """增量更新索引 - 只处理变化的页面""" + logger.info("Starting incremental update...") + + async with self._lock: + # 1. 获取所有 wiki 页面 + all_pages = self.scan_wiki_vault() + + # 2. 检查每个页面的哈希 + for page_path in all_pages: + try: + full_path = self.wiki_vault_path / page_path + with open(full_path, 'r', encoding='utf-8') as f: + content = f.read() + current_hash = compute_content_hash(content) + stored_hash = await self.db.get_page_hash(page_path) + + if stored_hash != current_hash: + await self.index_page(page_path) + except Exception as e: + logger.error(f"Failed to update {page_path}: {e}") + + # 3. 处理删除的页面 + indexed_paths = await self.db.get_all_indexed_paths() + for path in indexed_paths: + if path not in all_pages: + await self.db.delete_page(path) + await self.db.delete_fts_content(path) + logger.info(f"Deleted from index: {path}") + + # 更新元数据 + await self.db.set_meta("last_incremental_update", datetime.now().isoformat()) + + stats = await self.db.get_stats() + logger.info(f"Incremental update completed: {stats}") + return stats + + async def rebuild_index(self) -> dict: + """全量重建索引""" + logger.info("Starting full index rebuild...") + + async with self._lock: + # 清空现有索引 + await self.db.execute("DELETE FROM wiki_pages") + await self.db.execute("DELETE FROM wiki_content") + await self.db.execute("DELETE FROM wiki_fts") + await self.db.execute("DELETE FROM wiki_links") + await self.db.execute("DELETE FROM wiki_tags") + await self.db.execute("DELETE FROM wiki_page_tags") + logger.info("Cleared existing index") + + # 扫描并索引所有页面 + all_pages = self.scan_wiki_vault() + results = await self.index_batch(all_pages) + + # 更新元数据 + await self.db.set_meta("last_full_reindex", datetime.now().isoformat()) + + stats = await self.db.get_stats() + logger.info(f"Full rebuild completed: {stats}") + return {"results": results, "stats": stats} + + def scan_wiki_vault(self) -> List[str]: + """扫描 wiki vault，返回所有 .md 文件的相对路径""" + if not self.wiki_vault_path.exists(): + logger.error(f"Wiki vault path does not exist: {self.wiki_vault_path}") + return [] + + markdown_files = list(self.wiki_vault_path.rglob("*.md")) + relative_paths = [] + + for md_file in markdown_files: + try: + relative_path = md_file.relative_to(self.wiki_vault_path).as_posix() + relative_paths.append(relative_path) + except ValueError: + logger.warning(f"Failed to get relative path for: {md_file}") + + return sorted(relative_paths) + + async def get_index_status(self) -> dict: + """获取索引状态""" + stats = await self.db.get_stats() + dirty_pages = await self.get_dirty_pages() + + return { + "total_pages": stats.get("total_pages", 0), + "total_links": stats.get("total_links", 0), + "total_tags": stats.get("total_tags", 0), + "last_indexed": stats.get("last_indexed"), + "dirty_pages": len(dirty_pages), + "dirty_page_list": dirty_pages[:10] # 只返回前 10 个 + } diff --git a/mcp_server/services/parser.py b/mcp_server/services/parser.py new file mode 100644 index 0000000..e59f3dd --- /dev/null +++ b/mcp_server/services/parser.py @@ -0,0 +1,239 @@ +""" +Service Layer - ParserService（解析服务） + +解析 Markdown 文件，提取 frontmatter、链接、标签等。 + +参考设计文档：第 2.3 节 +""" + +import re +import logging +import json +from typing import List, Dict, Optional, Tuple +from datetime import datetime +from pathlib import Path + +logger = logging.getLogger(__name__) + + +class ParserService: + """解析服务 - 解析 Markdown 文件""" + + # Frontmatter 正则 + FRONTMATTER_PATTERN = re.compile(r'^---\s*\n(.*?)\n---\s*\n', re.DOTALL) + + # Wikilink 正则 + WIKILINK_PATTERN = re.compile(r'\[\[([^\]|]+)(?:\|[^\]]+)?\]\]') + + # Tag 正则（frontmatter 中的 tags 字段） + TAG_PATTERN = re.compile(r'tags:\s*\[(.*?)\]') + + def parse_frontmatter(self, content: str) -> Dict: + """ + 解析 frontmatter + + 返回格式： + { + "name": "...", + "description": "...", + "metadata": {...}, + "custom": {...} # 自定义字段 + } + """ + frontmatter = {} + match = self.FRONTMATTER_PATTERN.match(content) + + if match: + yaml_content = match.group(1) + # 简单解析（对于复杂情况，应该使用 PyYAML） + for line in yaml_content.split('\n'): + if ':' in line: + key, value = line.split(':', 1) + key = key.strip() + value = value.strip() + frontmatter[key] = value + + return frontmatter + + def extract_links(self, content: str) -> List[str]: + """ + 提取所有 [[wikilinks]] + + 返回链接目标列表（去重） + """ + links = self.WIKILINK_PATTERN.findall(content) + return list(set(links)) + + def extract_tags(self, content: str) -> List[str]: + """ + 从 frontmatter 中提取标签 + + 支持： + - tags: ["tag1", "tag2"] + - tags: [tag1, tag2] + """ + tags = [] + match = self.TAG_PATTERN.search(content) + + if match: + tag_content = match.group(1) + # 尝试 JSON 解析 + try: + tags = json.loads(f"[{tag_content}]") + except: + # 简单逗号分隔 + tags = [t.strip().strip('"\'') for t in tag_content.split(',')] + + return tags + + def extract_title(self, content: str) -> str: + """ + 提取标题（优先级：frontmatter > 第一个 # 标题） + + 返回标题或空字符串 + """ + # 先检查 frontmatter + frontmatter = self.parse_frontmatter(content) + if "title" in frontmatter: + return frontmatter["title"] + + # 检查第一个 # 标题 + lines = content.split('\n') + for line in lines: + line = line.strip() + if line.startswith('#'): + # 移除 # 符号和空格 + title = re.sub(r'^#+\s*', '', line) + return title + + return "" + + def extract_summary(self, content: str, max_length: int = 200) -> str: + """ + 提取摘要（优先级：frontmatter description > 第一段文字） + + 限制在 max_length 字符内 + """ + # 先检查 frontmatter + frontmatter = self.parse_frontmatter(content) + if "description" in frontmatter: + return frontmatter["description"][:max_length] + + # 移除 frontmatter 和代码块 + content_without_frontmatter = self.FRONTMATTER_PATTERN.sub('', content) + content_without_code = re.sub(r'```.*?```', '', content_without_frontmatter, flags=re.DOTALL) + + # 获取第一段 + lines = content_without_code.strip().split('\n\n') + for paragraph in lines: + paragraph = paragraph.strip() + if paragraph and not paragraph.startswith('#'): + # 移除 Markdown 格式 + clean_paragraph = re.sub(r'[*_`#\[\]]', '', paragraph) + if clean_paragraph: + return clean_paragraph[:max_length] + + return "" + + def infer_category(self, path: str) -> str: + """ + 从路径推断分类 + + 例如：practices/xxx.md -> practices + """ + path_obj = Path(path) + if path_obj.parent.name: + return path_obj.parent.name + return "uncategorized" + + def infer_source_tool(self, content: str) -> str: + """ + 从内容推断来源工具 + + 检查 frontmatter 中的特定字段： + - source_tool: claude/web_reader/gitea/other + """ + frontmatter = self.parse_frontmatter(content) + return frontmatter.get("source_tool", "other") + + def validate_page(self, path: str, content: str) -> List[str]: + """ + 验证页面，返回问题列表 + + 检查： + - 是否有标题 + - 是否有摘要 + - 是否有分类 + - frontmatter 是否有效 + """ + issues = [] + + # 检查标题 + title = self.extract_title(content) + if not title: + issues.append("Missing title") + + # 检查摘要 + summary = self.extract_summary(content) + if not summary: + issues.append("Missing summary") + + # 检查分类 + category = self.infer_category(path) + if category == "uncategorized": + issues.append("Uncategorized page") + + # 检查 frontmatter + frontmatter = self.parse_frontmatter(content) + if not frontmatter: + issues.append("Missing or invalid frontmatter") + + return issues + + def parse_wiki_page(self, path: str, content: str, created_at: Optional[str] = None, updated_at: Optional[str] = None) -> Dict: + """ + 完整解析 wiki 页面，返回结构化数据 + + 返回格式与 WikiPage 兼容 + """ + title = self.extract_title(content) + summary = self.extract_summary(content) + category = self.infer_category(path) + tags = self.extract_tags(content) + source_tool = self.infer_source_tool(content) + links = self.extract_links(content) + + now = datetime.now().isoformat() + + return { + "path": path, + "title": title, + "category": category, + "tags": tags, + "summary": summary, + "links": links, + "source_tool": source_tool, + "created_at": created_at or now, + "updated_at": updated_at or now, + "indexed_at": now + } + + def format_wikilink(self, path: str, title: Optional[str] = None) -> str: + """ + 格式化 wikilink + + 如果提供 title，使用 [[path|title]] 格式 + 否则使用 [[path]] 格式 + """ + if title: + return f"[[{path}|{title}]]" + return f"[[{path}]]" + + def resolve_wikilink_path(self, link: str, current_path: str) -> str: + """ + 解析相对 wikilink 路径 + + 例如：在 practices/a.md 中的 [[../concepts/x]] 解析为 concepts/x + """ + # TODO: 实现相对路径解析 + return link diff --git a/mcp_server/services/query.py b/mcp_server/services/query.py new file mode 100644 index 0000000..3fb6499 --- /dev/null +++ b/mcp_server/services/query.py @@ -0,0 +1,120 @@ +""" +Service Layer - QueryService（查询服务） + +负责所有查询逻辑，包括 FTS5 全文搜索、标签搜索、页面查询等。 + +参考设计文档：第 2.3 节 +""" + +import logging +from typing import List, Set, Optional +from ..storage import Database, WikiPage +from .cache import CacheService + +logger = logging.getLogger(__name__) + + +class QueryService: + """查询服务 - 负责所有查询逻辑""" + + def __init__(self, db: Database, cache: CacheService): + self.db = db + self.cache = cache + + async def search(self, query: str, limit: int = 10) -> List[WikiPage]: + """FTS5 全文搜索""" + # 1. 检查缓存 + cache_key = f"search:{query}:{limit}" + cached = await self.cache.get(cache_key) + if cached: + logger.debug(f"Cache hit for search: {query}") + return cached + + # 2. FTS5 搜索 + results = await self.db.fts_search(query, limit) + + # 3. 缓存结果 + await self.cache.set(cache_key, results, ttl=3600) + + return results + + async def search_by_tags(self, tags: List[str]) -> List[WikiPage]: + """按标签搜索""" + cache_key = f"tags:{','.join(tags)}" + cached = await self.cache.get(cache_key) + if cached: + return cached + + results = await self.db.search_by_tags(tags) + await self.cache.set(cache_key, results, ttl=3600) + return results + + async def search_by_source_tool(self, tool_name: str, limit: int = 50) -> List[WikiPage]: + """按来源工具搜索（memory_bridge 使用）""" + cache_key = f"tool:{tool_name}:{limit}" + cached = await self.cache.get(cache_key) + if cached: + return cached + + results = await self.db.search_by_source_tool(tool_name, limit) + await self.cache.set(cache_key, results, ttl=1800) # 30 分钟 TTL + return results + + async def get_page(self, path: str) -> Optional[WikiPage]: + """获取单个页面""" + cache_key = f"page:{path}" + cached = await self.cache.get(cache_key) + if cached: + return cached + + page = await self.db.get_page(path) + if page: + await self.cache.set(cache_key, page, ttl=3600) + return page + + async def get_links(self, path: str) -> Set[str]: + """获取页面的出链""" + cache_key = f"links:{path}" + cached = await self.cache.get(cache_key) + if cached: + return cached + + links = await self.db.get_links(path) + await self.cache.set(cache_key, links, ttl=3600) + return links + + async def get_backlinks(self, path: str) -> Set[str]: + """获取页面的反向链接""" + cache_key = f"backlinks:{path}" + cached = await self.cache.get(cache_key) + if cached: + return cached + + backlinks = await self.db.get_backlinks(path) + await self.cache.set(cache_key, backlinks, ttl=3600) + return backlinks + + async def find_orphans(self) -> Set[str]: + """查找孤立页面（无反向链接）""" + cache_key = "orphans" + cached = await self.cache.get(cache_key) + if cached: + return cached + + all_pages = await self.db.get_all_pages() + orphans = set() + for page in all_pages: + backlinks = await self.db.get_backlinks(page.path) + if not backlinks and page.path != "index.md": + orphans.add(page.path) + + await self.cache.set(cache_key, orphans, ttl=1800) + return orphans + + async def get_stats(self) -> dict: + """获取索引统计信息""" + return await self.db.get_stats() + + async def invalidate_cache(self, pattern: str = "*") -> int: + """使缓存失效""" + return await self.cache.invalidate(pattern) diff --git a/mcp_server/storage/__init__.py b/mcp_server/storage/__init__.py new file mode 100644 index 0000000..6673cb0 --- /dev/null +++ b/mcp_server/storage/__init__.py @@ -0,0 +1,14 @@ +""" +Storage Layer - 数据存储层 + +提供数据库访问和缓存功能。 +""" + +from .database import Database, WikiPage, fix_dirty_states, compute_content_hash + +__all__ = [ + "Database", + "WikiPage", + "fix_dirty_states", + "compute_content_hash" +] diff --git a/mcp_server/storage/database.py b/mcp_server/storage/database.py new file mode 100644 index 0000000..b465492 --- /dev/null +++ b/mcp_server/storage/database.py @@ -0,0 +1,531 @@ +""" +Storage Layer - Database 类 + +使用 aiosqlite 实现真正的异步支持，配合 WAL 模式和并发保护。 + +参考设计文档：第 2.4 节 +""" + +import aiosqlite +import asyncio +import logging +import time +from typing import List, Tuple, Dict, Set, Optional, Any +from datetime import datetime +from dataclasses import dataclass, asdict +import hashlib +import json + +logger = logging.getLogger(__name__) + + +@dataclass +class WikiPage: + """Wiki 页面模型""" + path: str + title: str + category: str + tags: List[str] + summary: str + content_hash: str + lifecycle: str + source_tool: str + created_at: str + updated_at: str + indexed_at: str + + def is_stale(self, days: int = 90) -> bool: + """检查页面是否过期""" + updated = datetime.fromisoformat(self.updated_at) + return (datetime.now() - updated).days > days + + def to_dict(self) -> dict: + """转换为字典""" + return asdict(self) + + +class Database: + """SQLite 数据库封装，使用 aiosqlite 实现真正的异步支持""" + + def __init__(self, path: str): + self.path = path + self._conn: Optional[aiosqlite.Connection] = None + self._lock = asyncio.Lock() + + async def connect(self) -> None: + """建立连接，启用 WAL 模式和并发保护""" + logger.info(f"Connecting to database: {self.path}") + self._conn = await aiosqlite.connect(self.path) + + # 启用 WAL 模式提升并发性能 + await self._conn.execute("PRAGMA journal_mode=WAL") + await self._conn.execute("PRAGMA busy_timeout=10000") # 10s + await self._conn.execute("PRAGMA synchronous=NORMAL") + await self._conn.commit() + + # 创建表结构 + await self._create_tables() + + logger.info("Database connected and initialized") + + async def _create_tables(self) -> None: + """创建所有表结构""" + + # 页面索引表 + await self._conn.execute(""" + CREATE TABLE IF NOT EXISTS wiki_pages ( + path TEXT PRIMARY KEY, + title TEXT NOT NULL, + category TEXT, + tags TEXT, + summary TEXT, + content_hash TEXT NOT NULL, + lifecycle TEXT DEFAULT 'draft', + source_tool TEXT DEFAULT 'other', + created_at TIMESTAMP, + updated_at TIMESTAMP, + indexed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """) + + # 索引 + await self._conn.execute("CREATE INDEX IF NOT EXISTS idx_pages_category ON wiki_pages(category)") + await self._conn.execute("CREATE INDEX IF NOT EXISTS idx_pages_lifecycle ON wiki_pages(lifecycle)") + await self._conn.execute("CREATE INDEX IF NOT EXISTS idx_pages_updated ON wiki_pages(updated_at)") + await self._conn.execute("CREATE INDEX IF NOT EXISTS idx_pages_source_tool ON wiki_pages(source_tool)") + + # 内容表（FTS5 外部内容表）- 必须先创建 + await self._conn.execute(""" + CREATE TABLE IF NOT EXISTS wiki_content ( + path TEXT PRIMARY KEY, + content TEXT NOT NULL + ) + """) + + # FTS5 全文搜索表 + await self._conn.execute(""" + CREATE VIRTUAL TABLE IF NOT EXISTS wiki_fts USING fts5( + path UNINDEXED, + title, + content, + summary, + content=wiki_content, + content_rowid=rowid, + tokenize = 'porter unicode61' + ) + """) + + # 链接关系表 + await self._conn.execute(""" + CREATE TABLE IF NOT EXISTS wiki_links ( + source TEXT NOT NULL, + target TEXT NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (source, target) + ) + """) + await self._conn.execute("CREATE INDEX IF NOT EXISTS idx_links_target ON wiki_links(target)") + + # 标签索引表 + await self._conn.execute(""" + CREATE TABLE IF NOT EXISTS wiki_tags ( + tag TEXT PRIMARY KEY, + count INTEGER DEFAULT 0, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """) + + # 页面-标签关联表 + await self._conn.execute(""" + CREATE TABLE IF NOT EXISTS wiki_page_tags ( + path TEXT NOT NULL, + tag TEXT NOT NULL, + PRIMARY KEY (path, tag), + FOREIGN KEY (path) REFERENCES wiki_pages(path) ON DELETE CASCADE, + FOREIGN KEY (tag) REFERENCES wiki_tags(tag) ON DELETE CASCADE + ) + """) + await self._conn.execute("CREATE INDEX IF NOT EXISTS idx_page_tags_tag ON wiki_page_tags(tag)") + + # 索引元数据表 + await self._conn.execute(""" + CREATE TABLE IF NOT EXISTS wiki_meta ( + key TEXT PRIMARY KEY, + value TEXT, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """) + + await self._conn.commit() + + async def execute(self, sql: str, params: tuple = ()) -> aiosqlite.Cursor: + """执行 SQL（带写入锁）""" + async with self._lock: + cursor = await self._conn.execute(sql, params) + await self._conn.commit() + return cursor + + async def execute_with_retry(self, sql: str, params: tuple = (), max_retries: int = 2) -> aiosqlite.Cursor: + """带重试的数据库操作（指数退避）""" + for attempt in range(max_retries): + try: + return await self.execute(sql, params) + except aiosqlite.OperationalError as e: + if "database is locked" in str(e) and attempt < max_retries - 1: + wait_time = 0.1 * (2 ** attempt) + logger.warning(f"Database locked, retrying in {wait_time}s (attempt {attempt + 1}/{max_retries})") + await asyncio.sleep(wait_time) + continue + raise + + async def fetch_all(self, sql: str, params: tuple = ()) -> List[Tuple]: + """查询所有结果（读操作无需锁，WAL 自动处理）""" + cursor = await self._conn.execute(sql, params) + return await cursor.fetchall() + + async def fetch_one(self, sql: str, params: tuple = ()) -> Optional[Tuple]: + """查询单个结果""" + rows = await self.fetch_all(sql, params) + return rows[0] if rows else None + + # === 页面操作 === + + async def get_page(self, path: str) -> Optional[WikiPage]: + """获取单个页面""" + row = await self.fetch_one( + "SELECT path, title, category, tags, summary, content_hash, lifecycle, source_tool, created_at, updated_at, indexed_at FROM wiki_pages WHERE path = ?", + (path,) + ) + if not row: + return None + + return WikiPage( + path=row[0], + title=row[1], + category=row[2], + tags=json.loads(row[3]) if row[3] else [], + summary=row[4], + content_hash=row[5], + lifecycle=row[6], + source_tool=row[7], + created_at=row[8], + updated_at=row[9], + indexed_at=row[10] + ) + + async def get_all_pages(self) -> List[WikiPage]: + """获取所有页面""" + rows = await self.fetch_all( + "SELECT path, title, category, tags, summary, content_hash, lifecycle, source_tool, created_at, updated_at, indexed_at FROM wiki_pages" + ) + return [ + WikiPage( + path=row[0], + title=row[1], + category=row[2], + tags=json.loads(row[3]) if row[3] else [], + summary=row[4], + content_hash=row[5], + lifecycle=row[6], + source_tool=row[7], + created_at=row[8], + updated_at=row[9], + indexed_at=row[10] + ) + for row in rows + ] + + async def get_page_hash(self, path: str) -> Optional[str]: + """获取页面内容哈希""" + row = await self.fetch_one("SELECT content_hash FROM wiki_pages WHERE path = ?", (path,)) + return row[0] if row else None + + async def upsert_page(self, page: WikiPage) -> None: + """插入或更新页面""" + await self.execute_with_retry(""" + INSERT INTO wiki_pages (path, title, category, tags, summary, content_hash, lifecycle, source_tool, created_at, updated_at, indexed_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(path) DO UPDATE SET + title = excluded.title, + category = excluded.category, + tags = excluded.tags, + summary = excluded.summary, + content_hash = excluded.content_hash, + lifecycle = excluded.lifecycle, + source_tool = excluded.source_tool, + updated_at = excluded.updated_at, + indexed_at = excluded.indexed_at + """, ( + page.path, page.title, page.category, json.dumps(page.tags), + page.summary, page.content_hash, page.lifecycle, page.source_tool, + page.created_at, page.updated_at, page.indexed_at + )) + + async def delete_page(self, path: str) -> None: + """删除页面（级联删除相关数据）""" + await self.execute_with_retry("DELETE FROM wiki_pages WHERE path = ?", (path,)) + + async def get_all_indexed_paths(self) -> Set[str]: + """获取所有已索引的页面路径""" + rows = await self.fetch_all("SELECT path FROM wiki_pages") + return {row[0] for row in rows} + + # === FTS5 搜索 === + + async def fts_search(self, query: str, limit: int = 10) -> List[WikiPage]: + """FTS5 全文搜索""" + # 先从 FTS5 获取匹配的路径 + fts_rows = await self.fetch_all( + "SELECT path FROM wiki_fts WHERE wiki_fts MATCH ? ORDER BY rank LIMIT ?", + (query, limit) + ) + paths = [row[0] for row in fts_rows] + + if not paths: + return [] + + # 从 wiki_pages 获取完整数据 + placeholders = ','.join('?' * len(paths)) + rows = await self.fetch_all( + f"SELECT path, title, category, tags, summary, content_hash, lifecycle, source_tool, created_at, updated_at, indexed_at FROM wiki_pages WHERE path IN ({placeholders})", + paths + ) + + return [ + WikiPage( + path=row[0], + title=row[1], + category=row[2], + tags=json.loads(row[3]) if row[3] else [], + summary=row[4], + content_hash=row[5], + lifecycle=row[6], + source_tool=row[7], + created_at=row[8], + updated_at=row[9], + indexed_at=row[10] + ) + for row in rows + ] + + async def search_by_tags(self, tags: List[str]) -> List[WikiPage]: + """按标签搜索""" + if not tags: + return [] + + placeholders = ','.join('?' * len(tags)) + rows = await self.fetch_all( + f"SELECT DISTINCT p.path, p.title, p.category, p.tags, p.summary, p.content_hash, p.lifecycle, p.source_tool, p.created_at, p.updated_at, p.indexed_at FROM wiki_pages p JOIN wiki_page_tags pt ON p.path = pt.path WHERE pt.tag IN ({placeholders})", + tags + ) + + return [ + WikiPage( + path=row[0], + title=row[1], + category=row[2], + tags=json.loads(row[3]) if row[3] else [], + summary=row[4], + content_hash=row[5], + lifecycle=row[6], + source_tool=row[7], + created_at=row[8], + updated_at=row[9], + indexed_at=row[10] + ) + for row in rows + ] + + async def search_by_source_tool(self, tool_name: str, limit: int = 50) -> List[WikiPage]: + """按来源工具搜索（memory_bridge 使用）""" + rows = await self.fetch_all( + "SELECT path, title, category, tags, summary, content_hash, lifecycle, source_tool, created_at, updated_at, indexed_at FROM wiki_pages WHERE source_tool = ? ORDER BY updated_at DESC LIMIT ?", + (tool_name, limit) + ) + + return [ + WikiPage( + path=row[0], + title=row[1], + category=row[2], + tags=json.loads(row[3]) if row[3] else [], + summary=row[4], + content_hash=row[5], + lifecycle=row[6], + source_tool=row[7], + created_at=row[8], + updated_at=row[9], + indexed_at=row[10] + ) + for row in rows + ] + + # === 链接操作 === + + async def get_links(self, path: str) -> Set[str]: + """获取页面的出链""" + rows = await self.fetch_all("SELECT target FROM wiki_links WHERE source = ?", (path,)) + return {row[0] for row in rows} + + async def get_backlinks(self, path: str) -> Set[str]: + """获取页面的反向链接""" + rows = await self.fetch_all("SELECT source FROM wiki_links WHERE target = ?", (path,)) + return {row[0] for row in rows} + + async def upsert_link(self, source: str, target: str) -> None: + """插入或更新链接""" + await self.execute_with_retry( + "INSERT INTO wiki_links (source, target) VALUES (?, ?) ON CONFLICT(source, target) DO NOTHING", + (source, target) + ) + + async def delete_links(self, path: str) -> None: + """删除页面的所有链接""" + await self.execute_with_retry("DELETE FROM wiki_links WHERE source = ?", (path,)) + + # === 标签操作 === + + async def upsert_tag(self, tag: str, count: int = 1) -> None: + """插入或更新标签""" + await self.execute_with_retry( + "INSERT INTO wiki_tags (tag, count) VALUES (?, ?) ON CONFLICT(tag) DO UPDATE SET count = count + excluded.count", + (tag, count) + ) + + async def get_all_tags(self) -> Dict[str, int]: + """获取所有标签""" + rows = await self.fetch_all("SELECT tag, count FROM wiki_tags") + return {row[0]: row[1] for row in rows} + + # === 元数据操作 === + + async def get_meta(self, key: str, default: str = "") -> str: + """获取元数据""" + row = await self.fetch_one("SELECT value FROM wiki_meta WHERE key = ?", (key,)) + return row[0] if row else default + + async def set_meta(self, key: str, value: str) -> None: + """设置元数据""" + await self.execute_with_retry( + "INSERT INTO wiki_meta (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value", + (key, value) + ) + + # === FTS5 内容更新 === + + async def update_fts_content(self, path: str, title: str, content: str, summary: str) -> None: + """更新 FTS5 内容""" + # 先更新内容表 + await self.execute_with_retry( + "INSERT INTO wiki_content (path, content) VALUES (?, ?) ON CONFLICT(path) DO UPDATE SET content = excluded.content", + (path, content) + ) + + # 再更新 FTS5 表（会自动从内容表同步） + await self.execute_with_retry( + "INSERT INTO wiki_fts (path, title, summary) VALUES (?, ?, ?) ON CONFLICT(path) DO UPDATE SET title = excluded.title, summary = excluded.summary", + (path, title, summary) + ) + + async def delete_fts_content(self, path: str) -> None: + """删除 FTS5 内容""" + await self.execute_with_retry("DELETE FROM wiki_content WHERE path = ?", (path,)) + await self.execute_with_retry("DELETE FROM wiki_fts WHERE path = ?", (path,)) + + # === 状态查询 === + + async def get_stats(self) -> Dict[str, Any]: + """获取索引统计信息""" + page_count = await self.fetch_one("SELECT COUNT(*) FROM wiki_pages") + link_count = await self.fetch_one("SELECT COUNT(*) FROM wiki_links") + tag_count = await self.fetch_one("SELECT COUNT(*) FROM wiki_tags") + last_indexed = await self.fetch_one("SELECT MAX(indexed_at) FROM wiki_pages") + + return { + "total_pages": page_count[0] if page_count else 0, + "total_links": link_count[0] if link_count else 0, + "total_tags": tag_count[0] if tag_count else 0, + "last_indexed": last_indexed[0] if last_indexed and last_indexed[0] else None + } + + async def get_recent_pages(self, days: int = 7, limit: int = 50) -> List[WikiPage]: + """获取最近更新的页面""" + rows = await self.fetch_all( + "SELECT path, title, category, tags, summary, content_hash, lifecycle, source_tool, created_at, updated_at, indexed_at FROM wiki_pages WHERE updated_at >= datetime('now', '-' || ? || ' days') ORDER BY updated_at DESC LIMIT ?", + (str(days), limit) + ) + + return [ + WikiPage( + path=row[0], + title=row[1], + category=row[2], + tags=json.loads(row[3]) if row[3] else [], + summary=row[4], + content_hash=row[5], + lifecycle=row[6], + source_tool=row[7], + created_at=row[8], + updated_at=row[9], + indexed_at=row[10] + ) + for row in rows + ] + + async def get_new_tags(self, days: int = 7) -> List[Tuple[str, str]]: + """获取新增标签""" + rows = await self.fetch_all( + "SELECT tag, updated_at FROM wiki_tags WHERE updated_at >= datetime('now', '-' || ? || ' days') ORDER BY updated_at DESC", + (str(days),) + ) + return list(rows) + + # === 完整性检查 === + + async def check_integrity(self) -> bool: + """检查数据库完整性""" + try: + result = await self.fetch_one("PRAGMA integrity_check") + if result and result[0] == "ok": + return True + logger.warning(f"Database integrity check failed: {result}") + return False + except Exception as e: + logger.error(f"Integrity check error: {e}") + return False + + # === 关闭连接 === + + async def close(self) -> None: + """关闭连接""" + if self._conn: + await self._conn.close() + self._conn = None + logger.info("Database connection closed") + + +async def fix_dirty_states(db: Database) -> None: + """启动时清理可能的脏状态""" + logger.info("Checking for dirty states...") + + try: + # 1. 检查 WAL 文件是否损坏 + if db._conn: + await db._conn.execute("PRAGMA wal_checkpoint(PASSIVE)") + await db._conn.commit() + + # 2. 检查数据库完整性 + if not await db.check_integrity(): + raise Exception("Database integrity check failed") + + logger.info("No dirty states found, database is healthy") + + except Exception as e: + logger.warning(f"Dirty states detected, attempting recovery: {e}") + # TODO: 实现自动重建逻辑 + # await rebuild_index(db) + + +def compute_content_hash(content: str) -> str: + """计算内容 MD5 哈希""" + return hashlib.md5(content.encode('utf-8')).hexdigest() diff --git a/mcp_server/tools/__init__.py b/mcp_server/tools/__init__.py new file mode 100644 index 0000000..f80e204 --- /dev/null +++ b/mcp_server/tools/__init__.py @@ -0,0 +1,25 @@ +""" +Tool Layer - 工具层 + +提供所有 MCP Tools。 +""" + +from .wiki_query import WikiQueryTool +from .memory_bridge import MemoryBridgeTool +from .wiki_status import WikiStatusTool +from .wiki_lint import WikiLintTool +from .cross_linker import CrossLinkerTool +from .tag_taxonomy import TagTaxonomyTool +from .wiki_synthesize import WikiSynthesizeTool +from .daily_update import DailyUpdateTool + +__all__ = [ + "WikiQueryTool", + "MemoryBridgeTool", + "WikiStatusTool", + "WikiLintTool", + "CrossLinkerTool", + "TagTaxonomyTool", + "WikiSynthesizeTool", + "DailyUpdateTool" +] diff --git a/mcp_server/tools/cross_linker.py b/mcp_server/tools/cross_linker.py new file mode 100644 index 0000000..de7a881 --- /dev/null +++ b/mcp_server/tools/cross_linker.py @@ -0,0 +1,87 @@ +""" +Tool Layer - cross_linker 工具 + +扫描 wiki，自动发现缺失的交叉引用。 + +参考设计文档：第 2.2 节 +""" + +import logging +from typing import Dict, Any, List +from ..services import GraphService + +logger = logging.getLogger(__name__) + + +class CrossLinkerTool: + """cross_linker 工具实现""" + + def __init__(self, graph_service: GraphService): + self.graph_service = graph_service + + async def handle(self, path: str = "", dry_run: bool = True) -> Dict[str, Any]: + """ + 扫描 wiki，发现缺失的交叉引用 + + Args: + path: 指定路径（空字符串表示全部） + dry_run: 是否为模拟运行（不实际修改） + + Returns: + { + "missing_links": [ + { + "source": "practices/a.md", + "target": "concepts/x", + "suggestion": "Consider creating [[concepts/x]]" + } + ], + "total": 12, + "dry_run": true + } + """ + # 获取缺失的链接 + all_missing = await self.graph_service.find_missing_links() + + # 过滤指定路径 + if path: + missing_links = [(s, t) for s, t in all_missing if s == path or t == path] + else: + missing_links = all_missing + + # 格式化结果 + missing_links_formatted = [] + for source, target in missing_links: + missing_links_formatted.append({ + "source": source, + "target": target, + "suggestion": f"Consider creating [[{target}]] or updating the link in [[{source}]]" + }) + + return { + "missing_links": missing_links_formatted[:50], # 最多返回 50 条 + "total": len(missing_links), + "dry_run": dry_run + } + + def get_schema(self) -> dict: + """返回 MCP Tool schema""" + return { + "name": "cross_linker", + "description": "扫描 wiki，自动发现缺失的交叉引用", + "inputSchema": { + "type": "object", + "properties": { + "path": { + "type": "string", + "default": "", + "description": "指定路径（空字符串表示全部）" + }, + "dry_run": { + "type": "boolean", + "default": True, + "description": "是否为模拟运行（不实际修改）" + } + } + } + } diff --git a/mcp_server/tools/daily_update.py b/mcp_server/tools/daily_update.py new file mode 100644 index 0000000..5dc0073 --- /dev/null +++ b/mcp_server/tools/daily_update.py @@ -0,0 +1,105 @@ +""" +Tool Layer - daily_update 工具 + +日常维护（检查源新鲜度、更新 index、重新生成 hot.md）。 + +参考设计文档：第 2.2 节 +""" + +import logging +from typing import Dict, Any +from datetime import datetime +from pathlib import Path +from ..services import QueryService, IndexerService + +logger = logging.getLogger(__name__) + + +class DailyUpdateTool: + """daily_update 工具实现""" + + def __init__(self, query_service: QueryService, indexer_service: IndexerService): + self.query_service = query_service + self.indexer_service = indexer_service + + async def handle(self) -> Dict[str, Any]: + """ + 日常维护 + + Returns: + { + "updated": 5, + "new": 2, + "hot_md_generated": true, + "index_status": {...} + } + """ + # 1. 增量更新索引 + index_status = await self.indexer_service.incremental_update() + + # 2. 获取最近更新的页面 + recent_pages = await self.query_service.db.get_recent_pages(days=7, limit=20) + + # 3. 获取新增标签 + new_tags = await self.query_service.db.get_new_tags(days=7) + + # 4. 查找孤立页面 + from ..services import GraphService + graph_service = GraphService(self.query_service.db) + orphans = await graph_service.find_orphans() + + # 5. 生成 hot.md + hot_md_generated = await self._generate_hot_md(recent_pages, new_tags, orphans) + + return { + "updated": len(recent_pages), + "new": len(new_tags), + "hot_md_generated": hot_md_generated, + "index_status": index_status + } + + async def _generate_hot_md(self, recent_pages, new_tags, orphans) -> bool: + """生成热点文件""" + try: + # 确定 hot.md 保存路径 + vault_path = Path(self.query_service.db.path).parent.parent / "wiki-vault" + hot_path = vault_path / "hot.md" + + # 格式化内容 + content = f"""# Wiki Hot - {datetime.now().strftime('%Y-%m-%d')} + +## 最近更新（7 天内） + +""" + for page in recent_pages[:10]: + content += f"- [[{page.path}|{page.title}]] - {page.updated_at}\n" + + content += "\n## 新增标签（7 天内）\n\n" + for tag, updated_at in new_tags[:10]: + content += f"- `{tag}` - {updated_at}\n" + + content += "\n## 待链接页面（孤立页面）\n\n" + for orphan in sorted(list(orphans))[:10]: + content += f"- [[{orphan}]]\n" + + # 写入文件 + with open(hot_path, 'w', encoding='utf-8') as f: + f.write(content) + + logger.info(f"Generated hot.md: {hot_path}") + return True + + except Exception as e: + logger.error(f"Failed to generate hot.md: {e}") + return False + + def get_schema(self) -> dict: + """返回 MCP Tool schema""" + return { + "name": "daily_update", + "description": "日常维护（检查源新鲜度、更新 index、重新生成 hot.md）", + "inputSchema": { + "type": "object", + "properties": {} + } + } diff --git a/mcp_server/tools/memory_bridge.py b/mcp_server/tools/memory_bridge.py new file mode 100644 index 0000000..2252bf6 --- /dev/null +++ b/mcp_server/tools/memory_bridge.py @@ -0,0 +1,112 @@ +""" +Tool Layer - memory_bridge 工具 + +按 AI 工具来源浏览和对比 wiki 知识。 + +参考设计文档：第 2.2 节 +""" + +import logging +from typing import Dict, Any, Optional +from datetime import datetime +from ..services import QueryService + +logger = logging.getLogger(__name__) + + +class MemoryBridgeTool: + """memory_bridge 工具实现""" + + def __init__(self, query_service: QueryService): + self.query_service = query_service + + async def handle(self, tool_name: str = "claude", date_range: str = "") -> Dict[str, Any]: + """ + 按 AI 工具来源浏览和对比 wiki 知识 + + Args: + tool_name: AI 工具名称（claude/web_reader/gitea/other） + date_range: 日期范围（如 "2024-01-01:2024-12-31" 或空字符串表示不限） + + Returns: + { + "entries": [ + { + "path": "practices/moziplus-orchestration.md", + "title": "moziplus 编排实践", + "summary": "...", + "updated_at": "2024-06-15T10:30:00", + "relevance_score": 0.85 + } + ], + "total": 12, + "tool_name": "claude", + "date_range": "2024-01-01:2024-12-31" + } + """ + # 按工具来源搜索 + pages = await self.query_service.search_by_source_tool(tool_name, limit=100) + + # 按日期范围过滤 + if date_range: + try: + start_str, end_str = date_range.split(":") + start_date = datetime.fromisoformat(start_str) + end_date = datetime.fromisoformat(end_str) + + filtered = [] + for page in pages: + updated = datetime.fromisoformat(page.updated_at) + if start_date <= updated <= end_date: + filtered.append(page) + pages = filtered + except Exception as e: + logger.warning(f"Invalid date_range format '{date_range}': {e}") + + # 格式化结果 + entries = [] + for page in pages: + # 计算相关性分数（基于摘要长度和更新时间） + recency_days = (datetime.now() - datetime.fromisoformat(page.updated_at)).days + relevance_score = max(0.1, 1.0 - recency_days / 365) # 简单衰减 + + entries.append({ + "path": page.path, + "title": page.title, + "summary": page.summary, + "updated_at": page.updated_at, + "relevance_score": round(relevance_score, 2) + }) + + # 按相关性排序 + entries.sort(key=lambda x: x["relevance_score"], reverse=True) + + return { + "entries": entries[:50], # 最多返回 50 条 + "total": len(entries), + "tool_name": tool_name, + "date_range": date_range + } + + def get_schema(self) -> dict: + """返回 MCP Tool schema""" + return { + "name": "memory_bridge", + "description": "按 AI 工具来源浏览和对比 wiki 知识", + "inputSchema": { + "type": "object", + "properties": { + "tool_name": { + "type": "string", + "default": "claude", + "description": "AI 工具名称（claude/web_reader/gitea/other）", + "enum": ["claude", "web_reader", "gitea", "other"] + }, + "date_range": { + "type": "string", + "default": "", + "description": "日期范围，格式：YYYY-MM-DD:YYYY-MM-DD" + } + } + } + } diff --git a/mcp_server/tools/tag_taxonomy.py b/mcp_server/tools/tag_taxonomy.py new file mode 100644 index 0000000..82ca5c3 --- /dev/null +++ b/mcp_server/tools/tag_taxonomy.py @@ -0,0 +1,117 @@ +""" +Tool Layer - tag_taxonomy 工具 + +用受控词表强制标签一致性。 + +参考设计文档：第 2.2 节 +""" + +import logging +from typing import Dict, Any, List, Set +from ..services import QueryService + +logger = logging.getLogger(__name__) + + +class TagTaxonomyTool: + """tag_taxonomy 工具实现""" + + # 受控词表（可配置） + CONTROLLED_VOCABULARY: Set[str] = { + # 分类 + "practices", "concepts", "entities", "projects", "skills", "references", + # 技术栈 + "python", "typescript", "rust", "go", "swift", + # 领域 + "ai", "ml", "database", "web", "mobile", "devops", + # 团队 + "sanguo", "moziplus", "bitnet", + # 通用 + "tutorial", "reference", "guide", "example", "draft" + } + + def __init__(self, query_service: QueryService): + self.query_service = query_service + + async def handle(self, path: str = "", enforce: bool = False) -> Dict[str, Any]: + """ + 用受控词表强制标签一致性 + + Args: + path: 指定路径（空字符串表示全部） + enforce: 是否强制修正（否则仅报告） + + Returns: + { + "conflicts": [ + { + "path": "practices/example.md", + "invalid_tags": ["CustomTag"], + "suggested_tags": ["practices"] + } + ], + "total_conflicts": 5, + "enforced": false + } + """ + conflicts = [] + + # 获取所有标签 + all_tags = await self.query_service.db.get_all_tags() + + # 检查非受控标签 + invalid_tags = set(all_tags.keys()) - self.CONTROLLED_VOCABULARY + + if invalid_tags: + # 获取使用这些标签的页面 + for tag in invalid_tags: + tag_pages = await self.query_service.search_by_tags([tag]) + + for page in tag_pages: + page_tags = set(page.tags) & invalid_tags + + if page_tags: + # 推断修正建议（基于页面分类） + suggested_tags = [page.category] if page.category else [] + + conflicts.append({ + "path": page.path, + "invalid_tags": list(page_tags), + "suggested_tags": suggested_tags + }) + + if enforce and path and page.path == path: + # TODO: 实际修正标签 + pass + + return { + "conflicts": conflicts[:50], + "total_conflicts": len(conflicts), + "enforced": enforce + } + + def get_schema(self) -> dict: + """返回 MCP Tool schema""" + return { + "name": "tag_taxonomy", + "description": "用受控词表强制标签一致性", + "inputSchema": { + "type": "object", + "properties": { + "path": { + "type": "string", + "default": "", + "description": "指定路径（空字符串表示全部）" + }, + "enforce": { + "type": "boolean", + "default": False, + "description": "是否强制修正（否则仅报告）" + } + } + } + } + + def get_controlled_vocabulary(self) -> List[str]: + """获取当前受控词表""" + return sorted(list(self.CONTROLLED_VOCABULARY)) diff --git a/mcp_server/tools/wiki_lint.py b/mcp_server/tools/wiki_lint.py new file mode 100644 index 0000000..4ad26bc --- /dev/null +++ b/mcp_server/tools/wiki_lint.py @@ -0,0 +1,164 @@ +""" +Tool Layer - wiki_lint 工具 + +审计 wiki 健康（格式、链接、frontmatter 规范）。 + +参考设计文档：第 2.2 节 +""" + +import logging +from typing import Dict, Any, List +from ..services import ParserService, QueryService, GraphService + +logger = logging.getLogger(__name__) + + +class WikiLintTool: + """wiki_lint 工具实现""" + + def __init__(self, parser: ParserService, query_service: QueryService, graph_service: GraphService): + self.parser = parser + self.query_service = query_service + self.graph_service = graph_service + + async def handle(self, path: str = "", level: str = "basic") -> Dict[str, Any]: + """ + 审计 wiki 健康 + + Args: + path: 指定路径（空字符串表示全部） + level: 检查级别（basic/strict） + + Returns: + { + "issues": [ + { + "path": "practices/example.md", + "type": "missing_title", + "message": "Missing title", + "severity": "warning" + } + ], + "fixes": [ + { + "path": "practices/example.md", + "action": "add_title", + "suggestion": "# Example Page" + } + ], + "summary": { + "total_issues": 5, + "critical": 0, + "warning": 5, + "info": 0 + } + } + """ + issues = [] + fixes = [] + + # 获取所有页面或指定页面 + if path: + pages = [await self.query_service.get_page(path)] + else: + pages = await self.query_service.db.get_all_pages() + + summary = {"critical": 0, "warning": 0, "info": 0} + + for page in pages: + # 读取内容 + try: + from pathlib import Path + vault_path = Path(self.query_service.db.path).parent.parent / "wiki-vault" + full_path = vault_path / page.path + with open(full_path, 'r', encoding='utf-8') as f: + content = f.read() + except Exception as e: + issues.append({ + "path": page.path, + "type": "file_error", + "message": f"Failed to read file: {e}", + "severity": "critical" + }) + summary["critical"] += 1 + continue + + # 验证页面 + validation_issues = self.parser.validate_page(page.path, content) + + for issue in validation_issues: + severity = "warning" if "Missing" in issue else "info" + issues.append({ + "path": page.path, + "type": issue.lower().replace(" ", "_"), + "message": issue, + "severity": severity + }) + summary[severity] += 1 + + # 生成修复建议 + if "Missing title" in issue: + title = self.parser.extract_title(content) + if not title: + title = page.path.replace(".md", "").replace("-", " ").replace("_", " ").title() + fixes.append({ + "path": page.path, + "action": "add_title", + "suggestion": f"# {title}" + }) + + # strict 模式额外检查 + if level == "strict": + # 检查孤立页面 + orphans = await self.graph_service.find_orphans() + for orphan in orphans: + issues.append({ + "path": orphan, + "type": "orphan", + "message": "No backlinks found", + "severity": "info" + }) + summary["info"] += 1 + + # 检查缺失链接 + missing_links = await self.graph_service.find_missing_links() + for source, target in missing_links[:10]: + issues.append({ + "path": source, + "type": "broken_link", + "message": f"Link to non-existent page: {target}", + "severity": "warning" + }) + summary["warning"] += 1 + + return { + "issues": issues, + "fixes": fixes, + "summary": { + "total_issues": len(issues), + **summary + } + } + + def get_schema(self) -> dict: + """返回 MCP Tool schema""" + return { + "name": "wiki_lint", + "description": "审计 wiki 健康（格式、链接、frontmatter 规范）", + "inputSchema": { + "type": "object", + "properties": { + "path": { + "type": "string", + "default": "", + "description": "指定路径（空字符串表示全部）" + }, + "level": { + "type": "string", + "default": "basic", + "enum": ["basic", "strict"], + "description": "检查级别" + } + } + } + } diff --git a/mcp_server/tools/wiki_query.py b/mcp_server/tools/wiki_query.py new file mode 100644 index 0000000..426eec8 --- /dev/null +++ b/mcp_server/tools/wiki_query.py @@ -0,0 +1,116 @@ +""" +Tool Layer - wiki_query 工具 + +基于 SQLite 索引查询 wiki，支持关键词（FTS5）和标签搜索。 + +参考设计文档：第 2.2 节 +""" + +import logging +from typing import List, Dict, Any +from ..services import QueryService + +logger = logging.getLogger(__name__) + + +class WikiQueryTool: + """wiki_query 工具实现""" + + def __init__(self, query_service: QueryService): + self.query_service = query_service + + async def handle(self, query: str = "", tags: List[str] = None, limit: int = 10) -> Dict[str, Any]: + """ + 查询 wiki 页面 + + Args: + query: 查询关键词（FTS5 全文搜索） + tags: 标签过滤 + limit: 返回数量限制 + + Returns: + { + "results": [ + { + "path": "practices/moziplus-orchestration.md", + "title": "moziplus 编排实践", + "category": "practices", + "tags": ["moziplus", "orchestration"], + "summary": "...", + "lifecycle": "verified", + "updated_at": "2024-06-15T10:30:00", + "wikilink": "[[practices/moziplus-orchestration.md|moziplus 编排实践]]" + } + ], + "total": 5, + "query": "...", + "tags": [...] + } + """ + if tags is None: + tags = [] + + results = [] + + # 组合查询 + if query and tags: + # 先 FTS5 搜索，再标签过滤 + fts_results = await self.query_service.search(query, limit * 2) + filtered = [r for r in fts_results if any(t in r.tags for t in tags)] + pages = filtered[:limit] + elif query: + # 仅 FTS5 搜索 + pages = await self.query_service.search(query, limit) + elif tags: + # 仅标签搜索 + pages = await self.query_service.search_by_tags(tags[:5]) # 限制标签数量 + pages = pages[:limit] + else: + # 无查询条件，返回最近更新的页面 + pages = [] + + # 格式化结果 + for page in pages: + results.append({ + "path": page.path, + "title": page.title, + "category": page.category, + "tags": page.tags, + "summary": page.summary, + "lifecycle": page.lifecycle, + "updated_at": page.updated_at, + "wikilink": f"[[{page.path}|{page.title}]]" + }) + + return { + "results": results, + "total": len(results), + "query": query, + "tags": tags + } + + def get_schema(self) -> dict: + """返回 MCP Tool schema""" + return { + "name": "wiki_query", + "description": "查询 wiki 页面，支持关键词（FTS5）和标签搜索", + "inputSchema": { + "type": "object", + "properties": { + "query": { + "type": "string", + "description": "查询关键词（FTS5 全文搜索）" + }, + "tags": { + "type": "array", + "items": {"type": "string"}, + "description": "标签过滤" + }, + "limit": { + "type": "integer", + "default": 10, + "description": "返回数量限制" + } + } + } + } diff --git a/mcp_server/tools/wiki_status.py b/mcp_server/tools/wiki_status.py new file mode 100644 index 0000000..d4299af --- /dev/null +++ b/mcp_server/tools/wiki_status.py @@ -0,0 +1,75 @@ +""" +Tool Layer - wiki_status 工具 + +显示 wiki 当前状态（页面数、待处理项、增量差异）。 + +参考设计文档：第 2.2 节 +""" + +import logging +from typing import Dict, Any +from ..services import QueryService, IndexerService, GraphService + +logger = logging.getLogger(__name__) + + +class WikiStatusTool: + """wiki_status 工具实现""" + + def __init__(self, query_service: QueryService, indexer_service: IndexerService, graph_service: GraphService): + self.query_service = query_service + self.indexer_service = indexer_service + self.graph_service = graph_service + + async def handle(self) -> Dict[str, Any]: + """ + 显示 wiki 当前状态 + + Returns: + { + "stats": { + "total_pages": 1353, + "total_links": 4200, + "total_tags": 156, + "last_indexed": "2024-06-26T10:30:00" + }, + "orphans": ["path1", "path2"], + "orphans_count": 5, + "dirty_pages": 12, + "cache_stats": { + "size": 100, + "max_size": 1000, + "utilization": 0.1 + } + } + """ + # 获取统计信息 + stats = await self.query_service.get_stats() + + # 获取孤立页面 + orphans = await self.graph_service.find_orphans() + + # 获取索引状态 + index_status = await self.indexer_service.get_index_status() + + # 获取缓存统计 + cache_stats = await self.query_service.cache.get_stats() + + return { + "stats": stats, + "orphans": sorted(list(orphans)), + "orphans_count": len(orphans), + "dirty_pages": index_status.get("dirty_pages", 0), + "cache_stats": cache_stats + } + + def get_schema(self) -> dict: + """返回 MCP Tool schema""" + return { + "name": "wiki_status", + "description": "显示 wiki 当前状态（页面数、待处理项、增量差异）", + "inputSchema": { + "type": "object", + "properties": {} + } + } diff --git a/mcp_server/tools/wiki_synthesize.py b/mcp_server/tools/wiki_synthesize.py new file mode 100644 index 0000000..71e5e0a --- /dev/null +++ b/mcp_server/tools/wiki_synthesize.py @@ -0,0 +1,105 @@ +""" +Tool Layer - wiki_synthesize 工具 + +发现跨概念的综合分析机会，生成 synthesis 页面。 + +参考设计文档：第 2.2 节 +""" + +import logging +from typing import Dict, Any, List +from ..services import QueryService, GraphService + +logger = logging.getLogger(__name__) + + +class WikiSynthesizeTool: + """wiki_synthesize 工具实现""" + + def __init__(self, query_service: QueryService, graph_service: GraphService): + self.query_service = query_service + self.graph_service = graph_service + + async def handle(self, concepts: List[str] = None, threshold: float = 0.3) -> Dict[str, Any]: + """ + 发现跨概念的综合分析机会 + + Args: + concepts: 概念列表（空表示自动发现） + threshold: 相关性阈值 + + Returns: + { + "synthesis": [ + { + "concept_a": "SQLite", + "concept_b": "Concurrency", + "common_references": ["page1", "page2"], + "strength": 0.8, + "suggestion": "Consider creating synthesis page: SQLite & Concurrency" + } + ], + "total": 5 + } + """ + synthesis = [] + + # 如果未提供概念，自动发现高相关性概念 + if not concepts: + # 获取高度连接的页面作为候选 + highly_connected = await self.graph_service.get_highly_connected_pages(threshold=10) + concepts = [item["title"] for item in highly_connected[:10]] + + # 分析概念间的关联 + for i, concept_a in enumerate(concepts): + for concept_b in concepts[i + 1:]: + # 查找同时引用两个概念的页面 + pages_a = await self.query_service.search(concept_a, limit=50) + pages_b = await self.query_service.search(concept_b, limit=50) + + paths_a = {p.path for p in pages_a} + paths_b = {p.path for p in pages_b} + + common = paths_a & paths_b + + if len(common) >= 2: # 至少 2 个共同引用 + strength = len(common) / max(len(paths_a), len(paths_b)) + + if strength >= threshold: + synthesis.append({ + "concept_a": concept_a, + "concept_b": concept_b, + "common_references": sorted(list(common)), + "strength": round(strength, 2), + "suggestion": f"Consider creating synthesis page: {concept_a} & {concept_b}" + }) + + # 按关联强度排序 + synthesis.sort(key=lambda x: x["strength"], reverse=True) + + return { + "synthesis": synthesis[:20], + "total": len(synthesis) + } + + def get_schema(self) -> dict: + """返回 MCP Tool schema""" + return { + "name": "wiki_synthesize", + "description": "发现跨概念的综合分析机会，生成 synthesis 页面", + "inputSchema": { + "type": "object", + "properties": { + "concepts": { + "type": "array", + "items": {"type": "string"}, + "description": "概念列表（空表示自动发现）" + }, + "threshold": { + "type": "number", + "default": 0.3, + "description": "相关性阈值" + } + } + } + } diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..9acafec --- /dev/null +++ b/requirements.txt @@ -0,0 +1,13 @@ +# MCP Server 核心依赖 +mcp>=0.1.0 + +# 数据库 +aiosqlite>=0.19.0 + +# 配置和日志 +pyyaml>=6.0 + +# 开发依赖（可选） +# pytest>=7.0.0 +# pytest-asyncio>=0.21.0 +# pytest-cov>=4.0.0