IdHub/idhub/admin/views.py

786 lines
23 KiB
Python

import os
import csv
import json
import copy
import logging
import pandas as pd
from pathlib import Path
from jsonschema import validate
from smtplib import SMTPException
from django.conf import settings
from django.utils.translation import gettext_lazy as _
from django.views.generic.base import TemplateView
from django.views.generic.edit import UpdateView, CreateView, DeleteView
from django.shortcuts import get_object_or_404, redirect
from django.urls import reverse_lazy
from django.http import HttpResponse
from django.contrib import messages
from apiregiter import iota
from idhub_auth.models import User
from idhub.mixins import AdminView
from idhub.email.views import NotifyActivateUserByEmail
from idhub.admin.forms import ImportForm
from idhub.models import (
DID,
File_datas,
Membership,
Rol,
Service,
Schemas,
UserRol,
VerificableCredential,
)
class AdminDashboardView(AdminView, TemplateView):
template_name = "idhub/admin/dashboard.html"
title = _('Dashboard')
subtitle = _('Success')
icon = 'bi bi-bell'
section = "Home"
class People(AdminView):
title = _("People Management")
section = "People"
class AccessControl(AdminView, TemplateView):
title = _("Access Control Management")
section = "AccessControl"
class Credentials(AdminView, TemplateView):
title = _("Credentials Management")
section = "Credentials"
class SchemasMix(AdminView, TemplateView):
title = _("Templates Management")
section = "Templates"
class ImportExport(AdminView, TemplateView):
title = _("Massive Data Management")
section = "ImportExport"
class AdminPeopleListView(People, TemplateView):
template_name = "idhub/admin/people.html"
subtitle = _('People list')
icon = 'bi bi-person'
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
context.update({
'users': User.objects.filter(),
})
return context
class AdminPeopleView(People, TemplateView):
template_name = "idhub/admin/user.html"
subtitle = _('User Profile')
icon = 'bi bi-person'
model = User
def get(self, request, *args, **kwargs):
self.pk = kwargs['pk']
self.object = get_object_or_404(self.model, pk=self.pk)
return super().get(request, *args, **kwargs)
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
context.update({
'object': self.object,
})
return context
class AdminPeopleActivateView(AdminPeopleView):
def get(self, request, *args, **kwargs):
self.pk = kwargs['pk']
self.object = get_object_or_404(self.model, pk=self.pk)
if self.object == self.request.user:
messages.error(self.request, _('Is not possible deactivate your account!'))
return redirect('idhub:admin_people', self.object.id)
if self.object.is_active:
self.object.is_active = False
else:
self.object.is_active = True
self.object.save()
return redirect('idhub:admin_people', self.object.id)
class AdminPeopleDeleteView(AdminPeopleView):
def get(self, request, *args, **kwargs):
self.pk = kwargs['pk']
self.object = get_object_or_404(self.model, pk=self.pk)
if self.object != self.request.user:
self.object.delete()
else:
messages.error(self.request, _('Is not possible delete your account!'))
return redirect('idhub:admin_people_list')
class AdminPeopleEditView(AdminPeopleView, UpdateView):
template_name = "idhub/admin/user_edit.html"
fields = ('first_name', 'last_name', 'email')
success_url = reverse_lazy('idhub:admin_people_list')
class AdminPeopleRegisterView(NotifyActivateUserByEmail, People, CreateView):
template_name = "idhub/admin/people_register.html"
subtitle = _('People Register')
icon = 'bi bi-person'
model = User
fields = ('first_name', 'last_name', 'email')
success_url = reverse_lazy('idhub:admin_people_list')
def get_success_url(self):
self.success_url = reverse_lazy(
'idhub:admin_people_membership_new',
kwargs={"pk": self.object.id}
)
return self.success_url
def form_valid(self, form):
user = form.save()
messages.success(self.request, _('The account is created successfully'))
if user.is_active:
try:
self.send_email(user)
except SMTPException as e:
messages.error(self.request, e)
return super().form_valid(form)
class AdminPeopleMembershipRegisterView(People, CreateView):
template_name = "idhub/admin/people_membership_register.html"
subtitle = _('People add membership')
icon = 'bi bi-person'
model = Membership
fields = ('type', 'start_date', 'end_date')
success_url = reverse_lazy('idhub:admin_people_list')
def get(self, request, *args, **kwargs):
self.pk = kwargs['pk']
self.user = get_object_or_404(User, pk=self.pk)
return super().get(request, *args, **kwargs)
def post(self, request, *args, **kwargs):
self.pk = kwargs['pk']
self.user = get_object_or_404(User, pk=self.pk)
return super().post(request, *args, **kwargs)
def get_form(self):
form = super().get_form()
form.fields['start_date'].widget.input_type = 'date'
form.fields['end_date'].widget.input_type = 'date'
return form
def get_form_kwargs(self):
self.object = self.model(user=self.user)
kwargs = super().get_form_kwargs()
return kwargs
def get_success_url(self):
self.success_url = reverse_lazy(
'idhub:admin_people_rol_new',
kwargs={"pk": self.user.id}
)
return self.success_url
class AdminPeopleMembershipEditView(People, CreateView):
template_name = "idhub/admin/people_membership_register.html"
subtitle = _('People add membership')
icon = 'bi bi-person'
model = Membership
fields = ('type', 'start_date', 'end_date')
success_url = reverse_lazy('idhub:admin_people_list')
def get_form(self):
form = super().get_form()
form.fields['start_date'].widget.input_type = 'date'
form.fields['end_date'].widget.input_type = 'date'
return form
def get_form_kwargs(self):
pk = self.kwargs.get('pk')
if pk:
self.object = get_object_or_404(self.model, pk=pk)
kwargs = super().get_form_kwargs()
return kwargs
class AdminPeopleMembershipDeleteView(AdminPeopleView):
model = Membership
def get(self, request, *args, **kwargs):
self.pk = kwargs['pk']
self.object = get_object_or_404(self.model, pk=self.pk)
if self.object != self.request.user:
user = self.object.user
self.object.delete()
else:
messages.error(self.request, _('Is not possible delete your account!'))
return redirect('idhub:admin_people_edit', user.id)
class AdminPeopleRolRegisterView(People, CreateView):
template_name = "idhub/admin/people_rol_register.html"
subtitle = _('Add Rol to User')
icon = 'bi bi-person'
model = UserRol
fields = ('service',)
def get(self, request, *args, **kwargs):
self.pk = kwargs['pk']
self.user = get_object_or_404(User, pk=self.pk)
return super().get(request, *args, **kwargs)
def post(self, request, *args, **kwargs):
self.pk = kwargs['pk']
self.user = get_object_or_404(User, pk=self.pk)
return super().post(request, *args, **kwargs)
def get_form_kwargs(self):
self.object = self.model(user=self.user)
kwargs = super().get_form_kwargs()
return kwargs
def get_success_url(self):
self.success_url = reverse_lazy(
'idhub:admin_people_edit',
kwargs={"pk": self.user.id}
)
return self.success_url
class AdminPeopleRolEditView(People, CreateView):
template_name = "idhub/admin/people_rol_register.html"
subtitle = _('Edit Rol to User')
icon = 'bi bi-person'
model = UserRol
fields = ('service',)
def get_form_kwargs(self):
pk = self.kwargs.get('pk')
if pk:
self.object = get_object_or_404(self.model, pk=pk)
kwargs = super().get_form_kwargs()
return kwargs
def get_success_url(self):
self.success_url = reverse_lazy(
'idhub:admin_people_edit',
kwargs={"pk": self.object.user.id}
)
return self.success_url
class AdminPeopleRolDeleteView(AdminPeopleView):
model = UserRol
def get(self, request, *args, **kwargs):
self.pk = kwargs['pk']
self.object = get_object_or_404(self.model, pk=self.pk)
user = self.object.user
self.object.delete()
return redirect('idhub:admin_people_edit', user.id)
class AdminRolesView(AccessControl):
template_name = "idhub/admin/roles.html"
subtitle = _('Roles Management')
icon = ''
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
context.update({
'roles': Rol.objects,
})
return context
class AdminRolRegisterView(AccessControl, CreateView):
template_name = "idhub/admin/rol_register.html"
subtitle = _('Add Rol')
icon = ''
model = Rol
fields = ('name',)
success_url = reverse_lazy('idhub:admin_roles')
object = None
class AdminRolEditView(AccessControl, CreateView):
template_name = "idhub/admin/rol_register.html"
subtitle = _('Edit Rol')
icon = ''
model = Rol
fields = ('name',)
success_url = reverse_lazy('idhub:admin_roles')
def get_form_kwargs(self):
pk = self.kwargs.get('pk')
if pk:
self.object = get_object_or_404(self.model, pk=pk)
kwargs = super().get_form_kwargs()
return kwargs
class AdminRolDeleteView(AccessControl):
model = Rol
def get(self, request, *args, **kwargs):
self.pk = kwargs['pk']
self.object = get_object_or_404(self.model, pk=self.pk)
self.object.delete()
return redirect('idhub:admin_roles')
class AdminServicesView(AccessControl):
template_name = "idhub/admin/services.html"
subtitle = _('Service Management')
icon = ''
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
context.update({
'services': Service.objects,
})
return context
class AdminServiceRegisterView(AccessControl, CreateView):
template_name = "idhub/admin/service_register.html"
subtitle = _('Add Service')
icon = ''
model = Service
fields = ('domain', 'description', 'rol')
success_url = reverse_lazy('idhub:admin_services')
object = None
class AdminServiceEditView(AccessControl, CreateView):
template_name = "idhub/admin/service_register.html"
subtitle = _('Edit Service')
icon = ''
model = Service
fields = ('domain', 'description', 'rol')
success_url = reverse_lazy('idhub:admin_services')
def get_form_kwargs(self):
pk = self.kwargs.get('pk')
if pk:
self.object = get_object_or_404(self.model, pk=pk)
kwargs = super().get_form_kwargs()
return kwargs
class AdminServiceDeleteView(AccessControl):
model = Service
def get(self, request, *args, **kwargs):
self.pk = kwargs['pk']
self.object = get_object_or_404(self.model, pk=self.pk)
self.object.delete()
return redirect('idhub:admin_services')
class AdminCredentialsView(Credentials):
template_name = "idhub/admin/credentials.html"
subtitle = _('Credentials list')
icon = ''
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
context.update({
'credentials': VerificableCredential.objects,
})
return context
class AdminCredentialView(Credentials):
template_name = "idhub/admin/issue_credentials.html"
subtitle = _('Change status of Credential')
icon = ''
model = VerificableCredential
def get(self, request, *args, **kwargs):
self.pk = kwargs['pk']
self.object = get_object_or_404(self.model, pk=self.pk)
return super().get(request, *args, **kwargs)
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
context.update({
'object': self.object,
})
return context
class AdminRevokeCredentialsView(Credentials):
template_name = "idhub/admin/revoke_credentials.html"
subtitle = _('Revoke Credentials')
icon = ''
class AdminDidsView(Credentials):
template_name = "idhub/admin/dids.html"
subtitle = _('Organization Identities (DID)')
icon = 'bi bi-patch-check-fill'
wallet = True
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
context.update({
'dids': DID.objects,
})
return context
class AdminDidRegisterView(Credentials, CreateView):
template_name = "idhub/admin/did_register.html"
subtitle = _('Add a new Organization Identities (DID)')
icon = 'bi bi-patch-check-fill'
wallet = True
model = DID
fields = ('did', 'label')
success_url = reverse_lazy('idhub:admin_dids')
object = None
def get_form_kwargs(self):
kwargs = super().get_form_kwargs()
kwargs['initial'] = {
'did': iota.issue_did()
}
return kwargs
def get_form(self):
form = super().get_form()
form.fields['did'].required = False
form.fields['did'].disabled = True
return form
def form_valid(self, form):
form.save()
messages.success(self.request, _('DID created successfully'))
return super().form_valid(form)
class AdminDidEditView(Credentials, UpdateView):
template_name = "idhub/admin/did_register.html"
subtitle = _('Organization Identities (DID)')
icon = 'bi bi-patch-check-fill'
wallet = True
model = DID
fields = ('did', 'label')
success_url = reverse_lazy('idhub:admin_dids')
def get(self, request, *args, **kwargs):
self.pk = kwargs['pk']
self.object = get_object_or_404(self.model, pk=self.pk)
return super().get(request, *args, **kwargs)
def get_form(self):
form = super().get_form()
form.fields['did'].required = False
form.fields['did'].disabled = True
return form
def form_valid(self, form):
user = form.save()
messages.success(self.request, _('DID updated successfully'))
return super().form_valid(form)
class AdminDidDeleteView(Credentials, DeleteView):
subtitle = _('Organization Identities (DID)')
icon = 'bi bi-patch-check-fill'
wallet = True
model = DID
success_url = reverse_lazy('idhub:admin_dids')
def get(self, request, *args, **kwargs):
self.pk = kwargs['pk']
self.object = get_object_or_404(self.model, pk=self.pk)
self.object.delete()
messages.success(self.request, _('DID delete successfully'))
return redirect(self.success_url)
class AdminWalletCredentialsView(Credentials):
template_name = "idhub/admin/wallet_credentials.html"
subtitle = _('Credentials')
icon = 'bi bi-patch-check-fill'
wallet = True
class AdminWalletConfigIssuesView(Credentials):
template_name = "idhub/admin/wallet_issues.html"
subtitle = _('Configure Issues')
icon = 'bi bi-patch-check-fill'
wallet = True
class AdminSchemasView(SchemasMix):
template_name = "idhub/admin/schemas.html"
subtitle = _('Template List')
icon = ''
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
context.update({
'schemas': Schemas.objects,
})
return context
class AdminSchemasDeleteView(SchemasMix):
def get(self, request, *args, **kwargs):
self.pk = kwargs['pk']
self.object = get_object_or_404(Schemas, pk=self.pk)
self.object.delete()
return redirect('idhub:admin_schemas')
class AdminSchemasDownloadView(SchemasMix):
def get(self, request, *args, **kwargs):
self.pk = kwargs['pk']
self.object = get_object_or_404(Schemas, pk=self.pk)
response = HttpResponse(self.object.data, content_type="application/json")
response['Content-Disposition'] = 'inline; filename={}'.format(self.object.file_schema)
return response
class AdminSchemasNewView(SchemasMix):
template_name = "idhub/admin/schemas_new.html"
subtitle = _('Upload Template')
icon = ''
success_url = reverse_lazy('idhub:admin_schemas')
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
context.update({
'form': SchemaForm(),
})
return context
def post(self, request, *args, **kwargs):
form = SchemaForm(request.POST, request.FILES)
if form.is_valid():
schema = self.handle_uploaded_file()
if not schema:
messages.error(request, _("There are some errors in the file"))
return super().get(request, *args, **kwargs)
return redirect(self.success_url)
else:
return super().get(request, *args, **kwargs)
return super().post(request, *args, **kwargs)
def handle_uploaded_file(self):
f = self.request.FILES.get('file_template')
if not f:
return
file_name = f.name
if Schemas.objects.filter(file_schema=file_name).exists():
messages.error(self.request, _("This template already exists!"))
return
try:
data = f.read().decode('utf-8')
json.loads(data)
except Exception:
messages.error(self.request, _('This is not a schema valid!'))
return
schema = Schemas.objects.create(file_schema=file_name, data=data)
schema.save()
return schema
class AdminSchemasImportView(SchemasMix):
template_name = "idhub/admin/schemas_import.html"
subtitle = _('Import Template')
icon = ''
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
context.update({
'schemas': self.get_schemas(),
})
return context
def get_schemas(self):
schemas_files = os.listdir(settings.SCHEMAS_DIR)
schemas = [x for x in schemas_files
if not Schemas.objects.filter(file_schema=x).exists()]
return schemas
class AdminSchemasImportAddView(SchemasMix):
def get(self, request, *args, **kwargs):
file_name = kwargs['file_schema']
schemas_files = os.listdir(settings.SCHEMAS_DIR)
if not file_name in schemas_files:
messages.error(self.request, f"The schema {file_name} not exist!")
return redirect('idhub:admin_schemas_import')
schema = self.create_schema(file_name)
if schema:
messages.success(self.request, _("The schema add successfully!"))
return redirect('idhub:admin_schemas_import')
def create_schema(self, file_name):
data = self.open_file(file_name)
try:
json.loads(data)
except Exception:
messages.error(self.request, _('This is not a schema valid!'))
return
schema = Schemas.objects.create(file_schema=file_name, data=data)
schema.save()
return schema
def open_file(self, file_name):
data = ''
filename = Path(settings.SCHEMAS_DIR).joinpath(file_name)
with filename.open() as schema_file:
data = schema_file.read()
return data
class AdminImportView(ImportExport):
template_name = "idhub/admin/import.html"
subtitle = _('Import')
icon = ''
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
context.update({
'dates': File_datas.objects,
})
return context
class AdminImportStep2View(ImportExport):
template_name = "idhub/admin/import_step2.html"
subtitle = _('Import')
icon = ''
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
context.update({
'schemas': Schemas.objects,
})
return context
class AdminImportStep3View(ImportExport):
template_name = "idhub/admin/import_step3.html"
subtitle = _('Import')
icon = ''
success_url = reverse_lazy('idhub:admin_import')
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
context.update({
'form': ImportForm(),
})
return context
def post(self, request, *args, **kwargs):
self.pk = kwargs['pk']
self.schema = get_object_or_404(Schemas, pk=self.pk)
form = ImportForm(request.POST, request.FILES)
if form.is_valid():
result = self.handle_uploaded_file()
if not result:
messages.error(request, _("There are some errors in the file"))
return super().get(request, *args, **kwargs)
return redirect(self.success_url)
else:
return super().get(request, *args, **kwargs)
return super().post(request, *args, **kwargs)
def handle_uploaded_file(self):
f = self.request.FILES.get('file_import')
if not f:
messages.error(self.request, _("There aren't file"))
return
file_name = f.name
if File_datas.objects.filter(file_name=file_name, success=True).exists():
messages.error(self.request, _("This file already exists!"))
return
self.json_schema = json.loads(self.schema.data)
df = pd.read_csv (f, delimiter="\t", quotechar='"', quoting=csv.QUOTE_ALL)
data_pd = df.fillna('').to_dict()
rows = {}
if not data_pd:
File_datas.objects.create(file_name=file_name, success=False)
return
for n in range(df.last_valid_index()+1):
row = {}
for k in data_pd.keys():
row[k] = data_pd[k][n]
user = self.validate(n, row)
if not user:
File_datas.objects.create(file_name=file_name, success=False)
return
rows[user] = row
File_datas.objects.create(file_name=file_name)
for k, v in rows.items():
self.create_credential(k, v)
return True
def validate(self, line, row):
try:
validate(instance=row, schema=self.json_schema)
except Exception as e:
messages.error(self.request, "line {}: {}".format(line+1, e))
return
user = User.objects.filter(email=row.get('email'))
if not user:
txt = _('The user not exist!')
messages.error(self.request, "line {}: {}".format(line+1, txt))
return
return user.first()
def create_credential(self, user, row):
d = copy.copy(self.json_schema)
d['instance'] = row
return VerificableCredential.objects.create(
verified=False,
user=user,
data=json.dumps(d)
)