文档首页/ 数据工坊 DWR/ 最佳实践/ 创建视频转码工作流
更新时间：2023-11-21 GMT+08:00
查看PDF
分享

创建视频转码工作流

场景介绍

针对使用DWR做视频转码的场景，如果需要每个对象转码任务的参数不同，可通过给对象增加对象元数据（例如：x-obs-meta-transcode-commands: base64(commands)），再使用DWR集成自定义函数的方式来实现视频转码。

具体的工作流视图如图1所示。

图1 工作流视图

资源和成本

表1 资源和成本规划

资源

资源说明

数量

每月费用

OBS

算子请求OBS API。

1

通过算子对数据进行处理，都会涉及到对OBS API的调用，每调用一次API都计算一次请求次数。对象存储服务OBS会根据调用API的请求次数进行费用收取，收取详情参见OBS请求费用说明。

FunctionGraph函数

算子使用FunctionGraph函数工作流。

1

通过算子对数据进行处理，会使用到函数工作流的资源，比如算子执行时长，函数工作流会根据资源使用情况进行收费，收费详情参见函数工作流计费说明。

视频转码

新建转码任务可以将视频进行转码，并在转码过程中压制水印、视频截图等。视频转码前需要配置转码模板。 待转码的音视频需要存储在与媒体处理服务同区域的OBS桶中，且该OBS桶已授权。

1

由媒体处理服务MPC进行收费，详情查看计费说明。

操作步骤

  1. 创建解析对象元数据并封装MPC转码任务的函数。

    在FunctionGraph创建函数，选择“Python 2.7”运行时语言，并为函数配置具有访问OBS权限的委托，最后配置函数环境变量：region_id=cn-north-4。创建函数请参见《函数工作流服务用户指南》。

    具体代码如下： 
      1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
    		 
    # -*- coding:utf-8 -*-
import urllib
import os
from obs import ObsClient  # Require public dependency:esdk_obs_python-3.x
import base64


class CreateTranscodingReq:
    def __init__(self, input=None, output=None, trans_template_id=None, av_parameters=None, output_filenames=None,
                 user_data=None, watermarks=None, thumbnail=None, priority=None, subtitle=None, encryption=None,
                 crop=None, audio_track=None, multi_audio=None, video_process=None, audio_process=None):
        if input is not None:
            self.input = input
        self.output = output
        if trans_template_id is not None:
            self.trans_template_id = trans_template_id
        if av_parameters is not None:
            self.av_parameters = av_parameters
        if output_filenames is not None:
            self.output_filenames = output_filenames
        if user_data is not None:
            self.user_data = user_data
        if watermarks is not None:
            self.watermarks = watermarks
        if thumbnail is not None:
            self.thumbnail = thumbnail
        if priority is not None:
            self.priority = priority
        if subtitle is not None:
            self.subtitle = subtitle
        if encryption is not None:
            self.encryption = encryption
        if crop is not None:
            self.crop = crop
        if audio_track is not None:
            self.audio_track = audio_track
        if multi_audio is not None:
            self.multi_audio = multi_audio
        if video_process is not None:
            self.video_process = video_process
        if audio_process is not None:
            self.audio_process = audio_process


