新增put测试接口

Controller/TestApi.py

@@ -11,6 +11,7 @@
 @file: Test.py
 @Contact: chanjunkai@163.com
 """
+import botocore
 from django.views.generic.base import View
 import os
 
@@ -41,7 +42,7 @@ from django.views.generic.base import View
 from Object.RedisObject import RedisObject
 from django.contrib.auth.hashers import make_password, check_password  # 对密码加密模块
 from Ansjer.config import OSS_STS_ACCESS_KEY, OSS_STS_ACCESS_SECRET, OSS_ROLE_ARN, SERVER_DOMAIN, PAYPAL_CRD, \
-    SERVER_DOMAIN_SSL
+    SERVER_DOMAIN_SSL, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY
 from Model.models import Device_Info, Order_Model, Store_Meal, VodHlsModel, OssCrdModel, UID_Bucket, CompanySerialModel, SerialNumberModel, CompanyModel, VPGModel
 from Object.ResponseObject import ResponseObject
 from Object.TokenObject import TokenObject
@@ -75,6 +76,12 @@ class testView(View):
         operation = kwargs.get('operation')
         return self.validation(request.POST, request, operation)
 
+    def put(self, request, *args, **kwargs):
+        request.encoding = 'utf-8'
+        operation = kwargs.get('operation')
+        response = ResponseObject()
+        return response.json(0,request.body)
+
     def validation(self, request_dict, request, operation):
         response = ResponseObject()
         # operation => cloudVod/path
@@ -105,7 +112,7 @@ class testView(View):
             return self.generate_token(request_dict,userID)
         elif operation == 'test_upload_s3':
             userID = '158943604783713800138000'
-            return self.test_upload_s3(request_dict)
+            return self.test_upload_s3(request_dict, response)
         elif operation == 'rekognition':
             userID = '158943604783713800138000'
             return self.testRekognition(request,request_dict)
@@ -698,42 +705,24 @@ class testView(View):
         return JsonResponse(status=200, data=res,safe=False)
 
 
-    def test_upload_s3(self,request_dict):
-        times = '2021-05-24 11:37:04+00:00'
-        test = CommonService.str_to_timestamp(times)
-        exit(test)
-        # redisObject = RedisObject()
-        # res = redisObject.rpush('serial_lock', 'serial_lock')
-        #
-        # exit(res)
-        #
-        # res = CompanyModel.objects.filter(id=3).delete()
-        # exit(res)
-        #
-        # testd = hmac.new('AWS4rf/xnQ3jIgY8bj7Sz4An4KbYct2sq2MbrfmP8rVV'.encode("utf-8"), '20210223'.encode("utf-8"), hashlib.sha256).digest()
-        # region_sign = hmac.new(testd, 'cn-northwest-1'.encode("utf-8"), hashlib.sha256).hexdigest()
-        # exit(region_sign)
-        # kService = sign(kRegion, 's3')
-        # kSigning = sign(kService, 'aws4_request')
-
-        # exit(CommonService.getIpIpInfo("120.197.196.156",'CN'))
-        exit(1123123)
-        aws_key = "ASIA2MMWBR4DXDM36CEN" #【你的 aws_access_key】
-        aws_secret = "qLmvltAthKkD5tGuFP/LE5YEmiD/iVBwuC3PneHm" #【你的 aws_access_key】
-        aws_session_token = "IQoJb3JpZ2luX2VjEFIaDmNuLW5vcnRod2VzdC0xIkgwRgIhAL60RpDvgWluwYehRrb3NgFDgMZYEOzqnwAgz3NH/2jiAiEA2G026WLnK4h38wctp6Wt7jnfJSUdyr7GJ7NKUCU4kKkq8QIIfxAAGgw3MTM4MTYxMTY5OTkiDNKIYzjvazWRomqCeSrOAgGuglgcUw3/QGPBLb1B5WpYU0RXw+UTfS4UIpmXHvJ8wUR6y50H0QxAWv5ADGXDam7gVV+jxcQpB6X9RZgHiDmnpzKO6PrRtevSoh89F519ffumHdoNpRNGA04vh4XcPIfLsaETgDAUFkWY2Ha3OxaDgF8+6R+0OsGNTSTmQO2yr2u3LfsoSVn6+vTLHOq8O5+xVMzFJzVzffJJenB5n2Tm2DiLoCPLZbKrm5i4bHPCb4+e6T9H8G/OqI15k8Tdu4xr6hkObTb2DmTwFcTRYXLy14iU6S+8j59UcBg1LUVz7Ygenoz6kvj0t4SGN8ogShFdZtxFp9yl5hVHzNg8/JiKbm1AI8h1TD4F3zN+va1qrUa3wBUKK8uSOzG7CG4yX42NK6q5bx9B6sI53vWLeBBREgYB4gD7yOOFKjz8MPukVgoWDLKMZJzTqnnAgjAwqpOthQY6mAGHp2YELSBx8McgvRcHmv/Ebp2vhWYZRXOuNrX574ZVAmHsIbzQFkZvEoXOu3uIHY17cqskeJna2lXaSwZDnm0wIDCVHorcN2n1GjEG/JA6CNv2jYBtiG7DzllLzUe0B1OPFQsO7W9ctkSSlA51njWVGHkKiRfml51XXvMzxOV7APllBvglvFgGr2q8FfayNKYtf7VoaBMDYg=="
-        session = Session(aws_access_key_id=aws_key,
-                          aws_secret_access_key=aws_secret,
-                          aws_session_token=aws_session_token,
-                          region_name="cn-northwest-1") # 此处根据自己的 s3 地区位置改变
-        s3 = session.resource("s3")
-        # client = session.client("s3")
-        bucket = "cn-formal-cs-experience-7day-3month" # 【你 bucket 的名字】 # 首先需要保.证 s3 上已经存在该存储桶，否则报错
-        upload_data = open("./././PushModel/tests.py", "rb")
-        # upload_key = "XKWZSC5FCJYT19B7111A/test/test6/test.log"
-        upload_key = "THY7M84KLDNZ9XE8111A/vod1/info.log"
-        file_obj = s3.Bucket(bucket).put_object(Key=upload_key, Body=upload_data)
-        print('---------------------')
-        exit(file_obj)
+    def test_upload_s3(self,request_dict , response):
+        aws_s3_guonei = boto3.client(
+            's3',
+            aws_access_key_id=AWS_ACCESS_KEY_ID[0],
+            aws_secret_access_key=AWS_SECRET_ACCESS_KEY[0],
+            config=botocore.client.Config(signature_version='s3v4'),
+            region_name='cn-northwest-1'
+        )
+        download_link = 'ipctest'
+        response_url = aws_s3_guonei.generate_presigned_url(
+            ClientMethod='put_object',
+            Params={
+                'Bucket': 'pc-package',
+                'Key': download_link
+            },
+            ExpiresIn=3600
+        )
+        return response.json(0, {'datas': response_url, 'count': 1})
 
 
     def testRekognition(self,request,request_dict):