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 = 20 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')) cli = Clientes.objects.get(RFC=data.get('client')) print('cli.id',cli.id) device = Device.objects.filter( Q(client=cli.id) #& Q(device_name=data.get('device_name')) #& Q(ip_address=data.get('ip_address')) #& Q(sistema= int(sis.id)) #& Q(macAddress__icontains=data.get('macAddress')) ).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}) print('client_version: ',client_version) print('server_version', server_version) 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})