forked from gergely/calendar-social
332 lines
9.8 KiB
Python
332 lines
9.8 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/>.
|
|
|
|
"""Main module for the Calendar.social app
|
|
"""
|
|
|
|
from datetime import datetime
|
|
import os
|
|
|
|
from flask import Flask, abort, current_app, redirect, render_template, request, url_for
|
|
from flask_babelex import Babel, get_locale as babel_get_locale
|
|
from flask_security import SQLAlchemyUserDatastore, current_user, login_required
|
|
from sqlalchemy.orm.exc import NoResultFound, MultipleResultsFound
|
|
|
|
|
|
def get_locale():
|
|
"""Locale selector
|
|
|
|
Selects the best locale based on values sent by the browser.
|
|
"""
|
|
|
|
supported_languages = ['en', 'hu']
|
|
|
|
if 'l' in request.args and request.args['l'].lower() in supported_languages:
|
|
return request.args['l'].lower()
|
|
|
|
return request.accept_languages.best_match(supported_languages)
|
|
|
|
|
|
def template_vars():
|
|
"""Function to inject global template variables
|
|
"""
|
|
|
|
now = datetime.utcnow()
|
|
|
|
return {
|
|
'lang': babel_get_locale().language,
|
|
'now': now,
|
|
'now_ts': now.timestamp(),
|
|
}
|
|
|
|
|
|
def route(*args, **kwargs):
|
|
"""Mark a function as a future route
|
|
|
|
Such functions will be iterated over when the application is initialised. ``*args`` and
|
|
``**kwargs`` will be passed verbatim to `Flask.route()`.
|
|
"""
|
|
|
|
def decorator(func): # pylint: disable=missing-docstring
|
|
setattr(func, 'routing', (args, kwargs))
|
|
|
|
return func
|
|
|
|
return decorator
|
|
|
|
|
|
class CalendarSocialApp(Flask):
|
|
"""The Calendar.social app
|
|
"""
|
|
|
|
def __init__(self, name, config=None): # pylint: disable=too-many-locals
|
|
from .forms import LoginForm
|
|
from .models import db, User, Role
|
|
from .security import security, AnonymousUser
|
|
|
|
Flask.__init__(self, name)
|
|
|
|
self._timezone = None
|
|
|
|
config_name = os.environ.get('ENV', config or 'dev')
|
|
self.config.from_pyfile(f'config_{config_name}.py', True)
|
|
# Make sure we look up users both by their usernames and email addresses
|
|
self.config['SECURITY_USER_IDENTITY_ATTRIBUTES'] = ('username', 'email')
|
|
db.init_app(self)
|
|
babel = Babel(app=self)
|
|
babel.localeselector(get_locale)
|
|
|
|
user_store = SQLAlchemyUserDatastore(db, User, Role)
|
|
security.init_app(self, datastore=user_store,
|
|
anonymous_user=AnonymousUser,
|
|
login_form=LoginForm)
|
|
|
|
self.context_processor(template_vars)
|
|
|
|
for attr_name in self.__dir__():
|
|
attr = getattr(self, attr_name)
|
|
|
|
if not callable(attr):
|
|
continue
|
|
|
|
args, kwargs = getattr(attr, 'routing', (None, None))
|
|
|
|
if args is None:
|
|
continue
|
|
|
|
self.route(*args, **kwargs)(attr)
|
|
|
|
@property
|
|
def timezone(self):
|
|
"""The default time zone of the app
|
|
"""
|
|
|
|
from warnings import warn
|
|
|
|
from flask import has_app_context
|
|
from pytz import timezone, utc
|
|
from pytz.exceptions import UnknownTimeZoneError
|
|
|
|
if not has_app_context():
|
|
return utc
|
|
|
|
if not self._timezone:
|
|
timezone_str = current_app.config.get('DEFAULT_TIMEZONE', 'UTC')
|
|
|
|
try:
|
|
self._timezone = timezone(timezone_str)
|
|
except UnknownTimeZoneError:
|
|
warn(f'Timezone of {self} (or the default timezone) "{timezone_str}" is invalid')
|
|
|
|
self._timezone = utc
|
|
|
|
return self._timezone
|
|
|
|
@staticmethod
|
|
@route('/')
|
|
def hello():
|
|
"""View for the main page
|
|
|
|
This will display a welcome message for users not logged in; for others, their main
|
|
calendar view is displayed.
|
|
"""
|
|
|
|
from .calendar_system.gregorian import GregorianCalendar
|
|
|
|
if not current_user.is_authenticated:
|
|
return render_template('welcome.html')
|
|
|
|
try:
|
|
timestamp = datetime.fromtimestamp(float(request.args.get('date')))
|
|
except TypeError:
|
|
timestamp = datetime.utcnow()
|
|
|
|
calendar = GregorianCalendar(timestamp.timestamp())
|
|
|
|
return render_template('index.html', calendar=calendar)
|
|
|
|
@staticmethod
|
|
@route('/register', methods=['POST', 'GET'])
|
|
def register():
|
|
"""View for user registration
|
|
|
|
If the ``REGISTRATION_FAILED`` configuration value is set to ``True`` it displays the
|
|
registration disabled template. Otherwise, it performs user registration.
|
|
"""
|
|
|
|
if not current_app.config['REGISTRATION_ENABLED']:
|
|
return render_template('registration-disabled.html')
|
|
|
|
from .forms import RegistrationForm
|
|
from .models import db, User
|
|
|
|
form = RegistrationForm()
|
|
|
|
if form.validate_on_submit():
|
|
# TODO: This might become False later, if we want registrations to be confirmed via
|
|
# e-mail
|
|
user = User(active=True)
|
|
|
|
form.populate_obj(user)
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
|
|
return redirect(url_for('hello'))
|
|
|
|
return render_template('registration.html', form=form)
|
|
|
|
@staticmethod
|
|
@route('/new-event', methods=['GET', 'POST'])
|
|
@login_required
|
|
def new_event():
|
|
"""View for creating a new event
|
|
|
|
This presents a form to the user that allows entering event details.
|
|
"""
|
|
|
|
from .forms import EventForm
|
|
from .models import db, Event
|
|
|
|
form = EventForm()
|
|
|
|
if form.validate_on_submit():
|
|
event = Event(user=current_user)
|
|
form.populate_obj(event)
|
|
|
|
db.session.add(event)
|
|
db.session.commit()
|
|
|
|
return redirect(url_for('hello'))
|
|
|
|
return render_template('event-edit.html', form=form)
|
|
|
|
@staticmethod
|
|
@route('/settings', methods=['GET', 'POST'])
|
|
@login_required
|
|
def settings():
|
|
"""View for user settings
|
|
"""
|
|
|
|
from .forms import SettingsForm
|
|
from .models import db
|
|
|
|
form = SettingsForm(current_user)
|
|
|
|
if form.validate_on_submit():
|
|
form.populate_obj(current_user)
|
|
|
|
db.session.commit()
|
|
|
|
return redirect(url_for('hello'))
|
|
|
|
return render_template('user-settings.html', form=form)
|
|
|
|
@staticmethod
|
|
@route('/event/<string:event_uuid>', methods=['GET', 'POST'])
|
|
def event_details(event_uuid):
|
|
"""View to display event details
|
|
"""
|
|
|
|
from .forms import InviteForm
|
|
from .models import db, Event, Invitation, Notification, NotificationAction
|
|
|
|
try:
|
|
event = Event.query.filter(Event.event_uuid == event_uuid).one()
|
|
except NoResultFound:
|
|
abort(404)
|
|
except MultipleResultsFound:
|
|
abort(500)
|
|
|
|
form = InviteForm(event)
|
|
|
|
if form.validate_on_submit():
|
|
invite = Invitation(event=event, sender=current_user.profile)
|
|
form.populate_obj(invite)
|
|
db.session.add(invite)
|
|
|
|
notification = Notification(profile=form.invitee.data,
|
|
actor=current_user.profile,
|
|
item=event,
|
|
action=NotificationAction.invite)
|
|
db.session.add(notification)
|
|
|
|
db.session.commit()
|
|
|
|
return redirect(url_for('event_details', event_uuid=event.event_uuid))
|
|
|
|
return render_template('event-details.html', event=event, form=form)
|
|
|
|
@staticmethod
|
|
@route('/profile/@<string:username>')
|
|
def display_profile(username):
|
|
"""View to display profile details
|
|
"""
|
|
|
|
from .models import Profile, User
|
|
|
|
try:
|
|
profile = Profile.query.join(User).filter(User.username == username).one()
|
|
except NoResultFound:
|
|
abort(404)
|
|
|
|
return render_template('profile-details.html', profile=profile)
|
|
|
|
@staticmethod
|
|
@route('/profile/@<string:username>/follow')
|
|
def follow_user(username):
|
|
"""View for following a user
|
|
"""
|
|
|
|
from .models import db, Profile, User, UserFollow, Notification, NotificationAction
|
|
|
|
try:
|
|
profile = Profile.query.join(User).filter(User.username == username).one()
|
|
except NoResultFound:
|
|
abort(404)
|
|
|
|
if profile.user != current_user:
|
|
follow = UserFollow(follower=current_user.profile,
|
|
followed=profile,
|
|
accepted_at=datetime.utcnow())
|
|
db.session.add(follow)
|
|
|
|
notification = Notification(profile=profile,
|
|
actor=current_user.profile,
|
|
item=profile,
|
|
action=NotificationAction.follow)
|
|
db.session.add(notification)
|
|
|
|
db.session.commit()
|
|
|
|
return redirect(url_for('display_profile', username=username))
|
|
|
|
@staticmethod
|
|
@route('/notifications')
|
|
def notifications():
|
|
"""View to list the notifications for the current user
|
|
"""
|
|
|
|
from .models import Notification
|
|
|
|
if current_user.is_authenticated:
|
|
notifs = Notification.query.filter(Notification.profile == current_user.profile)
|
|
else:
|
|
notifs = []
|
|
|
|
return render_template('notifications.html', notifs=notifs)
|
|
|
|
app = CalendarSocialApp(__name__)
|