Work migration

boofilsic/settings.py

@@ -121,6 +121,7 @@
     NEODB_SENTRY_SAMPLE_RATE=(float, 0),
     NEODB_FANOUT_LIMIT_DAYS=(int, 9),
     INDEX_ALIASES=(dict, {}),
+    SKIP_WORK_MIGRATION=(bool, False),
 )
 
 # ====== End of user configuration variables ======
@@ -629,3 +630,5 @@ def _init_language_settings(preferred_lanugages_env):
         send_default_pii=True,
         traces_sample_rate=env("NEODB_SENTRY_SAMPLE_RATE"),
     )
+
+SKIP_WORK_MIGRATION = env("SKIP_WORK_MIGRATION")

catalog/book/models.py

@@ -20,6 +20,8 @@
 from functools import cached_property
 from typing import TYPE_CHECKING
 
+from auditlog.models import QuerySet
+
 from django.core.validators import MaxValueValidator, MinValueValidator
 from django.db import models
 from django.utils.translation import gettext_lazy as _
@@ -109,6 +111,9 @@ class EditionSchema(EditionInSchema, BaseSchema):
 class Edition(Item):
     if TYPE_CHECKING:
         works: "models.ManyToManyField[Work, Edition]"
+    related_work = models.ForeignKey(
+        "Work", null=True, on_delete=models.SET_NULL, related_name="related_editions"
+    )
 
     class BookFormat(models.TextChoices):
         PAPERBACK = "paperback", _("Paperback")
@@ -263,13 +268,23 @@ def merge_to(self, to_item: "Edition | None"):  # type: ignore[reportIncompatibl
         if to_item:
             if self.merge_title():
                 self.save()
-            for work in self.works.all():
-                to_item.works.add(work)
-        self.works.clear()
+            if not self.related_work:
+                return
+            if to_item.related_work:
+                for edition in self.related_work.related_editions.exclude(
+                    pk=self.pk
+                ).all():
+                    edition.related_work = to_item.related_work
+                    edition.save()
+            else:
+                to_item.related_work = self.related_work
+                to_item.save()
+                self.related_work = None
+                self.save()
 
     def delete(self, using=None, keep_parents=False, soft=True, *args, **kwargs):
         if soft:
-            self.works.clear()
+            self.related_work = None
         return super().delete(using, soft, keep_parents, *args, **kwargs)
 
     def update_linked_items_from_external_resource(self, resource):
@@ -292,8 +307,13 @@ def update_linked_items_from_external_resource(self, resource):
                         primary_lookup_id_type=w["id_type"],
                         primary_lookup_id_value=w["id_value"],
                     ).first()
