Files
sanguo_llmwiki/mcp_server/services/parser.py
T
claude_dev dfd8421dc6 feat(§03): 实现 MCP Server 核心 - Storage/Service/Tool/MCP Protocol Layers
**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) <noreply@anthropic.com>
2026-06-26 11:32:43 +08:00

240 lines
7.0 KiB
Python

"""
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