def handler(event, context):
    # 获取上传桶、对象信息
    bucketName = event['Records'][0]['obs']['bucket']['name']
    objectKey = urllib.unquote(event['Records'][0]['obs']['object']['key'])  # a/b/c
    prefix = os.path.basename(objectKey)
    obsServer = 'obs.cn-north-7.ulanqab.huawei.com'
    # 使用obs sdk
    obsClient = newObsClient(context, obsServer)
    # 获取对象元数据
    resp = obsClient.getObjectMetadata(bucketName, objectKey)
    if resp.status < 300:
        print('headers: {}'.format(resp.header))
    else:
        print('errorCode: %s, errorMessage: %s', resp.errorCode, resp.errorMessage)
        return "ERROR"
    # 获取转码命令元数据
    command_base64 = dict(resp.header)["transcode-command"]
    print("command base64 {}".format(command_base64))
    if not command_base64:
        return "Error"
    # base64反解析
    command = base64.b64decode(command_base64)
    print("command :{}".format(command))
    # 解析转码命令
    command_map = trans(command)
    # 封装MPC转码任务参数
    av_parameters = []
    watermarks = []
    if "avthumb" in command_map:
        av_param = {}
        _type = 4
        if command_map["avthumb"] == "mp4":
            _type = 4
            av_param["video"] = {
                "output_policy": "transcode",
                "codec": 1,
                "bitrate": 40,
                "max_iframes_interval": 5,
            }
        elif command_map["avthumb"] == "mp3":
            _type = 5
            av_param["audio"] = {}
        av_param["common"] = {
            "pack_type": _type,
            "PVC": False,
            "hls_interval": 2,
            "dash_interval": 2,
        }
        av_parameters.append(av_param)
    # 文字水印参数封装
    if "wmText" in command_map:
        watermarks.append({
            "text_context": command_map["wmText"],
            "text_watermark": {
                "dx": command_map["wmOffsetX"],
                "dy": command_map["wmOffsetY"],
                "font_size": command_map["wmFontSize"],
            }
        })
    # 图片水印参数封装
    if "wmImage" in command_map:
        watermarks.append({
            "input": {
                "location": context.getUserData('region_id'),
                "bucket": bucketName,
                "object": objectKey
            },
            "image_watermark": {
                "dx": command_map["wmOffsetX"],
                "dy": command_map["wmOffsetY"],
            }
        })
    transcode_req = CreateTranscodingReq(input={
        "location": context.getUserData('region_id'),
        "bucket": bucketName,
        "object": objectKey
    }, output={
        "location": context.getUserData('region_id'),
        "bucket": bucketName,
        "object": prefix,
        "file_name": "demo.mp4"

    }, av_parameters=av_parameters,
        watermarks=watermarks)
    event["dynamic_source"] = {
        "transcodes": [
            transcode_req.__dict__
        ]
    }
    return event


def newObsClient(context, obsServer):
    ak = context.getAccessKey()
    sk = context.getSecretKey()
    return ObsClient(access_key_id=ak, secret_access_key=sk, server=obsServer)


def trans(command):
    commands = str(command).split("/")
    ret = {}
    key = ''
    for i in range(0, len(commands)):
        if i % 2 == 0:
            key = commands[i]
            continue
        ret[key] = commands[i]
    return ret

  2. 创建工作流。

    关联步骤1的自定义函数和“媒资转码”算子。拓扑图如图2所示。

    需要关闭自定义算子和“媒资转码”算子的动态参数开关。

    图2 创建工作流

  3. 配置触发器。

    触发器关联到需要做转码的桶，并根据业务需要指定对象前、后缀。

    填写配置信息如下：

    触发器名称：“test-trans”

    关联工作流：“trans-teat-meta”

    事件源存储桶：“dataplus-test”

    事件源类型：“ObjectCreated”

    前缀：“perfix”

    后缀：“suffix”

  4. 上传对象。

    上传对象时带上自定义对象元数据，具体代码如下：

     1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
    		 
    # -*- coding:utf-8 -*-
from obs import ObsClient
import base64

if __name__ == '__main__':
    # 上传后使用DWR做转码的参数
    command = "avthumb/mp4/wmText/dGVzdF90ZXh0/wmGravityText/SouthEast/wmFontSize/20/wmOffsetX/10/wmOffsetY/38"
    client = ObsClient(
        access_key_id="XXX",
        secret_access_key="XXX",
        server="https://obs.cn-north-4.myhuaweicloud.com"
    )
    # 上传对象，并配置转码对象命令元数据
    resp = client.putObject("dataplus-test", "test.sh", "content", metadata={
        "x-obs-meta-transcode-command": base64.b64encode(command)
    })
    print resp

相关文档