from datetime import datetime, timedelta, date from datetime import time as dtime import logging from django.db import models, transaction from django.utils import timezone from boards import utils from boards.mdx_neboard import Parser from boards.models import PostImage, Attachment import boards.models __author__ = 'vurdalak' NO_IP = '0.0.0.0' POSTS_PER_DAY_RANGE = 7 IMAGE_TYPES = ( 'jpeg', 'jpg', 'png', 'bmp', 'gif', ) 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=None): """ Creates new post """ is_banned = boards.models.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) post.threads.add(thread) logger = logging.getLogger('boards.post.create') logger.info('Created post {} by {}'.format(post, 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() post.set_global_id() # 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 @transaction.atomic def import_post(self, title: str, text: str, pub_time: str, global_id, opening_post=None, tags=list()): if opening_post is None: thread = boards.models.thread.Thread.objects.create( bump_time=pub_time, last_edit_time=pub_time) list(map(thread.tags.add, tags)) else: thread = opening_post.get_thread() post = self.create(title=title, text=text, pub_time=pub_time, poster_ip=NO_IP, last_edit_time=pub_time, thread_id=thread.id, global_id=global_id) post.build_url() post.connect_replies() post.connect_notifications()