このトピックでは、ApsaraVideo Media Processing(MPS)SDK for Python V2.0 を使用してメディアワークフローを作成する方法の例を示します。
サンプルコード
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:
"""
クライアントを初期化するために、AccessKey ID と AccessKey シークレットを使用します。
@return: Client
@throws Exception
"""
config = open_api_models.Config(
# 必須。 ALIBABA_CLOUD_ACCESS_KEY_ID 環境変数が設定されていることを確認します。
access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
# 必須。 ALIBABA_CLOUD_ACCESS_KEY_SECRET 環境変数が設定されていることを確認します。
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(
# メディアワークフローの名前。
name='mediaworkflow-example',
# メディアワークフローのトポロジー。
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": []}}',
# トリガーモード。
trigger_mode='OssAutoTrigger'
)
runtime = util_models.RuntimeOptions()
try:
# 必要に応じて、API操作のレスポンスを表示するための独自のコードを記述します。
client.add_media_workflow_with_options(add_media_workflow_request, runtime)
except Exception as error:
# 実際のビジネスシナリオでは、例外を慎重に処理し、プロジェクトで例外を無視しないでください。 この例では、エラーメッセージは参照用にのみ表示されます。
# エラーメッセージ。
print(error.message)
# 対応するエラー診断ページの URL。
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(
# メディアワークフローの名前。
name='mediaworkflow-example',
# メディアワークフローのトポロジー。
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": []}}',
# トリガーモード。
trigger_mode='OssAutoTrigger'
)
runtime = util_models.RuntimeOptions()
try:
# 必要に応じて、API操作のレスポンスを表示するための独自のコードを記述します。
await client.add_media_workflow_with_options_async(add_media_workflow_request, runtime)
except Exception as error:
# 実際のビジネスシナリオでは、例外を慎重に処理し、プロジェクトで例外を無視しないでください。 この例では、エラーメッセージは参照用にのみ表示されます。
# エラーメッセージ。
print(error.message)
# 対応するエラー診断ページの URL。
print(error.data.get("Recommend"))
UtilClient.assert_as_string(error.message)
if __name__ == '__main__':
Sample.main(sys.argv[1:])