134 lines
5.0 KiB
Python
134 lines
5.0 KiB
Python
from django.shortcuts import render
|
|
from django.utils import timezone
|
|
from rest_framework.response import Response
|
|
from rest_framework.viewsets import ModelViewSet, GenericViewSet
|
|
from rest_framework.mixins import UpdateModelMixin, RetrieveModelMixin, CreateModelMixin, ListModelMixin
|
|
from apps.hrm.filters import ClockRecordFilterSet
|
|
from apps.hrm.services import HRMService
|
|
from apps.system.mixins import CreateUpdateModelAMixin, OptimizationMixin
|
|
from apps.hrm.models import ClockRecord, Employee
|
|
from apps.hrm.serializers import ClockRecordListSerializer, EmployeeSerializer, FaceClockCreateSerializer, FaceLoginSerializer
|
|
import face_recognition
|
|
from django.conf import settings
|
|
from django.core.cache import cache
|
|
import logging
|
|
from rest_framework.generics import CreateAPIView
|
|
from rest_framework import status
|
|
from rest_framework_simplejwt.tokens import RefreshToken
|
|
from rest_framework import exceptions
|
|
from apps.system.models import User
|
|
from apps.system.serializers import UserSimpleSerializer
|
|
from rest_framework.permissions import AllowAny
|
|
logger = logging.getLogger('log')
|
|
|
|
|
|
def load_face_data(username:int, path:str):
|
|
"""
|
|
将某用户face_encoding加载进缓存
|
|
"""
|
|
face_datas = cache.get_or_set('face_datas', {}, timeout=None)
|
|
photo_path = settings.BASE_DIR + path
|
|
picture_of_me = face_recognition.load_image_file(photo_path)
|
|
my_face_encoding = face_recognition.face_encodings(picture_of_me)[0]
|
|
face_datas[username] = my_face_encoding
|
|
cache.set('face_datas', face_datas, timeout=None)
|
|
return my_face_encoding
|
|
|
|
# Create your views here.
|
|
class EmployeeViewSet(CreateUpdateModelAMixin, OptimizationMixin, UpdateModelMixin, RetrieveModelMixin, GenericViewSet):
|
|
"""
|
|
员工详细信息
|
|
"""
|
|
perms_map = {'get': '*', 'put': 'employee_update'}
|
|
queryset = Employee.objects.all()
|
|
serializer_class = EmployeeSerializer
|
|
ordering = ['-pk']
|
|
|
|
def perform_update(self, serializer):
|
|
instance = serializer.save(update_by = self.request.user)
|
|
try:
|
|
photo_path = settings.BASE_DIR + instance.photo
|
|
picture_of_me = face_recognition.load_image_file(photo_path)
|
|
my_face_encoding = face_recognition.face_encodings(picture_of_me)[0]
|
|
instance.face_data = my_face_encoding.tolist()
|
|
instance.save()
|
|
except:
|
|
logger.error('人脸识别出错')
|
|
|
|
|
|
class ClockRecordViewSet(CreateModelMixin, ListModelMixin, GenericViewSet):
|
|
"""
|
|
打卡记录
|
|
"""
|
|
perms_map = {'get':'*', 'post':'*'}
|
|
authentication_classes = []
|
|
permission_classes = [AllowAny]
|
|
queryset = ClockRecord.objects.select_related('create_by').all()
|
|
serializer_class = ClockRecordListSerializer
|
|
filterset_class = ClockRecordFilterSet
|
|
ordering = ['-pk']
|
|
|
|
def get_serializer_class(self):
|
|
if self.action == 'create':
|
|
return FaceClockCreateSerializer
|
|
return super().get_serializer_class()
|
|
|
|
def create(self, request, *args, **kwargs):
|
|
now = timezone.now()
|
|
now_local = timezone.localtime()
|
|
if 8<=now_local.hour<=17:
|
|
base64_data = base64.urlsafe_b64decode(tran64(
|
|
request.data.get('base64').replace(' ', '+')))
|
|
user, msg = HRMService.face_compare_from_base64(base64_data)
|
|
if user:
|
|
ins, created = ClockRecord.objects.get_or_create(
|
|
create_by=request.user, create_time__hour__range = [8,18],
|
|
create_time__year=now.year, create_time__month=now.month,
|
|
create_time__day=now.day,
|
|
defaults={
|
|
'type':ClockRecord.ClOCK_WORK1,
|
|
'create_by':user,
|
|
'create_time':now
|
|
})
|
|
if not created:
|
|
ins.create_time = now
|
|
ins.save()
|
|
# 设为在岗
|
|
user.is_atwork = True
|
|
user.save()
|
|
return Response(UserSimpleSerializer(instance=user).data)
|
|
return Response(msg, status=status.HTTP_400_BAD_REQUEST)
|
|
return Response('非打卡时间范围', status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
|
|
|
|
|
|
|
|
import base64
|
|
|
|
def tran64(s):
|
|
missing_padding = len(s) % 4
|
|
if missing_padding != 0:
|
|
s = s+'='* (4 - missing_padding)
|
|
return s
|
|
|
|
class FaceLogin(CreateAPIView):
|
|
authentication_classes = []
|
|
permission_classes = []
|
|
serializer_class = FaceLoginSerializer
|
|
|
|
|
|
def create(self, request, *args, **kwargs):
|
|
"""
|
|
人脸识别登录
|
|
"""
|
|
base64_data = base64.urlsafe_b64decode(tran64(request.data.get('base64').replace(' ', '+')))
|
|
user, msg = HRMService.face_compare_from_base64(base64_data)
|
|
if user:
|
|
refresh = RefreshToken.for_user(user)
|
|
return Response({
|
|
'refresh': str(refresh),
|
|
'access': str(refresh.access_token),
|
|
'username':user.username
|
|
})
|
|
return Response(msg, status=status.HTTP_400_BAD_REQUEST) |