consolidate mardown schemas

This commit is contained in:
phernandez
2024-12-22 16:27:20 -06:00
parent 017e155a54
commit 3d02827d95
5 changed files with 28 additions and 176 deletions
+6 -6
View File
@@ -61,13 +61,13 @@ class EntityParser(MarkdownParser[Entity]):
logger.error(f"Invalid entity frontmatter: {e}")
raise ParseError(f"Invalid entity frontmatter: {str(e)}") from e
async def parse_content(self, title: str, sections: Dict[str, List[str]]) -> EntityContent:
async def parse_content(self, title: str, sections: Dict[str, str]) -> EntityContent:
"""
Parse entity content section.
Args:
title: Document title
sections: Section name -> list of lines mapping
sections: Section name -> content mapping
Returns:
Parsed EntityContent
@@ -79,14 +79,14 @@ class EntityParser(MarkdownParser[Entity]):
# Get description (if any)
description = None
if "description" in sections:
description = " ".join(sections["description"])
description = sections["description"]
# Parse observations (required)
observations = []
if "observations" not in sections:
raise ParseError("Missing required observations section")
for line in sections["observations"]:
for line in sections["observations"].splitlines():
if line and not line.isspace():
observation = await self._parse_observation(line)
if observation:
@@ -95,7 +95,7 @@ class EntityParser(MarkdownParser[Entity]):
# Parse relations (optional)
relations = []
if "relations" in sections:
for line in sections["relations"]:
for line in sections["relations"].splitlines():
if line and not line.isspace():
relation = await self._parse_relation(line)
if relation:
@@ -204,7 +204,7 @@ class EntityParser(MarkdownParser[Entity]):
return None
return Relation(
relation_type=relation_type,
type=relation_type,
target=target,
context=context
)
@@ -1,16 +1,31 @@
"""Schema models for entity markdown files."""
from datetime import datetime
from typing import List, Optional, Dict, Any
from pydantic import BaseModel
from basic_memory.markdown.schemas.observation import Observation
from basic_memory.markdown.schemas.relation import Relation
from basic_memory.utils.file_utils import ParseError
class Observation(BaseModel):
"""An observation about an entity."""
category: Optional[str] = None
content: str
tags: Optional[List[str]] = None
context: Optional[str] = None
class Relation(BaseModel):
"""A relation between entities."""
type: str
target: str
context: Optional[str] = None
class EntityFrontmatter(BaseModel):
"""Required frontmatter fields for an entity."""
type: str
id: str
created: datetime
@@ -20,6 +35,7 @@ class EntityFrontmatter(BaseModel):
class EntityContent(BaseModel):
"""Content sections of an entity markdown file."""
title: str
description: Optional[str] = None
observations: List[Observation] = []
@@ -29,11 +45,13 @@ class EntityContent(BaseModel):
class EntityMetadata(BaseModel):
"""Optional metadata for an entity."""
metadata: Dict[str, Any] = {}
class Entity(BaseModel):
"""Complete entity combining frontmatter, content, and metadata."""
frontmatter: EntityFrontmatter
content: EntityContent
metadata: EntityMetadata = EntityMetadata()
metadata: EntityMetadata = EntityMetadata()
@@ -1,13 +0,0 @@
"""Schema models for markdown parsing."""
from basic_memory.markdown.schemas.entity import Entity, EntityContent, EntityFrontmatter, EntityMetadata
from basic_memory.markdown.schemas.observation import Observation
from basic_memory.markdown.schemas.relation import Relation
__all__ = [
'Entity',
'EntityContent',
'EntityFrontmatter',
'EntityMetadata',
'Observation',
'Relation',
]
@@ -1,85 +0,0 @@
"""Models for the markdown parser."""
import logging
import re
from typing import List, Optional
from pydantic import BaseModel
from basic_memory.utils.file_utils import ParseError
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 from_line(cls, line: str) -> Optional["Observation"]:
"""
Parse an observation from a line.
Format must be:
- [category] Content text #tag1 #tag2 (optional context)
"""
try:
# Skip blank lines
if not line.strip():
return None
# Remove leading/trailing whitespace and bullet
line = line.strip()
if not line.startswith("-"):
return None
line = line[1:].strip()
# Parse category [category]
match = re.match(r"^\[([^\]]+)\](.*)", line)
if not match:
raise ParseError("Invalid format - must start with '[category]'")
category = match.group(1).strip()
if not category:
raise ParseError("Category cannot be empty")
rest = match.group(2).strip()
# Parse content and tags
content_parts = []
tags = []
context = None
# Extract context if exists
if rest.endswith(")"):
context_start = rest.rfind("(")
if context_start > 0:
context = rest[context_start + 1:-1].strip()
rest = rest[:context_start].strip()
# Split remaining text and collect tags
for word in rest.split():
if word.startswith("#"):
tag = word[1:].strip()
if tag:
tags.append(tag)
else:
content_parts.append(word)
content = " ".join(content_parts)
if not content:
raise ParseError("Content cannot be empty")
return cls(
category=category,
content=content,
tags=tags,
context=context,
)
except Exception as e:
if not isinstance(e, ParseError):
raise ParseError(f"Failed to parse observation: {line}: {str(e)}") from e
raise
@@ -1,68 +0,0 @@
"""Models for the markdown parser."""
import logging
import re
from typing import Optional
from pydantic import BaseModel
from basic_memory.utils.file_utils import ParseError
logger = logging.getLogger(__name__)
class Relation(BaseModel):
"""A relation between entities."""
type: str
target: str
context: Optional[str] = None
@classmethod
def from_line(cls, line: str) -> Optional["Relation"]:
"""
Parse a relation from a line.
Format must be:
- relation_type [[Target Entity]] (optional context)
"""
try:
# Skip blank lines
if not line.strip():
return None
# Remove leading/trailing whitespace and bullet
line = line.strip()
if not line.startswith("-"):
return None
line = line[1:].strip()
# First, extract any context from parens at end
context = None
if line.endswith(")"):
context_start = line.rfind("(")
if context_start > 0:
context = line[context_start + 1:-1].strip()
line = line[:context_start].strip()
# Look for relation_type [[target]]
match = re.match(r"^(\w+)\s+\[\[([^\]]+)\]\]", line)
if not match:
raise ParseError("Invalid format - must be 'relation_type [[Target]]'")
rel_type = match.group(1).strip()
target = match.group(2).strip()
if not rel_type:
raise ParseError("Relation type cannot be empty")
if not target:
raise ParseError("Target cannot be empty")
return cls(
type=rel_type,
target=target,
context=context
)
except Exception as e:
if not isinstance(e, ParseError):
raise ParseError(f"Failed to parse relation: {line}: {str(e)}") from e
raise