import re from functools import cached_property from typing import TYPE_CHECKING, Any from django.conf import settings from django.db import models from django.utils.translation import gettext_lazy as _ from catalog.collection.models import Collection as CatalogCollection from catalog.common import jsondata from catalog.common.utils import piece_cover_path from catalog.models import Item from takahe.utils import Takahe from users.models import APIdentity from .common import Piece from .itemlist import List, ListMember from .renderers import render_md _RE_HTML_TAG = re.compile(r"<[^>]*>") class CollectionMember(ListMember): parent = models.ForeignKey( "Collection", related_name="members", on_delete=models.CASCADE ) note = jsondata.CharField(_("note"), null=True, blank=True) @property def ap_object(self): return { "id": self.absolute_url, "type": "CollectionItem", "collection": self.parent.absolute_url, "published": self.created_time.isoformat(), "updated": self.edited_time.isoformat(), "attributedTo": self.owner.actor_uri, "withRegardTo": self.item.absolute_url, "note": self.note, "href": self.absolute_url, } def to_indexable_doc(self) -> dict[str, Any]: return {} class Collection(List): if TYPE_CHECKING: members: models.QuerySet[CollectionMember] url_path = "collection" post_when_save = True index_when_save = True MEMBER_CLASS = CollectionMember catalog_item = models.OneToOneField( CatalogCollection, on_delete=models.PROTECT, related_name="journal_item" ) title = models.CharField(_("title"), max_length=1000, default="") brief = models.TextField(_("description"), blank=True, default="") cover = models.ImageField( upload_to=piece_cover_path, default=settings.DEFAULT_ITEM_COVER, blank=True ) items = models.ManyToManyField( Item, through="CollectionMember", related_name="collections" ) collaborative = models.PositiveSmallIntegerField( default=0 ) # 0: Editable by owner only / 1: Editable by bi-direction followers featured_by = models.ManyToManyField( to=APIdentity, related_name="featured_collections", through="FeaturedCollection" ) def __str__(self): return f"Collection:{self.uuid}@{self.owner_id}:{self.title}" @property def html_content(self): html = render_md(self.brief) return html @property def plain_content(self): html = render_md(self.brief) return _RE_HTML_TAG.sub(" ", html) def featured_since(self, owner: APIdentity): f = FeaturedCollection.objects.filter(target=self, owner=owner).first() return f.created_time if f else None def get_stats(self, owner: APIdentity): items = list(self.members.all().values_list("item_id", flat=True)) stats = {"total": len(items)} for st, shelf in owner.shelf_manager.shelf_list.items(): stats[st] = shelf.members.all().filter(item_id__in=items).count() stats["percentage"] = ( round(stats["complete"] * 100 / stats["total"]) if stats["total"] else 0 ) return stats def get_progress(self, owner: APIdentity): items = list(self.members.all().values_list("item_id", flat=True)) if len(items) == 0: return 0 shelf = owner.shelf_manager.shelf_list["complete"] return round( shelf.members.all().filter(item_id__in=items).count() * 100 / len(items) ) def save(self, *args, **kwargs): if getattr(self, "catalog_item", None) is None: self.catalog_item = CatalogCollection() if ( self.catalog_item.title != self.title or self.catalog_item.brief != self.brief ): self.catalog_item.title = self.title self.catalog_item.brief = self.brief self.catalog_item.cover = self.cover self.catalog_item.save() super().save(*args, **kwargs) def get_ap_data(self): return { "object": { # "tag": [item.ap_object_ref for item in collection.items], "relatedWith": [self.ap_object], } } def sync_to_timeline(self, update_mode: int = 0): existing_post = self.latest_post owner: APIdentity = self.owner user = owner.user v = Takahe.visibility_n2t(self.visibility, user.preference.post_public_mode) if existing_post and (update_mode == 1 or v != existing_post.visibility): Takahe.delete_posts([existing_post.pk]) existing_post = None data = self.get_ap_data() # if existing_post and existing_post.type_data == data: # return existing_post action = _("created collection") item_link = self.absolute_url prepend_content = f'{action} {self.title}
' content = self.plain_content if len(content) > 360: content = content[:357] + "..." post = Takahe.post( self.owner.pk, content, v, prepend_content, "", None, False, data, existing_post.pk if existing_post else None, self.created_time, language=owner.user.macrolanguage, ) if post and post != existing_post: self.link_post_id(post.pk) return post @property def ap_object(self): return { "id": self.absolute_url, "type": "Collection", "name": self.title, "content": self.brief, "mediaType": "text/markdown", "published": self.created_time.isoformat(), "updated": self.edited_time.isoformat(), "attributedTo": self.owner.actor_uri, "href": self.absolute_url, } def to_indexable_doc(self) -> dict[str, Any]: content = [self.title, self.brief] item_id = [] item_title = [] item_class = set() for m in self.members.all(): item_id.append(m.item.pk) item_title += m.item.to_indexable_titles() item_class |= {m.item.__class__.__name__} if m.note: content.append(m.note) return { "item_id": item_id, "item_class": list(item_class), "item_title": item_title, "content": content, } class FeaturedCollection(Piece): owner = models.ForeignKey(APIdentity, on_delete=models.CASCADE) target = models.ForeignKey(Collection, on_delete=models.CASCADE) created_time = models.DateTimeField(auto_now_add=True) edited_time = models.DateTimeField(auto_now=True) class Meta: unique_together = [["owner", "target"]] @property def visibility(self): return self.target.visibility @cached_property def progress(self): return self.target.get_progress(self.owner) def to_indexable_doc(self) -> dict[str, Any]: return {}