forms.py
369 lines
| 11.5 KiB
| text/x-python
|
PythonLexer
/ boards / forms.py
neko259
|
r69 | import re | ||
neko259
|
r527 | import time | ||
neko259
|
r1065 | import pytz | ||
neko259
|
r527 | |||
Ilyas
|
r14 | from django import forms | ||
neko259
|
r954 | from django.core.files.uploadedfile import SimpleUploadedFile | ||
neko259
|
r1077 | from django.core.exceptions import ObjectDoesNotExist | ||
neko259
|
r76 | from django.forms.util import ErrorList | ||
neko259
|
r205 | from django.utils.translation import ugettext_lazy as _ | ||
neko259
|
r954 | import requests | ||
neko259
|
r527 | |||
neko259
|
r438 | from boards.mdx_neboard import formatters | ||
neko259
|
r386 | from boards.models.post import TITLE_MAX_LENGTH | ||
neko259
|
r1077 | from boards.models import Tag, Post | ||
neko259
|
r35 | from neboard import settings | ||
neko259
|
r333 | import boards.settings as board_settings | ||
neko259
|
r76 | |||
neko259
|
r954 | |||
CONTENT_TYPE_IMAGE = ( | ||||
'image/jpeg', | ||||
'image/png', | ||||
'image/gif', | ||||
'image/bmp', | ||||
) | ||||
neko259
|
r937 | REGEX_TAGS = re.compile(r'^[\w\s\d]+$', re.UNICODE) | ||
neko259
|
r642 | VETERAN_POSTING_DELAY = 5 | ||
neko259
|
r517 | ATTRIBUTE_PLACEHOLDER = 'placeholder' | ||
neko259
|
r1006 | ATTRIBUTE_ROWS = 'rows' | ||
neko259
|
r517 | |||
LAST_POST_TIME = 'last_post_time' | ||||
LAST_LOGIN_TIME = 'last_login_time' | ||||
neko259
|
r937 | TEXT_PLACEHOLDER = _('Type message here. Use formatting panel for more advanced usage.') | ||
neko259
|
r521 | TAGS_PLACEHOLDER = _('tag1 several_words_tag') | ||
neko259
|
r211 | |||
neko259
|
r527 | LABEL_TITLE = _('Title') | ||
LABEL_TEXT = _('Text') | ||||
neko259
|
r566 | LABEL_TAG = _('Tag') | ||
neko259
|
r718 | LABEL_SEARCH = _('Search') | ||
neko259
|
r566 | |||
TAG_MAX_LENGTH = 20 | ||||
neko259
|
r965 | IMAGE_DOWNLOAD_CHUNK_BYTES = 100000 | ||
HTTP_RESULT_OK = 200 | ||||
neko259
|
r1006 | TEXTAREA_ROWS = 4 | ||
neko259
|
r153 | |||
neko259
|
r1065 | def get_timezones(): | ||
timezones = [] | ||||
for tz in pytz.common_timezones: | ||||
timezones.append((tz, tz),) | ||||
return timezones | ||||
neko259
|
r438 | class FormatPanel(forms.Textarea): | ||
neko259
|
r937 | """ | ||
Panel for text formatting. Consists of buttons to add different tags to the | ||||
form text area. | ||||
""" | ||||
neko259
|
r438 | def render(self, name, value, attrs=None): | ||
output = '<div id="mark-panel">' | ||||
for formatter in formatters: | ||||
neko259
|
r769 | output += '<span class="mark_btn"' + \ | ||
' onClick="addMarkToMsg(\'' + formatter.format_left + \ | ||||
neko259
|
r438 | '\', \'' + formatter.format_right + '\')">' + \ | ||
formatter.preview_left + formatter.name + \ | ||||
neko259
|
r769 | formatter.preview_right + '</span>' | ||
neko259
|
r438 | |||
output += '</div>' | ||||
output += super(FormatPanel, self).render(name, value, attrs=None) | ||||
return output | ||||
neko259
|
r76 | class PlainErrorList(ErrorList): | ||
def __unicode__(self): | ||||
return self.as_text() | ||||
def as_text(self): | ||||
neko259
|
r769 | return ''.join(['(!) %s ' % e for e in self]) | ||
neko259
|
r76 | |||
neko259
|
r16 | |||
neko259
|
r205 | class NeboardForm(forms.Form): | ||
neko259
|
r937 | """ | ||
Form with neboard-specific formatting. | ||||
""" | ||||
neko259
|
r205 | |||
neko259
|
r426 | def as_div(self): | ||
neko259
|
r438 | """ | ||
Returns this form rendered as HTML <as_div>s. | ||||
""" | ||||
neko259
|
r205 | return self._html_output( | ||
neko259
|
r425 | # TODO Do not show hidden rows in the list here | ||
neko259
|
r680 | normal_row='<div class="form-row"><div class="form-label">' | ||
neko259
|
r205 | '%(label)s' | ||
neko259
|
r680 | '</div></div>' | ||
'<div class="form-row"><div class="form-input">' | ||||
neko259
|
r205 | '%(field)s' | ||
neko259
|
r680 | '</div></div>' | ||
'<div class="form-row">' | ||||
neko259
|
r205 | '%(help_text)s' | ||
'</div>', | ||||
neko259
|
r426 | error_row='<div class="form-row">' | ||
'<div class="form-label"></div>' | ||||
'<div class="form-errors">%s</div>' | ||||
'</div>', | ||||
row_ender='</div>', | ||||
help_text_html='%s', | ||||
neko259
|
r205 | errors_on_separate_row=True) | ||
neko259
|
r533 | def as_json_errors(self): | ||
errors = [] | ||||
neko259
|
r771 | for name, field in list(self.fields.items()): | ||
neko259
|
r533 | if self[name].errors: | ||
errors.append({ | ||||
'field': name, | ||||
'errors': self[name].errors.as_text(), | ||||
}) | ||||
return errors | ||||
neko259
|
r438 | |||
neko259
|
r205 | class PostForm(NeboardForm): | ||
neko259
|
r76 | |||
neko259
|
r232 | title = forms.CharField(max_length=TITLE_MAX_LENGTH, required=False, | ||
neko259
|
r527 | label=LABEL_TITLE) | ||
neko259
|
r517 | text = forms.CharField( | ||
neko259
|
r1006 | widget=FormatPanel(attrs={ | ||
ATTRIBUTE_PLACEHOLDER: TEXT_PLACEHOLDER, | ||||
ATTRIBUTE_ROWS: TEXTAREA_ROWS, | ||||
}), | ||||
neko259
|
r527 | required=False, label=LABEL_TEXT) | ||
neko259
|
r675 | image = forms.ImageField(required=False, label=_('Image'), | ||
neko259
|
r721 | widget=forms.ClearableFileInput( | ||
attrs={'accept': 'image/*'})) | ||||
neko259
|
r954 | image_url = forms.CharField(required=False, label=_('Image URL'), | ||
widget=forms.TextInput( | ||||
attrs={ATTRIBUTE_PLACEHOLDER: | ||||
'http://example.com/image.png'})) | ||||
neko259
|
r29 | |||
neko259
|
r207 | # This field is for spam prevention only | ||
neko259
|
r232 | email = forms.CharField(max_length=100, required=False, label=_('e-mail'), | ||
widget=forms.TextInput(attrs={ | ||||
'class': 'form-email'})) | ||||
neko259
|
r1077 | threads = forms.CharField(required=False, label=_('Additional threads'), | ||
widget=forms.TextInput(attrs={ATTRIBUTE_PLACEHOLDER: | ||||
'123 456 789'})) | ||||
neko259
|
r207 | |||
neko259
|
r153 | session = None | ||
neko259
|
r271 | need_to_ban = False | ||
neko259
|
r153 | |||
neko259
|
r76 | def clean_title(self): | ||
title = self.cleaned_data['title'] | ||||
if title: | ||||
if len(title) > TITLE_MAX_LENGTH: | ||||
neko259
|
r211 | raise forms.ValidationError(_('Title must have less than %s ' | ||
'characters') % | ||||
str(TITLE_MAX_LENGTH)) | ||||
neko259
|
r76 | return title | ||
neko259
|
r29 | def clean_text(self): | ||
neko259
|
r678 | text = self.cleaned_data['text'].strip() | ||
neko259
|
r29 | if text: | ||
neko259
|
r333 | if len(text) > board_settings.MAX_TEXT_LENGTH: | ||
neko259
|
r211 | raise forms.ValidationError(_('Text must have less than %s ' | ||
'characters') % | ||||
neko259
|
r333 | str(board_settings | ||
.MAX_TEXT_LENGTH)) | ||||
neko259
|
r29 | return text | ||
def clean_image(self): | ||||
image = self.cleaned_data['image'] | ||||
neko259
|
r954 | |||
neko259
|
r1027 | if image: | ||
self.validate_image_size(image.size) | ||||
neko259
|
r954 | |||
return image | ||||
def clean_image_url(self): | ||||
url = self.cleaned_data['image_url'] | ||||
image = None | ||||
if url: | ||||
image = self._get_image_from_url(url) | ||||
if not image: | ||||
raise forms.ValidationError(_('Invalid URL')) | ||||
neko259
|
r1027 | else: | ||
self.validate_image_size(image.size) | ||||
neko259
|
r527 | |||
neko259
|
r29 | return image | ||
neko259
|
r1077 | def clean_threads(self): | ||
threads_str = self.cleaned_data['threads'] | ||||
if len(threads_str) > 0: | ||||
threads_id_list = threads_str.split(' ') | ||||
threads = list() | ||||
for thread_id in threads_id_list: | ||||
try: | ||||
thread = Post.objects.get(id=int(thread_id)) | ||||
if not thread.is_opening(): | ||||
raise ObjectDoesNotExist() | ||||
threads.append(thread) | ||||
except (ObjectDoesNotExist, ValueError): | ||||
raise forms.ValidationError(_('Invalid additional thread list')) | ||||
return threads | ||||
neko259
|
r29 | def clean(self): | ||
cleaned_data = super(PostForm, self).clean() | ||||
neko259
|
r211 | if not self.session: | ||
raise forms.ValidationError('Humans have sessions') | ||||
if cleaned_data['email']: | ||||
neko259
|
r271 | self.need_to_ban = True | ||
neko259
|
r211 | raise forms.ValidationError('A human cannot enter a hidden field') | ||
neko259
|
r207 | |||
neko259
|
r151 | if not self.errors: | ||
self._clean_text_image() | ||||
neko259
|
r77 | |||
neko259
|
r153 | if not self.errors and self.session: | ||
self._validate_posting_speed() | ||||
neko259
|
r76 | return cleaned_data | ||
neko259
|
r954 | def get_image(self): | ||
""" | ||||
Gets image from file or URL. | ||||
""" | ||||
image = self.cleaned_data['image'] | ||||
return image if image else self.cleaned_data['image_url'] | ||||
neko259
|
r76 | def _clean_text_image(self): | ||
text = self.cleaned_data.get('text') | ||||
neko259
|
r954 | image = self.get_image() | ||
neko259
|
r29 | |||
if (not text) and (not image): | ||||
neko259
|
r205 | error_message = _('Either text or image must be entered.') | ||
neko259
|
r77 | self._errors['text'] = self.error_class([error_message]) | ||
neko259
|
r153 | |||
def _validate_posting_speed(self): | ||||
can_post = True | ||||
neko259
|
r728 | posting_delay = settings.POSTING_DELAY | ||
neko259
|
r642 | |||
neko259
|
r725 | if board_settings.LIMIT_POSTING_SPEED and LAST_POST_TIME in \ | ||
self.session: | ||||
neko259
|
r153 | now = time.time() | ||
last_post_time = self.session[LAST_POST_TIME] | ||||
current_delay = int(now - last_post_time) | ||||
neko259
|
r642 | if current_delay < posting_delay: | ||
neko259
|
r211 | error_message = _('Wait %s seconds after last posting') % str( | ||
neko259
|
r642 | posting_delay - current_delay) | ||
neko259
|
r153 | self._errors['text'] = self.error_class([error_message]) | ||
can_post = False | ||||
if can_post: | ||||
self.session[LAST_POST_TIME] = time.time() | ||||
neko259
|
r29 | |||
neko259
|
r1027 | def validate_image_size(self, size: int): | ||
if size > board_settings.MAX_IMAGE_SIZE: | ||||
raise forms.ValidationError( | ||||
_('Image must be less than %s bytes') | ||||
% str(board_settings.MAX_IMAGE_SIZE)) | ||||
neko259
|
r954 | def _get_image_from_url(self, url: str) -> SimpleUploadedFile: | ||
""" | ||||
Gets an image file from URL. | ||||
""" | ||||
img_temp = None | ||||
try: | ||||
# Verify content headers | ||||
response_head = requests.head(url, verify=False) | ||||
neko259
|
r982 | content_type = response_head.headers['content-type'].split(';')[0] | ||
neko259
|
r954 | if content_type in CONTENT_TYPE_IMAGE: | ||
neko259
|
r962 | length_header = response_head.headers.get('content-length') | ||
if length_header: | ||||
length = int(length_header) | ||||
neko259
|
r1027 | self.validate_image_size(length) | ||
neko259
|
r954 | # Get the actual content into memory | ||
neko259
|
r965 | response = requests.get(url, verify=False, stream=True) | ||
neko259
|
r954 | |||
neko259
|
r965 | # Download image, stop if the size exceeds limit | ||
size = 0 | ||||
content = b'' | ||||
for chunk in response.iter_content(IMAGE_DOWNLOAD_CHUNK_BYTES): | ||||
size += len(chunk) | ||||
neko259
|
r1027 | self.validate_image_size(size) | ||
neko259
|
r965 | content += chunk | ||
neko259
|
r954 | |||
neko259
|
r965 | if response.status_code == HTTP_RESULT_OK and content: | ||
neko259
|
r954 | # Set a dummy file name that will be replaced | ||
# anyway, just keep the valid extension | ||||
filename = 'image.' + content_type.split('/')[1] | ||||
img_temp = SimpleUploadedFile(filename, content, | ||||
content_type) | ||||
except Exception: | ||||
neko259
|
r965 | # Just return no image | ||
neko259
|
r954 | pass | ||
return img_temp | ||||
neko259
|
r29 | |||
class ThreadForm(PostForm): | ||||
neko259
|
r232 | |||
neko259
|
r517 | tags = forms.CharField( | ||
widget=forms.TextInput(attrs={ATTRIBUTE_PLACEHOLDER: TAGS_PLACEHOLDER}), | ||||
neko259
|
r679 | max_length=100, label=_('Tags'), required=True) | ||
neko259
|
r31 | |||
def clean_tags(self): | ||||
neko259
|
r678 | tags = self.cleaned_data['tags'].strip() | ||
neko259
|
r69 | |||
neko259
|
r937 | if not tags or not REGEX_TAGS.match(tags): | ||
neko259
|
r679 | raise forms.ValidationError( | ||
_('Inappropriate characters in tags.')) | ||||
neko259
|
r31 | |||
neko259
|
r922 | required_tag_exists = False | ||
for tag in tags.split(): | ||||
tag_model = Tag.objects.filter(name=tag.strip().lower(), | ||||
neko259
|
r937 | required=True) | ||
neko259
|
r922 | if tag_model.exists(): | ||
required_tag_exists = True | ||||
neko259
|
r937 | break | ||
neko259
|
r922 | |||
if not required_tag_exists: | ||||
raise forms.ValidationError(_('Need at least 1 required tag.')) | ||||
neko259
|
r31 | return tags | ||
def clean(self): | ||||
cleaned_data = super(ThreadForm, self).clean() | ||||
neko259
|
r35 | return cleaned_data | ||
neko259
|
r205 | class SettingsForm(NeboardForm): | ||
neko259
|
r1065 | theme = forms.ChoiceField(choices=settings.THEMES, label=_('Theme')) | ||
neko259
|
r990 | username = forms.CharField(label=_('User name'), required=False) | ||
neko259
|
r1065 | timezone = forms.ChoiceField(choices=get_timezones(), label=_('Time zone')) | ||
neko259
|
r144 | |||
neko259
|
r995 | def clean_username(self): | ||
username = self.cleaned_data['username'] | ||||
if username and not REGEX_TAGS.match(username): | ||||
raise forms.ValidationError(_('Inappropriate characters.')) | ||||
return username | ||||
neko259
|
r144 | |||
neko259
|
r718 | class SearchForm(NeboardForm): | ||
neko259
|
r729 | query = forms.CharField(max_length=500, label=LABEL_SEARCH, required=False) | ||