claude_dev
62b8fef18a
- C1 (Critical): 添加 PyYAML 解析 frontmatter，支持多行值和列表 - M1 (Major): 添加 FTS5 查询验证，防止注入攻击 - M2/M3 (Major): 工具类使用显式 wiki_vault_path 参数 - M4 (Major): 集成 aiofiles 实现真正的异步文件 I/O - M5 (Major): 改用 logger.exception() 记录完整堆栈 Co-Authored-By: Claude Dev <noreply@anthropic.com>
267 lines
7.9 KiB
Python
267 lines
7.9 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
|
||
|
||
try:
|
||
import yaml
|
||
YAML_AVAILABLE = True
|
||
except ImportError:
|
||
YAML_AVAILABLE = False
|
||
logger = logging.getLogger(__name__)
|
||
logger.warning("PyYAML not available, using simple parser")
|
||
|
||
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（使用 PyYAML）
|
||
|
||
返回格式：
|
||
{
|
||
"name": "...",
|
||
"description": "...",
|
||
"metadata": {...},
|
||
"custom": {...} # 自定义字段
|
||
}
|
||
"""
|
||
frontmatter = {}
|
||
match = self.FRONTMATTER_PATTERN.match(content)
|
||
|
||
if match:
|
||
yaml_content = match.group(1)
|
||
|
||
# 使用 PyYAML 进行完整解析
|
||
if YAML_AVAILABLE:
|
||
try:
|
||
frontmatter = yaml.safe_load(yaml_content) or {}
|
||
except yaml.YAMLError as e:
|
||
logger.warning(f"Failed to parse YAML: {e}")
|
||
frontmatter = self._simple_parse(yaml_content)
|
||
else:
|
||
frontmatter = self._simple_parse(yaml_content)
|
||
|
||
return frontmatter
|
||
|
||
def _simple_parse(self, yaml_content: str) -> Dict:
|
||
"""简单 YAML 解析（fallback）"""
|
||
frontmatter = {}
|
||
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]
|
||
- YAML 列表格式
|
||
"""
|
||
tags = []
|
||
frontmatter = self.parse_frontmatter(content)
|
||
|
||
if "tags" in frontmatter:
|
||
tags_value = frontmatter["tags"]
|
||
|
||
if isinstance(tags_value, list):
|
||
tags = tags_value
|
||
elif isinstance(tags_value, str):
|
||
# 尝试解析字符串格式的标签
|
||
try:
|
||
tags = json.loads(tags_value) if tags_value.startswith('[') else tags_value.split(',')
|
||
except:
|
||
tags = [tags_value]
|
||
|
||
# 确保 tags 是字符串列表
|
||
return [str(tag).strip() for tag in tags if tag]
|
||
|
||
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
|