344 lines
14 KiB
Python
344 lines
14 KiB
Python
from django.shortcuts import render,redirect
|
|
from django.http import JsonResponse
|
|
from django.urls import reverse_lazy
|
|
from django.contrib.auth import logout
|
|
from django.views.generic.list import ListView
|
|
from django.views.generic.detail import DetailView
|
|
from django.views.generic.edit import CreateView
|
|
from rest_framework.authentication import TokenAuthentication
|
|
from rest_framework.decorators import authentication_classes,api_view,permission_classes
|
|
from rest_framework.exceptions import AuthenticationFailed
|
|
|
|
|
|
from rest_framework.views import APIView
|
|
from rest_framework.response import Response
|
|
from rest_framework.permissions import IsAuthenticated
|
|
from rest_framework import status, permissions
|
|
from django.contrib.auth.models import User
|
|
|
|
from django.contrib import messages
|
|
from django.contrib.auth.mixins import LoginRequiredMixin, UserPassesTestMixin
|
|
from .models import sistemas_por_cliente, DeviceHistory,Device
|
|
from .serializers import DeviceSerializer
|
|
from .permissions import HasAuthorizationHeader
|
|
from .models import Sistema, BitacoraErrores
|
|
from Clientes.models import Clientes
|
|
from .forms import SistemaForm,sistemas_por_clienteForm
|
|
from django.utils import timezone
|
|
import re
|
|
|
|
from rest_framework.authtoken.models import Token
|
|
from django.utils import timezone
|
|
from datetime import timedelta
|
|
|
|
from django.contrib.sessions.models import Session
|
|
|
|
from django.http.response import HttpResponse
|
|
import os
|
|
import mimetypes
|
|
import traceback
|
|
|
|
from django.core.files.base import ContentFile
|
|
from .customStorage import CustomStorage
|
|
|
|
from django.contrib.auth.decorators import login_required
|
|
|
|
from django.db.models import Q
|
|
|
|
@login_required
|
|
def uploadZipViewHTML(request):
|
|
if request.method=="GET":
|
|
user = request.user
|
|
if user.is_superuser:
|
|
token= Token.objects.get(user=user)
|
|
sistemas= Sistema.objects.all()
|
|
else:
|
|
redirect('index')
|
|
context = {
|
|
'token':token.key,
|
|
'sistemas':sistemas,
|
|
}
|
|
template_name= 'Sistemas/manageSystem/uploadZip.html'
|
|
return render(request, template_name,context)
|
|
return HttpResponse('Only GET request are allowed')
|
|
|
|
|
|
class UploadZipVersionView(APIView):
|
|
"""API CLASS for upload the CFDI Version into the server"""
|
|
authentication_classes = (TokenAuthentication,) #this is by default in settings.py
|
|
permissions_classes=[IsAuthenticated,]
|
|
def post(self,request,format=None):
|
|
try:
|
|
zip_file=request.FILES['zip_file']
|
|
if zip_file:
|
|
fileN, file_ext = os.path.splitext(zip_file.name)
|
|
if file_ext !=".zip":
|
|
messages.error(self.request, f'El archivo {fileN} no es .zip')
|
|
return redirect('uploadZipViewHTML')
|
|
version = request.POST.get('version')
|
|
sistema = request.POST.get('sistema')
|
|
except KeyError as ker:
|
|
BitacoraErrores.objects.create(level=2, message=f'keyError Try: {str(ker)}', traceback=traceback.format_exc(), view='Sistemas.UploadZipVersionView')
|
|
return Response({'error':'Zip file is required'},status=status.HTTP_400_BAD_REQUEST)
|
|
try:
|
|
versionSistema = Sistema.objects.get(id=sistema)
|
|
versionSistema.version=version
|
|
print('versionSistema: ',versionSistema.version)
|
|
versionSistema.save()
|
|
|
|
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
download_system_dir = os.path.join(BASE_DIR,'downloadSystems')
|
|
custom_storage = CustomStorage(location=download_system_dir)
|
|
#print('path exist? ',os.path.exists(download_system_dir))
|
|
fileName = custom_storage.save(zip_file.name, ContentFile(zip_file.read()))
|
|
messages.success(request, 'The files has been upload successfully')
|
|
except Exception as ex:
|
|
BitacoraErrores.objects.create(level=2, message=str(ex), traceback=traceback.format_exc(), view='Sistemas.UploadZipVersionView')
|
|
messages.error(request, f'was an error trying upload the zip file: {ex}')
|
|
return Response({'error':str(ex)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
|
|
return Response({'message':'File uploaded successfully'}, status=status.HTTP_201_CREATED)
|
|
|
|
#need to add it to C# header request or webclient in CFDI and Actualizador
|
|
@api_view(['GET'])
|
|
@authentication_classes([TokenAuthentication])
|
|
@permission_classes([IsAuthenticated])
|
|
def download_version_FromServer(request):
|
|
"""dw"""
|
|
try:
|
|
sys= request.GET.get('sys')
|
|
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
if sys:
|
|
filename=f'{sys}.zip'
|
|
else:
|
|
filename = 'CFDI.zip'
|
|
filepath = f'{BASE_DIR }/downloadSystems/{filename}'
|
|
with open(filepath, 'rb') as path:
|
|
mime_type, _ = mimetypes.guess_type(filepath)
|
|
response = HttpResponse(path.read(),content_type=mime_type)
|
|
response['Content-Disposition']= f'attachment; filename={filename}'
|
|
return response
|
|
except Exception as ex:
|
|
BitacoraErrores.objects.create(level=2, message=str(ex), traceback=traceback.format_exc(), view='Sistemas.download_version_FromServer')
|
|
return HttpResponse(status=204)
|
|
|
|
class UsersConnectedList(UserPassesTestMixin,LoginRequiredMixin,ListView):
|
|
"""CV"""
|
|
model = User
|
|
template_name= 'Sistemas/Usuarios/lista.html'
|
|
|
|
def get_queryset(self):
|
|
return self.get_logged_in_users()
|
|
|
|
def test_func(self):
|
|
res = self.request.user.groups.filter(name= 'admin_soft')
|
|
if not res:
|
|
messages.error(self.request, f'Lo sentimos. La página que buscas no está disponible, no cuentas con los permisos.')
|
|
return res
|
|
|
|
def get_logged_in_users(self):
|
|
""""""
|
|
sessions = Session.objects.filter(expire_date__gte=timezone.now())
|
|
|
|
# Get all non-expired tokens
|
|
#tokens = Token.objects.filter(created__gte=timezone.now() - timedelta(hours=1))
|
|
tokens = Token.objects.all()
|
|
|
|
# Get a list of logged-in user ids
|
|
uid_list = [token.user_id for token in tokens]
|
|
|
|
# Get the logged-in users
|
|
users = User.objects.filter(id__in=uid_list)
|
|
for user,session in zip(users,sessions):
|
|
user.session_data = session.get_decoded()
|
|
user.session_expire = session.expire_date
|
|
return users
|
|
|
|
class SistemasXCliente_ListView(UserPassesTestMixin,LoginRequiredMixin, ListView):
|
|
model = sistemas_por_cliente
|
|
paginate_by = 100
|
|
template_name = 'Sistemas/Xclientes/lista.html'
|
|
|
|
def test_func(self):
|
|
res = self.request.user.groups.filter(name= 'admin_soft')
|
|
if not res:
|
|
messages.error(self.request, f'Lo sentimos. La página que buscas no está disponible, no cuentas con los permisos.')
|
|
return res
|
|
|
|
class SistemasBitacora_ListView(UserPassesTestMixin,LoginRequiredMixin, ListView):
|
|
|
|
model = BitacoraErrores
|
|
paginate_by = 5
|
|
template_name = 'Sistemas/Xclientes/listaBitacora.html'
|
|
|
|
def test_func(self):
|
|
res = self.request.user.groups.filter(name= 'admin_soft')
|
|
if not res:
|
|
messages.error(self.request, f'Lo sentimos. La página que buscas no está disponible, no cuentas con los permisos.')
|
|
return res
|
|
|
|
class SistemasXCliente_DetailView(UserPassesTestMixin,LoginRequiredMixin, DetailView):
|
|
model = sistemas_por_cliente
|
|
template_name= 'Sistemas/Xclientes/detail.html'
|
|
def test_func(self):
|
|
"""Verifica que el usuario sea parte del grupo de admin_soft"""
|
|
res = self.request.user.groups.filter(name= 'admin_soft')
|
|
if not res:
|
|
messages.error(self.request, f'Lo sentimos. La página que buscas no está disponible, no cuentas con los permisos.')
|
|
return res
|
|
def get_context_data(self, **kwargs):
|
|
"""Filtra los Dispositivos por sistema del cliente"""
|
|
context = super().get_context_data(**kwargs)
|
|
sistemas_por_cliente_ = self.object #type: ignore
|
|
sistema = sistemas_por_cliente_.id_sistema
|
|
cliente = sistemas_por_cliente_.cliente
|
|
|
|
context['devices'] = Device.objects.filter(sistema=sistema, client=cliente.id)
|
|
return context
|
|
|
|
class SistemasxCliente_CreateView(CreateView):
|
|
model = sistemas_por_cliente
|
|
fields = ['id_sistema', 'cliente', 'num_licencias']
|
|
template_name = 'Sistemas/Xclientes/sistema_create.html'
|
|
success_url = reverse_lazy('lista_sistmas')
|
|
|
|
class Sistema_CreateView(CreateView):
|
|
model = Sistema
|
|
fields = ['nombre_sistema', 'version', 'id']
|
|
template_name = 'Sistemas/Xclientes/sistema_create_form.html'
|
|
success_url = '/sistemas/'
|
|
def form_valid(self, form):
|
|
response = super().form_valid(form)
|
|
data = {
|
|
'id': self.object.id,
|
|
'nombre_sistema': self.object.nombre_sistema,
|
|
'version': self.object.version,
|
|
}
|
|
return JsonResponse(data)
|
|
def form_invalid(self, form):
|
|
errors =form.errors.as_text()
|
|
return JsonResponse({'errors': f'{errors}'}, status=200)
|
|
'''
|
|
========================= API Views
|
|
'''
|
|
|
|
class GetDeviceToken(APIView):
|
|
"""Recobra el Token DRF del Device"""
|
|
authentication_classes= [TokenAuthentication]
|
|
permissions_classes=[IsAuthenticated, HasAuthorizationHeader]
|
|
def post(self,request):
|
|
try:
|
|
data = request.data
|
|
#sis = Sistema.objects.get(nombre_sistema=data.get('sistema'))
|
|
#ip_address = data.get('ip_address')
|
|
cli = Clientes.objects.get(RFC=data.get('client'))
|
|
device_name= data.get('device_name')
|
|
macAddress = data.get('macAddress')
|
|
database = data.get('database')
|
|
|
|
#username_ = f"Device_{cli.RFC}_{device_name}_{ip_address}_{macAddress}"
|
|
username_ = f"Device_{cli.RFC}_{device_name}_{macAddress}"
|
|
username_ = re.sub(r'\W+', '', username_)
|
|
|
|
device = Device.objects.filter(
|
|
username__username__icontains=username_,
|
|
database=database
|
|
).first()
|
|
print('device',device)
|
|
if device is not None:
|
|
token = {"token":str(device.token)}
|
|
else:
|
|
token = {"token":""}
|
|
return Response(token)
|
|
except Exception as ex:
|
|
BitacoraErrores.objects.create(
|
|
level=2,
|
|
message=str(ex),
|
|
traceback=traceback.format_exc(),
|
|
view='Sistemas.GetDeviceToken'
|
|
)
|
|
return Response({'Error':f'{ex}','isError':True}, status=status.HTTP_200_OK)
|
|
|
|
class RegisterDeviceView(APIView):
|
|
"""Registra al Device"""
|
|
#permissions_classes = (permissions.AllowAny,)
|
|
permissions_classes=[IsAuthenticated, HasAuthorizationHeader]
|
|
def post(self,request):
|
|
try:
|
|
serializer = DeviceSerializer(data=request.data,context={'request':request})
|
|
if serializer.is_valid():
|
|
serializer.save()
|
|
return Response(serializer.data, status=status.HTTP_200_OK)
|
|
else:
|
|
return Response({'Error':f'{serializer.errors}','isError':True}, status=status.HTTP_200_OK)
|
|
except Exception as ex:
|
|
BitacoraErrores.objects.create(level=2, message=str(ex), traceback=traceback.format_exc(),
|
|
view='Sistemas.RegisterDeviceView')
|
|
return Response({'Error':f'{ex}','isError':True}, status=status.HTTP_200_OK)
|
|
|
|
class AuthenticateDeviceView(APIView):
|
|
"""Autentica al device"""
|
|
authentication_classes= [TokenAuthentication]
|
|
permissions_classes=[IsAuthenticated, HasAuthorizationHeader]
|
|
|
|
def get(self, request):
|
|
try:
|
|
obj, created = DeviceHistory.objects.get_or_create(
|
|
device=request.user.device,
|
|
ip_address=request.META.get('REMOTE_ADDR'),
|
|
)
|
|
obj.last_authentication=timezone.now()
|
|
obj.save()
|
|
device_data=DeviceSerializer(request.user.device).data
|
|
if device_data.serializer.is_valid:
|
|
return Response(device_data.serializer.data, status=status.HTTP_200_OK)
|
|
else:
|
|
BitacoraErrores.objects.create(level=3, message=str(device_data.serializer.errors), traceback=traceback.format_exc(), view='Sistemas.AuthenticateDeviceView')
|
|
return Response(
|
|
{'Error':f'{device_data.serializer.errors}','isError':True}
|
|
, status=status.HTTP_200_OK)
|
|
except Exception as ex:
|
|
BitacoraErrores.objects.create(level=2, message=str(ex), traceback=traceback.format_exc(), view='Sistemas.AuthenticateDeviceView')
|
|
return Response(
|
|
{'Error':f'{ex}','isError':True}
|
|
, status=status.HTTP_200_OK
|
|
)
|
|
|
|
class LogoutView(APIView):
|
|
authentication_classes = (TokenAuthentication,)
|
|
permission_classes = (IsAuthenticated,HasAuthorizationHeader,)
|
|
|
|
def post(self, request):
|
|
logout(request)
|
|
return Response({'OK':'Dispositivo desautenticado'},status=200)
|
|
|
|
|
|
class CheckVersionView(APIView):
|
|
authentication_classes = (TokenAuthentication,)
|
|
permission_classes = (IsAuthenticated,HasAuthorizationHeader,)
|
|
def post(self, request,*args, **kwargs):
|
|
try:
|
|
version = request.data.get('version')
|
|
|
|
client_version = [int(x) for x in version.split(".")]
|
|
|
|
try:
|
|
ver = Sistema.objects.get(nombre_sistema="CFDI")
|
|
server_version = [int(x) for x in ver.version.split(".")]
|
|
except Exception as ex:
|
|
BitacoraErrores.objects.create(level=2, message=str(ex), traceback=traceback.format_exc(), view='Sistemas.CheckVersionView')
|
|
return Response({'Error':f'{ex}','isError':True})
|
|
|
|
result=False
|
|
for cont, ele in enumerate(client_version):
|
|
if client_version[cont] != server_version[cont]:
|
|
print('client version does not match server version')
|
|
result = True
|
|
break
|
|
else:
|
|
result = False
|
|
print('client version matches server version')
|
|
return Response({'success':True, 'actualizar':result})
|
|
except Exception as ex:
|
|
BitacoraErrores.objects.create(level=2, message=str(ex), traceback=traceback.format_exc(), view='Sistemas.CheckVersionView')
|
|
return Response({'Error':f'{ex}','isError':True})
|