lib.itmens/journal/apis/collection.py
2025-02-03 19:18:44 -05:00

213 lines
5.7 KiB
Python

from datetime import datetime
from typing import List
from django.core.cache import cache
from django.utils import timezone
from django.views.decorators.cache import cache_page
from ninja import Field, Schema
from ninja.decorators import decorate_view
from ninja.pagination import paginate
from catalog.common.models import Item, ItemSchema
from common.api import PageNumberPagination, Result, api
from ..models import Collection
class CollectionSchema(Schema):
uuid: str
url: str
visibility: int = Field(ge=0, le=2)
post_id: int | None = Field(alias="latest_post_id")
created_time: datetime
title: str
brief: str
cover: str
html_content: str
class CollectionInSchema(Schema):
title: str
brief: str
visibility: int = Field(ge=0, le=2)
class CollectionItemSchema(Schema):
item: ItemSchema
note: str
class CollectionItemInSchema(Schema):
item_uuid: str
note: str
@api.get(
"/me/collection/",
response={200: List[CollectionSchema], 401: Result, 403: Result},
tags=["collection"],
)
@paginate(PageNumberPagination)
def list_collections(request):
"""
Get collections created by current user
"""
queryset = Collection.objects.filter(owner=request.user.identity)
return queryset
@api.get(
"/me/collection/{collection_uuid}",
response={200: CollectionSchema, 401: Result, 403: Result, 404: Result},
tags=["collection"],
)
def get_collection(request, collection_uuid: str):
"""
Get collections by its uuid
"""
c = Collection.get_by_url(collection_uuid)
if not c:
return 404, {"message": "Collection not found"}
if c.owner != request.user.identity:
return 403, {"message": "Not owner"}
return c
@api.post(
"/me/collection/",
response={200: CollectionSchema, 401: Result, 403: Result, 404: Result},
tags=["collection"],
)
def create_collection(request, c_in: CollectionInSchema):
"""
Create collection.
`title`, `brief` (markdown formatted) and `visibility` are required;
"""
c = Collection.objects.create(
owner=request.user.identity,
title=c_in.title,
brief=c_in.brief,
visibility=c_in.visibility,
)
return c
@api.put(
"/me/collection/{collection_uuid}",
response={200: CollectionSchema, 401: Result, 403: Result, 404: Result},
tags=["collection"],
)
def update_collection(request, collection_uuid: str, c_in: CollectionInSchema):
"""
Update collection.
"""
c = Collection.get_by_url(collection_uuid)
if not c:
return 404, {"message": "Collection not found"}
if c.owner != request.user.identity:
return 403, {"message": "Not owner"}
c.title = c_in.title
c.brief = c_in.brief
c.visibility = c_in.visibility
c.save()
return c
@api.delete(
"/me/collection/{collection_uuid}",
response={200: Result, 401: Result, 403: Result, 404: Result},
tags=["collection"],
)
def delete_collection(request, collection_uuid: str):
"""
Remove a collection.
"""
c = Collection.get_by_url(collection_uuid)
if not c:
return 404, {"message": "Collection not found"}
if c.owner != request.user.identity:
return 403, {"message": "Not owner"}
c.delete()
return 200, {"message": "OK"}
@api.get(
"/me/collection/{collection_uuid}/item/",
response={200: List[CollectionItemSchema], 401: Result, 403: Result, 404: Result},
tags=["collection"],
)
@paginate(PageNumberPagination)
def collection_list_items(request, collection_uuid: str):
"""
Get items in a collection collections
"""
c = Collection.get_by_url(collection_uuid)
if not c:
return 404, {"message": "Collection not found"}
if c.owner != request.user.identity:
return 403, {"message": "Not owner"}
return c.ordered_members
@api.post(
"/me/collection/{collection_uuid}/item/",
response={200: Result, 401: Result, 403: Result, 404: Result},
tags=["collection"],
)
def collection_add_item(
request, collection_uuid: str, collection_item: CollectionItemInSchema
):
"""
Add an item to collection
"""
c = Collection.get_by_url(collection_uuid)
if not c:
return 404, {"message": "Collection not found"}
if c.owner != request.user.identity:
return 403, {"message": "Not owner"}
if not collection_item.item_uuid:
return 404, {"message": "Item not found"}
item = Item.get_by_url(collection_item.item_uuid)
if not item:
return 404, {"message": "Item not found"}
c.append_item(item, note=collection_item.note)
return 200, {"message": "OK"}
@api.delete(
"/me/collection/{collection_uuid}/item/{item_uuid}",
response={200: Result, 401: Result, 403: Result, 404: Result},
tags=["collection"],
)
def collection_delete_item(request, collection_uuid: str, item_uuid: str):
"""
Remove an item from collection
"""
c = Collection.get_by_url(collection_uuid)
if not c:
return 404, {"message": "Collection not found"}
if c.owner != request.user.identity:
return 403, {"message": "Not owner"}
item = Item.get_by_url(item_uuid)
if not item:
return 404, {"message": "Item not found"}
c.remove_item(item)
return 200, {"message": "OK"}
@api.get(
"/catalog/trending/collection/",
response={200: list[CollectionSchema]},
summary="Trending collection in catalog",
auth=None,
tags=["catalog"],
)
@decorate_view(cache_page(600))
def trending_collection(request):
rot = timezone.now().minute // 6
collection_ids = cache.get("featured_collections", [])
i = rot * len(collection_ids) // 10
collection_ids = collection_ids[i:] + collection_ids[:i]
featured_collections = Collection.objects.filter(pk__in=collection_ids)
return featured_collections