#!/usr/bin/env python
# vim: set fileencoding=utf-8 :
###############################################################################
# #
# Copyright (c) 2016 Idiap Research Institute, http://www.idiap.ch/ #
# Contact: beat.support@idiap.ch #
# #
# This file is part of the beat.web module of the BEAT platform. #
# #
# Commercial License Usage #
# Licensees holding valid commercial BEAT licenses may use this file in #
# accordance with the terms contained in a written agreement between you #
# and Idiap. For further information contact tto@idiap.ch #
# #
# Alternatively, this file may be used under the terms of the GNU Affero #
# Public License version 3 as published by the Free Software and appearing #
# in the file LICENSE.AGPL included in the packaging of this file. #
# The BEAT platform 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. #
# #
# You should have received a copy of the GNU Affero Public License along #
# with the BEAT platform. If not, see http://www.gnu.org/licenses/. #
# #
###############################################################################
import datetime
import logging
from django.conf import settings
from django.contrib import messages
from django.contrib.auth.decorators import login_required
from django.contrib.auth.forms import PasswordChangeForm
from django.contrib.auth.models import User
from django.core.mail import EmailMessage
from django.db.models import Q
from django.http import Http404
from django.http import HttpResponse
from django.shortcuts import get_object_or_404
from django.shortcuts import render
from django.template.loader import render_to_string
from rest_framework.authtoken.models import Token
from .. import __version__
from ..accounts.models import Profile
from ..accounts.models import SupervisionTrack
from ..accounts.views import inform_supervisor_new_supervisee_request
from ..common.models import Shareable
from ..utils import mail
from .registration.forms import BlockedUserRevalidationForm
logger = logging.getLogger(__name__)
# ----------------------------------------------------------
[docs]def index(request):
"""Our main index page"""
return render(request, "ui/index.html")
# ----------------------------------------------------------
[docs]def blocked_user_reactivation(request):
"""Reactivation page"""
if request.method == "POST":
form = BlockedUserRevalidationForm(request.POST)
if form.is_valid():
try:
user = User.objects.get(username=request.POST["username"])
if user.check_password(request.POST["password"]):
# Check if user is a blocked user
if user.profile.status == Profile.BLOCKED:
supervisor = User.objects.get(
username=request.POST["supervisor"]
)
# Check the supervisor
if supervisor.profile.status == Profile.ACCEPTED:
# Check if supervision track already exists
if user.profile.supervision_key is None:
# Create and assign key
supervisee = user
supervisee.profile.supervision_key = (
supervisee.profile._generate_current_supervision_key()
)
supervisiontrack = SupervisionTrack.objects.create(
supervisee=supervisee,
supervisor=supervisor,
is_valid=False,
)
# Assign key to supervision track
supervisiontrack.supervision_key = (
supervisee.profile.supervision_key
)
supervisiontrack.save()
supervisee.profile.supervision.add(supervisiontrack)
# Add a rejection date to the supervisee profile
now = datetime.datetime.now()
expiration_date_delta = datetime.timedelta(
days=settings.ACCOUNT_ACTIVATION_DAYS_FROM_SUPERVISOR
)
if supervisee.profile.rejection_date is None:
supervisee.profile.rejection_date = (
now + expiration_date_delta
)
supervisee.profile.save()
supervisee.save()
# Inform by email the supervisor that he has a new supervisee request
context = inform_supervisor_new_supervisee_request(
request, supervisiontrack, supervisor, supervisee
)
# inform the supervisee of his request
mail.send_email(
"registration/mail.supervisee_blocked_validation_wait.subject.txt",
"registration/mail.supervisee_blocked_state_wait_for_activation.message.txt",
context,
[supervisee.email],
)
messages.success(
request,
"Your supervision request has been successfully processed.",
)
else:
messages.error(
request,
"A supervision request already exists for this account, you need to wait for your supervisor's decision.",
)
else:
messages.error(
request, "The selected supervisor is not valid."
)
else:
messages.error(
request,
"Your profile is not blocked, you can go to the login page instead.",
)
else:
# Don't give too much details though we know the problem is the password only at this step
messages.error(request, "Wrong user or password combination!")
except User.DoesNotExist:
# Don't give too much details though we know the problem is the password only at this step
messages.error(request, "Wrong user or password combination!")
pass
# return redirect('blocked_user_reactivation', pk=post.pk)
else:
form = BlockedUserRevalidationForm()
return render(request, "registration/blocked_user_reactivate.html", {"form": form})
# ----------------------------------------------------------
[docs]def gather_contributions(requestor, author):
"""Gather contributions that are accessible to a certain requestor"""
from ..algorithms.models import Algorithm
from ..attestations.models import Attestation
from ..dataformats.models import DataFormat
from ..experiments.models import Experiment
from ..libraries.models import Library
from ..plotters.models import Plotter
from ..reports.models import Report
from ..search.models import Search
from ..team.models import Team
from ..toolchains.models import Toolchain
experiments = Experiment.objects.for_user(requestor).filter(author=author)
toolchains = Toolchain.objects.for_user(requestor).filter(author=author)
algorithms = Algorithm.objects.for_user(requestor).filter(author=author)
libraries = Library.objects.for_user(requestor).filter(author=author)
dataformats = DataFormat.objects.for_user(requestor).filter(author=author)
teams = Team.objects.for_user(requestor).filter(owner=author)
plotters = Plotter.objects.for_user(requestor).filter(author=author)
if requestor == author:
attestations = Attestation.objects.filter(experiment__author=author)
else:
attestations = Attestation.objects.published().filter(experiment__author=author)
searches = Search.objects.for_user(requestor).filter(author=author)
if requestor == author:
reports = Report.objects.filter(author=author)
else:
reports = Report.objects.published().filter(author=author)
return dict(
experiments=experiments,
toolchains=toolchains,
algorithms=algorithms,
libraries=libraries,
dataformats=dataformats,
teams=teams,
attestations=attestations,
searches=searches,
reports=reports,
plotters=plotters,
)
# ----------------------------------------------------------
[docs]def activity_stream(request, author_name):
"""User-specific activity stream"""
# check that the user exists on the system
author = get_object_or_404(User, username=author_name)
# gather leaderboards for the following conditions:
from ..search.models import Leaderboard
from ..search.models import Search
if request.user == author:
# 1. request.user == author AND user is subscribed
leaderboards = Leaderboard.objects.filter(
search__in=Search.objects.for_user(author, add_public=True),
notify__in=(author,),
).order_by("-changed")
else:
# 2. request.user != author
leaderboards = Leaderboard.objects.filter(
search__in=Search.objects.for_user(request.user).filter(author=author)
).order_by("-changed")
return render(
request,
"ui/activity_stream.html",
dict(
owner=(request.user == author),
author=author,
statistics=gather_contributions(request.user, author),
leaderboards=leaderboards,
),
)
# ----------------------------------------------------------
[docs]@login_required
def docreq(request, author_name):
"""Request a documentation for an existing platform object"""
url = request.META.get("HTTP_REFERER")
if not url:
raise Http404()
if author_name not in url:
raise Http404()
author = get_object_or_404(User, username=author_name)
to = "%s %s <%s>" % (author.first_name, author.last_name, author.email)
cc = "%s %s <%s>" % (
request.user.first_name,
request.user.last_name,
request.user.email,
)
try:
subject_template = "ui/docreq_email_subject.txt"
subject = render_to_string(
subject_template, {"user": request.user, "url": url},
).strip()
body_template = "ui/docreq_email_body.txt"
body = render_to_string(
body_template,
{"user": request.user, "url": url, "beat_version": __version__},
)
message = EmailMessage(
subject=subject,
body=body,
from_email=settings.DEFAULT_FROM_EMAIL,
to=[to],
reply_to=[cc],
)
message.send()
except Exception:
import traceback
logger.warn(
"Could not send e-mail to `%s' (cc: `%s') about "
"documentation request for `%s'. Exception caught: %s",
to,
cc,
url,
traceback.format_exc(),
)
return HttpResponse()
# ----------------------------------------------------------
[docs]@login_required
def user_settings(request):
"""User settings page (password and token change)"""
user = request.user
if request.method == "POST":
if "password" in request.POST:
password_change_form = PasswordChangeForm(data=request.POST, user=user)
if password_change_form.is_valid():
password_change_form.save()
messages.add_message(
request, messages.SUCCESS, "Password changed successfully"
)
elif "token" in request.POST:
user.auth_token.delete()
Token.objects.create(user=user)
password_change_form = PasswordChangeForm(user=user)
messages.add_message(
request, messages.SUCCESS, "Token changed successfully"
)
else:
password_change_form = PasswordChangeForm(user=user)
return render(
request,
"ui/user_settings.html",
{
"password_change_form": password_change_form,
"token": user.auth_token,
"statistics": {
"nb_experiments": user.experiments.count(),
"nb_public_experiments": user.experiments.filter(
sharing=Shareable.PUBLIC
).count(),
"nb_attested_experiments": user.experiments.filter(
~Q(attestation=None)
).count(),
"nb_toolchains": user.toolchains.count(),
"nb_public_toolchains": user.toolchains.filter(
sharing=Shareable.PUBLIC
).count(),
"nb_algorithms": user.algorithms.count(),
"nb_public_algorithms": user.algorithms.filter(
sharing=Shareable.PUBLIC
).count(),
"nb_libraries": user.algorithms.count(),
"nb_public_libraries": user.librarys.filter(
sharing=Shareable.PUBLIC
).count(),
"nb_dataformats": user.dataformats.count(),
"nb_public_dataformats": user.dataformats.filter(
sharing=Shareable.PUBLIC
).count(),
},
},
)
# ----------------------------------------------------------
[docs]def empty_error500_for_tests(request):
"""Custom error 500 view used ONLY DURING THE TESTS
Without it, Django tries to render the custom one from the website when an error
occurs in some tests, but fails because some variables needed by the template aren't
found.
"""
return HttpResponse()