<filename>fuchan/apps/posts/views.py
import os
import tempfile

from django.conf import settings
from django.core.files import File
from django.db import transaction
from django.db.models import Count, Q
from django.http import Http404, HttpResponseRedirect
from django.template.defaultfilters import filesizeformat
from django.urls import reverse
from django.views.generic import CreateView, DetailView, ListView

from fuchan.apps.common.models import Breadcrumb
from fuchan.apps.common.views import ViewDependencyInjectionMixin
from fuchan.apps.posts.forms import ReplyCreateForm, ThreadCreateForm
from fuchan.apps.posts.models import Post
from fuchan.apps.posts.utils import format_post_content
from fuchan.apps.topics.models import Topic
from fuchan.services.ff import FFMpeg, ImageMagick, MediaInfo
from fuchan.utils import get_ip_address_from_request


class ThreadPostListView(ViewDependencyInjectionMixin, ListView):
    model = Post
    template_name = "posts/thread_post_list.html"
    ordering = ("updated_at",)

    def initialize_dependencies(self, request, *args, **kwargs):
        self.topic = Topic.objects.filter(slug=kwargs["topic_slug"]).first()
        self.thread = (
            Post.objects.filter(parent__isnull=True, pk=self.kwargs["thread_id"])
            .annotate(
                media_count=Count("post", Q(post__media_file_original_name__isnull=False)) + 1,
                reply_count=Count("post"),
            )
            .first()
        )

        if not self.topic or not self.thread:
            raise Http404

    def get_queryset(self):
        return super().get_queryset().filter(parent=self.thread, topic=self.topic)

    def get_context_data(self, *, object_list=None, **kwargs):
        context = {
            "breadcrumb_list": [
                Breadcrumb(link=reverse("topic_list"), title="Home"),
                Breadcrumb(link=self.topic.get_absolute_url(), title=str(self.topic)),
                Breadcrumb(link=self.thread.get_absolute_url(), title=self.thread.title),
            ],
            "page_title": f"/{self.topic.slug}/ - {self.thread.title} - {self.topic.title}",
            "topic": self.topic,
            "topic_list": Topic.objects.all(),
            "thread": self.thread,
        }
        kwargs.update(context)
        return super().get_context_data(object_list=object_list, **kwargs)


class PostCreateBaseView(ViewDependencyInjectionMixin, CreateView):
    model = Post
    template_name = "error.html"

    def initialize_dependencies(self, request, *args, **kwargs):
        try:
            self.topic = Topic.objects.get(slug=kwargs["topic_slug"])
        except Topic.DoesNotExist as exc:
            raise Http404 from exc

    def get_success_url(self):
        return self.object.get_absolute_url(exclude_hash=False)

    def _form_valid(self, form):
        content, quoted_post_ids = format_post_content(form.instance.content)

        if content:
            form.instance.content = content

        self.object = form.save()

        if quoted_post_ids:
            self.object.quoted_posts.add(
                *quoted_post_ids,
                through_defaults={
                    "created_at": self.object.created_at,
                    "updated_at": self.object.updated_at,
                },
            )

    @transaction.atomic
    def form_valid(self, form):
        thread = getattr(self, "thread", None)
        user = self.request.user if self.request.user.is_authenticated else None
        form.instance.created_by = user
        form.instance.creator_ip_address = get_ip_address_from_request(self.request)
        form.instance.topic = self.topic

        if thread:
            form.instance.parent = thread

        if "media_file" in form.files:
            if form.files["media_file"].size > settings.MEDIA_MAX_SIZE:
                form.add_error(
                    "media_file", f"Media file is too big ({filesizeformat(settings.MEDIA_MAX_SIZE)} allowed)"
                )
                return self.form_invalid(form)

            with tempfile.NamedTemporaryFile(suffix=".webp") as thumb_tempf:
                with tempfile.NamedTemporaryFile() as media_tempf:
                    media_tempf.write(form.files["media_file"].read())
                    media_info = MediaInfo(media_tempf.name)
                    form.instance.is_video = media_info.is_video
                    form.instance.media_file_height = media_info.height
                    form.instance.media_file_width = media_info.width
                    form.instance.media_file_size = os.stat(media_tempf.name).st_size

                    if media_info.is_video:
                        FFMpeg().write_thumbnail(media_tempf.name, thumb_tempf.name)
                    else:
                        form.files["media_file"].seek(0)
                        thumb_tempf.write(form.files["media_file"].read())

                    ImageMagick().write_thumbnail(thumb_tempf.name, quality=60)

                    with open(thumb_tempf.name, "rb") as thumb_tempf_reopened:
                        form.instance.thumbnail_image = File(thumb_tempf_reopened)

                        # NOTE: Do not move outside 'with' statement, will
                        #       raise ValueError due to seek of closed file.
                        self._form_valid(form)
        else:
            self._form_valid(form)

        if thread:
            thread.thread_updated_at = self.object.updated_at
            thread.save()

        return HttpResponseRedirect(self.get_success_url())


