Misago/misago/threads/models/thread.py

from django.core.exceptions import ObjectDoesNotExist
from django.db import models
from django.utils import timezone
from django.utils.translation import ugettext_lazy as _

from misago.conf import settings
from misago.core.pgutils import PgPartialIndex
from misago.core.utils import slugify


class Thread(models.Model):
    WEIGHT_DEFAULT = 0
    WEIGHT_PINNED = 1
    WEIGHT_GLOBAL = 2

    WEIGHT_CHOICES = [
        (WEIGHT_DEFAULT, _("Don't pin thread")),
        (WEIGHT_PINNED, _("Pin thread within category")),
        (WEIGHT_GLOBAL, _("Pin thread globally")),
    ]

    category = models.ForeignKey(
        'misago_categories.Category',
        on_delete=models.CASCADE,
    )
    title = models.CharField(max_length=255)
    slug = models.CharField(max_length=255)
    replies = models.PositiveIntegerField(default=0, db_index=True)

    has_events = models.BooleanField(default=False)
    has_poll = models.BooleanField(default=False)
    has_reported_posts = models.BooleanField(default=False)
    has_open_reports = models.BooleanField(default=False)
    has_unapproved_posts = models.BooleanField(default=False)
    has_hidden_posts = models.BooleanField(default=False)

    started_on = models.DateTimeField(db_index=True)
    last_post_on = models.DateTimeField(db_index=True)

    first_post = models.ForeignKey(
        'misago_threads.Post',
        related_name='+',
        null=True,
        blank=True,
        on_delete=models.SET_NULL,
    )
    starter = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        null=True,
        blank=True,
        on_delete=models.SET_NULL,
    )
    starter_name = models.CharField(max_length=255)
    starter_slug = models.CharField(max_length=255)

    last_post = models.ForeignKey(
        'misago_threads.Post',
        related_name='+',
        null=True,
        blank=True,
        on_delete=models.SET_NULL,
    )
    last_post_is_event = models.BooleanField(default=False)
    last_poster = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        related_name='last_poster_set',
        null=True,
        blank=True,
        on_delete=models.SET_NULL,
    )
    last_poster_name = models.CharField(max_length=255, null=True, blank=True)
    last_poster_slug = models.CharField(max_length=255, null=True, blank=True)

    weight = models.PositiveIntegerField(default=WEIGHT_DEFAULT)

    is_unapproved = models.BooleanField(default=False, db_index=True)
    is_hidden = models.BooleanField(default=False)
    is_closed = models.BooleanField(default=False)

    best_answer = models.ForeignKey(
        'misago_threads.Post',
        related_name='+',
        null=True,
        blank=True,
        on_delete=models.SET_NULL,
    )
    best_answer_is_protected =  models.BooleanField(default=False)
    best_answer_marked_on = models.DateTimeField(null=True, blank=True)
    best_answer_marked_by = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        related_name='marked_best_answer_set',
        null=True,
        blank=True,
        on_delete=models.SET_NULL,
    )
    best_answer_marked_by_name = models.CharField(max_length=255, null=True, blank=True)
    best_answer_marked_by_slug = models.CharField(max_length=255, null=True, blank=True)

    participants = models.ManyToManyField(
        settings.AUTH_USER_MODEL,
        related_name='privatethread_set',
        through='ThreadParticipant',
        through_fields=('thread', 'user'),
    )

    class Meta:
        indexes = [
            PgPartialIndex(
                fields=['weight'],
                where={'weight': 2},
            ),
            PgPartialIndex(
                fields=['weight'],
                where={'weight': 1},
            ),
            PgPartialIndex(
                fields=['weight'],
                where={'weight': 0},
            ),
            PgPartialIndex(
                fields=['weight'],
                where={'weight__lt': 2},
            ),
            PgPartialIndex(
                fields=['has_reported_posts'],
                where={'has_reported_posts': True},
            ),
            PgPartialIndex(
                fields=['has_unapproved_posts'],
                where={'has_unapproved_posts': True},
            ),
            PgPartialIndex(
                fields=['is_hidden'],
                where={'is_hidden': False},
            ),
        ]

        index_together = [
            ['category', 'id'],
            ['category', 'last_post_on'],
            ['category', 'replies'],
        ]

    def __str__(self):
        return self.title

    def delete(self, *args, **kwargs):
        from misago.threads.signals import delete_thread
        delete_thread.send(sender=self)

        super().delete(*args, **kwargs)

    def merge(self, other_thread):
        if self.pk == other_thread.pk:
            raise ValueError("thread can't be merged with itself")

        from misago.threads.signals import merge_thread
        merge_thread.send(sender=self, other_thread=other_thread)

    def move(self, new_category):
        from misago.threads.signals import move_thread

        self.category = new_category
        move_thread.send(sender=self)

    def synchronize(self):
        try:
            self.has_poll = bool(self.poll)
        except ObjectDoesNotExist:
            self.has_poll = False

        self.replies = self.post_set.filter(is_event=False, is_unapproved=False).count()

        if self.replies > 0:
            self.replies -= 1

        reported_post_qs = self.post_set.filter(has_reports=True)
        self.has_reported_posts = reported_post_qs.exists()

        if self.has_reported_posts:
            open_reports_qs = self.post_set.filter(has_open_reports=True)
            self.has_open_reports = open_reports_qs.exists()
        else:
            self.has_open_reports = False

        unapproved_post_qs = self.post_set.filter(is_unapproved=True)
        self.has_unapproved_posts = unapproved_post_qs.exists()

        hidden_post_qs = self.post_set.filter(is_hidden=True)[:1]
        self.has_hidden_posts = hidden_post_qs.exists()

        posts = self.post_set.order_by('id')

        first_post = posts.first()
        self.set_first_post(first_post)

        last_post = posts.filter(is_unapproved=False).last()
        if last_post:
            self.set_last_post(last_post)
        else:
            self.set_last_post(first_post)

        self.has_events = False
        if last_post:
            if last_post.is_event:
                self.has_events = True
            else:
                self.has_events = self.post_set.filter(is_event=True).exists()

    @property
    def has_best_answer(self):
        return bool(self.best_answer_id)

    @property
    def thread_type(self):
        return self.category.thread_type

    def get_api_url(self):
        return self.thread_type.get_thread_api_url(self)

    def get_editor_api_url(self):
        return self.thread_type.get_thread_editor_api_url(self)

    def get_merge_api_url(self):
        return self.thread_type.get_thread_merge_api_url(self)

    def get_posts_api_url(self):
        return self.thread_type.get_thread_posts_api_url(self)

    def get_post_merge_api_url(self):
        return self.thread_type.get_post_merge_api_url(self)

    def get_post_move_api_url(self):
        return self.thread_type.get_post_move_api_url(self)

    def get_post_split_api_url(self):
        return self.thread_type.get_post_split_api_url(self)

    def get_poll_api_url(self):
        return self.thread_type.get_thread_poll_api_url(self)

    def get_absolute_url(self, page=1):
        return self.thread_type.get_thread_absolute_url(self, page)

    def get_new_post_url(self):
        return self.thread_type.get_thread_new_post_url(self)

    def get_last_post_url(self):
        return self.thread_type.get_thread_last_post_url(self)

    def get_best_answer_url(self):
        return self.thread_type.get_thread_best_answer_url(self)

    def get_unapproved_post_url(self):
        return self.thread_type.get_thread_unapproved_post_url(self)

    def set_title(self, title):
        self.title = title
        self.slug = slugify(title)

    def set_first_post(self, post):
        self.started_on = post.posted_on
        self.first_post = post
        self.starter = post.poster
        self.starter_name = post.poster_name
        if post.poster:
            self.starter_slug = post.poster.slug
        else:
            self.starter_slug = slugify(post.poster_name)

        self.is_unapproved = post.is_unapproved
        self.is_hidden = post.is_hidden

    def set_last_post(self, post):
        self.last_post_on = post.posted_on
        self.last_post_is_event = post.is_event
        self.last_post = post
        self.last_poster = post.poster
        self.last_poster_name = post.poster_name
        if post.poster:
            self.last_poster_slug = post.poster.slug
        else:
            self.last_poster_slug = slugify(post.poster_name)

    def set_best_answer(self, user, post):
        if post.thread_id != self.id:
            raise ValueError("post to set as best answer must be in same thread")
        if post.is_first_post:
            raise ValueError("post to set as best answer can't be first post")
        if post.is_hidden:
            raise ValueError("post to set as best answer can't be hidden")
        if post.is_unapproved:
            raise ValueError("post to set as best answer can't be unapproved")

        self.best_answer = post
        self.best_answer_is_protected = post.is_protected
        self.best_answer_marked_on = timezone.now()
        self.best_answer_marked_by = user
        self.best_answer_marked_by_name = user.username
        self.best_answer_marked_by_slug = user.slug

    def clear_best_answer(self):
        self.best_answer = None
        self.best_answer_is_protected = False
        self.best_answer_marked_on = None
        self.best_answer_marked_by = None
        self.best_answer_marked_by_name = None
        self.best_answer_marked_by_slug = None