from django.shortcuts import render,redirect from django.contrib.auth import logout from django.views.generic.list import ListView from django.views.generic.detail import DetailView from rest_framework.authentication import TokenAuthentication 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 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.storage import default_storage from django.core.files.base import ContentFile from .customStorage import CustomStorage from django.contrib.auth.decorators import login_required @login_required def uploadZipViewHTML(request): if request.method=="GET": user = request.user if user.is_superuser: token= Token.objects.get(user=user) else: redirect('index') context = { 'token':token.key, } 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""" permissions_classes=[IsAuthenticated,] #authentication_classes = (TokenAuthentication,) #this is by default in settings.py def post(self,request,format=None): try: zip_file=request.FILES['zip_file'] except KeyError: return Response({'error':'Zip file is required'},status=status.HTTP_400_BAD_REQUEST) try: 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())) except Exception as ex: BitacoraErrores.objects.create(level=2, message=str(ex), traceback=traceback.format_exc(), view='Sistemas.UploadZipVersionView') return Response({'error':str(ex)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR) return Response({'message':'File uploaded successfully'}, status=status.HTTP_201_CREATED) def download_version_FromServer(request): try: BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) print('BASE_DIR',BASE_DIR) filename = 'CFDI.zip' filepath = f'{BASE_DIR }/downloadSystems/{filename}' #path = open(filepath, 'r') 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) def get_logged_in_users(): 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 UsersConnectedList(UserPassesTestMixin,LoginRequiredMixin,ListView): model = User template_name= 'Sistemas/Usuarios/lista.html' def get_queryset(self): return 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 class SistemasXCliente_ListView(UserPassesTestMixin,LoginRequiredMixin, ListView): model = sistemas_por_cliente paginate_by = 5 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 SistemasXCliente_DetailView(UserPassesTestMixin,LoginRequiredMixin, DetailView): model = sistemas_por_cliente template_name= 'Sistemas/Xclientes/detail.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 ''' ========================= API Views ''' class RegisterDeviceView(APIView): permissions_classes = (permissions.AllowAny,) 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): 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(".")] #print('client_version: ',client_version) 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}) print('server_version', server_version) result=False #for i in range(len(client_version)): for cont, ele in enumerate(client_version): if client_version[cont] < server_version[cont]: print('server verion is grater') result = True break else: result= False print('equal') 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})