327 lines
12 KiB
Python
327 lines
12 KiB
Python
# welco - multichannel request processing
|
|
# Copyright (C) 2015 Entr'ouvert
|
|
#
|
|
# 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 <http://www.gnu.org/licenses/>.
|
|
|
|
import json
|
|
import logging
|
|
|
|
from django.conf import settings
|
|
from django.contrib.auth.decorators import login_required
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.core.urlresolvers import reverse
|
|
from django.db.models import Q
|
|
from django.http import HttpResponse, HttpResponseRedirect
|
|
from django.utils.translation import ugettext_lazy as _
|
|
from django.views.generic import TemplateView, DetailView
|
|
|
|
from hobo.agent.common.models import Role
|
|
|
|
from .models import Inbox
|
|
|
|
from welco.sources.mail.models import Mail
|
|
from welco.sources.mail.views import Home as MailOriginalHome
|
|
from welco.qualif.models import Association
|
|
from welco.utils import get_wcs_data, get_wcs_options, response_for_json
|
|
from welco.views import HomeMail as HomeScreen
|
|
|
|
|
|
class MailHome(MailOriginalHome):
|
|
def get_template(self):
|
|
return super(MailHome, self).get_template()
|
|
|
|
|
|
class DgsMailHome(MailHome):
|
|
display_filter = True
|
|
allow_reject = False
|
|
|
|
def get_queryset(self):
|
|
return Mail.objects.filter(status='done-qualif')
|
|
|
|
|
|
class DgHomeScreen(HomeScreen):
|
|
def get_template_names(self):
|
|
if self.kwargs.get('table') == 'table/':
|
|
return ['alfortville/dg-table.html']
|
|
return [self.template_name]
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super(DgHomeScreen, self).get_context_data(**kwargs)
|
|
context['home_screen_title'] = self.home_screen_title
|
|
context['mails'] = context['source'].get_queryset()
|
|
return context
|
|
|
|
|
|
class Dgs(DgHomeScreen):
|
|
source_klass = DgsMailHome
|
|
home_screen_title = _('DGS Validation')
|
|
|
|
def check_user_ok(self):
|
|
return 'DGS' in [x.name for x in self.request.user.groups.all()]
|
|
|
|
dgs = login_required(Dgs.as_view())
|
|
|
|
|
|
def is_dga_role(role_name):
|
|
return bool(role_name.startswith('DGA') or role_name.startswith('DGD') or
|
|
role_name.startswith('Cabinet'))
|
|
|
|
|
|
class DgaMailHome(MailHome):
|
|
display_filter = True
|
|
allow_reject = False
|
|
|
|
def filter_formdef_condition(self, formdef):
|
|
roles = set()
|
|
for function_role in formdef.get('functions', {}).values():
|
|
if is_dga_role(function_role.get('role', {}).get('name', '')):
|
|
roles.add(function_role.get('role').get('slug'))
|
|
return self.user_roles.intersection(roles)
|
|
|
|
def get_queryset(self):
|
|
mellon = self.request.session['mellon_session']
|
|
params = {'NameID': mellon['name_id_content']}
|
|
self.user_roles = set([x['slug'] for x in get_wcs_data('api/user/', params).get('user_roles')])
|
|
formdef_references = []
|
|
for category in get_wcs_options('api/formdefs/', self.filter_formdef_condition):
|
|
formdef_references.extend([x[0] for x in category[1]])
|
|
return Mail.objects.filter(status='done-dgs',
|
|
associations__formdef_reference__in=formdef_references)
|
|
|
|
class Dga(DgHomeScreen):
|
|
source_klass = DgaMailHome
|
|
home_screen_title = _('DGA Validation')
|
|
|
|
def check_user_ok(self):
|
|
user_roles = [x.name for x in self.request.user.groups.all()]
|
|
return any([x for x in user_roles if is_dga_role(x)])
|
|
|
|
dga = login_required(Dga.as_view())
|
|
|
|
|
|
class QualifMixin(object):
|
|
def get(self, request, *args, **kwargs):
|
|
return HttpResponseRedirect(self.get_redirect_url())
|
|
|
|
def post(self, request, *args, **kwargs):
|
|
source_pks = request.POST.getlist('pks')
|
|
logger = logging.getLogger(__name__)
|
|
validation_steps = settings.VALIDATION_STEPS.get('mail')
|
|
mail_content_type = ContentType.objects.get_for_model(Mail)
|
|
allowed_mails = [str(x.id) for x in self.source_klass(request).get_queryset()]
|
|
mail_ids = set(source_pks).intersection(set(allowed_mails))
|
|
for mail in Mail.objects.filter(id__in=mail_ids):
|
|
if mail.status:
|
|
mail.status = validation_steps[
|
|
validation_steps.index(mail.status)+1]
|
|
else:
|
|
mail.status = validation_steps[0]
|
|
if mail.status == validation_steps[-1]:
|
|
for association in Association.objects.filter(
|
|
source_type=mail_content_type,
|
|
source_pk=mail.id):
|
|
try:
|
|
association.push(request)
|
|
except Exception, e:
|
|
logger.exception('error pushing mail')
|
|
continue
|
|
mail.save()
|
|
return HttpResponseRedirect(self.get_redirect_url())
|
|
|
|
|
|
class DgaQualifMany(QualifMixin, Dga):
|
|
def get_redirect_url(self):
|
|
return reverse('alfortville-dga', kwargs={'table': 'table/'})
|
|
|
|
dga_qualif_many = login_required(DgaQualifMany.as_view())
|
|
|
|
|
|
class DgsQualifMany(QualifMixin, Dgs):
|
|
def get_redirect_url(self):
|
|
return reverse('alfortville-dgs', kwargs={'table': 'table/'})
|
|
|
|
dgs_qualif_many = login_required(DgsQualifMany.as_view())
|
|
|
|
|
|
class Copies(DetailView):
|
|
model = Mail
|
|
template_name = 'alfortville/copies.html'
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super(Copies, self).get_context_data(**kwargs)
|
|
checked_dicts = {
|
|
Inbox.MANDATORY_AVIS: {},
|
|
Inbox.AVIS: {},
|
|
Inbox.INFO: {},
|
|
}
|
|
for inbox in Inbox.objects.filter(source_pk=self.object.id):
|
|
checked_dicts[inbox.subtype][inbox.role_slug] = True
|
|
context['checked_info'] = checked_dicts[Inbox.INFO]
|
|
context['checked_avis'] = checked_dicts[Inbox.AVIS]
|
|
context['checked_mandatory_avis'] = checked_dicts[Inbox.MANDATORY_AVIS]
|
|
context['roles'] = []
|
|
all_roles = get_wcs_data('api/roles').get('data')
|
|
for start in ('Maire', 'Cabinet', 'Adjoint', 'Conseiller', 'Elu', 'DGS', 'DGD',
|
|
'DGA', 'Direction'):
|
|
roles = [x for x in all_roles if x['text'].startswith(start)]
|
|
roles.sort(lambda x, y: cmp(x['text'], y['text']))
|
|
context['roles'].extend(roles)
|
|
return context
|
|
|
|
def post(self, request, *args, **kwargs):
|
|
lists = {
|
|
Inbox.MANDATORY_AVIS: request.POST.getlist('mandatory_avis'),
|
|
Inbox.AVIS: request.POST.getlist('avis'),
|
|
Inbox.INFO: request.POST.getlist('info'),
|
|
}
|
|
for subtype in lists.keys():
|
|
Inbox.objects.filter(subtype=subtype).filter(
|
|
source_pk=kwargs['pk']).exclude(
|
|
role_slug__in=lists[subtype]).delete()
|
|
mail_content_type = ContentType.objects.get_for_model(Mail)
|
|
for subtype in lists.keys():
|
|
for role_slug in lists[subtype]:
|
|
Inbox.objects.get_or_create(source_type=mail_content_type,
|
|
source_pk=kwargs['pk'], role_slug=role_slug, subtype=subtype)
|
|
return HttpResponseRedirect('.')
|
|
|
|
copies = login_required(Copies.as_view())
|
|
|
|
|
|
def copies_ajax(request, *args, **kwargs):
|
|
roles_by_slug = {}
|
|
for role in get_wcs_data('api/roles').get('data'):
|
|
roles_by_slug[role['slug']] = role
|
|
lists = {
|
|
Inbox.MANDATORY_AVIS: [],
|
|
Inbox.AVIS: [],
|
|
Inbox.INFO: []
|
|
}
|
|
for inbox in Inbox.objects.filter(source_pk=kwargs.get('pk')):
|
|
lists[inbox.subtype].append(roles_by_slug[inbox.role_slug]['text'])
|
|
response = HttpResponse(content_type='application/json')
|
|
json.dump({'info': ', '.join(lists[Inbox.INFO]) or '-',
|
|
'avis': ', '.join(lists[Inbox.AVIS]) or '-',
|
|
'mandatory_avis': ', '.join(lists[Inbox.MANDATORY_AVIS]) or '-'},
|
|
response, indent=2)
|
|
return response
|
|
|
|
|
|
class MailTable(TemplateView):
|
|
template_name = 'alfortville/mail-table.html'
|
|
button_text = _('Next')
|
|
|
|
def get_context_data(self, *args, **kwargs):
|
|
context = super(MailTable, self).get_context_data(**kwargs)
|
|
params = {}
|
|
if self.request.session.get('mellon_session'):
|
|
mellon = self.request.session['mellon_session']
|
|
params['NameID'] = mellon['name_id_content']
|
|
|
|
user_roles = [x.uuid for x in Role.objects.filter(user=self.request.user)]
|
|
|
|
mails = Mail.objects.filter(status='done-dga')
|
|
content_type = ContentType.objects.get_for_model(Mail)
|
|
context['objects'] = Inbox.objects.filter(
|
|
role_slug__in=user_roles,
|
|
source_type=content_type,
|
|
source_pk__in=[x.id for x in mails],
|
|
subtype=self.subtype, done=False).order_by('source_pk').distinct('source_pk')
|
|
|
|
context['subtype'] = self.subtype
|
|
context['button_text'] = self.button_text
|
|
context['title'] = self.title
|
|
context['display_comments_field'] = True
|
|
|
|
return context
|
|
|
|
def post(self, request, *args, **kwargs):
|
|
user_roles = [x.uuid for x in Role.objects.filter(user=self.request.user)]
|
|
content_type = ContentType.objects.get_for_model(Mail)
|
|
Inbox.objects.filter(
|
|
source_type=content_type,
|
|
source_pk__in=request.POST.getlist('object-pk'),
|
|
role_slug__in=user_roles,
|
|
subtype=self.subtype).update(
|
|
done=True,
|
|
comments=request.POST.get('comments'))
|
|
return HttpResponseRedirect('.')
|
|
|
|
|
|
class TableMandatoryAvis(MailTable):
|
|
subtype = Inbox.MANDATORY_AVIS
|
|
title = _('Mandatory Avis Copies')
|
|
|
|
table_mandatory_avis = login_required(TableMandatoryAvis.as_view())
|
|
|
|
|
|
class TableAvis(MailTable):
|
|
subtype = Inbox.AVIS
|
|
title = _('Avis Copies')
|
|
|
|
table_avis = login_required(TableAvis.as_view())
|
|
|
|
|
|
class TableInfo(MailTable):
|
|
subtype = Inbox.INFO
|
|
title = _('Info Copies')
|
|
|
|
table_info = login_required(TableInfo.as_view())
|
|
|
|
|
|
class TableWaiting(TemplateView):
|
|
template_name = 'alfortville/mail-table-waiting.html'
|
|
|
|
def get_context_data(self, *args, **kwargs):
|
|
context = super(TableWaiting, self).get_context_data(**kwargs)
|
|
|
|
content_type = ContentType.objects.get_for_model(Mail)
|
|
context['objects'] = Mail.objects.filter(
|
|
status__in=('', 'done-qualif', 'done-dgs')).order_by('creation_timestamp')
|
|
return context
|
|
|
|
table_waiting = login_required(TableWaiting.as_view())
|
|
|
|
|
|
@login_required
|
|
def count_dgs(request, *args, **kwargs):
|
|
count = DgsMailHome(request).get_queryset().count()
|
|
return response_for_json(request, {'count': count})
|
|
|
|
@login_required
|
|
def count_dga(request, *args, **kwargs):
|
|
count = DgaMailHome(request).get_queryset().count()
|
|
return response_for_json(request, {'count': count})
|
|
|
|
@login_required
|
|
def count_info(request, *args, **kwargs):
|
|
table = TableInfo()
|
|
table.request = request
|
|
count = table.get_context_data().get('objects').count()
|
|
return response_for_json(request, {'count': count})
|
|
|
|
@login_required
|
|
def count_avis(request, *args, **kwargs):
|
|
table = TableAvis()
|
|
table.request = request
|
|
count = table.get_context_data().get('objects').count()
|
|
return response_for_json(request, {'count': count})
|
|
|
|
@login_required
|
|
def count_mandatory_avis(request, *args, **kwargs):
|
|
table = TableMandatoryAvis()
|
|
table.request = request
|
|
count = table.get_context_data().get('objects').count()
|
|
return response_for_json(request, {'count': count})
|