新增调试OCI预认证请求URL接口

Controller/TestApi.py

@@ -58,6 +58,7 @@ from Object.AWS.AmazonS3Util import AmazonS3Util
 from Object.CeleryBeatObject import CeleryBeatObj
 from Object.ContentSecurityObject import ContentSecurity
 from Object.IPWeatherObject import IPQuery, OpenWeatherMap, GeoIP2
+from Object.OCIObjectStorage import OCIObjectStorage
 from Object.m3u8generate import PlaylistGenerator
 from Object.RedisObject import RedisObject
 from Object.ResponseObject import ResponseObject
@@ -183,9 +184,43 @@ class testView(View):
             return self.celery(request_dict, response)
         elif operation == 'tz':
             return self.tz(request_dict, response)
+        elif operation == 'createPreauthRequest':
+            return self.create_preauth_request(request_dict, response)
+        elif operation == 'getOciUrl':
+            return self.get_oci_url(request_dict, request, response)
         else:
             return response.json(414)
 
+    @classmethod
+    def create_preauth_request(cls, request_dict, response):
+
+        oci = OCIObjectStorage(request_dict.get('region'))
+        bucket_name = request_dict.get('bucketName')
+        name = 'oci-test'
+        object_name = request_dict.get('objectName')
+        time_expires = datetime.datetime.utcnow() + datetime.timedelta(minutes=10)
+        result = oci.create_ereauthenticated_request(bucket_name, name, object_name, time_expires)
+        return response.json(0, {'fullPath': result.full_path + result.object_name})
+
+    @classmethod
+    def get_oci_url(cls, request_dict, request, response):
+        """
+        获取Oci存储地址
+        """
+        try:
+            object_name = request_dict.get('objectName', None)
+            if not object_name:
+                return response.json(404)
+            oci = OCIObjectStorage(request_dict.get('region'))
+            bucket_name = request_dict.get('bucketName')
+            name = 'oci-test'
+            time_expires = datetime.datetime.utcnow() + datetime.timedelta(minutes=10)
+            result = oci.get_preauthenticated_request_url(bucket_name, name, object_name, time_expires)
+            return response.json(0, {'data': result.full_path})
+        except Exception as e:
+            print(repr(e))
+            return response.json(500)
+
     @staticmethod
     def generate_token(request_dict):
         tko = TokenObject()