##// END OF EJS Templates
Moved exceptions to a separate module
Moved exceptions to a separate module

File last commit:

r1602:beb32860 default
r1602:beb32860 default
Show More
manager.py
181 lines | 5.6 KiB | text/x-python | PythonLexer
import logging
from datetime import datetime, timedelta, date
from datetime import time as dtime
from boards.abstracts.exceptions import BannedException, ArchiveException
from django.db import models, transaction
from django.utils import timezone
from django.dispatch import Signal
import boards
from boards.models.user import Ban
from boards.mdx_neboard import Parser
from boards.models import Attachment
from boards import utils
__author__ = 'neko259'
POSTS_PER_DAY_RANGE = 7
NO_IP = '0.0.0.0'
post_import_deps = Signal()
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='', monochrome=False, images=[]):
"""
Creates new post
"""
if thread is not None and thread.is_archived():
raise ArchiveException('Cannot post into an archived thread')
if not utils.is_anonymous_mode():
is_banned = Ban.objects.filter(ip=ip).exists()
else:
is_banned = False
if is_banned:
raise BannedException("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,
monochrome=monochrome)
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))
if file:
self._add_file_to_post(file, post)
for image in images:
post.images.add(image)
post.connect_threads(opening_posts)
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(), files=list(),
tripcode=None, version=1):
is_opening = opening_post is None
if is_opening:
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,
global_id=global_id,
opening=is_opening,
thread=thread,
tripcode=tripcode,
version=version)
for file in files:
self._add_file_to_post(file, post)
post.threads.add(thread)
url_to_post = '[post]{}[/post]'.format(str(global_id))
replies = self.filter(text__contains=url_to_post)
for reply in replies:
post_import_deps.send(reply.__class__)
@transaction.atomic
def update_post(self, post, title: str, text: str, pub_time: str,
tags=list(), files=list(), tripcode=None, version=1):
post.title = title
post.text = text
post.pub_time = pub_time
post.tripcode = tripcode
post.version = version
post.save()
post.clear_cache()
post.images.clear()
post.attachments.clear()
for file in files:
self._add_file_to_post(file, post)
thread = post.get_thread()
thread.tags.clear()
list(map(thread.tags.add, tags))
def _add_file_to_post(self, file, post):
post.attachments.add(Attachment.objects.create_with_hash(file))