All Products
Search
Document Center

ApsaraVideo Media Processing:Add watermarks

Last Updated:Mar 14, 2024

You can add visible watermarks such as company logos or TV station logos to your videos to enhance brand visibility, protect copyrights, and boost product recognition. ApsaraVideo Media Processing (MPS) supports image watermarks, animated watermarks, and text watermarks. You can add watermarks based on your requirements. This topic provides sample code for using MPS SDK for Python to add watermarks.

Sample code

import os
import json

from urllib.parse import quote
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.acs_exception.exceptions import ClientException
from aliyunsdkcore.acs_exception.exceptions import ServerException
from aliyunsdkcore.auth.credentials import AccessKeyCredential
from aliyunsdkmts.request.v20140618.SubmitJobsRequest import SubmitJobsRequest

# Obtain an AccessKey ID and an AccessKey secret from environment variables.
credentials = AccessKeyCredential(os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'], os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET'])
# The region_id parameter specifies the ID of the region in which your MPS service is deployed. For information about supported regions, visit https://www.alibabacloud.com/help/zh/apsaravideo-for-media-processing/latest/service-regions.
client = AcsClient(region_id = 'cn-shanghai', credential = credentials)

mps_region_id = 'cn-shanghai'
pipeline_id = '9bad1a2e3c2c4df9c6****'
# The ID of the watermark template. You can create a watermark template in the MPS console. Watermark templates are supported only for image watermarks.
watermark_template_id = '9bad1a2e3c2c4df9c6****'
template_id = 'S00000001-200030'
oss_location = 'oss-cn-shanghai'
oss_bucket = '<your bucket name>'
oss_input_object = 'input.mp4'
oss_output_object = 'output.mp4'
image_watermark_object = 'logo.png'
video_watermark_object = 'logo.mov'

request = SubmitJobsRequest()
request.set_accept_format('json')
# The job input.
job_input = {'Location': oss_location,
             'Bucket': oss_bucket,
             'Object': quote(oss_input_object) }
request.set_Input(json.dumps(job_input))
# The job output.
output = {'OutputObject': quote(oss_output_object)}
output['TemplateId'] = template_id
## The configurations of the image watermark.
image_watermark_input = {'Location': oss_location,
                         'Bucket': oss_bucket,
                         'Object': quote(image_watermark_object) }
image_watermark = {
                  'WaterMarkTemplateId': watermark_template_id,
                  'Type': 'Image',
                  'InputFile': image_watermark_input,
                  'ReferPos': 'TopRight',
                  'Width': 0.05,
                  'Dx': 0,
                  'Dy': 0
                 }
## The configurations of the text watermark.
text_config = {
               # The content of the text watermark, which must be Base64-encoded.
               'Content': '5rWL6K+V5paH5a2X5rC05Y2w',
               'FontName': 'SimSun',
               'FontSize': 16,
               'FontColor': 'Red',
               'FontAlpha': 0.5,
               'Top': 10,
               'Left': 10
              }
text_watermark = {
                  'WaterMarkTemplateId': watermark_template_id,
                  'Type': 'Text',
                  'TextWaterMark': text_config
                 }
## The configurations of the animated image watermark.
video_watermark_input = {'Location': oss_location,
                         'Bucket': oss_bucket,
                         'Object': quote(video_watermark_object) }
video_watermark = {
                  'WaterMarkTemplateId': watermark_template_id,
                  'Type': 'Image',
                  'InputFile': video_watermark_input,
                  'ReferPos': 'BottomLeft',
                  'Height': 240,
                  'Dx': 0,
                  'Dy': 0
                 }
# Output->Watermarks
watermarks = [image_watermark, text_watermark, video_watermark]
output['WaterMarks'] = watermarks
# The response to the request.
outputs = [output]
request.set_Outputs(json.dumps(outputs))
request.set_OutputBucket(oss_bucket)
request.set_OutputLocation(oss_location)
request.set_PipelineId(pipeline_id)


response = client.do_action_with_exception(request)
# Display the response.
print(str(response, encoding='utf-8'))

References