manager.py
125 lines
| 3.7 KiB
| text/x-python
|
PythonLexer
|
r1234 | 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 | ||||
import boards.models | ||||
__author__ = 'vurdalak' | ||||
NO_IP = '0.0.0.0' | ||||
POSTS_PER_DAY_RANGE = 7 | ||||
class PostManager(models.Manager): | ||||
@transaction.atomic | ||||
def create_post(self, title: str, text: str, image=None, thread=None, | ||||
ip=NO_IP, tags: list=None, opening_posts: list=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() | ||||
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() | ||||
else: | ||||
thread.last_edit_time = posting_time | ||||
thread.bump() | ||||
thread.save() | ||||
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) | ||||
post.threads.add(thread) | ||||
post.set_global_id() | ||||
logger = logging.getLogger('boards.post.create') | ||||
logger.info('Created post {} by {}'.format(post, post.poster_ip)) | ||||
if image: | ||||
post.images.add(PostImage.objects.create_with_hash(image)) | ||||
post.build_url() | ||||
post.connect_replies() | ||||
post.connect_threads(opening_posts) | ||||
post.connect_notifications() | ||||
return post | ||||
@transaction.atomic | ||||
def import_post(self, title: str, text: str, pub_time: str, | ||||
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) | ||||
post.build_url() | ||||
post.connect_replies() | ||||
post.connect_notifications() | ||||
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 | ||||