class ThreadReplyCreateView(PostCreateBaseView):
    form_class = ReplyCreateForm

    def initialize_dependencies(self, request, *args, **kwargs):
        try:
            self.topic = Topic.objects.get(slug=kwargs["topic_slug"])
            self.thread = Post.objects.get(is_locked=False, parent__isnull=True, pk=self.kwargs["thread_id"])
        except (Post.DoesNotExist, Topic.DoesNotExist) as exc:
            raise Http404 from exc


class TopicThreadCreateView(PostCreateBaseView):
    form_class = ThreadCreateForm


class TopicThreadListView(ViewDependencyInjectionMixin, ListView):  # pylint: disable=too-many-ancestors
    model = Post
    paginate_by = 20
    template_name = "posts/topic_thread_list.html"
    ordering = ("-is_sticky", "-thread_updated_at")

    def initialize_dependencies(self, request, *args, **kwargs):
        self.topic = (
            Topic.objects.filter(slug=kwargs["topic_slug"])
            .annotate(
                media_count=Count("post", filter=Q(post__media_file_original_name__isnull=False)),
                post_count=Count("post", filter=Q(post__parent__isnull=True)),
                reply_count=Count("post", filter=Q(post__parent__isnull=False)),
            )
            .first()
        )

        if not self.topic:
            raise Http404

    def get_queryset(self):
        # latest_reply_qs = Post.objects.filter(parent=OuterRef('pk')).order_by('-created_at')

        return (
            super()
            .get_queryset()
            .filter(parent__isnull=True, topic=self.topic)
            .annotate(
                media_count=Count("post", Q(post__media_file_original_name__isnull=False)) + 1,
                reply_count=Count("post"),
                # last_reply_created_at=Subquery(latest_reply_qs.values('updated_at')[:1]),
            )
        )

    def get_context_data(self, *, object_list=None, **kwargs):
        context = {
            "breadcrumb_list": [
                Breadcrumb(link=reverse("topic_list"), title="Home"),
                Breadcrumb(link=self.topic.get_absolute_url(), title=str(self.topic)),
            ],
            "page_title": str(self.topic),
            "topic": self.topic,
            "topic_list": Topic.objects.all(),
        }
        kwargs.update(context)

        return super().get_context_data(object_list=object_list, **kwargs)


class PostCardView(DetailView):
    model = Post
    template_name = "components/post.html"
    pk_url_kwarg = "post_id"
    context_object_name = "item"

    def get_context_data(self, **kwargs):
        context = {"ajax": True}
        kwargs.update(context)
        return super().get_context_data(**kwargs)


class PostLinkGeneratorView(DetailView):
    model = Post
    template_name = "blank.html"
    pk_url_kwarg = "post_id"

    def get(self, request, *args, **kwargs):
        self.object = self.get_object()
        return HttpResponseRedirect(self.object.get_absolute_url(exclude_hash=False))

<filename>fuchan/apps/posts/views.py
import os
import tempfile