-                if work and work not in self.works.all():
-                    self.works.add(work)
+                if not work:
+                    return
+                if not self.related_work:
+                    self.related_work = work
+                    self.save()
+                elif work.pk != self.related_work.pk:
+                    work.merge_to(self.related_work)
 
     def merge_data_from_external_resource(
         self, p: "ExternalResource", ignore_existing_content: bool = False
@@ -317,9 +337,8 @@ def merge_title(self) -> bool:
 
     @property
     def sibling_items(self):
-        works = list(self.works.all())
         return (
-            Edition.objects.filter(works__in=works)
+            Edition.objects.filter(related_work__in=[self.related_work])
             .exclude(pk=self.pk)
             .exclude(is_deleted=True)
             .exclude(merged_to_item__isnull=False)
@@ -339,41 +358,45 @@ def title_deco(self):
         return f"({' '.join(a)})" if a else ""
 
     def has_related_books(self):
-        works = list(self.works.all())
-        if not works:
+        if not self.related_work:
             return False
-        return Edition.objects.filter(works__in=works).exclude(pk=self.pk).exists()
+        return (
+            Edition.objects.filter(related_work__in=[self.related_work])
+            .exclude(pk=self.pk)
+            .exists()
+        )
 
     def link_to_related_book(self, target: "Edition") -> bool:
         if target == self or target.is_deleted or target.merged_to_item:
             return False
-        if target.works.all().exists():
-            for work in target.works.all():
-                self.works.add(work)
-                work.localized_title = uniq(work.localized_title + self.localized_title)
-                work.save()
-        elif self.works.all().exists():
-            for work in self.works.all():
-                target.works.add(work)
-                work.localized_title = uniq(
-                    work.localized_title + target.localized_title
-                )
-                work.save()
+        if target.related_work:
+            self.related_work = target.related_work
+            self.save()
+            target.related_work.localized_title = uniq(
+                target.related_work.localized_title + self.localized_title
+            )
+            target.related_work.save()
         else:
             work = Work.objects.create(localized_title=self.localized_title)
-            work.editions.add(self, target)
+            self.related_work = work
+            self.save()
+            target.related_work = work
+            target.save()
             # work.localized_title = self.localized_title
             # work.save()
         return True
 
     def unlink_from_all_works(self):
-        self.works.clear()
+        self.related_work = None
+        self.save()
 
     def has_works(self):
-        return self.works.all().exists()
+        return self.related_work is not None
 
 
 class Work(Item):
+    if TYPE_CHECKING:
+        related_editions: QuerySet[Edition]
     category = ItemCategory.Book
     url_path = "book/work"
     douban_work = PrimaryLookupIdDescriptor(IdType.DoubanBook_Work)
@@ -417,24 +440,25 @@ def merge_to(self, to_item: "Work | None"):  # type: ignore[reportIncompatibleMe
         super().merge_to(to_item)
         if not to_item:
             return
-        for edition in self.editions.all():
-            to_item.editions.add(edition)
-        self.editions.clear()
+        for edition in self.related_editions.all():
+            edition.related_work = to_item
+            edition.save()
         to_item.language = uniq(to_item.language + self.language)  # type: ignore
         to_item.localized_title = uniq(to_item.localized_title + self.localized_title)
         to_item.save()
 
     def delete(self, using=None, keep_parents=False, soft=True, *args, **kwargs):
         if soft:
-            self.editions.clear()
+            for edition in self.related_editions.all():
+                edition.related_work = None
         return super().delete(using, keep_parents, soft, *args, **kwargs)
 
     @property
     def cover_image_url(self):
         url = super().cover_image_url
         if url:
             return url
-        e = next(filter(lambda e: e.cover_image_url, self.editions.all()), None)
+        e = next(filter(lambda e: e.cover_image_url, self.related_editions.all()), None)
         return e.cover_image_url if e else None
 
     def update_linked_items_from_external_resource(self, resource):
@@ -457,8 +481,9 @@ def update_linked_items_from_external_resource(self, resource):
                         primary_lookup_id_type=e["id_type"],
                         primary_lookup_id_value=e["id_value"],
                     ).first()
-                if edition and edition not in self.editions.all():
-                    self.editions.add(edition)
+                if edition and edition not in self.related_editions.all():
+                    edition.related_work = self
+                    edition.save()
 
 
 class Series(Item):

catalog/book/tests.py

@@ -105,7 +105,7 @@ def setUp(self):
 
     def test_work(self):
         self.assertFalse(self.hyperion_print.has_related_books())
-        self.hyperion.editions.add(self.hyperion_print)
+        self.hyperion.related_editions.add(self.hyperion_print)
         self.assertFalse(self.hyperion_print.has_related_books())
 
     def test_merge(self):
@@ -122,7 +122,7 @@ def test_link(self):
         self.assertTrue(self.hyperion_ebook.has_related_books())
         self.assertTrue(self.hyperion_print.has_works())
         self.assertEqual(
-            self.hyperion_print.works.first().display_title,
+            self.hyperion_print.related_work.display_title,
             self.hyperion_print.display_title,
         )
         self.hyperion_print.unlink_from_all_works()
@@ -140,9 +140,8 @@ def test_link3(self):
         self.hyperion_ebook.link_to_related_book(self.hyperion_hardcover)
         self.hyperion_print.link_to_related_book(self.hyperion_hardcover)
         self.assertTrue(self.hyperion_print.has_works())
-        self.assertEqual(self.hyperion_print.works.all().count(), 1)
         self.assertEqual(
-            self.hyperion_ebook.works.all().first().editions.all().count(), 3
+            self.hyperion_ebook.related_work.related_editions.all().count(), 3
         )
 
 
@@ -371,12 +370,14 @@ def test_work(self):
         url2 = "https://book.douban.com/subject/2037260/"
         p1 = SiteManager.get_site_by_url(url1).get_resource_ready()
         p2 = SiteManager.get_site_by_url(url2).get_resource_ready()
-        w1 = p1.item.works.all().first()
-        w2 = p2.item.works.all().first()
+        w1 = p1.item.related_work
+        w2 = p2.item.related_work
         self.assertEqual(w1.display_title, "黄金时代")
         self.assertEqual(w2.display_title, "黄金时代")
         self.assertEqual(w1, w2)
-        editions = sorted(list(w1.editions.all()), key=lambda e: e.display_title)
+        editions = sorted(
+            list(w1.related_editions.all()), key=lambda e: e.display_title
+        )
         self.assertEqual(len(editions), 2)
         self.assertEqual(editions[0].display_title, "Wang in Love and Bondage")
         self.assertEqual(editions[1].display_title, "黄金时代")
@@ -523,26 +524,24 @@ def test_works(self):
         p1 = SiteManager.get_site_by_url(
             url1
         ).get_resource_ready()  # lxml bug may break this
-        w1 = p1.item.works.all().first()
+        w1 = p1.item.related_work
         p2 = SiteManager.get_site_by_url(url2).get_resource_ready()
-        w2 = p2.item.works.all().first()
+        w2 = p2.item.related_work
         self.assertEqual(w1, w2)
-        self.assertEqual(p1.item.works.all().count(), 1)
+        self.assertNotEqual(w1, None)
         p3 = SiteManager.get_site_by_url(url3).get_resource_ready()
-        w3 = p3.item.works.all().first()
+        w3 = p3.item.related_work
         self.assertNotEqual(w3, w2)
         p4 = SiteManager.get_site_by_url(url4).get_resource_ready()
         self.assertEqual(p4.item.id, p1.item.id)
-        self.assertEqual(p4.item.works.all().count(), 2)
-        self.assertEqual(p1.item.works.all().count(), 2)
-        w2e = sorted(list(w2.editions.all()), key=lambda e: e.display_title)
-        self.assertEqual(len(w2e), 2)
-        self.assertEqual(w2e[0].display_title, "Wang in Love and Bondage")
-        self.assertEqual(w2e[1].display_title, "黄金时代")
-        w3e = sorted(list(w3.editions.all()), key=lambda e: e.display_title)
-        self.assertEqual(len(w3e), 2)
-        self.assertEqual(w3e[0].display_title, "Golden Age: A Novel")
-        self.assertEqual(w3e[1].display_title, "黄金时代")
+        self.assertEqual(p4.item.related_work, p1.item.related_work)
+        w2e = sorted(list(w2.related_editions.all()), key=lambda e: e.display_title)
+        self.assertEqual(len(w2e), 3)
+        self.assertEqual(w2e[0].display_title, "Golden Age: A Novel")
+        self.assertEqual(w2e[1].display_title, "Wang in Love and Bondage")
+        self.assertEqual(w2e[2].display_title, "黄金时代")
+        w3e = sorted(list(w3.related_editions.all()), key=lambda e: e.display_title)
+        self.assertEqual(len(w3e), 0)  # w3 is merged to w2
         e = Edition.objects.get(primary_lookup_id_value=9781662601217)
         self.assertEqual(e.display_title, "Golden Age: A Novel")
 
@@ -556,35 +555,35 @@ def test_works_merge(self):
         p1 = SiteManager.get_site_by_url(
             url1
         ).get_resource_ready()  # lxml bug may break this
-        w1 = p1.item.works.all().first()
+        w1 = p1.item.related_work
         p2 = SiteManager.get_site_by_url(url2).get_resource_ready()
-        w2 = p2.item.works.all().first()
+        w2 = p2.item.related_work
         self.assertEqual(w1, w2)
-        self.assertEqual(p1.item.works.all().count(), 1)
+        self.assertNotEqual(w1, None)
         p3 = SiteManager.get_site_by_url(url3).get_resource_ready()
-        w3 = p3.item.works.all().first()
+        w3 = p3.item.related_work
         self.assertNotEqual(w3, w2)
         self.assertEqual(w2.external_resources.all().count(), 1)
         self.assertEqual(w3.external_resources.all().count(), 1)
         w3.merge_to(w2)
         self.assertEqual(w2.external_resources.all().count(), 2)
         self.assertEqual(w3.external_resources.all().count(), 0)
-        self.assertEqual(w2.editions.all().count(), 3)
-        self.assertEqual(w3.editions.all().count(), 0)
+        self.assertEqual(w2.related_editions.all().count(), 3)
+        self.assertEqual(w3.related_editions.all().count(), 0)
         p4 = SiteManager.get_site_by_url(url4).get_resource_ready()
         self.assertEqual(p4.item.id, p1.item.id)
-        self.assertEqual(p4.item.works.all().count(), 1)
-        self.assertEqual(p1.item.works.all().count(), 1)
-        w2e = sorted(list(w2.editions.all()), key=lambda e: e.display_title)
+        self.assertNotEqual(p1.item.related_work, None)
+        self.assertNotEqual(p4.item.related_work, None)
+        w2e = sorted(list(w2.related_editions.all()), key=lambda e: e.display_title)
         self.assertEqual(len(w2e), 3)
         self.assertEqual(w2e[0].display_title, "Golden Age: A Novel")
         self.assertEqual(w2e[1].display_title, "Wang in Love and Bondage")
         self.assertEqual(w2e[2].display_title, "黄金时代")
-        w3e = w3.editions.all().order_by("title")
+        w3e = w3.related_editions.all().order_by("title")
         self.assertEqual(w3e.count(), 0)
         e = Edition.objects.get(primary_lookup_id_value=9781662601217)
         self.assertEqual(e.display_title, "Golden Age: A Novel")
         w2e[1].delete()
-        self.assertEqual(w2.editions.all().count(), 2)
-        w2.editions.all().delete()
+        self.assertEqual(w2.related_editions.all().count(), 2)
+        w2.related_editions.all().delete()
         self.assertEqual(p1.item.works.all().count(), 0)