__init__.py
360 lines
| 11.2 KiB
| text/x-python
|
PythonLexer
|
r1156 | import uuid | ||
|
r1986 | |||
|
r1638 | import hashlib | ||
import re | ||||
|
r1986 | 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 | ||||
|
r1156 | |||
|
r1296 | from boards.abstracts.tripcode import Tripcode | ||
|
r1590 | from boards.models import Attachment, KeyPair, GlobalId | ||
from boards.models.attachment import FILE_TYPES_IMAGE | ||||
|
r1156 | from boards.models.base import Viewable | ||
from boards.models.post.export import get_exporter, DIFF_TYPE_JSON | ||||
|
r1678 | from boards.models.post.manager import PostManager, NO_IP | ||
|
r1533 | from boards.utils import datetime_to_epoch | ||
|
r1156 | |||
|
r1366 | CSS_CLS_HIDDEN_POST = 'hidden_post' | ||
CSS_CLS_DEAD_POST = 'dead_post' | ||||
CSS_CLS_ARCHIVE_POST = 'archive_post' | ||||
CSS_CLS_POST = 'post' | ||||
|
r1434 | CSS_CLS_MONOCHROME = 'monochrome' | ||
|
r1366 | |||
|
r1358 | TITLE_MAX_WORDS = 10 | ||
|
r1156 | APP_LABEL_BOARDS = 'boards' | ||
BAN_REASON_AUTO = 'Auto' | ||||
TITLE_MAX_LENGTH = 200 | ||||
REGEX_REPLY = re.compile(r'\[post\](\d+)\[/post\]') | ||||
|
r1157 | 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*)?') | ||||
|
r1156 | 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' | ||||
|
r1166 | PARAMETER_NEED_OP_DATA = 'need_op_data' | ||
|
r1156 | |||
|
r1180 | POST_VIEW_PARAMS = ( | ||
'need_op_data', | ||||
'reply_link', | ||||
'need_open_link', | ||||
'truncated', | ||||
'mode_tree', | ||||
|
r1388 | 'perms', | ||
|
r1473 | 'tree_depth', | ||
|
r1180 | ) | ||
|
r1358 | |||
|
r1156 | class Post(models.Model, Viewable): | ||
"""A post is a message.""" | ||||
objects = PostManager() | ||||
class Meta: | ||||
app_label = APP_LABEL_BOARDS | ||||
ordering = ('id',) | ||||
|
r1751 | title = models.CharField(max_length=TITLE_MAX_LENGTH, blank=True, default='') | ||
|
r1667 | pub_time = models.DateTimeField(db_index=True) | ||
|
r1751 | text = TextField(blank=True, default='') | ||
|
r1156 | _text_rendered = TextField(blank=True, null=True, editable=False) | ||
|
r1273 | attachments = models.ManyToManyField(Attachment, null=True, blank=True, | ||
|
r1358 | related_name='attachment_posts') | ||
|
r1156 | |||
poster_ip = models.GenericIPAddressField() | ||||
|
r1601 | # Used for cache and threads updating | ||
|
r1156 | last_edit_time = models.DateTimeField() | ||
referenced_posts = models.ManyToManyField('Post', symmetrical=False, | ||||
null=True, | ||||
|
r1180 | blank=True, related_name='refposts', | ||
|
r1156 | db_index=True) | ||
refmap = models.TextField(null=True, blank=True) | ||||
|
r1986 | thread = models.ForeignKey('Thread', on_delete=models.CASCADE, | ||
db_index=True, related_name='replies') | ||||
|
r1156 | |||
url = models.TextField() | ||||
|
r1835 | uid = models.TextField() | ||
|
r1156 | |||
|
r1157 | # Global ID with author key. If the message was downloaded from another | ||
# server, this indicates the server. | ||||
|
r1511 | global_id = models.OneToOneField(GlobalId, null=True, blank=True, | ||
on_delete=models.CASCADE) | ||||
|
r1157 | |||
|
r1367 | tripcode = models.CharField(max_length=50, blank=True, default='') | ||
|
r1381 | opening = models.BooleanField(db_index=True) | ||
|
r1366 | hidden = models.BooleanField(default=False) | ||
|
r1293 | |||
|
r1156 | def __str__(self): | ||
|
r1358 | return 'P#{}/{}'.format(self.id, self.get_title()) | ||
|
r1156 | |||
def get_title(self) -> str: | ||||
|
r1358 | return self.title | ||
|
r1156 | |||
|
r1358 | def get_title_or_text(self): | ||
title = self.get_title() | ||||
|
r1156 | if not title: | ||
|
r1358 | title = truncatewords(striptags(self.get_text()), TITLE_MAX_WORDS) | ||
|
r1156 | |||
return title | ||||
|
r1619 | def build_refmap(self, excluded_ids=None) -> None: | ||
|
r1156 | """ | ||
Builds a replies map string from replies list. This is a cache to stop | ||||
the server from recalculating the map on every post show. | ||||
""" | ||||
|
r1619 | 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] | ||||
|
r1156 | |||
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. | ||||
""" | ||||
|
r1337 | return self.opening | ||
|
r1156 | |||
|
r1365 | def get_absolute_url(self, thread=None): | ||
|
r1668 | # Url is cached only for the "main" thread. When getting url | ||
# for other threads, do it manually. | ||||
|
r1669 | return self.url | ||
|
r1192 | |||
|
r1156 | def get_thread(self): | ||
return self.thread | ||||
|
r1455 | def get_thread_id(self): | ||
return self.thread_id | ||||
|
r1531 | def _get_cache_key(self): | ||
return [datetime_to_epoch(self.last_edit_time)] | ||||
|
r1648 | def get_view_params(self, *args, **kwargs): | ||
|
r1651 | """ | ||
Gets the parameters required for viewing the post based on the arguments | ||||
given and the post itself. | ||||
""" | ||||
|
r1670 | thread = kwargs.get('thread') or self.get_thread() | ||
|
r1156 | |||
|
r1366 | css_classes = [CSS_CLS_POST] | ||
|
r1414 | if thread.is_archived(): | ||
|
r1366 | css_classes.append(CSS_CLS_ARCHIVE_POST) | ||
|
r1156 | elif not thread.can_bump(): | ||
|
r1366 | css_classes.append(CSS_CLS_DEAD_POST) | ||
if self.is_hidden(): | ||||
css_classes.append(CSS_CLS_HIDDEN_POST) | ||||
|
r1434 | if thread.is_monochrome(): | ||
css_classes.append(CSS_CLS_MONOCHROME) | ||||
|
r1156 | |||
|
r1180 | params = dict() | ||
for param in POST_VIEW_PARAMS: | ||||
if param in kwargs: | ||||
params[param] = kwargs[param] | ||||
params.update({ | ||||
|
r1156 | PARAMETER_POST: self, | ||
|
r1337 | PARAMETER_IS_OPENING: self.is_opening(), | ||
|
r1156 | PARAMETER_THREAD: thread, | ||
|
r1366 | PARAMETER_CSS_CLASS: ' '.join(css_classes), | ||
|
r1156 | }) | ||
|
r1648 | 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. | ||||
""" | ||||
|
r1650 | params = self.get_view_params(*args, **kwargs) | ||
|
r1648 | |||
|
r1180 | return render_to_string('boards/post.html', params) | ||
|
r1755 | def get_images(self) -> Attachment: | ||
return self.attachments.filter(mimetype__in=FILE_TYPES_IMAGE) | ||||
|
r1590 | def get_first_image(self) -> Attachment: | ||
|
r1725 | try: | ||
|
r1755 | return self.get_images().earliest('-id') | ||
|
r1725 | except Attachment.DoesNotExist: | ||
|
r1755 | return None | ||
|
r1156 | |||
|
r1157 | def set_global_id(self, key_pair=None): | ||
|
r1156 | """ | ||
|
r1157 | Sets global id based on the given key pair. If no key pair is given, | ||
default one is used. | ||||
|
r1156 | """ | ||
|
r1157 | 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, | ||||
|
r1229 | local_id=self.id) | ||
|
r1157 | global_id.save() | ||
self.global_id = global_id | ||||
self.save(update_fields=['global_id']) | ||||
|
r1229 | def get_pub_time_str(self): | ||
return str(self.pub_time) | ||||
|
r1156 | |||
|
r1157 | def get_replied_ids(self): | ||
""" | ||||
Gets ID list of the posts that this post replies. | ||||
""" | ||||
raw_text = self.get_raw_text() | ||||
|
r1278 | |||
|
r1157 | 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] | ||||
|
r1156 | |||
|
r1157 | try: | ||
global_id = GlobalId.objects.get(key_type=key_type, | ||||
|
r1241 | key=key, local_id=local_id) | ||
|
r1157 | 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 | ||||
|
r1156 | |||
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) | ||||
|
r1669 | def _build_url(self): | ||
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 | ||||
|
r1156 | |||
def save(self, force_insert=False, force_update=False, using=None, | ||||
update_fields=None): | ||||
|
r1440 | new_post = self.id is None | ||
|
r1156 | self.uid = str(uuid.uuid4()) | ||
if update_fields is not None and 'uid' not in update_fields: | ||||
update_fields += ['uid'] | ||||
|
r1440 | if not new_post: | ||
|
r1704 | thread = self.get_thread() | ||
if thread: | ||||
|
r1156 | thread.last_edit_time = self.last_edit_time | ||
|
r1414 | thread.save(update_fields=['last_edit_time', 'status']) | ||
|
r1156 | |||
super().save(force_insert, force_update, using, update_fields) | ||||
|
r1669 | if new_post: | ||
self.url = self._build_url() | ||||
super().save(update_fields=['url']) | ||||
|
r1440 | |||
|
r1156 | def get_text(self) -> str: | ||
return self._text_rendered | ||||
def get_raw_text(self) -> str: | ||||
return self.text | ||||
|
r1228 | 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()): | ||||
|
r1575 | try: | ||
absolute_post_id = str(Post.objects.get(id=post_id).global_id) | ||||
replacements[post_id] = absolute_post_id | ||||
except Post.DoesNotExist: | ||||
pass | ||||
|
r1228 | |||
|
r1511 | text = self.get_raw_text() or '' | ||
|
r1228 | for key in replacements: | ||
text = text.replace('[post]{}[/post]'.format(key), | ||||
'[post]{}[/post]'.format(replacements[key])) | ||||
|
r1504 | text = text.replace('\r\n', '\n').replace('\r', '\n') | ||
|
r1228 | |||
return text | ||||
|
r1296 | def get_tripcode(self): | ||
if self.tripcode: | ||||
return Tripcode(self.tripcode) | ||||
|
r1309 | |||
def get_link_view(self): | ||||
""" | ||||
Gets view of a reflink to the post. | ||||
""" | ||||
result = '<a href="{}">>>{}</a>'.format(self.get_absolute_url(), | ||||
|
r1323 | self.id) | ||
|
r1309 | if self.is_opening(): | ||
result = '<b>{}</b>'.format(result) | ||||
return result | ||||
|
r1366 | |||
def is_hidden(self) -> bool: | ||||
return self.hidden | ||||
def set_hidden(self, hidden): | ||||
self.hidden = hidden | ||||
|
r1569 | |||
def clear_cache(self): | ||||
|
r1586 | """ | ||
Clears sync data (content cache, signatures etc). | ||||
""" | ||||
|
r1569 | global_id = self.global_id | ||
if global_id is not None and global_id.is_local()\ | ||||
and global_id.content is not None: | ||||
|
r1586 | global_id.clear_cache() | ||
|
r1620 | |||
def get_tags(self): | ||||
return self.get_thread().get_tags() | ||||
|
r1638 | |||
def get_ip_color(self): | ||||
return hashlib.md5(self.poster_ip.encode()).hexdigest()[:6] | ||||
|
r1678 | |||
def has_ip(self): | ||||
return self.poster_ip != NO_IP | ||||