from django.conf import settings
from django.core.files import File
from django.db import transaction
from django.db.models import Count, Q
from django.http import Http404, HttpResponseRedirect
from django.template.defaultfilters import filesizeformat
from django.urls import reverse
from django.views.generic import CreateView, DetailView, ListView

from fuchan.apps.common.models import Breadcrumb
from fuchan.apps.common.views import ViewDependencyInjectionMixin
from fuchan.apps.posts.forms import ReplyCreateForm, ThreadCreateForm
from fuchan.apps.posts.models import Post
from fuchan.apps.posts.utils import format_post_content
from fuchan.apps.topics.models import Topic
from fuchan.services.ff import FFMpeg, ImageMagick, MediaInfo
from fuchan.utils import get_ip_address_from_request


class ThreadPostListView(ViewDependencyInjectionMixin, ListView):
    model = Post
    template_name = "posts/thread_post_list.html"
    ordering = ("updated_at",)

    def initialize_dependencies(self, request, *args, **kwargs):
        self.topic = Topic.objects.filter(slug=kwargs["topic_slug"]).first()
        self.thread = (
            Post.objects.filter(parent__isnull=True, pk=self.kwargs["thread_id"])
            .annotate(
                media_count=Count("post", Q(post__media_file_original_name__isnull=False)) + 1,
                reply_count=Count("post"),
            )
            .first()
        )

        if not self.topic or not self.thread:
            raise Http404

    def get_queryset(self):
        return super().get_queryset().filter(parent=self.thread, topic=self.topic)

    def get_context_data(self, *, object_list=None, **kwargs):
        context = {
            "breadcrumb_list": [
                Breadcrumb(link=reverse("topic_list"), title="Home"),
                Breadcrumb(link=self.topic.get_absolute_url(), title=str(self.topic)),
                Breadcrumb(link=self.thread.get_absolute_url(), title=self.thread.title),
            ],
            "page_title": f"/{self.topic.slug}/ - {self.thread.title} - {self.topic.title}",
            "topic": self.topic,
            "topic_list": Topic.objects.all(),
            "thread": self.thread,
        }
        kwargs.update(context)
        return super().get_context_data(object_list=object_list, **kwargs)


class PostCreateBaseView(ViewDependencyInjectionMixin, CreateView):
    model = Post
    template_name = "error.html"

    def initialize_dependencies(self, request, *args, **kwargs):
        try:
            self.topic = Topic.objects.get(slug=kwargs["topic_slug"])
        except Topic.DoesNotExist as exc:
            raise Http404 from exc

    def get_success_url(self):
        return self.object.get_absolute_url(exclude_hash=False)

    def _form_valid(self, form):
        content, quoted_post_ids = format_post_content(form.instance.content)

        if content:
            form.instance.content = content

        self.object = form.save()

        if quoted_post_ids:
            self.object.quoted_posts.add(
                *quoted_post_ids,
                through_defaults={
                    "created_at": self.object.created_at,
                    "updated_at": self.object.updated_at,
                },
            )

    @transaction.atomic
    def form_valid(self, form):
        thread = getattr(self, "thread", None)
        user = self.request.user if self.request.user.is_authenticated else None
        form.instance.created_by = user
        form.instance.creator_ip_address = get_ip_address_from_request(self.request)
        form.instance.topic = self.topic

        if thread:
            form.instance.parent = thread

        if "media_file" in form.files:
            if form.files["media_file"].size > settings.MEDIA_MAX_SIZE:
                form.add_error(
                    "media_file", f"Media file is too big ({filesizeformat(settings.MEDIA_MAX_SIZE)} allowed)"
                )
                return self.form_invalid(form)

            with tempfile.NamedTemporaryFile(suffix=".webp") as thumb_tempf:
                with tempfile.NamedTemporaryFile() as media_tempf:
                    media_tempf.write(form.files["media_file"].read())
                    media_info = MediaInfo(media_tempf.name)
                    form.instance.is_video = media_info.is_video
                    form.instance.media_file_height = media_info.height
                    form.instance.media_file_width = media_info.width
                    form.instance.media_file_size = os.stat(media_tempf.name).st_size

                    if media_info.is_video:
                        FFMpeg().write_thumbnail(media_tempf.name, thumb_tempf.name)
                    else:
                        form.files["media_file"].seek(0)
                        thumb_tempf.write(form.files["media_file"].read())

                    ImageMagick().write_thumbnail(thumb_tempf.name, quality=60)

                    with open(thumb_tempf.name, "rb") as thumb_tempf_reopened:
                        form.instance.thumbnail_image = File(thumb_tempf_reopened)

                        # NOTE: Do not move outside 'with' statement, will
                        #       raise ValueError due to seek of closed file.
                        self._form_valid(form)
        else:
            self._form_valid(form)

        if thread:
            thread.thread_updated_at = self.object.updated_at
            thread.save()

        return HttpResponseRedirect(self.get_success_url())


