文档

Python SDK

更新时间:

阿里云API调试工具OpenAPI Explore提供OpenAPI文档、API调试和SDK示例,帮助您快速入门API开发。本文介绍如何安装和使用智能媒体管理Python SDK。

前提条件

安装SDK

重要

调用2020-09-30版本的API,需要使用2020-09-30版本的SDK。

使用智能媒体管理Python SDK的操作步骤请参见快速入门

使用方法

如下以使用1.27.3版本的Python SDK在华东1(杭州)地域中创建文档转换任务为例介绍如何使用Python SDK。

  1. 创建demo.py文件并写入如下内容。

    说明

    复制使用如下示例代码时请将访问域名换成您的IMM项目所在Region的域名,并且IMM项目和OSS文件Region需要保持一致。

    # -*- coding: utf-8 -*-
    
    import os
    import sys
    import json
    import time
    
    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
    
    # 导入下载的MNS SDK。
    from mns.account import Account
    
    
    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-hangzhou.aliyuncs.com'
            return imm20200930Client(config)
    
        @staticmethod
        def main() -> 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")
            # 填写MNS的AccessKey ID、AccessKey Secret、Endpoint、queue名称、topic名称,建议从配置中读取。
            mns_access_key_id = os.getenv("MnsAccessKeyId")
            mns_access_key_secret = os.getenv("MnsAccessKeySecret")
            mns_endpoint = "https://1111111111.mns.cn-hangzhou.aliyuncs.com"
            mns_queue_name = "queue"
            mns_topic_name = "topic1"
            # 初始化客户端。
            client = Sample.create_client(imm_access_key_id, imm_access_key_secret)
            # 创建API请求。
            request = imm_20200930_models.CreateOfficeConversionTaskRequest()
            params = {
                # 填写IMM项目名称。
                "ProjectName": "immtest",
                # 填写待转换的文件OSS地址。
                "SourceURI": "oss://your-bucket-name/your-path/your-doc.doc",
                # 填写目标文件类型。
                "TargetType": "pdf",
                # 填写目标文件路径前缀。
                "TargetURIPrefix": "oss://your-bucket-name/your-path/your-doc-output",
                # 填写MNS Endpoint。
                "NotifyEndpoint": mns_endpoint,
                # 填写MNS topic名称。
                "NotifyTopicName": mns_topic_name
            }
            request.from_map(params)
            runtime = util_models.RuntimeOptions()
            try:
                # 发起请求,打印API的返回值。
                response = client.create_office_conversion_task_with_options(request, runtime)
                print(json.dumps(response.body.to_map(), indent=4))
                event_id = response.body.event_id
    
                # 获取该任务的MNS消息,文档转换结果通过MNS消息返回。
                # 初始化my_account和my_queue。
                my_account = Account(mns_endpoint, mns_access_key_id, mns_access_key_secret)
                my_queue = my_account.get_queue(mns_queue_name)
                my_queue.set_encoding(False)
                # 循环读取删除消息直到读取到该任务的MNS消息,如果循环时间超过300秒则退出。
                receive_mns_start_time = int(time.time())
                wait_seconds = 3
                while True:
                    # 防止queue长时间未收到该消息一直卡着,超过5分钟则返回。
                    if int(time.time()) - receive_mns_start_time > 300:
                        raise Exception("Receive Message Timeout.")
                    # 读取消息。
                    try:
                        recv_msg = my_queue.receive_message(wait_seconds)
                        print(
                            "Receive Message Succeed! ReceiptHandle:%s MessageBody:%s MessageID:%s" % (
                            recv_msg.receipt_handle, recv_msg.message_body,
                            recv_msg.message_id))
                        msg_body = json.loads(recv_msg.message_body)
                        mns_message = json.loads(msg_body.get("Message"))
                        mns_event_id = mns_message.get("id")
                        if mns_event_id == event_id:
                            # 找到该任务的MNS消息,打印消息内容。
                            print(json.dumps(mns_message, indent=4))
                            # 删除消息。
                            try:
                                my_queue.delete_message(recv_msg.receipt_handle)
                                print(
                                    "Delete Message Succeed!  ReceiptHandle:%s" % recv_msg.receipt_handle)
                            except Exception as e:
                                print("Delete Message Fail! Exception:%s\n" % e)
                            break
                    except Exception as e:
                        if e.type == u"QueueNotExist":
                            print("Queue not exist, please create queue before receive message.")
                            sys.exit(0)
                        elif e.type == u"MessageNotExist":
                            print("Queue is empty!")
                        print("Receive Message Fail! Exception:%s\n" % e)
                        continue
    
            except Exception as error:
                # 如有需要,请打印错误信息。
                UtilClient.assert_as_string(error.message)
                print(error)
    
    
    if __name__ == '__main__':
        Sample.main()                 
  2. 执行python demo.py命令,控制台输出结果如下。

    // 文档转换API返回值。
    {
        "TaskId": "OfficeConversion-65ab9c57-4598-4eb4-bf73-****",
        "RequestId": "A637DBBA-3E98-00F5-B0F2-****",
        "EventId": "011-1LNuW4Ws2lRfjGsZl4a****"
    }
    
    // MNS消息,Status为Succeeded表示文档转换任务执行成功。
    {
        "ProjectName": "immtest",
        "DatasetName": "",
        "RequestId": "A637DBBA-3E98-00F5-B0F2-****",
        "StartTime": "2022-07-19T10:41:44.649Z",
        "EndTime": "2022-07-19T10:41:45.26Z",
        "UserData": "",
        "TaskType": "OfficeConversion",
        "TaskId": "OfficeConversion-65ab9c57-4598-4eb4-bf73-****",
        "Status": "Succeeded",
        "Code": "",
        "Message": ""
    }

  • 本页导读 (1)
文档反馈