455 lines
13 KiB
Python
455 lines
13 KiB
Python
# Calendar.social
|
||
# Copyright (C) 2018 Gergely Polonkai
|
||
#
|
||
# This program is free software: you can redistribute it and/or modify
|
||
# it under the terms of the GNU Affero General Public License as published by
|
||
# the Free Software Foundation, either version 3 of the License, or
|
||
# (at your option) any later version.
|
||
#
|
||
# This program is distributed in the hope that it will be useful,
|
||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||
# GNU Affero General Public License for more details.
|
||
#
|
||
# You should have received a copy of the GNU Affero General Public License
|
||
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||
|
||
"""Forms for Calendar.social
|
||
"""
|
||
|
||
from enum import Enum
|
||
|
||
from flask_babelex import lazy_gettext as _
|
||
from flask_security.forms import LoginForm as BaseLoginForm
|
||
from flask_wtf import FlaskForm
|
||
import pytz
|
||
from wtforms import BooleanField, PasswordField, SelectField, StringField, RadioField
|
||
from wtforms.ext.dateutil.fields import DateTimeField
|
||
from wtforms.validators import DataRequired, Email, StopValidation, ValidationError
|
||
from wtforms.widgets import TextArea
|
||
|
||
from calsocial.models import EventVisibility, EVENT_VISIBILITY_TRANSLATIONS, GroupVisibility, \
|
||
GROUP_VISIBILITY_TRANSLATIONS
|
||
|
||
|
||
class UsernameAvailable: # pylint: disable=too-few-public-methods
|
||
"""Checks if a username is available
|
||
"""
|
||
|
||
def __init__(self, message=None):
|
||
self.message = message
|
||
|
||
def __call__(self, form, field):
|
||
from sqlalchemy.orm.exc import NoResultFound
|
||
|
||
from calsocial.models import User
|
||
|
||
# If there is no data, we don’t raise an error; it’s not the task of this validator to
|
||
# check the validity of the username
|
||
if not field.data:
|
||
return
|
||
|
||
try:
|
||
User.query.filter(User.username == field.data).one()
|
||
except NoResultFound:
|
||
return
|
||
|
||
if self.message is None:
|
||
message = field.gettext('This username is not available')
|
||
else:
|
||
message = self.message
|
||
|
||
field.errors[:] = []
|
||
raise StopValidation(message)
|
||
|
||
|
||
class EmailAvailable: # pylint: disable=too-few-public-methods
|
||
"""Checks if an email address is available
|
||
"""
|
||
|
||
def __init__(self, message=None):
|
||
self.message = message
|
||
|
||
def __call__(self, form, field):
|
||
from sqlalchemy.orm.exc import NoResultFound
|
||
|
||
from calsocial.models import User
|
||
|
||
# If there is no data, we don’t raise an error; it’s not the task of this validator to
|
||
# check the validity of the username
|
||
if not field.data:
|
||
return
|
||
|
||
try:
|
||
User.query.filter(User.email == field.data).one()
|
||
except NoResultFound:
|
||
return
|
||
|
||
if self.message is None:
|
||
message = field.gettext('This email address can not be used')
|
||
else:
|
||
message = self.message
|
||
|
||
field.errors[:] = []
|
||
raise StopValidation(message)
|
||
|
||
|
||
class RegistrationForm(FlaskForm):
|
||
"""Registration form
|
||
"""
|
||
|
||
username = StringField(_('Username'), validators=[DataRequired(), UsernameAvailable()])
|
||
email = StringField(_('Email address'), validators=[Email(), EmailAvailable()])
|
||
password = PasswordField(_('Password'), validators=[DataRequired()])
|
||
password_retype = PasswordField(_('Password, once more'), validators=[DataRequired()])
|
||
|
||
def validate_password_retype(self, field):
|
||
"""Validate the ``password_retype`` field
|
||
|
||
Its value must match the ``password`` field’s.
|
||
"""
|
||
|
||
if field.data != self.password.data:
|
||
raise ValidationError(_('The two passwords must match!'))
|
||
|
||
|
||
class TimezoneField(SelectField):
|
||
"""Field for selecting a time zone
|
||
|
||
Note: this field overrides whatever is passed to the ``choices`` parameter, and fills
|
||
``choices`` with the common timezones of pytz. In every other aspects, it behaves exactly
|
||
like `SelectField`.
|
||
"""
|
||
|
||
def __init__(self, *args, **kwargs):
|
||
self.data = None
|
||
|
||
kwargs.update({
|
||
'choices': [
|
||
(pytz.timezone(tz), tz.replace('_', ' '))
|
||
for tz in pytz.common_timezones
|
||
],
|
||
})
|
||
|
||
SelectField.__init__(self, *args, **kwargs)
|
||
|
||
def process_formdata(self, valuelist):
|
||
if not valuelist:
|
||
self.data = None
|
||
|
||
return
|
||
|
||
try:
|
||
self.data = pytz.timezone(valuelist[0])
|
||
except pytz.exceptions.UnknownTimeZoneError:
|
||
self.data = None
|
||
|
||
raise ValueError('Unknown time zone')
|
||
|
||
@staticmethod
|
||
def is_pytz_instance(value):
|
||
"""Check if ``value`` is a valid pytz time zone instance
|
||
"""
|
||
|
||
return value is pytz.UTC or isinstance(value, pytz.tzinfo.BaseTzInfo)
|
||
|
||
def process_data(self, value):
|
||
if value is None:
|
||
self.data = None
|
||
|
||
return
|
||
|
||
if self.is_pytz_instance(value):
|
||
self.data = value
|
||
|
||
return
|
||
|
||
try:
|
||
self.data = pytz.timezone(value)
|
||
except pytz.exceptions.UnknownTimeZoneError:
|
||
raise ValueError(f'Unknown time zone {value}')
|
||
|
||
def iter_choices(self):
|
||
for value, label in self.choices:
|
||
yield (value, label, value == self.data)
|
||
|
||
|
||
class EnumField(SelectField):
|
||
"""Field that allows selecting one value from an ``Enum`` class
|
||
|
||
:param enum_type: an ``Enum`` type
|
||
:type enum_type: type(Enum)
|
||
:param translations: translatable labels for enum values
|
||
:type translations: dict
|
||
:param args: passed verbatim to the constructor of `SelectField`
|
||
:param kwargs: passed verbatim to the constructor of `SelectField`
|
||
"""
|
||
|
||
def __init__(self, enum_type, translations, *args, **kwargs):
|
||
if not issubclass(enum_type, Enum):
|
||
raise TypeError('enum_type must be a subclass of Enum')
|
||
|
||
kwargs.update({'choices': [(value, None) for value in enum_type]})
|
||
self.data = None
|
||
self.enum_type = enum_type
|
||
self.translations = translations
|
||
SelectField.__init__(self, *args, **kwargs)
|
||
|
||
def process_formdata(self, valuelist):
|
||
if not valuelist:
|
||
self.data = None
|
||
|
||
return
|
||
|
||
try:
|
||
self.data = self.enum_type[valuelist[0]]
|
||
except KeyError:
|
||
raise ValueError('Unknown value')
|
||
|
||
def iter_choices(self):
|
||
for value in self.enum_type:
|
||
label = self.gettext(self.translations[value]) if self.translations else value.name
|
||
|
||
yield (value.name, label, value == self.data)
|
||
|
||
|
||
class EventForm(FlaskForm):
|
||
"""Form for event creation/editing
|
||
"""
|
||
|
||
title = StringField(_('Title'), validators=[DataRequired()])
|
||
time_zone = TimezoneField(_('Time zone'), validators=[DataRequired()])
|
||
start_time = DateTimeField(_('Start time'), validators=[DataRequired()])
|
||
end_time = DateTimeField(_('End time'), validators=[DataRequired()])
|
||
all_day = BooleanField(_('All day'))
|
||
description = StringField(_('Description'), widget=TextArea())
|
||
visibility = EnumField(EventVisibility, EVENT_VISIBILITY_TRANSLATIONS, label=_('Visibility'))
|
||
|
||
def __init__(self, *args, **kwargs):
|
||
from flask_security import current_user
|
||
|
||
self.time_zone.kwargs['default'] = current_user.timezone # pylint: disable=no-member
|
||
|
||
FlaskForm.__init__(self, *args, **kwargs)
|
||
|
||
def populate_obj(self, obj):
|
||
"""Populate ``obj`` with event data
|
||
"""
|
||
|
||
FlaskForm.populate_obj(self, obj)
|
||
|
||
timezone = self.time_zone.data
|
||
|
||
obj.time_zone = str(timezone)
|
||
obj.start_time = timezone.localize(self.start_time.data).astimezone(pytz.utc)
|
||
obj.end_time = timezone.localize(self.end_time.data).astimezone(pytz.utc)
|
||
|
||
def validate_end_time(self, field):
|
||
"""Validate the ``end_time`` field
|
||
|
||
Its value must be later than the value of the ``start_time`` field.
|
||
"""
|
||
|
||
if field.data < self.start_time.data:
|
||
raise ValidationError(_('End time must be later than start time!'))
|
||
|
||
|
||
class SettingsForm(FlaskForm):
|
||
"""Form for user settings
|
||
"""
|
||
|
||
timezone = TimezoneField(_('Time zone'), validators=[DataRequired()])
|
||
|
||
def __init__(self, user, *args, **kwargs):
|
||
self.user = user
|
||
|
||
kwargs.setdefault('timezone', user.timezone)
|
||
|
||
FlaskForm.__init__(self, *args, **kwargs)
|
||
|
||
def populate_obj(self, user):
|
||
"""Populate ``obj`` with event data
|
||
"""
|
||
|
||
for name, field in self._fields.items():
|
||
if not (hasattr(self.__class__, name) and not hasattr(FlaskForm, name)):
|
||
continue
|
||
|
||
user.settings[name] = str(field.data)
|
||
|
||
|
||
class LoginForm(BaseLoginForm):
|
||
"""Login form for Calendar.social
|
||
"""
|
||
|
||
email = StringField(_('Username or email'), validators=[DataRequired()])
|
||
|
||
def __init__(self, *args, **kwargs):
|
||
super(LoginForm, self).__init__(*args, **kwargs)
|
||
|
||
self.user = None
|
||
|
||
def validate(self):
|
||
from flask_security.utils import _datastore
|
||
from flask_security.utils import verify_and_update_password
|
||
|
||
from .models import AuditLog
|
||
|
||
ret = super(LoginForm, self).validate()
|
||
|
||
if self.user is None:
|
||
self.user = _datastore.get_user(self.email.data)
|
||
|
||
if self.user is None:
|
||
# We can’t figure out the user that’s trying to log in, just return
|
||
return ret
|
||
|
||
if not verify_and_update_password(self.password.data, self.user):
|
||
AuditLog.log(self.user, AuditLog.TYPE_LOGIN_FAIL)
|
||
|
||
return ret
|
||
|
||
|
||
class ProfileField(StringField):
|
||
"""Input field for profiles
|
||
"""
|
||
|
||
def process_formdata(self, valuelist):
|
||
from sqlalchemy.orm.exc import NoResultFound
|
||
|
||
from .models import User, Profile
|
||
|
||
if not valuelist:
|
||
self.data = None
|
||
|
||
return
|
||
|
||
value = valuelist[0]
|
||
|
||
if value.startswith('@'):
|
||
value = value[1:]
|
||
|
||
try:
|
||
self.data = Profile.query.join(User).filter(User.username == value).one()
|
||
except NoResultFound:
|
||
self.data = None
|
||
|
||
raise ValueError('Unknown user')
|
||
|
||
def _value(self):
|
||
if self.data:
|
||
return self.data.fqn
|
||
|
||
return ''
|
||
|
||
|
||
class InviteForm(FlaskForm):
|
||
"""Form for event invitations
|
||
"""
|
||
|
||
invitee = ProfileField(validators=[DataRequired()])
|
||
|
||
def __init__(self, event, *args, **kwargs):
|
||
FlaskForm.__init__(self, *args, **kwargs)
|
||
|
||
self.event = event
|
||
|
||
def validate_invitee(self, field):
|
||
"""Validate the value of the invitee field
|
||
|
||
:raises ValidationError: If the given user is already invited
|
||
"""
|
||
|
||
from sqlalchemy.orm.exc import NoResultFound
|
||
|
||
from .models import Invitation
|
||
|
||
try:
|
||
Invitation.query \
|
||
.filter(Invitation.event == self.event) \
|
||
.filter(Invitation.invitee == field.data) \
|
||
.one()
|
||
|
||
raise ValidationError(_('User is already invited'))
|
||
except NoResultFound:
|
||
pass
|
||
|
||
|
||
class FirstStepsForm(FlaskForm):
|
||
"""Form for the initial profile setup
|
||
"""
|
||
|
||
display_name = StringField(
|
||
label=_('Display name'),
|
||
validators=[DataRequired()],
|
||
# pylint: disable=line-too-long
|
||
description=_('This will be shown to other users as your name. You can use your real name, or any nickname you like.'))
|
||
time_zone = TimezoneField(
|
||
label=_('Your time zone'),
|
||
validators=[DataRequired()],
|
||
description=_('The start and end times of events will be displayed in this time zone.'))
|
||
|
||
|
||
class ProfileForm(FlaskForm):
|
||
"""Form for editing a user profile
|
||
"""
|
||
|
||
display_name = StringField(label=_('Display name'), validators=[DataRequired()])
|
||
builtin_avatar = RadioField(label=_('Use a built-in avatar'))
|
||
locked = BooleanField(label=_('Lock profile'))
|
||
|
||
def __init__(self, profile, *args, **kwargs):
|
||
from flask import current_app
|
||
|
||
kwargs.update(
|
||
{
|
||
'display_name': profile.display_name,
|
||
'locked': profile.locked,
|
||
'builtin_avatar': profile.builtin_avatar,
|
||
})
|
||
FlaskForm.__init__(self, *args, **kwargs)
|
||
|
||
self.builtin_avatar.choices = [(name, name)
|
||
for name in current_app.config['BUILTIN_AVATARS']]
|
||
self.profile = profile
|
||
|
||
|
||
class HandleAvailable: # pylint: disable=too-few-public-methods
|
||
"""Checks if a group handle is available
|
||
"""
|
||
|
||
def __init__(self, message=None):
|
||
self.message = message
|
||
|
||
def __call__(self, form, field):
|
||
from sqlalchemy.orm.exc import NoResultFound
|
||
|
||
from calsocial.models import Group
|
||
|
||
# If there is no data, we don’t raise an error; it’s not the task of this validator to
|
||
# check the validity of the username
|
||
if not field.data:
|
||
return
|
||
|
||
try:
|
||
Group.query.filter(Group.handle == field.data).one()
|
||
except NoResultFound:
|
||
return
|
||
|
||
if self.message is None:
|
||
message = field.gettext('This group handle is not available')
|
||
else:
|
||
message = self.message
|
||
|
||
field.errors[:] = []
|
||
raise StopValidation(message)
|
||
|
||
|
||
class GroupForm(FlaskForm):
|
||
"""Form for editing a group
|
||
"""
|
||
|
||
handle = StringField(label=_('Handle'), validators=[DataRequired(), HandleAvailable()])
|
||
display_name = StringField(label=_('Display name'))
|
||
visibility = EnumField(GroupVisibility, GROUP_VISIBILITY_TRANSLATIONS, label=_('Visibility'))
|