models.py
322 lines
| 9.0 KiB
| text/x-python
|
PythonLexer
/ boards / models.py
|
r29 | import os | ||
|
r50 | from random import random | ||
|
r31 | import re | ||
|
r29 | import time | ||
|
r46 | import math | ||
|
r71 | |||
from django.db import models | ||||
from django.http import Http404 | ||||
from django.utils import timezone | ||||
from markupfield.fields import MarkupField | ||||
|
r95 | from threading import Thread | ||
|
r4 | |||
|
r22 | from neboard import settings | ||
|
r71 | import thumbs | ||
|
r22 | |||
|
r79 | IMAGE_THUMB_SIZE = (200, 150) | ||
TITLE_MAX_LENGTH = 50 | ||||
DEFAULT_MARKUP_TYPE = 'markdown' | ||||
|
r22 | |||
|
r4 | NO_PARENT = -1 | ||
NO_IP = '0.0.0.0' | ||||
UNKNOWN_UA = '' | ||||
|
r71 | ALL_PAGES = -1 | ||
|
r85 | OPENING_POST_POPULARITY_WEIGHT = 2 | ||
|
r71 | IMAGES_DIRECTORY = 'images/' | ||
FILE_EXTENSION_DELIMITER = '.' | ||||
REGEX_PRETTY = re.compile(r'^\d(0)+$') | ||||
REGEX_SAME = re.compile(r'^(.)\1+$') | ||||
|
r22 | |||
|
r0 | |||
|
r2 | class PostManager(models.Manager): | ||
|
r22 | def create_post(self, title, text, image=None, parent_id=NO_PARENT, | ||
|
r24 | ip=NO_IP, tags=None): | ||
|
r22 | post = self.create(title=title, | ||
text=text, | ||||
pub_time=timezone.now(), | ||||
parent=parent_id, | ||||
image=image, | ||||
poster_ip=ip, | ||||
|
r25 | poster_user_agent=UNKNOWN_UA, | ||
last_edit_time=timezone.now()) | ||||
|
r0 | |||
|
r24 | if tags: | ||
|
r79 | map(post.tags.add, tags) | ||
|
r24 | |||
|
r25 | if parent_id != NO_PARENT: | ||
|
r38 | self._bump_thread(parent_id) | ||
|
r28 | else: | ||
self._delete_old_threads() | ||||
|
r25 | |||
|
r1 | return post | ||
|
r0 | |||
def delete_post(self, post): | ||||
|
r22 | children = self.filter(parent=post.id) | ||
|
r3 | for child in children: | ||
self.delete_post(child) | ||||
|
r2 | post.delete() | ||
|
r0 | def delete_posts_by_ip(self, ip): | ||
|
r22 | posts = self.filter(poster_ip=ip) | ||
|
r1 | for post in posts: | ||
self.delete_post(post) | ||||
|
r0 | |||
|
r87 | def get_threads(self, tag=None, page=ALL_PAGES, | ||
order_by='-last_edit_time'): | ||||
|
r27 | if tag: | ||
threads = self.filter(parent=NO_PARENT, tags=tag) | ||||
|
r87 | |||
# TODO Throw error 404 if no threads for tag found? | ||||
|
r27 | else: | ||
|
r22 | threads = self.filter(parent=NO_PARENT) | ||
|
r79 | |||
|
r87 | threads = threads.order_by(order_by) | ||
|
r5 | |||
|
r71 | if page != ALL_PAGES: | ||
|
r46 | thread_count = len(threads) | ||
if page < self.get_thread_page_count(tag=tag): | ||||
start_thread = page * settings.THREADS_PER_PAGE | ||||
end_thread = min(start_thread + settings.THREADS_PER_PAGE, | ||||
thread_count) | ||||
threads = threads[start_thread:end_thread] | ||||
|
r5 | return threads | ||
def get_thread(self, opening_post_id): | ||||
|
r79 | try: | ||
|
r86 | opening_post = self.get(id=opening_post_id, parent=NO_PARENT) | ||
|
r79 | except Post.DoesNotExist: | ||
|
r71 | raise Http404 | ||
|
r33 | if opening_post.parent == NO_PARENT: | ||
replies = self.filter(parent=opening_post_id) | ||||
|
r5 | |||
|
r33 | thread = [opening_post] | ||
thread.extend(replies) | ||||
|
r17 | |||
|
r33 | return thread | ||
|
r5 | |||
|
r8 | def exists(self, post_id): | ||
|
r22 | posts = self.filter(id=post_id) | ||
|
r8 | |||
|
r58 | return posts.count() > 0 | ||
|
r8 | |||
|
r46 | def get_thread_page_count(self, tag=None): | ||
if tag: | ||||
threads = self.filter(parent=NO_PARENT, tags=tag) | ||||
else: | ||||
threads = self.filter(parent=NO_PARENT) | ||||
|
r58 | return int(math.ceil(threads.count() / float( | ||
settings.THREADS_PER_PAGE))) | ||||
|
r46 | |||
|
r28 | def _delete_old_threads(self): | ||
""" | ||||
Preserves maximum thread count. If there are too many threads, | ||||
delete the old ones. | ||||
""" | ||||
# TODO Move old threads to the archive instead of deleting them. | ||||
# Maybe make some 'old' field in the model to indicate the thread | ||||
# must not be shown and be able for replying. | ||||
threads = self.get_threads() | ||||
thread_count = len(threads) | ||||
if thread_count > settings.MAX_THREAD_COUNT: | ||||
num_threads_to_delete = thread_count - settings.MAX_THREAD_COUNT | ||||
|
r58 | old_threads = threads[thread_count - num_threads_to_delete:] | ||
|
r28 | |||
for thread in old_threads: | ||||
self.delete_post(thread) | ||||
|
r38 | def _bump_thread(self, thread_id): | ||
thread = self.get(id=thread_id) | ||||
|
r46 | if thread.can_bump(): | ||
|
r38 | thread.last_edit_time = timezone.now() | ||
thread.save() | ||||
|
r5 | |||
|
r33 | class TagManager(models.Manager): | ||
def get_not_empty_tags(self): | ||||
|
r34 | all_tags = self.all().order_by('name') | ||
|
r33 | tags = [] | ||
for tag in all_tags: | ||||
if not tag.is_empty(): | ||||
tags.append(tag) | ||||
return tags | ||||
|
r57 | def get_popular_tags(self): | ||
all_tags = self.get_not_empty_tags() | ||||
sorted_tags = sorted(all_tags, key=lambda tag: tag.get_popularity(), | ||||
reverse=True) | ||||
return sorted_tags[:settings.POPULAR_TAGS] | ||||
|
r33 | |||
|
r4 | class Tag(models.Model): | ||
""" | ||||
A tag is a text node assigned to the post. The tag serves as a board | ||||
section. There can be multiple tags for each message | ||||
""" | ||||
|
r33 | objects = TagManager() | ||
|
r22 | name = models.CharField(max_length=100) | ||
|
r28 | # TODO Connect the tag to its posts to check the number of threads for | ||
# the tag. | ||||
|
r4 | |||
|
r31 | def __unicode__(self): | ||
return self.name | ||||
|
r32 | def is_empty(self): | ||
|
r33 | return self.get_post_count() == 0 | ||
def get_post_count(self): | ||||
|
r32 | posts_with_tag = Post.objects.get_threads(tag=self) | ||
|
r58 | return posts_with_tag.count() | ||
|
r32 | |||
|
r57 | def get_popularity(self): | ||
posts_with_tag = Post.objects.get_threads(tag=self) | ||||
reply_count = 0 | ||||
for post in posts_with_tag: | ||||
reply_count += post.get_reply_count() | ||||
|
r85 | reply_count += OPENING_POST_POPULARITY_WEIGHT | ||
|
r57 | |||
return reply_count | ||||
|
r8 | |||
|
r2 | class Post(models.Model): | ||
|
r4 | """A post is a message.""" | ||
|
r0 | |||
|
r2 | objects = PostManager() | ||
|
r0 | |||
|
r50 | def _update_image_filename(self, filename): | ||
"""Get unique image filename""" | ||||
|
r71 | path = IMAGES_DIRECTORY | ||
|
r50 | new_name = str(int(time.mktime(time.gmtime()))) | ||
new_name += str(int(random() * 1000)) | ||||
|
r71 | new_name += FILE_EXTENSION_DELIMITER | ||
new_name += filename.split(FILE_EXTENSION_DELIMITER)[-1:][0] | ||||
|
r50 | |||
return os.path.join(path, new_name) | ||||
|
r79 | title = models.CharField(max_length=TITLE_MAX_LENGTH) | ||
|
r2 | pub_time = models.DateTimeField() | ||
|
r79 | text = MarkupField(default_markup_type=DEFAULT_MARKUP_TYPE, | ||
|
r90 | escape_html=False) | ||
|
r50 | image = thumbs.ImageWithThumbsField(upload_to=_update_image_filename, | ||
|
r79 | blank=True, sizes=(IMAGE_THUMB_SIZE,)) | ||
|
r2 | poster_ip = models.IPAddressField() | ||
poster_user_agent = models.TextField() | ||||
parent = models.BigIntegerField() | ||||
|
r7 | tags = models.ManyToManyField(Tag) | ||
|
r25 | last_edit_time = models.DateTimeField() | ||
|
r0 | |||
def __unicode__(self): | ||||
|
r79 | return '#' + str(self.id) + ' ' + self.title + ' (' + self.text.raw + \ | ||
')' | ||||
|
r9 | |||
|
r30 | def _get_replies(self): | ||
return Post.objects.filter(parent=self.id) | ||||
def get_reply_count(self): | ||||
|
r58 | return self._get_replies().count() | ||
|
r30 | |||
def get_images_count(self): | ||||
images_count = 1 if self.image else 0 | ||||
for reply in self._get_replies(): | ||||
if reply.image: | ||||
images_count += 1 | ||||
return images_count | ||||
def get_gets_count(self): | ||||
gets_count = 1 if self.is_get() else 0 | ||||
for reply in self._get_replies(): | ||||
if reply.is_get(): | ||||
gets_count += 1 | ||||
return gets_count | ||||
def is_get(self): | ||||
"""If the post has pretty id (1, 1000, 77777), than it is called GET""" | ||||
|
r39 | first = self.id == 1 | ||
|
r30 | |||
|
r46 | id_str = str(self.id) | ||
|
r71 | pretty = REGEX_PRETTY.match(id_str) | ||
same_digits = REGEX_SAME.match(id_str) | ||||
|
r46 | |||
return first or pretty or same_digits | ||||
|
r31 | |||
|
r46 | def can_bump(self): | ||
"""Check if the thread can be bumped by replying""" | ||||
replies_count = len(Post.objects.get_thread(self.id)) | ||||
return replies_count <= settings.MAX_POSTS_PER_THREAD | ||||
|
r31 | |||
|
r59 | def get_last_replies(self): | ||
if settings.LAST_REPLIES_COUNT > 0: | ||||
reply_count = self.get_reply_count() | ||||
if reply_count > 0: | ||||
reply_count_to_show = min(settings.LAST_REPLIES_COUNT, | ||||
reply_count) | ||||
last_replies = self._get_replies()[reply_count | ||||
- reply_count_to_show:] | ||||
return last_replies | ||||
|
r9 | |||
|
r11 | class Admin(models.Model): | ||
|
r9 | """ | ||
Model for admin users | ||||
""" | ||||
|
r22 | name = models.CharField(max_length=100) | ||
password = models.CharField(max_length=100) | ||||
|
r9 | |||
def __unicode__(self): | ||||
|
r72 | return self.name + '/' + '*' * len(self.password) | ||
|
r110 | |||
class Setting(models.Model): | ||||
name = models.CharField(max_length=50) | ||||
value = models.CharField(max_length=50) | ||||
class User(models.Model): | ||||
user_id = models.CharField(max_length=20) | ||||
settings = models.ManyToManyField(Setting) | ||||
def save_setting(self, name, value): | ||||
setting, created = self.settings.get_or_create(name=name) | ||||
setting.value = value | ||||
setting.save() | ||||
return setting | ||||
def get_setting(self, name): | ||||
settings = self.settings.filter(name=name) | ||||
if len(settings) > 0: | ||||
setting = settings[0] | ||||
else: | ||||
setting = None | ||||
if setting: | ||||
setting_value = setting.value | ||||
else: | ||||
setting_value = None | ||||
return setting_value | ||||