From 9e3f71cb87bcea01c4c494886c67fcfbaa688c4f Mon Sep 17 00:00:00 2001 From: phernandez Date: Thu, 20 Feb 2025 22:29:40 -0600 Subject: [PATCH] file sync WIP --- src/basic_memory/markdown/entity_parser.py | 6 +- src/basic_memory/models/knowledge.py | 5 + .../repository/search_repository.py | 17 +- src/basic_memory/services/entity_service.py | 8 +- src/basic_memory/services/file_service.py | 55 ++++- src/basic_memory/services/search_service.py | 54 ++++- src/basic_memory/sync/sync_service.py | 219 ++++++++++-------- src/basic_memory/sync/sync_service.py.bak | 174 ++++++++++++++ tests/conftest.py | 2 + tests/sync/test_sync_service.py | 2 +- 10 files changed, 419 insertions(+), 123 deletions(-) create mode 100644 src/basic_memory/sync/sync_service.py.bak diff --git a/src/basic_memory/markdown/entity_parser.py b/src/basic_memory/markdown/entity_parser.py index 2fa73c85..da628d22 100644 --- a/src/basic_memory/markdown/entity_parser.py +++ b/src/basic_memory/markdown/entity_parser.py @@ -88,10 +88,10 @@ class EntityParser: return parsed return None - async def parse_file(self, file_path: Path) -> EntityMarkdown: + async def parse_file(self, path: Path | str) -> EntityMarkdown: """Parse markdown file into EntityMarkdown.""" - absolute_path = self.base_path / file_path + absolute_path = self.base_path / path # Parse frontmatter and content using python-frontmatter post = frontmatter.load(str(absolute_path)) @@ -99,7 +99,7 @@ class EntityParser: file_stats = absolute_path.stat() metadata = post.metadata - metadata["title"] = post.metadata.get("title", file_path.name) + metadata["title"] = post.metadata.get("title", absolute_path.name) metadata["type"] = post.metadata.get("type", "note") metadata["tags"] = parse_tags(post.metadata.get("tags", [])) diff --git a/src/basic_memory/models/knowledge.py b/src/basic_memory/models/knowledge.py index 5d1f0573..cf9f3cbb 100644 --- a/src/basic_memory/models/knowledge.py +++ b/src/basic_memory/models/knowledge.py @@ -79,6 +79,11 @@ class Entity(Base): """Get all relations (incoming and outgoing) for this entity.""" return self.incoming_relations + self.outgoing_relations + @property + def is_markdown(self): + """Check if the entity is a markdown file.""" + return self.content_type == "text/markdown" + def __repr__(self) -> str: return f"Entity(id={self.id}, name='{self.title}', type='{self.entity_type}'" diff --git a/src/basic_memory/repository/search_repository.py b/src/basic_memory/repository/search_repository.py index de87dab2..eb2bc70b 100644 --- a/src/basic_memory/repository/search_repository.py +++ b/src/basic_memory/repository/search_repository.py @@ -240,10 +240,10 @@ class SearchRepository: """Index or update a single item.""" async with db.scoped_session(self.session_maker) as session: # Delete existing record if any - await session.execute( - text("DELETE FROM search_index WHERE permalink = :permalink"), - {"permalink": search_index_row.permalink}, - ) + # await session.execute( + # text("DELETE FROM search_index WHERE permalink = :permalink"), + # {"permalink": search_index_row.permalink}, + # ) # Insert new record await session.execute( @@ -265,6 +265,15 @@ class SearchRepository: logger.debug(f"indexed row {search_index_row}") await session.commit() + async def delete_by_entity_id(self, entity_id: int): + """Delete an item from the search index by entity_id.""" + async with db.scoped_session(self.session_maker) as session: + await session.execute( + text("DELETE FROM search_index WHERE entity_id = :entity_id"), + {"entity_id": entity_id}, + ) + await session.commit() + async def delete_by_permalink(self, permalink: str): """Delete an item from the search index.""" async with db.scoped_session(self.session_maker) as session: diff --git a/src/basic_memory/services/entity_service.py b/src/basic_memory/services/entity_service.py index 5055aa3f..6a4e2a10 100644 --- a/src/basic_memory/services/entity_service.py +++ b/src/basic_memory/services/entity_service.py @@ -256,13 +256,13 @@ class EntityService(BaseService[EntityModel]): async def update_entity_relations( self, - file_path: Path, + path: str, markdown: EntityMarkdown, ) -> EntityModel: """Update relations for entity""" - logger.debug(f"Updating relations for entity: {file_path}") + logger.debug(f"Updating relations for entity: {path}") - db_entity = await self.repository.get_by_file_path(str(file_path)) + db_entity = await self.repository.get_by_file_path(path) # Clear existing relations first await self.relation_repository.delete_outgoing_relations_from_entity(db_entity.id) @@ -296,4 +296,4 @@ class EntityService(BaseService[EntityModel]): ) continue - return await self.repository.get_by_file_path(str(file_path)) + return await self.repository.get_by_file_path(path) diff --git a/src/basic_memory/services/file_service.py b/src/basic_memory/services/file_service.py index 03ea48fb..dd3b148a 100644 --- a/src/basic_memory/services/file_service.py +++ b/src/basic_memory/services/file_service.py @@ -1,7 +1,8 @@ """Service for file operations with checksum tracking.""" - +import mimetypes +from os import stat_result from pathlib import Path -from typing import Tuple, Union +from typing import Tuple, Union, Dict, Any from loguru import logger @@ -174,3 +175,53 @@ class FileService: path = Path(path) full_path = path if path.is_absolute() else self.base_path / path full_path.unlink(missing_ok=True) + + async def update_frontmatter(self, path: Union[Path, str], updates: Dict[str, Any]) -> str: + """ + Update frontmatter fields in a file while preserving all content. + """ + + path = Path(path) + full_path = path if path.is_absolute() else self.base_path / path + return await file_utils.update_frontmatter(full_path, updates) + + async def compute_checksum(self, path: Union[Path, str]) -> str: + """ + Compute SHA-256 checksum of content. + """ + path = Path(path) + full_path = path if path.is_absolute() else self.base_path / path + await file_utils.compute_checksum(full_path.read_text()) + + + async def file_stats(self, path: Union[Path, str]) -> stat_result: + """ + Return file stats for a given path. + :param path: + :return: + """ + path = Path(path) + full_path = path if path.is_absolute() else self.base_path / path + # get file timestamps + return full_path.stat() + + async def content_type(self, path: Union[Path, str]) -> stat_result: + """ + Return content_type for a given path. + :param path: + :return: + """ + path = Path(path) + full_path = path if path.is_absolute() else self.base_path / path + # get file timestamps + mime_type, _ = mimetypes.guess_type(full_path.name) + content_type = mime_type or "text/plain" + return content_type + + async def is_markdown(self, path: Union[Path, str]) -> stat_result: + """ + Return content_type for a given path. + :param path: + :return: + """ + return self.content_type(path) == "text/markdown" diff --git a/src/basic_memory/services/search_service.py b/src/basic_memory/services/search_service.py index 738f3ba3..4a900b61 100644 --- a/src/basic_memory/services/search_service.py +++ b/src/basic_memory/services/search_service.py @@ -118,6 +118,48 @@ class SearchService: self, entity: Entity, background_tasks: Optional[BackgroundTasks] = None, + ) -> None: + if background_tasks: + background_tasks.add_task(self.index_entity_data, entity) + else: + await self.index_entity_data(entity) + + async def index_entity_data( + self, + entity: Entity, + ) -> None: + + # delete all search index data associated with entity + await self.repository.delete_by_entity_id(entity_id=entity.id) + + # reindex + await self.index_entity_markdown( + entity + ) if entity.is_markdown else await self.index_entity_file(entity) + + async def index_entity_file( + self, + entity: Entity, + ) -> None: + # Index entity file with no content + await self.repository.index_item( + SearchIndexRow( + id=entity.id, + type=SearchItemType.ENTITY.value, + title=entity.title, + permalink=entity.permalink, + file_path=entity.file_path, + metadata={ + "entity_type": entity.entity_type, + }, + created_at=entity.created_at, + updated_at=entity.updated_at, + ) + ) + + async def index_entity_markdown( + self, + entity: Entity, ) -> None: """Index an entity and all its observations and relations. @@ -136,16 +178,6 @@ class SearchService: Each type gets its own row in the search index with appropriate metadata. """ - if background_tasks: - background_tasks.add_task(self.index_entity_data, entity) - else: - await self.index_entity_data(entity) - - async def index_entity_data( - self, - entity: Entity, - ) -> None: - """Actually perform the indexing.""" content_parts = [] title_variants = self._generate_variants(entity.title) @@ -169,6 +201,7 @@ class SearchService: content=entity_content, permalink=entity.permalink, file_path=entity.file_path, + entity_id=entity.id, metadata={ "entity_type": entity.entity_type, }, @@ -214,6 +247,7 @@ class SearchService: permalink=rel.permalink, file_path=entity.file_path, type=SearchItemType.RELATION.value, + entity_id=entity.id, from_id=rel.from_id, to_id=rel.to_id, relation_type=rel.relation_type, diff --git a/src/basic_memory/sync/sync_service.py b/src/basic_memory/sync/sync_service.py index 5159231d..48a29cb1 100644 --- a/src/basic_memory/sync/sync_service.py +++ b/src/basic_memory/sync/sync_service.py @@ -1,28 +1,25 @@ """Service for syncing files between filesystem and database.""" +import mimetypes from pathlib import Path -from typing import Dict +from typing import Tuple import logfire from loguru import logger from sqlalchemy.exc import IntegrityError from basic_memory import file_utils -from basic_memory.markdown import EntityParser, EntityMarkdown +from basic_memory.markdown import EntityParser from basic_memory.repository import EntityRepository, RelationRepository -from basic_memory.services import EntityService +from basic_memory.services import EntityService, FileService from basic_memory.services.search_service import SearchService from basic_memory.sync import FileChangeScanner from basic_memory.sync.utils import SyncReport +from basic_memory.models import Entity class SyncService: - """Syncs documents and knowledge files with database. - - Implements two-pass sync strategy for knowledge files to handle relations: - 1. First pass creates/updates entities without relations - 2. Second pass processes relations after all entities exist - """ + """Syncs documents and knowledge files with database.""" def __init__( self, @@ -32,6 +29,7 @@ class SyncService: entity_repository: EntityRepository, relation_repository: RelationRepository, search_service: SearchService, + file_service: FileService, ): self.scanner = scanner self.entity_service = entity_service @@ -39,9 +37,87 @@ class SyncService: self.entity_repository = entity_repository self.relation_repository = relation_repository self.search_service = search_service + self.file_service = file_service + + async def sync_file(self, path: str) -> Tuple[Entity, str]: + """Sync a single file completely.""" + + try: + if self.file_service.is_markdown(path): + entity, checksum = await self.sync_markdown_file(path) + else: + entity, checksum = await self.sync_regular_file(path) + await self.search_service.index_entity(entity) + return entity, checksum + + except Exception as e: + logger.error(f"Failed to sync {path}: {e}") + raise + + async def sync_markdown_file(self, path: str) -> Tuple[Entity, str]: + """Sync a markdown file with full processing.""" + + # Parse markdown first to get any existing permalink + entity_markdown = await self.entity_parser.parse_file(path) + + # Resolve permalink - this handles all the cases including conflicts + permalink = await self.entity_service.resolve_permalink(path, markdown=entity_markdown) + + # If permalink changed, update the file + if permalink != entity_markdown.frontmatter.permalink: + logger.info(f"Updating permalink in {path}: {permalink}") + entity_markdown.frontmatter.metadata["permalink"] = permalink + checksum = await self.file_service.update_frontmatter(path, {"permalink": permalink}) + else: + checksum = await self.file_service.compute_checksum(path) + + # Create/update entity with resolved permalink + entity = await self.entity_service.create_entity_from_markdown(path, entity_markdown) + + # Update relations and search index + entity = await self.entity_service.update_entity_relations(path, entity_markdown) + + return entity, checksum + + async def sync_regular_file(self, path: Path) -> Tuple[Entity, str]: + """Sync a non-markdown file with basic tracking.""" + + checksum = await self.file_service.compute_checksum(path) + existing = await self.entity_repository.get_by_file_path(path) + if not existing: + # Generate permalink from path + permalink = await self.entity_service.resolve_permalink(path) + + # get file timestamps + file_stats = self.file_service.file_stats(path) + + # get mime type + mime_type, _ = mimetypes.guess_type(path.name) + content_type = mime_type or "text/plain" + + entity = await self.entity_repository.add( + Entity( + entity_type="file", + file_path=path, + permalink=permalink, + checksum=checksum, + title=path.name, + created_at=file_stats.st_ctime, + updated_at=file_stats.st_mtime, + content_type=content_type, + ) + ) + else: + entity = await self.entity_repository.update( + existing.id, {"file_path": path, "checksum": checksum} + ) + + await self.search_service.index_entity(entity) + return entity, checksum async def handle_entity_deletion(self, file_path: str): """Handle complete entity deletion including search index cleanup.""" + # First get entity to get permalink before deletion entity = await self.entity_repository.get_by_file_path(file_path) if entity: @@ -61,9 +137,9 @@ class SyncService: await self.search_service.delete_by_permalink(permalink) async def sync(self, directory: Path) -> SyncReport: - """Sync knowledge files with database.""" + """Sync all files with database.""" - with logfire.span("sync", directory=directory): # pyright: ignore [reportGeneralTypeIssues] + with logfire.span("sync", directory=directory): changes = await self.scanner.find_knowledge_changes(directory) logger.info(f"Found {changes.total_changes} knowledge changes") @@ -73,102 +149,47 @@ class SyncService: entity = await self.entity_repository.get_by_file_path(old_path) if entity: # Update file_path but keep the same permalink for link stability - updated = await self.entity_repository.update( + await self.entity_repository.update( entity.id, {"file_path": new_path, "checksum": changes.checksums[new_path]} ) # update search index - if updated: - await self.search_service.index_entity(updated) + await self.search_service.index_entity(entity) # Handle deletions next - # remove rows from db for files no longer present for path in changes.deleted: await self.handle_entity_deletion(path) - # Parse files that need updating - parsed_entities: Dict[str, EntityMarkdown] = {} - + # Handle new and modified files for path in [*changes.new, *changes.modified]: - entity_markdown = await self.entity_parser.parse_file(directory / path) - parsed_entities[path] = entity_markdown - - # First pass: Create/update entities - # entities will have a null checksum to indicate they are not complete - for path, entity_markdown in parsed_entities.items(): - # Get unique permalink and update markdown if needed - permalink = await self.entity_service.resolve_permalink( - Path(path), markdown=entity_markdown - ) - - if permalink != entity_markdown.frontmatter.permalink: - # Add/update permalink in frontmatter - logger.info(f"Adding permalink '{permalink}' to file: {path}") - - # update markdown - entity_markdown.frontmatter.metadata["permalink"] = permalink - - # update file frontmatter - updated_checksum = await file_utils.update_frontmatter( - directory / path, {"permalink": permalink} - ) - - # Update checksum in changes report since file was modified - changes.checksums[path] = updated_checksum - - # if the file is new, create an entity - if path in changes.new: - # Create entity with final permalink - logger.debug(f"Creating new entity_markdown: {path}") - await self.entity_service.create_entity_from_markdown( - Path(path), entity_markdown - ) - # otherwise we need to update the entity and observations - else: - logger.debug(f"Updating entity_markdown: {path}") - await self.entity_service.update_entity_and_observations( - Path(path), entity_markdown - ) - - # Second pass - for path, entity_markdown in parsed_entities.items(): - logger.debug(f"Updating relations for: {path}") - - # Process relations - checksum = changes.checksums[path] - entity = await self.entity_service.update_entity_relations( - Path(path), entity_markdown - ) - - # add to search index - await self.search_service.index_entity(entity) - - # Set final checksum to mark sync complete - await self.entity_repository.update(entity.id, {"checksum": checksum}) - - # Third pass: Try to resolve any forward references - logger.debug("Attempting to resolve forward references") - for relation in await self.relation_repository.find_unresolved_relations(): - target_entity = await self.entity_service.link_resolver.resolve_link( - relation.to_name - ) - # check we found a link that is not the source - if target_entity and target_entity.id != relation.from_id: - logger.debug( - f"Resolved forward reference: {relation.to_name} -> {target_entity.permalink}" - ) - - try: - await self.relation_repository.update( - relation.id, - { - "to_id": target_entity.id, - "to_name": target_entity.title, # Update to actual title - }, - ) - except IntegrityError: - logger.debug(f"Ignoring duplicate relation {relation}") - - # update search index - await self.search_service.index_entity(target_entity) + logger.debug(f"Syncing file: {path}") + entity, checksum = await self.sync_file(path) + changes.checksums[path] = checksum + await self.resolve_relations() return changes + + async def resolve_relations(self): + """Try to resolve any unresolved relations""" + + logger.debug("Attempting to resolve forward references") + for relation in await self.relation_repository.find_unresolved_relations(): + resolved_entity = await self.entity_service.link_resolver.resolve_link(relation.to_name) + + # ignore reference to self + if resolved_entity and resolved_entity.id != relation.from_id: + logger.debug( + f"Resolved forward reference: {relation.to_name} -> {resolved_entity.title}" + ) + try: + await self.relation_repository.update( + relation.id, + { + "to_id": resolved_entity.id, + "to_name": resolved_entity.title, + }, + ) + except IntegrityError: + logger.debug(f"Ignoring duplicate relation {relation}") + + # update search index + await self.search_service.index_entity(resolved_entity) diff --git a/src/basic_memory/sync/sync_service.py.bak b/src/basic_memory/sync/sync_service.py.bak new file mode 100644 index 00000000..5159231d --- /dev/null +++ b/src/basic_memory/sync/sync_service.py.bak @@ -0,0 +1,174 @@ +"""Service for syncing files between filesystem and database.""" + +from pathlib import Path +from typing import Dict + +import logfire +from loguru import logger +from sqlalchemy.exc import IntegrityError + +from basic_memory import file_utils +from basic_memory.markdown import EntityParser, EntityMarkdown +from basic_memory.repository import EntityRepository, RelationRepository +from basic_memory.services import EntityService +from basic_memory.services.search_service import SearchService +from basic_memory.sync import FileChangeScanner +from basic_memory.sync.utils import SyncReport + + +class SyncService: + """Syncs documents and knowledge files with database. + + Implements two-pass sync strategy for knowledge files to handle relations: + 1. First pass creates/updates entities without relations + 2. Second pass processes relations after all entities exist + """ + + def __init__( + self, + scanner: FileChangeScanner, + entity_service: EntityService, + entity_parser: EntityParser, + entity_repository: EntityRepository, + relation_repository: RelationRepository, + search_service: SearchService, + ): + self.scanner = scanner + self.entity_service = entity_service + self.entity_parser = entity_parser + self.entity_repository = entity_repository + self.relation_repository = relation_repository + self.search_service = search_service + + async def handle_entity_deletion(self, file_path: str): + """Handle complete entity deletion including search index cleanup.""" + # First get entity to get permalink before deletion + entity = await self.entity_repository.get_by_file_path(file_path) + if entity: + logger.debug(f"Deleting entity and cleaning up search index: {file_path}") + + # Delete from db (this cascades to observations/relations) + await self.entity_service.delete_entity_by_file_path(file_path) + + # Clean up search index + permalinks = ( + [entity.permalink] + + [o.permalink for o in entity.observations] + + [r.permalink for r in entity.relations] + ) + logger.debug(f"Deleting from search index: {permalinks}") + for permalink in permalinks: + await self.search_service.delete_by_permalink(permalink) + + async def sync(self, directory: Path) -> SyncReport: + """Sync knowledge files with database.""" + + with logfire.span("sync", directory=directory): # pyright: ignore [reportGeneralTypeIssues] + changes = await self.scanner.find_knowledge_changes(directory) + logger.info(f"Found {changes.total_changes} knowledge changes") + + # Handle moves first + for old_path, new_path in changes.moves.items(): + logger.debug(f"Moving entity: {old_path} -> {new_path}") + entity = await self.entity_repository.get_by_file_path(old_path) + if entity: + # Update file_path but keep the same permalink for link stability + updated = await self.entity_repository.update( + entity.id, {"file_path": new_path, "checksum": changes.checksums[new_path]} + ) + # update search index + if updated: + await self.search_service.index_entity(updated) + + # Handle deletions next + # remove rows from db for files no longer present + for path in changes.deleted: + await self.handle_entity_deletion(path) + + # Parse files that need updating + parsed_entities: Dict[str, EntityMarkdown] = {} + + for path in [*changes.new, *changes.modified]: + entity_markdown = await self.entity_parser.parse_file(directory / path) + parsed_entities[path] = entity_markdown + + # First pass: Create/update entities + # entities will have a null checksum to indicate they are not complete + for path, entity_markdown in parsed_entities.items(): + # Get unique permalink and update markdown if needed + permalink = await self.entity_service.resolve_permalink( + Path(path), markdown=entity_markdown + ) + + if permalink != entity_markdown.frontmatter.permalink: + # Add/update permalink in frontmatter + logger.info(f"Adding permalink '{permalink}' to file: {path}") + + # update markdown + entity_markdown.frontmatter.metadata["permalink"] = permalink + + # update file frontmatter + updated_checksum = await file_utils.update_frontmatter( + directory / path, {"permalink": permalink} + ) + + # Update checksum in changes report since file was modified + changes.checksums[path] = updated_checksum + + # if the file is new, create an entity + if path in changes.new: + # Create entity with final permalink + logger.debug(f"Creating new entity_markdown: {path}") + await self.entity_service.create_entity_from_markdown( + Path(path), entity_markdown + ) + # otherwise we need to update the entity and observations + else: + logger.debug(f"Updating entity_markdown: {path}") + await self.entity_service.update_entity_and_observations( + Path(path), entity_markdown + ) + + # Second pass + for path, entity_markdown in parsed_entities.items(): + logger.debug(f"Updating relations for: {path}") + + # Process relations + checksum = changes.checksums[path] + entity = await self.entity_service.update_entity_relations( + Path(path), entity_markdown + ) + + # add to search index + await self.search_service.index_entity(entity) + + # Set final checksum to mark sync complete + await self.entity_repository.update(entity.id, {"checksum": checksum}) + + # Third pass: Try to resolve any forward references + logger.debug("Attempting to resolve forward references") + for relation in await self.relation_repository.find_unresolved_relations(): + target_entity = await self.entity_service.link_resolver.resolve_link( + relation.to_name + ) + # check we found a link that is not the source + if target_entity and target_entity.id != relation.from_id: + logger.debug( + f"Resolved forward reference: {relation.to_name} -> {target_entity.permalink}" + ) + + try: + await self.relation_repository.update( + relation.id, + { + "to_id": target_entity.id, + "to_name": target_entity.title, # Update to actual title + }, + ) + except IntegrityError: + logger.debug(f"Ignoring duplicate relation {relation}") + + # update search index + await self.search_service.index_entity(target_entity) + + return changes diff --git a/tests/conftest.py b/tests/conftest.py index 23947e8d..fe919995 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -152,6 +152,7 @@ async def sync_service( entity_repository: EntityRepository, relation_repository: RelationRepository, search_service: SearchService, + file_service: FileService, ) -> SyncService: """Create sync service for testing.""" return SyncService( @@ -161,6 +162,7 @@ async def sync_service( relation_repository=relation_repository, entity_parser=entity_parser, search_service=search_service, + file_service=file_service, ) diff --git a/tests/sync/test_sync_service.py b/tests/sync/test_sync_service.py index 36357f6f..ddd9631d 100644 --- a/tests/sync/test_sync_service.py +++ b/tests/sync/test_sync_service.py @@ -74,7 +74,7 @@ Target content @pytest.mark.asyncio -async def test_sync_knowledge( +async def test_sync( sync_service: SyncService, test_config: ProjectConfig, entity_service: EntityService ): """Test basic knowledge sync functionality."""