sanguo_llmwiki/mcp_server/services/query.py

"""
Service Layer - QueryService（查询服务）

负责所有查询逻辑，包括 FTS5 全文搜索、标签搜索、页面查询等。

参考设计文档：第 2.3 节
"""

import logging
import re
from typing import List, Set, Optional, Dict
from ..storage import Database, WikiPage
from .cache import CacheService

logger = logging.getLogger(__name__)


# 中文同义词映射（用于扩展搜索查询）
CHINESE_SYNONYMS: Dict[str, List[str]] = {
    "多智能体": ["multi-agent", "多agent", "multi agent"],
    "编排": ["orchestration", "调度"],
    "协作": ["collaboration", "协同"],
    "工具": ["tool", "工具调用"],
    "模式": ["pattern", "设计模式"],
    "实践": ["practice", "最佳实践"],
    "架构": ["architecture", "系统架构"],
    "代理": ["agent", "智能体"],
    "大模型": ["llm", "language model", "语言模型"],
    "向量": ["vector", "embedding", "嵌入"],
    "检索": ["retrieval", "search", "搜索"],
    "生成": ["generation", "生成式"],
    "推理": ["inference", "reasoning"],
}


class QueryService:
    """查询服务 - 负责所有查询逻辑"""

    def __init__(self, db: Database, cache: CacheService):
        self.db = db
        self.cache = cache

    def _expand_query_with_synonyms(self, query: str) -> str:
        """使用中文同义词扩展查询

        例如："多智能体编排" → "多智能体编排 multi-agent orchestration"
        """
        expanded_terms = []
        query_lower = query.lower()

        # 检查每个中文词是否有对应的英文同义词
        for chinese_term, english_synonyms in CHINESE_SYNONYMS.items():
            if chinese_term in query:
                expanded_terms.extend(english_synonyms)

        if expanded_terms:
            # 构建扩展查询（OR 连接）
            expanded_query = f"{query} OR {' OR '.join(expanded_terms)}"
            logger.debug(f"Expanded query: {query} → {expanded_query}")
            return expanded_query

        return query

    async def search(self, query: str, limit: int = 10, use_synonyms: bool = True) -> List[WikiPage]:
        """FTS5 全文搜索

        Args:
            query: 搜索查询
            limit: 返回结果数量限制
            use_synonyms: 是否使用中文同义词扩展
        """
        # 1. 检查缓存
        cache_key = f"search:{query}:{limit}:{use_synonyms}"
        cached = await self.cache.get(cache_key)
        if cached:
            logger.debug(f"Cache hit for search: {query}")
            return cached

        # 2. 查询扩展（中文同义词）
        search_query = query
        if use_synonyms:
            search_query = self._expand_query_with_synonyms(query)

        # 3. FTS5 搜索
        results = await self.db.fts_search(search_query, limit)

        # 4. 缓存结果
        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:
        """使缓存失效

        如果 pattern 是具体路径（不是通配符），自动尝试匹配所有缓存前缀：
        - page:{pattern}
        - links:{pattern}
        - search:{pattern}
        """
        # 清除所有缓存
        if pattern == "*":
            return await self.cache.invalidate(pattern)

        # 对于具体路径，尝试所有可能的前缀
        prefixes = ["page:", "links:", "search:"]
        total_invalidated = 0

        for prefix in prefixes:
            count = await self.cache.invalidate(f"{prefix}{pattern}")
            total_invalidated += count

        # 也尝试原始模式（兼容直接传入完整 key 的情况）
        total_invalidated += await self.cache.invalidate(pattern)

        return total_invalidated