Use MPS SDK for Python to capture video snapshots - ApsaraVideo Media Processing

You can use the video snapshot feature to capture snapshots of a specific size at specific points in time of a video. The snapshots are used in scenarios such as video thumbnails, sprites, and progress bar thumbnails. You can specify the points in time when snapshots are captured, the interval between two consecutive snapshots, the number of snapshots to be captured, the types of snapshots to be captured, and whether to compose multiple snapshots into one image sprite. You can submit snapshot jobs in the ApsaraVideo Media Processing (MPS) console or by using the API or SDKs. This topic provides an example on how to use MPS SDK for Python V2.0 to capture snapshots.

Sample code

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:
        """
        Use your AccessKey ID and AccessKey secret to initialize a client.
        @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()
        submit_snapshot_job_request = mts_20140618_models.SubmitSnapshotJobRequest(
            # The job input.
            input='{"Bucket":"example-bucket","Location":"example-location","Object":"example%2Ftest.flv"}',
            # The snapshot configuration.
            snapshot_config='{"OutputFile":{"Bucket":"example-001","Location":"example-location","Object":"{Count}.jpg"},"Time":"5","Num":"10","Interval":"20"}',
            # The user-defined data.
            user_data='testid-001',
            # The ID of the MPS queue.
            pipeline_id='dd3dae411e704030b921e52698e5****'
        )
        runtime = util_models.RuntimeOptions()
        try:
            # Write your own code to display the response of the API operation if necessary.
            client.submit_snapshot_job_with_options(submit_snapshot_job_request, runtime)
        except Exception as error:
            # Handle exceptions with caution in actual business scenarios and never ignore exceptions in your project. In this example, error messages are displayed for reference 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()
        submit_snapshot_job_request = mts_20140618_models.SubmitSnapshotJobRequest(
            # The job input.
            input='{"Bucket":"example-bucket","Location":"example-location","Object":"example%2Ftest.flv"}',
            # The snapshot configuration.
            snapshot_config='{"OutputFile":{"Bucket":"example-001","Location":"example-location","Object":"{Count}.jpg"},"Time":"5","Num":"10","Interval":"20"}',
            # The user-defined data.
            user_data='testid-001',
            # The ID of the MPS queue.
            pipeline_id='dd3dae411e704030b921e52698e5****'
        )
        runtime = util_models.RuntimeOptions()
        try:
            # Write your own code to display the response of the API operation if necessary.
            await client.submit_snapshot_job_with_options_async(submit_snapshot_job_request, runtime)
        except Exception as error:
            # Handle exceptions with caution in actual business scenarios and never ignore exceptions in your project. In this example, error messages are displayed for reference 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:])

ApsaraVideo Media Processing:Capture snapshots

Sample code

References