import json import datetime import cv2 import numpy as np import base64 import os from django.http import JsonResponse from django.views.decorators.csrf import csrf_exempt from django.conf import settings from django.core.files.base import ContentFile from proctoring.models.image import Image from proctoring.models.imageTypes import ImageTypes from proctoring.services.face_compare_v2 import compare_id_images, FaceVerificationError from proctoring.services.identity_messages import ImageCheckCode, ImageCheckMessage from urllib.parse import urlparse def _storage_path_from_url(full_url: str) -> str: """ Convert a full URL like https://cdn.example.com/proctoring_images/123/identity-check/face/xxx.jpg into a storage path: proctoring_images/123/identity-check/face/xxx.jpg """ parsed = urlparse(full_url) return parsed.path.lstrip("/") @csrf_exempt def check_image(request): if request.method != "POST": return JsonResponse( { "code": ImageCheckCode.INVALID_REQUEST_METHOD.value, "message": ImageCheckMessage.INVALID_REQUEST_METHOD.value, }, status=403, ) try: data = json.loads(request.body) except json.JSONDecodeError: return JsonResponse( { "code": ImageCheckCode.INVALID_JSON.value, "message": ImageCheckMessage.INVALID_JSON.value, }, status=400, ) base64image = (data.get("base64_image") or "").strip() shareId = (data.get("shareId") or "").strip() isIdentityCheck = bool(data.get("isIdentityCheck", False)) compareWithIdentity = bool(data.get("compareWithIdentity", False)) if not base64image or not shareId: return JsonResponse( { "code": ImageCheckCode.MISSING_PARAMETERS.value, "message": ImageCheckMessage.MISSING_PARAMETERS.value, }, status=400, ) ai_model_dir = os.path.join(settings.BASE_DIR, "ml", "models","object_detection") net = settings.net if cv2.cuda.getCudaEnabledDeviceCount() > 0: net.setPreferableBackend(cv2.dnn.DNN_BACKEND_CUDA) net.setPreferableTarget(cv2.dnn.DNN_TARGET_CUDA) try: with open(os.path.join(ai_model_dir, "classes.txt"), "r") as f: classes = f.read().splitlines() except Exception: return JsonResponse( { "pass": False, "code": ImageCheckCode.FAILED_LOAD_CLASSES.value, "message": ImageCheckMessage.FAILED_LOAD_CLASSES.value, }, status=200, ) try: img_bytes = base64.b64decode(base64image) img_np = np.frombuffer(img_bytes, dtype=np.uint8) img = cv2.imdecode(img_np, cv2.IMREAD_COLOR) if img is None: raise ValueError("cv2.imdecode returned None") except Exception: return JsonResponse( { "pass": False, "code": ImageCheckCode.INVALID_IMAGE_DATA.value, "message": ImageCheckMessage.INVALID_IMAGE_DATA.value, }, status=200, ) height, width, _ = img.shape blob = cv2.dnn.blobFromImage( img, 1 / 255.0, (416, 416), (0, 0, 0), swapRB=True, crop=False ) net.setInput(blob) output_layers_names = net.getUnconnectedOutLayersNames() layerOutputs = net.forward(output_layers_names) boxes = [] confidences = [] class_ids = [] for output in layerOutputs: for detection in output: scores = detection[5:] class_id = int(np.argmax(scores)) confidence = float(scores[class_id]) if confidence > 0.2: center_x = int(detection[0] * width) center_y = int(detection[1] * height) w = int(detection[2] * width) h = int(detection[3] * height) x = int(center_x - w / 2) y = int(center_y - h / 2) boxes.append([x, y, w, h]) confidences.append(confidence) class_ids.append(class_id) indexes = cv2.dnn.NMSBoxes(boxes, confidences, 0.2, 0.4) myClasses = ["book", "laptop", "cell phone", "person"] personCount = 0 prohibitedObjectsCount = 0 if len(indexes) > 0: for i in indexes.flatten(): label = str(classes[class_ids[i]]) if label not in myClasses: continue if label == "person": personCount += 1 else: prohibitedObjectsCount += 1 passed = (personCount == 1 and prohibitedObjectsCount == 0) distance = None if compareWithIdentity: try: ref_image_obj = ( Image.objects .filter(shareId=shareId, type=ImageTypes.FaceCheck) .order_by("id") .first() ) if ref_image_obj is not None: storage_path = _storage_path_from_url(ref_image_obj.path) with settings.s3_storage.open(storage_path, mode="rb") as f: ref_bytes = f.read() ref_base64 = base64.b64encode(ref_bytes).decode("utf-8") result = compare_id_images( id_card_image_base64=ref_base64, id_holder_image_base64=base64image, ) if result.get("status") == "ok": distance = float(result.get("distance", 0.0)) else: distance = float(result.get("distance", 0.0)) else: distance = None except FaceVerificationError: distance = None except Exception: distance = None try: imgWidth = 240 imgHeight = 135 img_resized = cv2.resize(img, (imgWidth, imgHeight), interpolation=cv2.INTER_AREA) outputImage = cv2.imencode(".jpg", img_resized)[1].tobytes() save_folder = os.path.join("proctoring_images", shareId) filename = "{date:%Y-%m-%d_%H.%M.%S}.jpg".format(date=datetime.datetime.now()) filepath = "" imageType = None if not isIdentityCheck: if passed: passed_folder = os.path.join(save_folder, "passed") filepath = os.path.join(passed_folder, filename) imageType = ImageTypes.Acceptable else: failed_folder = os.path.join(save_folder, "failed") filepath = os.path.join(failed_folder, filename) imageType = ImageTypes.AttentionRequired content_file = ContentFile(outputImage, name=filename) filepath = settings.s3_storage.save(filepath, content_file) image = Image( path=f"{settings.AWS_S3_ACCESS_DOMAIN}/{filepath}", personCount=personCount, prohibitedObjectsCount=prohibitedObjectsCount, type=imageType, shareId=shareId, ) image.save() else: if passed: identity_folder = os.path.join(save_folder, "identity-check", "face") filepath = os.path.join(identity_folder, filename) content_file = ContentFile(outputImage, name=filename) filepath = settings.s3_storage.save(filepath, content_file) imageType = ImageTypes.FaceCheck image = Image( path=f"{settings.AWS_S3_ACCESS_DOMAIN}/{filepath}", personCount=personCount, prohibitedObjectsCount=prohibitedObjectsCount, type=imageType, shareId=shareId, ) image.save() except Exception: return JsonResponse( { "pass": False, "code": ImageCheckCode.FAILED_SAVE_IMAGE.value, "message": ImageCheckMessage.FAILED_SAVE_IMAGE.value, "personCount": personCount, "prohibitedObjectsCount": prohibitedObjectsCount, "isIdentityCheck": isIdentityCheck, "compareWithIdentity": compareWithIdentity, "distance": distance, }, status=200, ) response = { "pass": passed, "code": ( ImageCheckCode.SUCCESS.value if passed else ImageCheckCode.ATTENTION_REQUIRED.value ), "message": ( ImageCheckMessage.SUCCESS.value if passed else ImageCheckMessage.ATTENTION_REQUIRED.value ), "personCount": personCount, "prohibitedObjectsCount": prohibitedObjectsCount, "isIdentityCheck": isIdentityCheck, "compareWithIdentity": compareWithIdentity, "distance": distance, } return JsonResponse(response, status=200)