mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
markdown parsing
This commit is contained in:
phernandez
+127
-134
@@ -1,6 +1,6 @@
|
||||
"""
|
||||
Parser for Basic Memory entity markdown files.
|
||||
"""
|
||||
"""Parser for Basic Memory entity markdown files."""
|
||||
import re
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
@@ -9,6 +9,9 @@ import frontmatter
|
||||
from markdown_it import MarkdownIt
|
||||
from pydantic import BaseModel
|
||||
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class ParseError(Exception):
|
||||
"""Raised when parsing fails"""
|
||||
pass
|
||||
@@ -29,6 +32,7 @@ class Relation(BaseModel):
|
||||
class EntityFrontmatter(BaseModel):
|
||||
"""Frontmatter metadata for an entity."""
|
||||
type: str
|
||||
id: str
|
||||
created: datetime
|
||||
modified: datetime
|
||||
tags: List[str]
|
||||
@@ -63,119 +67,106 @@ class EntityParser:
|
||||
def __init__(self):
|
||||
self.md = MarkdownIt()
|
||||
|
||||
def _parse_observation(self, line: str) -> Observation:
|
||||
"""Parse a single observation line.
|
||||
|
||||
Format: - [category] content #tag1 #tag2 "optional context"
|
||||
"""
|
||||
# Remove leading "- " if present
|
||||
content = line.strip()
|
||||
if content.startswith("- "):
|
||||
content = content[2:].strip()
|
||||
|
||||
# Extract category
|
||||
if not content.startswith("["):
|
||||
raise ParseError(f"Invalid observation format, missing category: {line}")
|
||||
category_end = content.find("]")
|
||||
if category_end == -1:
|
||||
raise ParseError(f"Invalid observation format, unclosed category: {line}")
|
||||
category = content[1:category_end].strip()
|
||||
|
||||
# Remove category from content
|
||||
content = content[category_end + 1:].strip()
|
||||
|
||||
# Extract tags
|
||||
tags = []
|
||||
words = content.split()
|
||||
filtered_words = []
|
||||
for word in words:
|
||||
if word.startswith("#"):
|
||||
tags.append(word[1:]) # Remove # from tag
|
||||
else:
|
||||
filtered_words.append(word)
|
||||
content = " ".join(filtered_words)
|
||||
|
||||
# Extract context if present
|
||||
context = None
|
||||
if content.endswith('"'):
|
||||
last_quote = content.rfind('"')
|
||||
second_last_quote = content.rfind('"', 0, last_quote)
|
||||
if second_last_quote != -1:
|
||||
context = content[second_last_quote + 1:last_quote]
|
||||
content = content[:second_last_quote].strip()
|
||||
|
||||
return Observation(
|
||||
category=category,
|
||||
content=content,
|
||||
tags=tags,
|
||||
context=context
|
||||
)
|
||||
def _parse_observation(self, content: str) -> Optional[Observation]:
|
||||
"""Parse an observation line."""
|
||||
try:
|
||||
if not content.strip():
|
||||
return None
|
||||
|
||||
def _parse_relation(self, line: str) -> Relation:
|
||||
"""Parse a single relation line.
|
||||
|
||||
Format: - [[Entity]] #relation_type "optional context"
|
||||
"""
|
||||
# Remove leading "- " if present
|
||||
content = line.strip()
|
||||
if content.startswith("- "):
|
||||
content = content[2:].strip()
|
||||
|
||||
# Extract target
|
||||
if not content.startswith("[["):
|
||||
raise ParseError(f"Invalid relation format, missing [[ : {line}")
|
||||
link_end = content.find("]]")
|
||||
if link_end == -1:
|
||||
raise ParseError(f"Invalid relation format, missing ]] : {line}")
|
||||
target = content[2:link_end].strip()
|
||||
|
||||
# Move past ]]
|
||||
content = content[link_end + 2:].strip()
|
||||
|
||||
# Extract relation type
|
||||
if not content.startswith("#"):
|
||||
raise ParseError(f"Invalid relation format, missing relation type: {line}")
|
||||
|
||||
words = content.split()
|
||||
rel_type = words[0][1:] # Remove # from type
|
||||
|
||||
# Extract context if present
|
||||
context = None
|
||||
remaining = " ".join(words[1:])
|
||||
if remaining:
|
||||
if remaining.startswith('"') and remaining.endswith('"'):
|
||||
context = remaining[1:-1]
|
||||
|
||||
return Relation(
|
||||
target=target,
|
||||
type=rel_type,
|
||||
context=context
|
||||
)
|
||||
# Parse category [type]
|
||||
match = re.match(r'^\s*(?:-\s*)?\[([^\]]+)\](.*)', content)
|
||||
if not match:
|
||||
return None
|
||||
category = match.group(1).strip()
|
||||
content = match.group(2).strip()
|
||||
|
||||
def _parse_metadata_line(self, line: str) -> tuple[str, str]:
|
||||
"""Parse a single metadata line."""
|
||||
if ":" not in line:
|
||||
return None, None
|
||||
# Parse tags and content
|
||||
tags = []
|
||||
words = []
|
||||
for word in content.split():
|
||||
if word.startswith('#'):
|
||||
# Handle #tag1#tag2#tag3
|
||||
for tag in word.lstrip('#').split('#'):
|
||||
if tag:
|
||||
tags.append(tag)
|
||||
else:
|
||||
words.append(word)
|
||||
|
||||
content = ' '.join(words)
|
||||
|
||||
key, value = line.split(":", 1)
|
||||
return key.strip(), value.strip()
|
||||
# Extract context in parentheses
|
||||
context = None
|
||||
if content.endswith(')'):
|
||||
ctx_start = content.rfind('(')
|
||||
if ctx_start != -1:
|
||||
context = content[ctx_start + 1:-1].strip()
|
||||
content = content[:ctx_start].strip()
|
||||
|
||||
def parse_file(self, path: Path) -> Entity:
|
||||
return Observation(
|
||||
category=category,
|
||||
content=content,
|
||||
tags=tags,
|
||||
context=context
|
||||
)
|
||||
except Exception as e:
|
||||
logger.exception("Failed to parse observation: %s", content)
|
||||
return None
|
||||
|
||||
def _parse_relation(self, content: str) -> Optional[Relation]:
|
||||
"""Parse a relation line."""
|
||||
try:
|
||||
if not content.strip():
|
||||
return None
|
||||
|
||||
# Find the link
|
||||
match = re.search(r'\[\[([^\]]+)\]\]', content)
|
||||
if not match:
|
||||
return None
|
||||
|
||||
target = match.group(1).strip()
|
||||
before_link = content[:match.start()].strip(' -')
|
||||
after_link = content[match.end():].strip()
|
||||
|
||||
# Everything before the link is the type
|
||||
rel_type = before_link.strip()
|
||||
if not rel_type:
|
||||
return None
|
||||
|
||||
# Check for context in parentheses
|
||||
context = None
|
||||
if after_link.startswith('(') and after_link.endswith(')'):
|
||||
context = after_link[1:-1].strip()
|
||||
|
||||
return Relation(
|
||||
target=target,
|
||||
type=rel_type,
|
||||
context=context
|
||||
)
|
||||
except Exception as e:
|
||||
logger.exception("Failed to parse relation: %s", content)
|
||||
return None
|
||||
|
||||
def parse_file(self, path: Path, encoding: str = 'utf-8') -> Entity:
|
||||
"""Parse an entity markdown file."""
|
||||
if not path.exists():
|
||||
raise ParseError(f"File does not exist: {path}")
|
||||
|
||||
try:
|
||||
# Parse frontmatter and content
|
||||
post = frontmatter.load(path)
|
||||
# Read and parse frontmatter
|
||||
with open(path, 'r', encoding=encoding) as f:
|
||||
content = f.read()
|
||||
post = frontmatter.loads(content)
|
||||
|
||||
# Parse frontmatter
|
||||
frontmatter_data = EntityFrontmatter(**post.metadata)
|
||||
# Handle frontmatter
|
||||
metadata = dict(post.metadata)
|
||||
if isinstance(metadata.get('tags'), str):
|
||||
metadata['tags'] = [t.strip() for t in metadata['tags'].split(',')]
|
||||
frontmatter_data = EntityFrontmatter(**metadata)
|
||||
|
||||
# Parse markdown
|
||||
tokens = self.md.parse(post.content)
|
||||
|
||||
# Extract title, description, observations, etc
|
||||
# State for parsing
|
||||
title = ""
|
||||
description = ""
|
||||
observations = []
|
||||
@@ -184,43 +175,42 @@ class EntityParser:
|
||||
metadata = {}
|
||||
|
||||
current_section = None
|
||||
collecting_description = False
|
||||
current_list = []
|
||||
in_list = False
|
||||
|
||||
# Process tokens
|
||||
for token in tokens:
|
||||
if token.type == "heading_open" and token.tag == "h1":
|
||||
# Next token will be title
|
||||
current_section = "title"
|
||||
elif token.type == "heading_open" and token.tag == "h2":
|
||||
# Next token will be section name
|
||||
current_section = "section_name"
|
||||
collecting_description = False
|
||||
elif token.type == "inline":
|
||||
if current_section == "title":
|
||||
if token.type == 'heading_open':
|
||||
if token.tag == 'h1':
|
||||
current_section = 'title'
|
||||
elif token.tag == 'h2':
|
||||
current_section = 'section_name'
|
||||
elif token.type == 'inline':
|
||||
if current_section == 'title':
|
||||
title = token.content
|
||||
current_section = None
|
||||
elif current_section == "section_name":
|
||||
section_name = token.content.lower()
|
||||
if section_name == "description":
|
||||
collecting_description = True
|
||||
current_section = section_name
|
||||
elif collecting_description:
|
||||
elif current_section == 'section_name':
|
||||
section = token.content.lower()
|
||||
current_section = section
|
||||
if section in ['observations', 'relations']:
|
||||
in_list = False
|
||||
current_list = []
|
||||
elif current_section == 'description':
|
||||
description = token.content
|
||||
elif current_section == "observations":
|
||||
if token.content.strip(): # Skip empty lines
|
||||
observations.append(self._parse_observation(token.content))
|
||||
elif current_section == "relations":
|
||||
if token.content.strip(): # Skip empty lines
|
||||
relations.append(self._parse_relation(token.content))
|
||||
elif current_section == "context":
|
||||
elif current_section == 'observations' and token.content.strip():
|
||||
if obs := self._parse_observation(token.content):
|
||||
observations.append(obs)
|
||||
elif current_section == 'relations' and token.content.strip():
|
||||
if rel := self._parse_relation(token.content):
|
||||
relations.append(rel)
|
||||
elif current_section == 'context':
|
||||
context = token.content
|
||||
elif current_section == "metadata":
|
||||
# Process each line of metadata separately
|
||||
for line in token.content.split("\n"):
|
||||
key, value = self._parse_metadata_line(line.strip())
|
||||
if key and value:
|
||||
metadata[key] = value
|
||||
|
||||
# Create EntityContent
|
||||
elif token.type == 'bullet_list_open':
|
||||
in_list = True
|
||||
elif token.type == 'bullet_list_close':
|
||||
in_list = False
|
||||
|
||||
# Create entity
|
||||
content_data = EntityContent(
|
||||
title=title,
|
||||
description=description,
|
||||
@@ -230,11 +220,14 @@ class EntityParser:
|
||||
metadata=metadata
|
||||
)
|
||||
|
||||
# Return complete Entity
|
||||
return Entity(
|
||||
frontmatter=frontmatter_data,
|
||||
content=content_data
|
||||
)
|
||||
|
||||
except UnicodeError as e:
|
||||
if encoding == 'utf-8':
|
||||
return self.parse_file(path, encoding='utf-16')
|
||||
raise ParseError(f"Failed to read {path} with encoding {encoding}: {str(e)}")
|
||||
except Exception as e:
|
||||
raise ParseError(f"Failed to parse {path}: {str(e)}") from e
|
||||
Reference in New Issue
Block a user