boards/models/post/manager.py

import logging

from datetime import datetime, timedelta, date
from datetime import time as dtime

from django.db import models, transaction
from django.utils import timezone

import boards

from boards.models.user import Ban
from boards.mdx_neboard import Parser
from boards.models import PostImage, Attachment
from boards import utils

__author__ = 'neko259'

IMAGE_TYPES = (
    'jpeg',
    'jpg',
    'png',
    'bmp',
    'gif',
)

POSTS_PER_DAY_RANGE = 7
NO_IP = '0.0.0.0'


class PostManager(models.Manager):
    @transaction.atomic
    def create_post(self, title: str, text: str, file=None, thread=None,
                    ip=NO_IP, tags: list=None, opening_posts: list=None,
                    tripcode=''):
        """
        Creates new post
        """

        if not utils.is_anonymous_mode():
            is_banned = Ban.objects.filter(ip=ip).exists()

        # TODO Raise specific exception and catch it in the views
        if is_banned:
            raise Exception("This user is banned")

        if not tags:
            tags = []
        if not opening_posts:
            opening_posts = []

        posting_time = timezone.now()
        new_thread = False
        if not thread:
            thread = boards.models.thread.Thread.objects.create(
                bump_time=posting_time, last_edit_time=posting_time)
            list(map(thread.tags.add, tags))
            boards.models.thread.Thread.objects.process_oldest_threads()
            new_thread = True

        pre_text = Parser().preparse(text)

        post = self.create(title=title,
                           text=pre_text,
                           pub_time=posting_time,
                           poster_ip=ip,
                           thread=thread,
                           last_edit_time=posting_time,
                           tripcode=tripcode,
                           opening=new_thread)
        post.threads.add(thread)

        logger = logging.getLogger('boards.post.create')

        logger.info('Created post {} with text {} by {}'.format(post,
            post.get_text(),post.poster_ip))

        # TODO Move this to other place
        if file:
            file_type = file.name.split('.')[-1].lower()
            if file_type in IMAGE_TYPES:
                post.images.add(PostImage.objects.create_with_hash(file))
            else:
                post.attachments.add(Attachment.objects.create_with_hash(file))

        post.build_url()
        post.connect_replies()
        post.connect_threads(opening_posts)
        post.connect_notifications()

        # Thread needs to be bumped only when the post is already created
        if not new_thread:
            thread.last_edit_time = posting_time
            thread.bump()
            thread.save()

        return post

    def delete_posts_by_ip(self, ip):
        """
        Deletes all posts of the author with same IP
        """

        posts = self.filter(poster_ip=ip)
        for post in posts:
            post.delete()

    @utils.cached_result()
    def get_posts_per_day(self) -> float:
        """
        Gets average count of posts per day for the last 7 days
        """

        day_end = date.today()
        day_start = day_end - timedelta(POSTS_PER_DAY_RANGE)

        day_time_start = timezone.make_aware(datetime.combine(
            day_start, dtime()), timezone.get_current_timezone())
        day_time_end = timezone.make_aware(datetime.combine(
            day_end, dtime()), timezone.get_current_timezone())

        posts_per_period = float(self.filter(
            pub_time__lte=day_time_end,
            pub_time__gte=day_time_start).count())

        ppd = posts_per_period / POSTS_PER_DAY_RANGE

        return ppd