mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
split up parsing logic
This commit is contained in:
phernandez
@@ -1,130 +1,40 @@
|
||||
"""Parser for Basic Memory entity markdown files."""
|
||||
|
||||
import logging
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Optional, List, Tuple, Dict
|
||||
from typing import List
|
||||
|
||||
import frontmatter
|
||||
from markdown_it import MarkdownIt
|
||||
|
||||
from basic_memory.markdown.exceptions import ParseError
|
||||
from basic_memory.markdown.models import (
|
||||
from basic_memory.markdown.schemas import (
|
||||
Observation,
|
||||
Relation,
|
||||
Entity,
|
||||
EntityFrontmatter,
|
||||
EntityContent,
|
||||
EntityMetadata,
|
||||
)
|
||||
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def debug_sections(text):
|
||||
"""Debug helper to show section contents."""
|
||||
sections = text.split("---\n")
|
||||
logger.debug("File sections:")
|
||||
for i, section in enumerate(sections):
|
||||
logger.debug(f"\n=== Section {i} ===\n{section.strip()}\n")
|
||||
return sections
|
||||
|
||||
|
||||
class EntityParser:
|
||||
"""Parser for entity markdown files."""
|
||||
|
||||
def __init__(self):
|
||||
self.md = MarkdownIt()
|
||||
|
||||
def _parse_observation(self, content: str) -> Optional[Observation]:
|
||||
"""Parse an observation line."""
|
||||
try:
|
||||
if not content.strip():
|
||||
return None
|
||||
|
||||
# Check for unclosed bracket first
|
||||
if "[" in content and "]" not in content:
|
||||
raise ParseError("unclosed category")
|
||||
|
||||
# Parse category [type]
|
||||
match = re.match(r"^\s*(?:-\s*)?\[([^\]]*)\](.*)", content)
|
||||
if not match:
|
||||
raise ParseError("missing category")
|
||||
|
||||
category = match.group(1).strip()
|
||||
if not category:
|
||||
return None
|
||||
|
||||
content = match.group(2).strip()
|
||||
|
||||
# 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)
|
||||
|
||||
# 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()
|
||||
|
||||
return Observation(category=category, content=content, tags=tags, context=context)
|
||||
except ParseError:
|
||||
raise
|
||||
except Exception:
|
||||
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
|
||||
|
||||
# Check for unclosed [[
|
||||
if "[[" in content and "]]" not in content:
|
||||
raise ParseError("missing ]]")
|
||||
|
||||
# Find the link
|
||||
match = re.search(r"\[\[([^\]]+)\]\]", content)
|
||||
if not match:
|
||||
raise ParseError("missing [[")
|
||||
|
||||
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 ParseError:
|
||||
raise
|
||||
except Exception:
|
||||
logger.exception("Failed to parse relation: %s", content)
|
||||
return None
|
||||
|
||||
def _parse_metadata_line(self, line: str) -> Optional[Tuple[str, str]]:
|
||||
"""Parse a metadata line into key-value pair."""
|
||||
if not line.strip():
|
||||
return None
|
||||
try:
|
||||
# Split on first colon
|
||||
if ":" not in line:
|
||||
return None
|
||||
key, value = line.split(":", 1)
|
||||
return key.strip(), value.strip()
|
||||
except ValueError:
|
||||
return None
|
||||
self.debug = False
|
||||
|
||||
def parse_file(self, path: Path, encoding: str = "utf-8") -> Entity:
|
||||
"""Parse an entity markdown file."""
|
||||
@@ -132,117 +42,109 @@ class EntityParser:
|
||||
raise ParseError(f"File does not exist: {path}")
|
||||
|
||||
try:
|
||||
# Read and parse frontmatter
|
||||
# Read file content
|
||||
with open(path, "r", encoding=encoding) as f:
|
||||
content = f.read()
|
||||
post = frontmatter.loads(content)
|
||||
raw_content = f.read()
|
||||
|
||||
# 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)
|
||||
# Split into sections and debug
|
||||
sections = debug_sections(raw_content)
|
||||
|
||||
# Parse markdown
|
||||
tokens = self.md.parse(post.content)
|
||||
if len(sections) < 4: # Needs at least empty,frontmatter,content,empty for no metadata
|
||||
raise ParseError("Missing required document sections")
|
||||
|
||||
# State for parsing
|
||||
# Parse frontmatter (first yaml section)
|
||||
frontmatter_text = sections[1].strip()
|
||||
frontmatter_data = {}
|
||||
|
||||
for line in frontmatter_text.split("\n"):
|
||||
if ":" in line:
|
||||
key, value = line.split(":", 1)
|
||||
frontmatter_data[key.strip()] = value.strip()
|
||||
|
||||
if isinstance(frontmatter_data.get("tags"), str):
|
||||
frontmatter_data["tags"] = [t.strip() for t in frontmatter_data["tags"].split(",")]
|
||||
|
||||
frontmatter = EntityFrontmatter(**frontmatter_data)
|
||||
|
||||
# Parse markdown content (middle section)
|
||||
content_tokens = self.md.parse(sections[2].strip())
|
||||
|
||||
# State for content parsing
|
||||
title = ""
|
||||
description = ""
|
||||
observations: List[Observation] = []
|
||||
relations: List[Relation] = []
|
||||
context = ""
|
||||
metadata = {}
|
||||
|
||||
current_section = None
|
||||
description_tokens = []
|
||||
|
||||
# Track list item state
|
||||
# Track list items
|
||||
in_list_item = False
|
||||
list_item_tokens = []
|
||||
list_item_level = None
|
||||
base_list_level = None
|
||||
|
||||
# Track metadata state
|
||||
in_metadata_para = False
|
||||
|
||||
for token in tokens:
|
||||
for token in content_tokens:
|
||||
if token.type == "heading_open":
|
||||
if token.tag == "h1":
|
||||
current_section = "title"
|
||||
elif token.tag == "h2":
|
||||
# Handle previous section
|
||||
if current_section == "description":
|
||||
description = " ".join(t.content for t in description_tokens)
|
||||
current_section = "section_name"
|
||||
description_tokens = []
|
||||
|
||||
|
||||
elif token.type == "inline":
|
||||
content = token.content.strip()
|
||||
|
||||
|
||||
if current_section == "title":
|
||||
title = content
|
||||
current_section = "description"
|
||||
elif current_section == "section_name":
|
||||
section = content.lower()
|
||||
current_section = section
|
||||
current_section = content.lower()
|
||||
elif current_section == "description":
|
||||
description_tokens.append(token)
|
||||
elif current_section == "metadata":
|
||||
if parsed := self._parse_metadata_line(content):
|
||||
metadata[parsed[0]] = parsed[1]
|
||||
if description:
|
||||
description += " "
|
||||
description += content
|
||||
elif in_list_item:
|
||||
list_item_tokens.append(token)
|
||||
|
||||
|
||||
elif token.type == "list_item_open":
|
||||
in_list_item = True
|
||||
list_item_level = token.level
|
||||
if base_list_level is None:
|
||||
base_list_level = token.level
|
||||
list_item_tokens = []
|
||||
|
||||
elif token.type == "list_item_close" and in_list_item:
|
||||
# Only process top-level items
|
||||
if base_list_level is None or list_item_level == base_list_level:
|
||||
item_content = " ".join(t.content for t in list_item_tokens)
|
||||
try:
|
||||
if current_section == "observations":
|
||||
if obs := self._parse_observation(item_content):
|
||||
observations.append(obs)
|
||||
elif current_section == "relations":
|
||||
if rel := self._parse_relation(item_content):
|
||||
relations.append(rel)
|
||||
except ParseError:
|
||||
# Skip malformed items
|
||||
pass
|
||||
|
||||
elif token.type == "list_item_close":
|
||||
item_content = " ".join(t.content for t in list_item_tokens)
|
||||
try:
|
||||
if current_section == "observations":
|
||||
if obs := Observation.parse_observation(item_content):
|
||||
observations.append(obs)
|
||||
elif current_section == "relations":
|
||||
if rel := Relation.parse_relation(item_content):
|
||||
relations.append(rel)
|
||||
except ParseError:
|
||||
# Skip malformed items
|
||||
pass
|
||||
in_list_item = False
|
||||
list_item_tokens = []
|
||||
|
||||
elif token.type == "paragraph_open":
|
||||
in_metadata_para = current_section == "metadata"
|
||||
|
||||
elif token.type == "paragraph_close":
|
||||
in_metadata_para = False
|
||||
|
||||
# Handle any remaining description
|
||||
if current_section == "description" and description_tokens:
|
||||
description = " ".join(t.content for t in description_tokens)
|
||||
|
||||
# Create entity
|
||||
content_data = EntityContent(
|
||||
# Create content object
|
||||
content = EntityContent(
|
||||
title=title,
|
||||
description=description,
|
||||
observations=observations,
|
||||
relations=relations,
|
||||
context=context,
|
||||
metadata=metadata,
|
||||
)
|
||||
|
||||
return Entity(frontmatter=frontmatter_data, content=content_data)
|
||||
# Parse metadata (final section if exists)
|
||||
metadata = {}
|
||||
if len(sections) >= 5: # Has backmatter section
|
||||
metadata_text = sections[4].strip()
|
||||
logger.debug(f"Metadata text: {metadata_text}")
|
||||
for line in metadata_text.split("\n"):
|
||||
if ":" in line:
|
||||
key, value = line.split(":", 1)
|
||||
metadata[key.strip()] = value.strip()
|
||||
|
||||
metadata_obj = EntityMetadata(metadata=metadata)
|
||||
|
||||
return Entity(frontmatter=frontmatter, content=content, metadata=metadata_obj)
|
||||
|
||||
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
|
||||
raise ParseError(f"Failed to parse {path}: {str(e)}") from e
|
||||
|
||||
@@ -0,0 +1,18 @@
|
||||
from basic_memory.markdown.schemas.entity import (
|
||||
Entity,
|
||||
EntityFrontmatter,
|
||||
EntityContent,
|
||||
EntityMetadata,
|
||||
)
|
||||
from basic_memory.markdown.schemas.observation import Observation
|
||||
from basic_memory.markdown.schemas.relation import Relation
|
||||
|
||||
|
||||
__all__ = [
|
||||
"Entity",
|
||||
"EntityFrontmatter",
|
||||
"EntityContent",
|
||||
"EntityMetadata",
|
||||
"Observation",
|
||||
"Relation",
|
||||
]
|
||||
@@ -10,23 +10,6 @@ logging.basicConfig(level=logging.DEBUG)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Observation(BaseModel):
|
||||
"""An observation about an entity."""
|
||||
|
||||
category: str
|
||||
content: str
|
||||
tags: List[str]
|
||||
context: Optional[str] = None
|
||||
|
||||
|
||||
class Relation(BaseModel):
|
||||
"""A relation between entities."""
|
||||
|
||||
target: str # The entity being linked to
|
||||
type: str # The type of relation
|
||||
context: Optional[str] = None
|
||||
|
||||
|
||||
class EntityFrontmatter(BaseModel):
|
||||
"""Required frontmatter fields for an entity."""
|
||||
|
||||
@@ -57,5 +40,5 @@ class Entity(BaseModel):
|
||||
"""Complete entity combining frontmatter, content, and metadata."""
|
||||
|
||||
frontmatter: EntityFrontmatter
|
||||
content: EntityContent
|
||||
metadata: EntityMetadata = EntityMetadata()
|
||||
content: EntityContent
|
||||
metadata: EntityMetadata = EntityMetadata()
|
||||
@@ -0,0 +1,68 @@
|
||||
"""Models for the markdown parser."""
|
||||
|
||||
import logging
|
||||
import re
|
||||
from typing import List, Optional
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
from basic_memory.markdown import ParseError
|
||||
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Observation(BaseModel):
|
||||
"""An observation about an entity."""
|
||||
|
||||
category: str
|
||||
content: str
|
||||
tags: List[str]
|
||||
context: Optional[str] = None
|
||||
|
||||
@classmethod
|
||||
def parse_observation(cls, content: str) -> Optional["Observation"]:
|
||||
"""Parse an observation line."""
|
||||
try:
|
||||
if not content.strip():
|
||||
return None
|
||||
|
||||
# Parse category [type]
|
||||
match = re.match(r"^\s*(?:-\s*)?\[([^\]]*)\](.*)", content)
|
||||
if not match:
|
||||
raise ParseError("missing category")
|
||||
|
||||
category = match.group(1).strip()
|
||||
if not category:
|
||||
return None
|
||||
|
||||
content = match.group(2).strip()
|
||||
|
||||
# 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)
|
||||
|
||||
# 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()
|
||||
|
||||
return Observation(category=category, content=content, tags=tags, context=context)
|
||||
except ParseError:
|
||||
raise
|
||||
except Exception:
|
||||
logger.exception("Failed to parse observation: %s", content)
|
||||
return None
|
||||
@@ -0,0 +1,57 @@
|
||||
"""Models for the markdown parser."""
|
||||
|
||||
import logging
|
||||
import re
|
||||
from typing import Optional
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
from basic_memory.markdown import ParseError
|
||||
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Relation(BaseModel):
|
||||
"""A relation between entities."""
|
||||
|
||||
target: str # The entity being linked to
|
||||
type: str # The type of relation
|
||||
context: Optional[str] = None
|
||||
|
||||
@classmethod
|
||||
def parse_relation(cls, content: str) -> Optional["Relation"]:
|
||||
"""Parse a relation line."""
|
||||
try:
|
||||
if not content.strip():
|
||||
return None
|
||||
|
||||
# Check for unclosed [[
|
||||
if "[[" in content and "]]" not in content:
|
||||
raise ParseError("missing ]]")
|
||||
|
||||
# Find the link
|
||||
match = re.search(r"\[\[([^\]]+)\]\]", content)
|
||||
if not match:
|
||||
raise ParseError("missing [[")
|
||||
|
||||
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 ParseError:
|
||||
raise
|
||||
except Exception:
|
||||
logger.exception("Failed to parse relation: %s", content)
|
||||
return None
|
||||
Reference in New Issue
Block a user