Misago/misago/threads/viewmodels/threads.py

from datetime import timedelta

from django.core.exceptions import PermissionDenied
from django.db.models import F, Q
from django.http import Http404
from django.utils import timezone
from django.utils.translation import ugettext as _
from django.utils.translation import ugettext_lazy

from misago.acl import add_acl
from misago.conf import settings
from misago.core.shortcuts import paginate, pagination_dict
from misago.readtracker import threadstracker
from misago.readtracker.dates import get_cutoff_date
from misago.threads.models import Post, Thread
from misago.threads.participants import make_participants_aware
from misago.threads.permissions import exclude_invisible_posts, exclude_invisible_threads
from misago.threads.serializers import ThreadsListSerializer
from misago.threads.subscriptions import make_subscription_aware
from misago.threads.utils import add_categories_to_items


__all__ = ['ForumThreads', 'PrivateThreads', 'filter_read_threads_queryset']

LISTS_NAMES = {
    'all': None,
    'my': ugettext_lazy("Your threads"),
    'new': ugettext_lazy("New threads"),
    'unread': ugettext_lazy("Unread threads"),
    'subscribed': ugettext_lazy("Subscribed threads"),
    'unapproved': ugettext_lazy("Unapproved content"),
}

LIST_DENIED_MESSAGES = {
    'my': ugettext_lazy("You have to sign in to see list of threads that you have started."),
    'new': ugettext_lazy("You have to sign in to see list of threads you haven't read."),
    'unread': ugettext_lazy("You have to sign in to see list of threads with new replies."),
    'subscribed': ugettext_lazy("You have to sign in to see list of threads you are subscribing."),
    'unapproved': ugettext_lazy("You have to sign in to see list of threads with unapproved posts."),
}


class ViewModel(object):
    def __init__(self, request, category, list_type, page):
        self.allow_see_list(request, category, list_type)

        category_model = category.unwrap()

        base_queryset = self.get_base_queryset(request, category.categories, list_type)
        base_queryset = base_queryset.select_related('starter', 'last_poster')

        threads_categories = [category_model] + category.subcategories

        threads_queryset = self.get_remaining_threads_queryset(
            base_queryset, category_model, threads_categories
        )

        list_page = paginate(
            threads_queryset, page, settings.MISAGO_THREADS_PER_PAGE, settings.MISAGO_THREADS_TAIL
        )
        paginator = pagination_dict(list_page)

        if list_page.number > 1:
            threads = list(list_page.object_list)
        else:
            pinned_threads = list(
                self.get_pinned_threads(base_queryset, category_model, threads_categories)
            )
            threads = list(pinned_threads) + list(list_page.object_list)

        add_categories_to_items(category_model, category.categories, threads)
        add_acl(request.user, threads)
        make_subscription_aware(request.user, threads)

        if list_type in ('new', 'unread'):
            # we already know all threads on list are unread
            for thread in threads:
                thread.is_read = False
                thread.is_new = True
        else:
            threadstracker.make_read_aware(request.user, threads)

        self.filter_threads(request, threads)

        # set state on object for easy access from hooks
        self.category = category
        self.threads = threads
        self.list_type = list_type
        self.paginator = paginator

    def allow_see_list(self, request, category, list_type):
        if list_type not in LISTS_NAMES:
            raise Http404()

        if request.user.is_anonymous:
            if list_type in LIST_DENIED_MESSAGES:
                raise PermissionDenied(LIST_DENIED_MESSAGES[list_type])
        else:
            has_permission = request.user.acl_cache['can_see_unapproved_content_lists']
            if list_type == 'unapproved' and not has_permission:
                raise PermissionDenied(
                    _("You don't have permission to see unapproved content lists.")
                )

    def get_list_name(self, list_type):
        return LISTS_NAMES[list_type]

    def get_base_queryset(self, request, threads_categories, list_type):
        return get_threads_queryset(
            request.user,
            threads_categories,
            list_type,
        ).order_by('-last_post_id')

    def get_pinned_threads(self, queryset, category, threads_categories):
        return []

    def get_remaining_threads_queryset(self, queryset, category, threads_categories):
        return []

    def filter_threads(self, request, threads):
        pass  # hook for custom thread types to add features to extend threads

    def get_frontend_context(self):
        context = {
            'THREADS': {
                'results': ThreadsListSerializer(self.threads, many=True).data,
                'subcategories': [c.pk for c in self.category.children],
            },
        }

        context['THREADS'].update(self.paginator)
        return context

    def get_template_context(self):
        return {
            'list_name': self.get_list_name(self.list_type),
            'list_type': self.list_type,
            'threads': self.threads,
            'paginator': self.paginator,
        }


class ForumThreads(ViewModel):
    def get_pinned_threads(self, queryset, category, threads_categories):
        if category.level:
            return list(queryset.filter(weight=2)
                        ) + list(queryset.filter(weight=1, category__in=threads_categories))
        else:
            return queryset.filter(weight=2)

    def get_remaining_threads_queryset(self, queryset, category, threads_categories):
        if category.level:
            return queryset.filter(
                weight=0,
                category__in=threads_categories,
            )
        else:
            return queryset.filter(
                weight__lt=2,
                category__in=threads_categories,
            )


class PrivateThreads(ViewModel):
    def get_base_queryset(self, request, threads_categories, list_type):
        queryset = super().get_base_queryset(request, threads_categories, list_type)

        # limit queryset to threads we are participant of
        participated_threads = request.user.threadparticipant_set.values('thread_id')

        if request.user.acl_cache['can_moderate_private_threads']:
            queryset = queryset.filter(Q(id__in=participated_threads) | Q(has_reported_posts=True))
        else:
            queryset = queryset.filter(id__in=participated_threads)

        return queryset

    def get_remaining_threads_queryset(self, queryset, category, threads_categories):
        return queryset.filter(category__in=threads_categories)

    def filter_threads(self, request, threads):
        make_participants_aware(request.user, threads)


def get_threads_queryset(user, categories, list_type):
    queryset = exclude_invisible_threads(user, categories, Thread.objects)

    if list_type == 'all':
        return queryset
    else:
        return filter_threads_queryset(user, categories, list_type, queryset)


def filter_threads_queryset(user, categories, list_type, queryset):
    if list_type == 'my':
        return queryset.filter(starter=user)
    elif list_type == 'subscribed':
        subscribed_threads = user.subscription_set.values('thread_id')
        return queryset.filter(id__in=subscribed_threads)
    elif list_type == 'unapproved':
        return queryset.filter(has_unapproved_posts=True)
    elif list_type in ('new', 'unread'):
        return filter_read_threads_queryset(user, categories, list_type, queryset)
    else:
        return queryset


def filter_read_threads_queryset(user, categories, list_type, queryset):
    # grab cutoffs for categories
    cutoff_date = get_cutoff_date(user)

    visible_posts = Post.objects.filter(posted_on__gt=cutoff_date)
    visible_posts = exclude_invisible_posts(user, categories, visible_posts)

    queryset = queryset.filter(id__in=visible_posts.distinct().values('thread'))

    read_posts = visible_posts.filter(id__in=user.postread_set.values('post'))

    if list_type == 'new':
        # new threads have no entry in reads table
        return queryset.exclude(id__in=read_posts.distinct().values('thread'))

    if list_type == 'unread':
        # unread threads were read in past but have new posts
        unread_posts = visible_posts.exclude(id__in=user.postread_set.values('post'))
        queryset = queryset.filter(id__in=read_posts.distinct().values('thread'))
        queryset = queryset.filter(id__in=unread_posts.distinct().values('thread'))
        return queryset