ApsaraVideo Media Processing:Crie um fluxo de trabalho de mídia

import os
import sys

from typing import List

from alibabacloud_mts20140618.client import Client as Mts20140618Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_mts20140618 import models as mts_20140618_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() -> Mts20140618Client:
        """
        Initialize an MTS client using AccessKey credentials from environment variables.
        @return: Client
        @throws Exception
        """
        config = open_api_models.Config(
            # Required. Make sure that the ALIBABA_CLOUD_ACCESS_KEY_ID environment variable is configured. ,
            access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
            # Required. Make sure that the ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variable is configured. ,
            access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
        )
        config.endpoint = f'mts.cn-hangzhou.aliyuncs.com'
        return Mts20140618Client(config)

    @staticmethod
    def main(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        add_media_workflow_request = mts_20140618_models.AddMediaWorkflowRequest(
            # The name of the media workflow.
            name='mediaworkflow-example',
            # The topology of the media workflow, defined as a JSON string.
            # It contains three activities: Act-Start (OSS trigger), mediaworkflow-example (transcode), and Act-Report (manual publish).
            topology='{"Activities": {"mediaworkflow-example": {"Parameters": {"Outputs": "[{\"OutputObject\":\"examplebucket/output/{RunId}/TRANSCODE_165941222****/{FileName}\",\"TemplateId\":\"S00000001-200010\",\"TemplateName\":\"MP4-Low definition\"}]","OutputBucket": "examplebucket","OutputLocation": "oss-cn-shanghai"},"Type": "Transcode"},"Act-Start": {"Parameters": {"PipelineId": "a7d481f07d8c45da88c71853ce7d****","InputFile": "{\"Bucket\":\"example-input\",\"Location\":\"oss-cn-shanghai\",\"ObjectPrefix\":\"mps-test/input/\"}"},"Type": "Start"},"Act-Report": {"Parameters": {"PublishType": "Manual"},"Type": "Report"}},"Dependencies": {"mediaworkflow-example": ["Act-Report"],"Act-Start": ["mediaworkflow-example"],"Act-Report": []}}',
            # The trigger mode. OssAutoTrigger starts the workflow when an object is uploaded to the input bucket.
            trigger_mode='OssAutoTrigger'
        )
        runtime = util_models.RuntimeOptions()
        try:
            # The API response is not printed here. Add your own output logic if needed.
            client.add_media_workflow_with_options(add_media_workflow_request, runtime)
        except Exception as error:
            # Handle exceptions appropriately in production. The following output is for debugging only.
            # The error message.
            print(error.message)
            # The URL of the corresponding error diagnostics page.
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)

    @staticmethod
    async def main_async(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        add_media_workflow_request = mts_20140618_models.AddMediaWorkflowRequest(
            # The name of the media workflow.
            name='mediaworkflow-example',
            # The topology of the media workflow, defined as a JSON string.
            # It contains three activities: Act-Start (OSS trigger), mediaworkflow-example (transcode), and Act-Report (manual publish).
            topology='{"Activities": {"mediaworkflow-example": {"Parameters": {"Outputs": "[{\"OutputObject\":\"examplebucket/output/{RunId}/TRANSCODE_165941222****/{FileName}\",\"TemplateId\":\"S00000001-200010\",\"TemplateName\":\"MP4-Low definition\"}]","OutputBucket": "examplebucket","OutputLocation": "oss-cn-shanghai"},"Type": "Transcode"},"Act-Start": {"Parameters": {"PipelineId": "a7d481f07d8c45da88c71853ce7d****","InputFile": "{\"Bucket\":\"example-input\",\"Location\":\"oss-cn-shanghai\",\"ObjectPrefix\":\"mps-test/input/\"}"},"Type": "Start"},"Act-Report": {"Parameters": {"PublishType": "Manual"},"Type": "Report"}},"Dependencies": {"mediaworkflow-example": ["Act-Report"],"Act-Start": ["mediaworkflow-example"],"Act-Report": []}}',
            # The trigger mode. OssAutoTrigger starts the workflow when an object is uploaded to the input bucket.
            trigger_mode='OssAutoTrigger'
        )
        runtime = util_models.RuntimeOptions()
        try:
            # The API response is not printed here. Add your own output logic if needed.
            await client.add_media_workflow_with_options_async(add_media_workflow_request, runtime)
        except Exception as error:
            # Handle exceptions appropriately in production. The following output is for debugging only.
            # The error message.
            print(error.message)
            # The URL of the corresponding error diagnostics page.
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)

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