sanguo_llmwiki/mcp_server/tools/wiki_lint.py

"""
Tool Layer - wiki_lint 工具

审计 wiki 健康（格式、链接、frontmatter 规范）。

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

import logging
import os
from typing import Dict, Any, List
from pathlib import Path
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, wiki_vault_path: str = None):
        self.parser = parser
        self.query_service = query_service
        self.graph_service = graph_service
        # 优先使用传入的路径，否则使用环境变量
        self.wiki_vault_path = Path(wiki_vault_path or os.environ.get("WIKI_VAULT_PATH", "/Volumes/KnowledgeBase/wiki-vault"))

    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:
                full_path = self.wiki_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": "检查级别"
                    }
                }
            }
        }