|
|
import uuid
|
|
|
|
|
|
import hashlib
|
|
|
import re
|
|
|
from django.db import models
|
|
|
from django.db.models import TextField
|
|
|
from django.template.defaultfilters import truncatewords, striptags
|
|
|
from django.template.loader import render_to_string
|
|
|
from django.urls import reverse
|
|
|
|
|
|
from boards.abstracts.constants import REGEX_REPLY
|
|
|
from boards.abstracts.tripcode import Tripcode
|
|
|
from boards.models import Attachment, KeyPair, GlobalId
|
|
|
from boards.models.attachment import FILE_TYPES_IMAGE
|
|
|
from boards.models.base import Viewable
|
|
|
from boards.models.post.export import get_exporter, DIFF_TYPE_JSON
|
|
|
from boards.models.post.manager import PostManager, NO_IP
|
|
|
from boards.utils import datetime_to_epoch
|
|
|
|
|
|
CSS_CLS_HIDDEN_POST = 'hidden_post'
|
|
|
CSS_CLS_DEAD_POST = 'dead_post'
|
|
|
CSS_CLS_ARCHIVE_POST = 'archive_post'
|
|
|
CSS_CLS_POST = 'post'
|
|
|
CSS_CLS_MONOCHROME = 'monochrome'
|
|
|
|
|
|
CSS_CLASS_DELIMITER = ' '
|
|
|
|
|
|
TITLE_MAX_WORDS = 10
|
|
|
|
|
|
APP_LABEL_BOARDS = 'boards'
|
|
|
|
|
|
BAN_REASON_AUTO = 'Auto'
|
|
|
|
|
|
TITLE_MAX_LENGTH = 200
|
|
|
|
|
|
REGEX_GLOBAL_REPLY = re.compile(r'\[post\](\w+)::([^:]+)::(\d+)\[/post\]')
|
|
|
REGEX_URL = re.compile(r'https?\://[a-zA-Z0-9\-\.]+\.[a-zA-Z]{2,3}(/\S*)?')
|
|
|
REGEX_NOTIFICATION = re.compile(r'\[user\](\w+)\[/user\]')
|
|
|
|
|
|
PARAMETER_TRUNCATED = 'truncated'
|
|
|
PARAMETER_TAG = 'tag'
|
|
|
PARAMETER_OFFSET = 'offset'
|
|
|
PARAMETER_DIFF_TYPE = 'type'
|
|
|
PARAMETER_CSS_CLASS = 'css_class'
|
|
|
PARAMETER_THREAD = 'thread'
|
|
|
PARAMETER_IS_OPENING = 'is_opening'
|
|
|
PARAMETER_POST = 'post'
|
|
|
PARAMETER_OP_ID = 'opening_post_id'
|
|
|
PARAMETER_NEED_OPEN_LINK = 'need_open_link'
|
|
|
PARAMETER_REPLY_LINK = 'reply_link'
|
|
|
PARAMETER_NEED_OP_DATA = 'need_op_data'
|
|
|
|
|
|
POST_VIEW_PARAMS = (
|
|
|
'need_op_data',
|
|
|
'reply_link',
|
|
|
'need_open_link',
|
|
|
'truncated',
|
|
|
'mode_tree',
|
|
|
'perms',
|
|
|
'tree_depth',
|
|
|
)
|
|
|
|
|
|
TEMPLATE_POST = 'boards/post.html'
|
|
|
|
|
|
|
|
|
class Post(models.Model, Viewable):
|
|
|
"""A post is a message."""
|
|
|
|
|
|
objects = PostManager()
|
|
|
|
|
|
class Meta:
|
|
|
app_label = APP_LABEL_BOARDS
|
|
|
ordering = ('id',)
|
|
|
|
|
|
title = models.CharField(max_length=TITLE_MAX_LENGTH, blank=True, default='')
|
|
|
pub_time = models.DateTimeField(db_index=True)
|
|
|
text = TextField(blank=True, default='')
|
|
|
_text_rendered = TextField(blank=True, null=True, editable=False)
|
|
|
|
|
|
attachments = models.ManyToManyField(Attachment, blank=True,
|
|
|
related_name='attachment_posts')
|
|
|
|
|
|
poster_ip = models.GenericIPAddressField()
|
|
|
|
|
|
# Used for cache and threads updating
|
|
|
last_edit_time = models.DateTimeField()
|
|
|
|
|
|
referenced_posts = models.ManyToManyField('Post', symmetrical=False,
|
|
|
blank=True, related_name='refposts',
|
|
|
db_index=True)
|
|
|
refmap = models.TextField(null=True, blank=True)
|
|
|
thread = models.ForeignKey('Thread', on_delete=models.CASCADE,
|
|
|
db_index=True, related_name='replies')
|
|
|
|
|
|
url = models.TextField()
|
|
|
uid = models.TextField()
|
|
|
|
|
|
# Global ID with author key. If the message was downloaded from another
|
|
|
# server, this indicates the server.
|
|
|
global_id = models.OneToOneField(GlobalId, null=True, blank=True,
|
|
|
on_delete=models.CASCADE)
|
|
|
|
|
|
tripcode = models.CharField(max_length=50, blank=True, default='')
|
|
|
opening = models.BooleanField(db_index=True)
|
|
|
hidden = models.BooleanField(default=False)
|
|
|
|
|
|
def __str__(self):
|
|
|
return 'P#{}/{}'.format(self.id, self.get_title())
|
|
|
|
|
|
def get_title(self) -> str:
|
|
|
return self.title
|
|
|
|
|
|
def get_title_or_text(self):
|
|
|
title = self.get_title()
|
|
|
if not title:
|
|
|
title = truncatewords(striptags(self.get_text()), TITLE_MAX_WORDS)
|
|
|
|
|
|
return title
|
|
|
|
|
|
def build_refmap(self, excluded_ids=None) -> None:
|
|
|
"""
|
|
|
Builds a replies map string from replies list. This is a cache to stop
|
|
|
the server from recalculating the map on every post show.
|
|
|
"""
|
|
|
|
|
|
replies = self.referenced_posts
|
|
|
if excluded_ids is not None:
|
|
|
replies = replies.exclude(id__in=excluded_ids)
|
|
|
else:
|
|
|
replies = replies.all()
|
|
|
|
|
|
post_urls = [refpost.get_link_view() for refpost in replies]
|
|
|
|
|
|
self.refmap = ', '.join(post_urls)
|
|
|
|
|
|
def is_referenced(self) -> bool:
|
|
|
return self.refmap and len(self.refmap) > 0
|
|
|
|
|
|
def is_opening(self) -> bool:
|
|
|
"""
|
|
|
Checks if this is an opening post or just a reply.
|
|
|
"""
|
|
|
|
|
|
return self.opening
|
|
|
|
|
|
def get_absolute_url(self, thread=None):
|
|
|
# Url is cached only for the "main" thread. When getting url
|
|
|
# for other threads, do it manually.
|
|
|
return self.url or self._build_url()
|
|
|
|
|
|
def get_thread(self):
|
|
|
return self.thread
|
|
|
|
|
|
def get_thread_id(self):
|
|
|
return self.thread_id
|
|
|
|
|
|
def _get_cache_key(self):
|
|
|
return [datetime_to_epoch(self.last_edit_time)]
|
|
|
|
|
|
def get_view_params(self, *args, **kwargs):
|
|
|
"""
|
|
|
Gets the parameters required for viewing the post based on the arguments
|
|
|
given and the post itself.
|
|
|
"""
|
|
|
thread = kwargs.get('thread') or self.get_thread()
|
|
|
|
|
|
css_classes = [CSS_CLS_POST]
|
|
|
if thread.is_archived():
|
|
|
css_classes.append(CSS_CLS_ARCHIVE_POST)
|
|
|
elif not thread.can_bump():
|
|
|
css_classes.append(CSS_CLS_DEAD_POST)
|
|
|
if self.is_hidden():
|
|
|
css_classes.append(CSS_CLS_HIDDEN_POST)
|
|
|
if thread.is_monochrome():
|
|
|
css_classes.append(CSS_CLS_MONOCHROME)
|
|
|
|
|
|
params = dict()
|
|
|
for param in POST_VIEW_PARAMS:
|
|
|
if param in kwargs:
|
|
|
params[param] = kwargs[param]
|
|
|
|
|
|
params.update({
|
|
|
PARAMETER_POST: self,
|
|
|
PARAMETER_IS_OPENING: self.is_opening(),
|
|
|
PARAMETER_THREAD: thread,
|
|
|
PARAMETER_CSS_CLASS: CSS_CLASS_DELIMITER.join(css_classes),
|
|
|
})
|
|
|
|
|
|
return params
|
|
|
|
|
|
def get_view(self, *args, **kwargs) -> str:
|
|
|
"""
|
|
|
Renders post's HTML view. Some of the post params can be passed over
|
|
|
kwargs for the means of caching (if we view the thread, some params
|
|
|
are same for every post and don't need to be computed over and over.
|
|
|
"""
|
|
|
params = self.get_view_params(*args, **kwargs)
|
|
|
|
|
|
return render_to_string(TEMPLATE_POST, params)
|
|
|
|
|
|
def get_images(self) -> Attachment:
|
|
|
return self.attachments.filter(mimetype__in=FILE_TYPES_IMAGE)
|
|
|
|
|
|
def get_first_image(self) -> Attachment:
|
|
|
try:
|
|
|
return self.get_images().earliest('-id')
|
|
|
except Attachment.DoesNotExist:
|
|
|
return None
|
|
|
|
|
|
def set_global_id(self, key_pair=None):
|
|
|
"""
|
|
|
Sets global id based on the given key pair. If no key pair is given,
|
|
|
default one is used.
|
|
|
"""
|
|
|
|
|
|
if key_pair:
|
|
|
key = key_pair
|
|
|
else:
|
|
|
try:
|
|
|
key = KeyPair.objects.get(primary=True)
|
|
|
except KeyPair.DoesNotExist:
|
|
|
# Do not update the global id because there is no key defined
|
|
|
return
|
|
|
global_id = GlobalId(key_type=key.key_type,
|
|
|
key=key.public_key,
|
|
|
local_id=self.id)
|
|
|
global_id.save()
|
|
|
|
|
|
self.global_id = global_id
|
|
|
|
|
|
self.save(update_fields=['global_id'])
|
|
|
|
|
|
def get_pub_time_str(self):
|
|
|
return str(self.pub_time)
|
|
|
|
|
|
def get_replied_ids(self):
|
|
|
"""
|
|
|
Gets ID list of the posts that this post replies.
|
|
|
"""
|
|
|
|
|
|
raw_text = self.get_raw_text()
|
|
|
|
|
|
local_replied = REGEX_REPLY.findall(raw_text)
|
|
|
global_replied = []
|
|
|
for match in REGEX_GLOBAL_REPLY.findall(raw_text):
|
|
|
key_type = match[0]
|
|
|
key = match[1]
|
|
|
local_id = match[2]
|
|
|
|
|
|
try:
|
|
|
global_id = GlobalId.objects.get(key_type=key_type,
|
|
|
key=key, local_id=local_id)
|
|
|
for post in Post.objects.filter(global_id=global_id).only('id'):
|
|
|
global_replied.append(post.id)
|
|
|
except GlobalId.DoesNotExist:
|
|
|
pass
|
|
|
return local_replied + global_replied
|
|
|
|
|
|
def get_post_data(self, format_type=DIFF_TYPE_JSON, request=None,
|
|
|
include_last_update=False) -> str:
|
|
|
"""
|
|
|
Gets post HTML or JSON data that can be rendered on a page or used by
|
|
|
API.
|
|
|
"""
|
|
|
|
|
|
return get_exporter(format_type).export(self, request,
|
|
|
include_last_update)
|
|
|
|
|
|
def _build_url(self):
|
|
|
if self.id:
|
|
|
opening = self.is_opening()
|
|
|
opening_id = self.id if opening else self.get_thread().get_opening_post_id()
|
|
|
url = reverse('thread', kwargs={'post_id': opening_id})
|
|
|
if not opening:
|
|
|
url += '#' + str(self.id)
|
|
|
|
|
|
return url
|
|
|
|
|
|
def save(self, force_insert=False, force_update=False, using=None,
|
|
|
update_fields=None):
|
|
|
new_post = self.id is None
|
|
|
|
|
|
self.uid = str(uuid.uuid4())
|
|
|
if update_fields is not None and 'uid' not in update_fields:
|
|
|
update_fields += ['uid']
|
|
|
|
|
|
if not new_post:
|
|
|
thread = self.get_thread()
|
|
|
if thread:
|
|
|
thread.last_edit_time = self.last_edit_time
|
|
|
thread.save(update_fields=['last_edit_time', 'status'])
|
|
|
|
|
|
super().save(force_insert, force_update, using, update_fields)
|
|
|
|
|
|
if new_post:
|
|
|
if not self.url:
|
|
|
self.url = self._build_url()
|
|
|
super().save(update_fields=['url'])
|
|
|
|
|
|
def get_text(self) -> str:
|
|
|
return self._text_rendered
|
|
|
|
|
|
def get_raw_text(self) -> str:
|
|
|
return self.text
|
|
|
|
|
|
def get_sync_text(self) -> str:
|
|
|
"""
|
|
|
Returns text applicable for sync. It has absolute post reflinks.
|
|
|
"""
|
|
|
|
|
|
replacements = dict()
|
|
|
for post_id in REGEX_REPLY.findall(self.get_raw_text()):
|
|
|
try:
|
|
|
absolute_post_id = str(Post.objects.get(id=post_id).global_id)
|
|
|
replacements[post_id] = absolute_post_id
|
|
|
except Post.DoesNotExist:
|
|
|
pass
|
|
|
|
|
|
text = self.get_raw_text() or ''
|
|
|
for key in replacements:
|
|
|
text = text.replace('[post]{}[/post]'.format(key),
|
|
|
'[post]{}[/post]'.format(replacements[key]))
|
|
|
text = text.replace('\r\n', '\n').replace('\r', '\n')
|
|
|
|
|
|
return text
|
|
|
|
|
|
def get_tripcode(self):
|
|
|
if self.tripcode:
|
|
|
return Tripcode(self.tripcode)
|
|
|
|
|
|
def get_link_view(self):
|
|
|
"""
|
|
|
Gets view of a reflink to the post.
|
|
|
"""
|
|
|
result = '<a href="{}">>>{}</a>'.format(self.get_absolute_url(),
|
|
|
self.id)
|
|
|
if self.is_opening():
|
|
|
result = '<b>{}</b>'.format(result)
|
|
|
|
|
|
return result
|
|
|
|
|
|
def is_hidden(self) -> bool:
|
|
|
return self.hidden
|
|
|
|
|
|
def set_hidden(self, hidden):
|
|
|
self.hidden = hidden
|
|
|
|
|
|
def clear_cache(self):
|
|
|
"""
|
|
|
Clears sync data (content cache, signatures etc).
|
|
|
"""
|
|
|
global_id = self.global_id
|
|
|
if global_id is not None and global_id.is_local()\
|
|
|
and global_id.content is not None:
|
|
|
global_id.clear_cache()
|
|
|
|
|
|
def get_tags(self):
|
|
|
return self.get_thread().get_tags()
|
|
|
|
|
|
def get_ip_color(self):
|
|
|
return hashlib.md5(self.poster_ip.encode()).hexdigest()[:6]
|
|
|
|
|
|
def has_ip(self):
|
|
|
return self.poster_ip != NO_IP
|
|
|
|
|
|
def has_global_id(self):
|
|
|
return self.global_id_id is not None
|
|
|
|
