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