class ThreadReplyCreateView(PostCreateBaseView):
    form_class = ReplyCreateForm

    def initialize_dependencies(self, request, *args, **kwargs):
        try:
            self.topic = Topic.objects.get(slug=kwargs["topic_slug"])
            self.thread = Post.objects.get(is_locked=False, parent__isnull=True, pk=self.kwargs["thread_id"])
        except (Post.DoesNotExist, Topic.DoesNotExist) as exc:
            raise Http404 from exc


class TopicThreadCreateView(PostCreateBaseView):
    form_class = ThreadCreateForm


class TopicThreadListView(ViewDependencyInjectionMixin, ListView):  # pylint: disable=too-many-ancestors
    model = Post
    paginate_by = 20
    template_name = "posts/topic_thread_list.html"
    ordering = ("-is_sticky", "-thread_updated_at")

    def initialize_dependencies(self, request, *args, **kwargs):
        self.topic = (
            Topic.objects.filter(slug=kwargs["topic_slug"])
            .annotate(
                media_count=Count("post", filter=Q(post__media_file_original_name__isnull=False)),
                post_count=Count("post", filter=Q(post__parent__isnull=True)),
                reply_count=Count("post", filter=Q(post__parent__isnull=False)),
            )
            .first()
        )

        if not self.topic:
            raise Http404

    def get_queryset(self):
        # latest_reply_qs = Post.objects.filter(parent=OuterRef('pk')).order_by('-created_at')

        return (
            super()
            .get_queryset()
            .filter(parent__isnull=True, topic=self.topic)
            .annotate(
                media_count=Count("post", Q(post__media_file_original_name__isnull=False)) + 1,
                reply_count=Count("post"),
                # last_reply_created_at=Subquery(latest_reply_qs.values('updated_at')[:1]),
            )
        )

    def get_context_data(self, *, object_list=None, **kwargs):
        context = {
            "breadcrumb_list": [
                Breadcrumb(link=reverse("topic_list"), title="Home"),
                Breadcrumb(link=self.topic.get_absolute_url(), title=str(self.topic)),
            ],
            "page_title": str(self.topic),
            "topic": self.topic,
            "topic_list": Topic.objects.all(),
        }
        kwargs.update(context)

        return super().get_context_data(object_list=object_list, **kwargs)


class PostCardView(DetailView):
    model = Post
    template_name = "components/post.html"
    pk_url_kwarg = "post_id"
    context_object_name = "item"

    def get_context_data(self, **kwargs):
        context = {"ajax": True}
        kwargs.update(context)
        return super().get_context_data(**kwargs)


class PostLinkGeneratorView(DetailView):
    model = Post
    template_name = "blank.html"
    pk_url_kwarg = "post_id"

    def get(self, request, *args, **kwargs):
        self.object = self.get_object()
        return HttpResponseRedirect(self.object.get_absolute_url(exclude_hash=False))

# NOTE: Do not move outside 'with' statement, will #       raise ValueError due to seek of closed file. # pylint: disable=too-many-ancestors # latest_reply_qs = Post.objects.filter(parent=OuterRef('pk')).order_by('-created_at') # last_reply_created_at=Subquery(latest_reply_qs.values('updated_at')[:1]),