fix: relink unlinked entry to existing entry without sql error

tagstudio/src/core/library/alchemy/library.py

@@ -490,6 +490,32 @@ def get_entries_full(self, entry_ids: list[int] | set[int]) -> Iterator[Entry]:
                 yield entry
                 session.expunge(entry)
 
+    def get_entry_full_by_path(self, path: Path) -> Entry | None:
+        """Get the entry with the corresponding path."""
+        with Session(self.engine) as session:
+            stmt = select(Entry).where(Entry.path == path)
+            stmt = (
+                stmt.outerjoin(Entry.text_fields)
+                .outerjoin(Entry.datetime_fields)
+                .options(selectinload(Entry.text_fields), selectinload(Entry.datetime_fields))
+            )
+            stmt = (
+                stmt.outerjoin(Entry.tags)
+                .outerjoin(TagAlias)
+                .options(
+                    selectinload(Entry.tags).options(
+                        joinedload(Tag.aliases),
+                        joinedload(Tag.parent_tags),
+                    )
+                )
+            )
+            entry = session.scalar(stmt)
+            if not entry:
+                return None
+            session.expunge(entry)
+            make_transient(entry)
+            return entry
+
     @property
     def entries_count(self) -> int:
         with Session(self.engine) as session:
@@ -698,7 +724,13 @@ def search_tags(
 
             return res
 
-    def update_entry_path(self, entry_id: int | Entry, path: Path) -> None:
+    def update_entry_path(self, entry_id: int | Entry, path: Path) -> bool:
+        """Set the path field of an entry.
+
+        Returns True if the action succeeded and False if the path already exists.
+        """
+        if self.has_path_entry(path):
+            return False
         if isinstance(entry_id, Entry):
             entry_id = entry_id.id
 
@@ -715,6 +747,7 @@ def update_entry_path(self, entry_id: int | Entry, path: Path) -> None:
 
             session.execute(update_stmt)
             session.commit()
+        return True
 
     def remove_tag(self, tag: Tag):
         with Session(self.engine, expire_on_commit=False) as session:
@@ -1185,6 +1218,18 @@ def mirror_entry_fields(self, *entries: Entry) -> None:
                         value=field.value,
                     )
 
+    def merge_entries(self, from_entry: Entry, into_entry: Entry) -> None:
+        """Add fields and tags from the first entry to the second, and then delete the first."""
+        for field in from_entry.fields:
+            self.add_field_to_entry(
+                entry_id=into_entry.id,
+                field_id=field.type_key,
+                value=field.value,
+            )
+        tag_ids = [tag.id for tag in from_entry.tags]
+        self.add_tags_to_entry(into_entry.id, tag_ids)
+        self.remove_entries([from_entry.id])
+
     @property
     def tag_color_groups(self) -> dict[str, list[TagColorGroup]]:
         """Return every TagColorGroup in the library."""

tagstudio/src/core/utils/missing_files.py

@@ -53,8 +53,18 @@ def fix_missing_files(self) -> Iterator[int]:
         for i, entry in enumerate(self.missing_files, start=1):
             item_matches = self.match_missing_file(entry)
             if len(item_matches) == 1:
-                logger.info("fix_missing_files", entry=entry, item_matches=item_matches)
-                self.library.update_entry_path(entry.id, item_matches[0])
+                logger.info(
+                    "fix_missing_files",
+                    entry=entry.path.as_posix(),
+                    item_matches=item_matches[0].as_posix(),
+                )
+                if not self.library.update_entry_path(entry.id, item_matches[0]):
+                    try:
+                        match = self.library.get_entry_full_by_path(item_matches[0])
+                        entry_full = self.library.get_entry_full(entry.id)
+                        self.library.merge_entries(entry_full, match)
+                    except AttributeError:
+                        continue
                 self.files_fixed_count += 1
                 # remove fixed file
                 self.missing_files.remove(entry)

tagstudio/tests/test_library.py

@@ -309,6 +309,41 @@ def test_mirror_entry_fields(library: Library, entry_full):
     }
 
 
+def test_merge_entries(library: Library):
+    a = Entry(
+        folder=library.folder,
+        path=Path("a"),
+        fields=[
+            TextField(type_key=_FieldID.AUTHOR.name, value="Author McAuthorson", position=0),
+            TextField(type_key=_FieldID.DESCRIPTION.name, value="test description", position=2),
+        ],
+    )
+    b = Entry(
+        folder=library.folder,
+        path=Path("b"),
+        fields=[TextField(type_key=_FieldID.NOTES.name, value="test note", position=1)],
+    )
+    try:
+        ids = library.add_entries([a, b])
+        entry_a = library.get_entry_full(ids[0])
+        entry_b = library.get_entry_full(ids[1])
+        tag_0 = library.add_tag(Tag(id=1000, name="tag_0"))
+        tag_1 = library.add_tag(Tag(id=1001, name="tag_1"))
+        tag_2 = library.add_tag(Tag(id=1002, name="tag_2"))
+        library.add_tags_to_entry(ids[0], [tag_0.id, tag_2.id])
+        library.add_tags_to_entry(ids[1], [tag_1.id])
+        library.merge_entries(entry_a, entry_b)
+        assert library.has_path_entry(Path("b"))
+        assert not library.has_path_entry(Path("a"))
+        fields = [field.value for field in entry_a.fields]
+        assert "Author McAuthorson" in fields
+        assert "test description" in fields
+        assert "test note" in fields
+        assert b.has_tag(tag_0) and b.has_tag(tag_1) and b.has_tag(tag_2)
+    except AttributeError:
+        AssertionError()
+
+
 def test_remove_tag_from_entry(library, entry_full):
     removed_tag_id = -1
     for tag in entry_full.tags: