如何对OSS Bucket中文件进行在线压缩和解压缩_智能媒体管理(IMM)-阿里云帮助中心

本文介绍如何操作用户Bucket中的压缩包和待压缩文件，包括查看压缩包内容、将压缩文件进行解压缩处理以及将用户Bucket内的多个文件进行压缩。适用于各种场景，如数据备份、存档管理、文件传输等。通过使用这些方法，用户可以更高效地管理和处理存储在Bucket中的数据，提高数据传输速度、减少存储空间占用，以及方便地进行文件归档和共享。

功能简介

智能媒体管理服务提供了云上资源的压缩和解压缩功能，该功能帮助用户高效、便捷地进行云上文件管理，用户可以根据需要使用该功能对OSS中的文件进行在线压缩和解压缩。

前提条件

已创建并获取AccessKey。具体操作，请参见创建AccessKey。
已开通OSS服务、创建存储空间并上传文件到存储空间。具体操作，请参见控制台上传文件。
已开通智能媒体管理服务。具体操作，请参见开通产品。
已通过智能媒体管理控制台创建项目。具体操作，请参见创建项目。
说明
- 您也可以调用API接口创建项目。具体操作，请参见CreateProject - 创建项目。
- 您可以调用ListProjects - 列出所有项目信息的列表接口列出指定地域下已创建的所有项目信息。

注意事项

压缩、查看压缩包内容、解压缩各任务开始执行后，任务信息只保存7天，超过7天则无法再获取。您可以通过以下四种方式及时获取任务信息：

调用GetTask - 获取任务信息或ListTasks - 列出任务接口获取返回的TaskId，查看任务信息。
在与智能媒体管理相同的地域下开通MNS服务，并配置订阅关系，及时获取任务信息通知，异步通知消息格式请参见异步通知消息格式。关于MNS SDK的更多信息，请参见步骤四：接收和删除消息。
在与智能媒体管理相同的地域下开通RocketMQ服务，并创建RocketMQ 4.0实例、Topic、Group，及时获取任务信息通知，异步通知消息格式请参见异步通知消息格式。关于RocketMQ使用的更多信息，请参见调用HTTP协议的SDK收发普通消息。
在与智能媒体管理相同的地域下开通并接入事件总线EventBridge服务，及时获取任务信息通知。更多信息，请参见智能媒体管理IMM事件。

压缩

调用CreateFileCompressionTask - 创建文件压缩任务接口将OSS Bucket中的多个文件进行压缩处理。

压缩信息

项目名称：test-project
待压缩文件：
- oss://test-bucket/test-object1.jpg
- oss://test-bucket/test-object2.jpg
- oss://test-bucket/test-object3.jpg
- oss://test-bucket/test-object4.jpg
输出压缩文件的OSS地址：oss://test-bucket/test-target-object.zip

请求示例

{
    "ProjectName": "test-project",
    "CompressedFormat": "zip",
    "Sources": "[{\"URI\":\"oss://test-bucket/test-object1.jpg\"},{\"URI\":\"oss://test-bucket/test-object2.jpg\"},{\"URI\":\"oss://test-bucket/test-object3.jpg\"},{\"URI\":\"oss://test-bucket/test-object4.jpg\"}]",
    "TargetURI": "oss://test-bucket/test-target-object.zip"
}

返回示例

{
  "TaskId": "FileCompression-7e3e2b8c-81bd-495f-a338-25b3762a****",
  "RequestId": "C54C5888-9C96-5A35-AF73-ABD6EF15****",
  "EventId": "164-1S6uXTYhUU4qAFCPOTcYdcb****"
}

示例代码

压缩文件的完整示例代码如下。

# -*- coding: utf-8 -*-
# This file is auto-generated, don't edit it. Thanks.
import os
import sys

from typing import List

from alibabacloud_imm20200930.client import Client as imm20200930Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_imm20200930 import models as imm_20200930_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient


class Sample:
    def __init__(self):
        pass

    @staticmethod
    def create_client(
        access_key_id: str,
        access_key_secret: str,
    ) -> imm20200930Client:
        """
        使用AccessKey ID&AccessKey Secret初始化账号Client。
        @param access_key_id:
        @param access_key_secret:
        @return: Client
        @throws Exception
        """
        config = open_api_models.Config(
            access_key_id=access_key_id,
            access_key_secret=access_key_secret
        )
        # 填写访问的IMM域名。
        config.endpoint = f'imm.cn-beijing.aliyuncs.com'
        return imm20200930Client(config)

    @staticmethod
    def main(
        args: List[str],
    ) -> None:
        # 阿里云账号AccessKey拥有所有API的访问权限，建议您使用RAM用户进行API访问或日常运维。
        # 强烈建议不要把AccessKey ID和AccessKey Secret保存到工程代码里，否则可能导致AccessKey泄露，威胁您账号下所有资源的安全。
        # 本示例通过从环境变量中读取AccessKey，来实现API访问的身份验证。如何配置环境变量，请参见https://help.aliyun.com/document_detail/2361894.html。
        imm_access_key_id = os.getenv("AccessKeyId")
        imm_access_key_secret = os.getenv("AccessKeySecret")
        client = Sample.create_client(imm_access_key_id, imm_access_key_secret)
        sources_0 = imm_20200930_models.CreateFileCompressionTaskRequestSources(
            uri='oss://test-bucket/test-object1.jpg'
        )
        sources_1 = imm_20200930_models.CreateFileCompressionTaskRequestSources(
            uri='oss://test-bucket/test-object2.jpg'
        )
        sources_2 = imm_20200930_models.CreateFileCompressionTaskRequestSources(
            uri='oss://test-bucket/test-object3.jpg'
        )
        sources_3 = imm_20200930_models.CreateFileCompressionTaskRequestSources(
            uri='oss://test-bucket/test-object4.jpg'
        )
        create_file_compression_task_request = imm_20200930_models.CreateFileCompressionTaskRequest(
            project_name='test-project',
            sources=[
                sources_0,
                sources_1,
                sources_2,
                sources_3
            ],
            target_uri='oss://test-bucket/test-target-object.zip',
            compressed_format='zip',
        )
        runtime = util_models.RuntimeOptions()
        try:
            # 复制代码运行请自行打印API的返回值。
            client.create_file_compression_task_with_options(create_file_compression_task_request, runtime)
        except Exception as error:
            # 如有需要，请打印错误信息。
            UtilClient.assert_as_string(error.message)

    @staticmethod
    async def main_async(
        args: List[str],
    ) -> None:
        # 阿里云账号AccessKey拥有所有API的访问权限，建议您使用RAM用户进行API访问或日常运维。
        # 强烈建议不要把AccessKey ID和AccessKey Secret保存到工程代码里，否则可能导致AccessKey泄露，威胁您账号下所有资源的安全。
        # 本示例通过从环境变量中读取AccessKey，来实现API访问的身份验证。如何配置环境变量，请参见https://help.aliyun.com/document_detail/2361894.html。
        imm_access_key_id = os.getenv("AccessKeyId")
        imm_access_key_secret = os.getenv("AccessKeySecret")
        client = Sample.create_client(imm_access_key_id, imm_access_key_secret)
        sources_0 = imm_20200930_models.CreateFileCompressionTaskRequestSources(
            uri='oss://test-bucket/test-object1.jpg'
        )
        sources_1 = imm_20200930_models.CreateFileCompressionTaskRequestSources(
            uri='oss://test-bucket/test-object2.jpg'
        )
        sources_2 = imm_20200930_models.CreateFileCompressionTaskRequestSources(
            uri='oss://test-bucket/test-object3.jpg'
        )
        sources_3 = imm_20200930_models.CreateFileCompressionTaskRequestSources(
            uri='oss://test-bucket/test-object4.jpg'
        )
        create_file_compression_task_request = imm_20200930_models.CreateFileCompressionTaskRequest(
            project_name='test-project',
            sources=[
                sources_0,
                sources_1,
                sources_2,
                sources_3
            ],
            target_uri='oss://test-bucket/test-target-object.zip',
            compressed_format='zip',
        )
        runtime = util_models.RuntimeOptions()
        try:
            # 复制代码运行请自行打印API的返回值。
            await client.create_file_compression_task_with_options_async(create_file_compression_task_request, runtime)
        except Exception as error:
            # 如有需要，请打印错误信息。
            UtilClient.assert_as_string(error.message)


if __name__ == '__main__':
    Sample.main(sys.argv[1:])

查看压缩包内容

调用CreateArchiveFileInspectionTask - 创建压缩包解析任务接口查看OSS压缩包内容。

压缩包信息

项目名称：test-project
压缩包路径：oss://test-bucket/test-target-object.zip
任务执行结果保存地址：oss://test-bucket/test-target-object.json

请求示例

{
    "ProjectName": "test-project",
    "SourceURI": "oss://test-bucket/test-target-object.zip",
    "TargetURI": "oss://test-bucket/test-target-object.json"
}

返回示例

{
    "TaskId": "ArchiveFileInspection-4a7c308a-5671-4d34-ba41-86a942d0****",
    "RequestId": "20D23855-C9E6-5193-8939-3A8797E1****",
    "EventId": "2AB-1S53L6FTdU6yMxkc1JgpwcV****"
}

示例代码

查看压缩包内容的完整示例代码如下。

# -*- coding: utf-8 -*-
# This file is auto-generated, don't edit it. Thanks.
import os
import sys

from typing import List

from alibabacloud_imm20200930.client import Client as imm20200930Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_imm20200930 import models as imm_20200930_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient


class Sample:
    def __init__(self):
        pass

    @staticmethod
    def create_client(
        access_key_id: str,
        access_key_secret: str,
    ) -> imm20200930Client:
        """
        使用AccessKey ID&AccessKey Secret初始化账号Client。
        @param access_key_id:
        @param access_key_secret:
        @return: Client
        @throws Exception
        """
        config = open_api_models.Config(
            access_key_id=access_key_id,
            access_key_secret=access_key_secret
        )
        # 填写访问的IMM域名。
        config.endpoint = f'imm.cn-beijing.aliyuncs.com'
        return imm20200930Client(config)

    @staticmethod
    def main(
        args: List[str],
    ) -> None:
        # 阿里云账号AccessKey拥有所有API的访问权限，建议您使用RAM用户进行API访问或日常运维。
        # 强烈建议不要把AccessKey ID和AccessKey Secret保存到工程代码里，否则可能导致AccessKey泄露，威胁您账号下所有资源的安全。
        # 本示例通过从环境变量中读取AccessKey，来实现API访问的身份验证。如何配置环境变量，请参见https://help.aliyun.com/document_detail/2361894.html。
        imm_access_key_id = os.getenv("AccessKeyId")
        imm_access_key_secret = os.getenv("AccessKeySecret")
        client = Sample.create_client(imm_access_key_id, imm_access_key_secret)
        create_archive_file_inspection_task_request = imm_20200930_models.CreateArchiveFileInspectionTaskRequest(
            project_name='test-project',
            source_uri='oss://test-bucket/test-target-object.zip',
            target_uri='oss://test-bucket/test-target-object.json'
        )
        runtime = util_models.RuntimeOptions()
        try:
            # 复制代码运行请自行打印API的返回值。
            client.create_archive_file_inspection_task_with_options(create_archive_file_inspection_task_request, runtime)
        except Exception as error:
            # 如有需要，请打印错误信息。
            UtilClient.assert_as_string(error.message)

    @staticmethod
    async def main_async(
        args: List[str],
    ) -> None:
        # 阿里云账号AccessKey拥有所有API的访问权限，建议您使用RAM用户进行API访问或日常运维。
        # 强烈建议不要把AccessKey ID和AccessKey Secret保存到工程代码里，否则可能导致AccessKey泄露，威胁您账号下所有资源的安全。
        # 本示例通过从环境变量中读取AccessKey，来实现API访问的身份验证。如何配置环境变量，请参见https://help.aliyun.com/document_detail/2361894.html。
        imm_access_key_id = os.getenv("AccessKeyId")
        imm_access_key_secret = os.getenv("AccessKeySecret")
        client = Sample.create_client(imm_access_key_id, imm_access_key_secret)
        create_archive_file_inspection_task_request = imm_20200930_models.CreateArchiveFileInspectionTaskRequest(
            project_name='test-project',
            source_uri='oss://test-bucket/test-target-object.zip',
            target_uri='oss://test-bucket/test-target-object.json',
        )
        runtime = util_models.RuntimeOptions()
        try:
            # 复制代码运行请自行打印API的返回值。
            await client.create_archive_file_inspection_task_with_options_async(create_archive_file_inspection_task_request, runtime)
        except Exception as error:
            # 如有需要，请打印错误信息。
            UtilClient.assert_as_string(error.message)


if __name__ == '__main__':
    Sample.main(sys.argv[1:])

解压缩

调用CreateFileUncompressionTask - 创建解压缩任务接口对OSS中的压缩包进行解压缩处理。

解压缩信息

项目名称：test-project
压缩包路径：oss://test-bucket/test-target-object.zip
待解压缩文件：test-object1.jpg
解压缩文件的存放路径：oss://test-bucket/test-target-object

请求示例

{
    "ProjectName": "test-project",
    "SourceURI": "oss://test-bucket/test-target-object.zip",
    "Target": "{\"URI\":\"oss://test-bucket/test-target-object\"}",
    "SelectedFiles": "[\"test-object1.jpg\"]"
}

返回示例

{
  "TaskId": "FileUncompression-f31061ab-eaeb-469d-ac8f-9af10e5d****",
  "RequestId": "20115435-9F59-58C0-8D15-E6ECC096****",
  "EventId": "3B6-1S6ywKAB2ogXJnrzmk6tFUF****"
}

示例代码

文件解压缩的完整示例代码如下。

# -*- coding: utf-8 -*-
# This file is auto-generated, don't edit it. Thanks.
import os
import sys

from typing import List

from alibabacloud_imm20200930.client import Client as imm20200930Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_imm20200930 import models as imm_20200930_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient


class Sample:
    def __init__(self):
        pass

    @staticmethod
    def create_client(
            access_key_id: str,
            access_key_secret: str,
    ) -> imm20200930Client:
        """
        使用AK&SK初始化账号Client
        @param access_key_id:
        @param access_key_secret:
        @return: Client
        @throws Exception
        """
        config = open_api_models.Config(
            access_key_id=access_key_id,
            access_key_secret=access_key_secret
        )
        # 访问的域名
        config.endpoint = f'imm.cn-beijing.aliyuncs.com'
        return imm20200930Client(config)

    @staticmethod
    def main(
            args: List[str],
    ) -> None:
        # 阿里云账号AccessKey拥有所有API的访问权限，建议您使用RAM用户进行API访问或日常运维。
        # 强烈建议不要把AccessKey ID和AccessKey Secret保存到工程代码里，否则可能导致AccessKey泄露，威胁您账号下所有资源的安全。
        # 本示例通过从环境变量中读取AccessKey，来实现API访问的身份验证。如何配置环境变量，请参见https://help.aliyun.com/document_detail/2361894.html。
        imm_access_key_id = os.getenv("AccessKeyId")
        imm_access_key_secret = os.getenv("AccessKeySecret")
        client = Sample.create_client(imm_access_key_id, imm_access_key_secret)
        target = imm_20200930_models.CreateFileUncompressionTaskRequestTarget(
            uri='oss://test-bucket/test'
        )
        create_file_uncompression_task_request = imm_20200930_models.CreateFileUncompressionTaskRequest(
            project_name='test-project',
            target=target,
            source_uri='oss://test-bucket/test-target-object.zip'
        )
        runtime = util_models.RuntimeOptions()
        try:
            # 复制代码运行请自行打印 API 的返回值
            response = client.create_file_uncompression_task_with_options(create_file_uncompression_task_request, runtime)
        except Exception as error:
            # 如有需要，请打印 error
            UtilClient.assert_as_string(error.message)

    @staticmethod
    async def main_async(
            args: List[str],
    ) -> None:
        # 阿里云账号AccessKey拥有所有API的访问权限，建议您使用RAM用户进行API访问或日常运维。
        # 强烈建议不要把AccessKey ID和AccessKey Secret保存到工程代码里，否则可能导致AccessKey泄露，威胁您账号下所有资源的安全。
        # 本示例通过从环境变量中读取AccessKey，来实现API访问的身份验证。如何配置环境变量，请参见https://help.aliyun.com/document_detail/2361894.html。
        imm_access_key_id = os.getenv("AccessKeyId")
        imm_access_key_secret = os.getenv("AccessKeySecret")
        client = Sample.create_client(imm_access_key_id, imm_access_key_secret)
        target = imm_20200930_models.CreateFileUncompressionTaskRequestTarget(
            uri='oss://test-bucket/test'
        )
        create_file_uncompression_task_request = imm_20200930_models.CreateFileUncompressionTaskRequest(
            project_name='test-project',
            target=target,
            source_uri='oss://test-bucket/test-target-object.zip'
        )
        runtime = util_models.RuntimeOptions()
        try:
            # 复制代码运行请自行打印 API 的返回值
            await client.create_file_uncompression_task_with_options_async(create_file_uncompression_task_request, runtime)
        except Exception as error:
            # 如有需要，请打印 error
            UtilClient.assert_as_string(error.message)


if __name__ == '__main__':
    Sample.main(sys.argv[1:])

FAQ

通过CreateFileCompressionTask接口创建zip成功后，需要多久才能获取到压缩后的zip文件？
- 可以通过GetTask - 获取任务信息接口轮询任务状态，或者通过异步任务MNS消息处理示例异步接收成功的消息。