新增测试服务接口,AI测试接口

AdminController/TestServeController.py

@@ -0,0 +1,76 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+
+import json
+import time
+import urllib
+import uuid
+import boto3
+import threading
+import logging
+from boto3.session import Session
+from django.http import JsonResponse, HttpResponseRedirect, HttpResponse
+from django.views.generic.base import View
+from Model.models import Device_Info, Role, MenuModel
+from Object.ResponseObject import ResponseObject
+from Object.TokenObject import TokenObject
+from Object.UidTokenObject import UidTokenObject
+from Service.CommonService import CommonService
+from django.db.models import Q, F
+from time import strftime
+
+
+class TestServeView(View):
+    def get(self, request, *args, **kwargs):
+        request.encoding = 'utf-8'
+        operation = kwargs.get('operation')
+        return self.validation(request.GET, request, operation)
+
+    def post(self, request, *args, **kwargs):
+        request.encoding = 'utf-8'
+        operation = kwargs.get('operation')
+        return self.validation(request.POST, request, operation)
+
+    def validation(self, request_dict, request, operation):
+        language = request_dict.get('language', 'en')
+        response = ResponseObject(language, 'pc')
+        if operation == '??':
+            return 0
+        else:
+            tko = TokenObject(
+                request.META.get('HTTP_AUTHORIZATION'),
+                returntpye='pc')
+            # if tko.code != 0:
+            #     return response.json(tko.code)
+            response.lang = tko.lang
+            userID = tko.userID
+            if operation == 'AItest':
+                return self.AItest(userID, request, request_dict, response)
+            if operation == 'doEdit':
+                return self.doEdit(userID, request_dict, response)
+            if operation == 'doDelete':
+                return self.doDelete(userID, request_dict, response)
+            else:
+                return response.json(404)
+
+    def AItest(self, userID, request, request_dict, response):
+        file = request.FILES.get('file', None)
+        print('request_dict: ', request_dict)
+        maxLabels = int(request_dict.get('maxLabels', 5))
+        minConfidence = float(request_dict.get('minConfidence', 55))
+
+        client = boto3.client(
+            'rekognition',
+            aws_access_key_id='AKIA2E67UIMD6JD6TN3J',
+            aws_secret_access_key='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',
+            region_name='us-east-1')
+        # doc:
+        # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/rekognition.html#Rekognition.Client.detect_labels
+        rekognition_res = client.detect_labels(
+            Image={
+                'Bytes': file.read()},
+            MaxLabels=maxLabels,
+            MinConfidence=minConfidence)
+        print('rekognition_res: ', rekognition_res)
+        return response.json(0, {'rekognition_res': rekognition_res})

Ansjer/urls.py

@@ -19,7 +19,7 @@ from Controller import FeedBack, EquipmentOTA, EquipmentInfo, AdminManage, AppIn
     OrderTaskController, HistoryUIDController, UIDManageUserController, SerialNumberController, CompanyController, \
     RegionController, VPGController, LanguageController, TestController, DeviceConfirmRegion, S3GetStsController, \
     DetectControllerV2, ShadowController, TestDetectController, PcInfo, PctestController, DeviceDebug
-from AdminController import UserManageController, RoleController, MenuController
+from AdminController import UserManageController, RoleController, MenuController, TestServeController
 
 urlpatterns = [
     url(r'^testApi/(?P<operation>.*)$', TestApi.testView.as_view()),
@@ -347,6 +347,9 @@ urlpatterns = [
     #菜单管理
     re_path('menuManagement/(?P<operation>.*)', MenuController.MenuView.as_view()),
 
+    # 测试服务
+    re_path('testServe/(?P<operation>.*)', TestServeController.TestServeView.as_view()),
+
     #后台界面接口 -----------------------------------------------------
     re_path('(?P<path>.*)', LogManager.errorPath),