合并testapi

Controller/TestApi.py

@@ -17,8 +17,10 @@ import traceback
 import botocore
 import cv2
 from botocore import client
+from django.db import transaction
 
 from Ansjer.cn_config.config_formal import CONFIG_INFO
+from Ansjer.config import CONFIG_INFO
 from Controller.DeviceConfirmRegion import Device_Region
 from Object.AWS.AmazonS3Util import AmazonS3Util
 from Object.RedisObject import RedisObject
@@ -42,10 +44,10 @@ from django.utils.decorators import method_decorator
 from django.views.decorators.csrf import csrf_exempt
 from django.views.generic.base import View
 from django.contrib.auth.hashers import make_password  # 对密码加密模块
-from Ansjer.config import OSS_STS_ACCESS_KEY, OSS_STS_ACCESS_SECRET, OSS_ROLE_ARN, AWS_ACCESS_KEY_ID, \
+from Ansjer.config import OSS_STS_ACCESS_KEY, OSS_STS_ACCESS_SECRET, AWS_ACCESS_KEY_ID, \
     AWS_SECRET_ACCESS_KEY, SERVER_TYPE, AWS_SES_ACCESS_REGION
 from Model.models import Order_Model, Store_Meal, DeviceLogModel, VodBucketModel, \
-    UIDCompanySerialModel, CompanySerialModel, LogModel
+    TestSerialRepetition, UIDCompanySerialModel, CompanySerialModel, LogModel
 from Object.ResponseObject import ResponseObject
 from Object.TokenObject import TokenObject
 from Service.CommonService import CommonService
@@ -158,6 +160,12 @@ class testView(View):
             return self.play_m3u8(request_dict, response)
         elif operation == 'generate_video':
             return self.generate_video(request_dict, response)
+        elif operation == 'serial-repetition':  # 用与测试序列号重复接口
+            response = ResponseObject('cn')
+            return response.json(475)
+        elif operation == 'v2/serial-repetition':  # 用与测试序列号重复接口
+            response = ResponseObject('cn')
+            return self.serial_repetition_test_v2(request_dict, response)
         elif operation == 'getSerialNumberInfo':  # 序列号信息查询
             return self.getSerialNumberInfo(request_dict, response)
         elif operation == 'get-serial-details':  # 序列号信息查询
@@ -165,6 +173,60 @@ class testView(View):
         else:
             return 123
 
+    @classmethod
+    def serial_repetition_test_v2(cls, request_dict, response):
+        try:
+            serial_no = request_dict.get('serialNo', None)
+            phone_model = request_dict.get('phoneModel', None)
+            if not serial_no:
+                return response.json(444)
+            with transaction.atomic():
+                first_serial = serial_no[:6]
+                first_serial_qs = TestSerialRepetition.objects.filter(serial_number__icontains=first_serial)
+                if first_serial_qs.exists():
+                    result = {'serialNumber': first_serial_qs.first().serial_number,
+                              'phoneModel': first_serial_qs.first().phone_model,
+                              'createdTime': first_serial_qs.first().created_time}
+                    return response.json(174, result)
+
+                serial_qs = TestSerialRepetition.objects.filter(serial_number=serial_no)
+                if not serial_qs.exists():
+                    n_time = int(time.time())
+                    params = {'serial_number': serial_no, 'created_time': n_time}
+                    if phone_model:
+                        params['phone_model'] = phone_model
+                    TestSerialRepetition.objects.create(**params)
+                    return response.json(0)
+                else:
+                    return response.json(174)
+        except Exception as e:
+            logging.info('异常错误,errLine:{}, errMsg:{}'.format(e.__traceback__.tb_lineno, repr(e)))
+            return response.json(178, e)
+
+    @classmethod
+    def serial_repetition_test(cls, request_dict, response):
+        return response.json()
+        # try:
+        #     serial_no = request_dict.get('serialNo', None)
+        #     if not serial_no:
+        #         return response.json(444)
+        #     with transaction.atomic():
+        #         first_serial = serial_no[:6]
+        #         first_serial_qs = TestSerialRepetition.objects.filter(serial_number__icontains=first_serial)
+        #         if first_serial_qs.exists():
+        #             return response.json(174)
+        #         serial_qs = TestSerialRepetition.objects.filter(serial_number=serial_no)
+        #         if not serial_qs.exists():
+        #             n_time = int(time.time())
+        #             params = {'serial_number': serial_no, 'created_time': n_time}
+        #             TestSerialRepetition.objects.create(**params)
+        #             return response.json(0)
+        #         else:
+        #             return response.json(174)
+        # except Exception as e:
+        #     logging.info('异常错误,errLine:{}, errMsg:{}'.format(e.__traceback__.tb_lineno, repr(e)))
+        #     return response.json(178, e)
+
     @classmethod
     def generate_video(cls, request_dict, response):
         # 设计抽取图片规则通过消息随机还是时间随机，调试copy S3对象查询是否携带失效时间
@@ -844,18 +906,18 @@ class testView(View):
         serial_number = serial_number[:6]
         try:
             uid_user_message = {
-                    "uid": "",
-                    "serialNumber": "",
-                    "userID": "",
-                    "username": "",
-                    "primaryUserID": ""
+                "uid": "",
+                "serialNumber": "",
+                "userID": "",
+                "username": "",
+                "primaryUserID": ""
             }
             data = {
                 "uid": "",
                 "serialNumber": serialNumber,
                 "status": "",
                 "uid_user_message": uid_user_message
-                }
+            }
             company_serial_qs = CompanySerialModel.objects.filter(serial_number=serial_number).values('status')
             if not company_serial_qs.exists():
                 return response.json(173)
@@ -868,14 +930,14 @@ class testView(View):
                 if company_serial_qs[0]['status'] == 1:
                     data['status'] = '已分配'
                 if company_serial_qs[0]['status'] == 2:
-                    data['status'] = '已绑定uid'
+                    data['status'] = '绑定uid'
                 if company_serial_qs[0]['status'] == 3:
                     data['status'] = '已占用'
                 return response.json(0, data)
             for uid_company_serial in uid_company_serial_qs:
                 data['uid'] = uid_company_serial['uid__uid']
                 data['serialNumber'] = serialNumber
-                data['status'] =  uid_company_serial['uid__status']
+                data['status'] = uid_company_serial['uid__status']
                 if company_serial_qs[0]['status'] == 1:
                     data['status'] = '已分配'
                 if company_serial_qs[0]['status'] == 2:
@@ -917,11 +979,15 @@ class testView(View):
             local_response = cls.getSerialNumberInfo(request_dict, response)
             res = json.loads(local_response.content)
             res['result']['server'] = 1
+            res['result']['serverName'] = '中国服'
+            res['result']['domainName'] = 'https://www.zositechc.cn'
             results_data.append(res['result'])
             res1 = requests.post("http://www.dvema.com/testApi/getSerialNumberInfo",
                                  data={'serialNumber': serial_number}, timeout=3)
             results1 = json.loads(res1.text)
             results1['result']['server'] = 2
+            results1['result']['serverName'] = '美国服'
+            results1['result']['domainName'] = 'https://www.dvema.com'
             results_data.append(results1['result'])
             # res2 = requests.post("https://api.zositeche.com/testApi/getSerialNumberInfo",
             #                      data={'serialNumber': serial_number}, timeout=5)