簡單視頻處理系統

前提條件

• 函數計算：開通函數計算服務。

• 對象存儲OSS：控制臺創建存儲空間。

操作步驟

1. 創建服務。

   1. 登錄函數計算控制臺，在左側導航欄，單擊服務及函數。

   2. 在頂部菜單欄，選擇地域，然后在服務列表頁面，單擊創建服務。

   3. 在創建服務面板，填寫服務名稱和描述，并按需設置配置項，然后單擊確定。

      本文設置服務角色為AliyunFCDefaultRole，并在權限策略中增加AliyunOSSFullAccess。

      關于創建服務的詳細配置信息，請參見創建服務。

2. 創建函數。

   1. 在函數管理頁面，單擊創建函數。

   2. 在創建函數頁面，按需選擇創建函數的方式，配置以下配置項，然后單擊創建。

      • 創建函數方式：使用內置運行時創建。

      • 基本設置：配置函數的基本信息，包括函數名稱和請求處理程序類型。請求處理程序類型選擇處理事件請求。

      • 函數代碼：函數的運行環境選擇Python 3.9，代碼上傳方式選擇使用示例代碼。

      • 高級配置：由于處理視頻文件比較耗時，并且要考慮視頻文件的大小，所以本文選擇vCPU規格為4核，內存規格為8 GB，臨時硬盤大小為10 GB，執行超時時間為7200秒。

      其他配置項使用默認值，關于創建函數的詳細配置信息，請參見創建函數。

3. 創建OSS觸發器。

   1. 在函數詳情頁面，單擊觸發器管理頁簽，從版本或別名下拉列表選擇要創建觸發器的版本或別名，然后單擊創建觸發器。

   2. 在創建觸發器面板，填寫相關信息。然后單擊確定。

      配置項	操作	本文示例
      觸發器類型	選擇對象存儲 OSS。	對象存儲 OSS
      名稱	填寫自定義的觸發器名稱。	oss-trigger
      版本或別名	默認值為LATEST，如果您需要創建其他版本或別名的觸發器，需先在函數詳情頁的版本或別名下拉列表選擇該版本。關于版本和別名的簡介，請參見管理版本和管理別名。	LATEST
      Bucket 名稱	選擇已創建的OSS Bucket。	testbucket
      文件前綴	輸入要匹配的文件名稱的前綴。建議您配置文件的前綴和后綴，避免觸發事件嵌套循環觸發引起額外費用。此外，一個Bucket的不同觸發器如果指定了相同的事件類型，則前綴和后綴不能重復。詳細信息，請參見OSS觸發器觸發規則。 重要 文件前綴不能以/開頭，否則會導致OSS觸發器無法被觸發。	source
      文件后綴	輸入要匹配的文件名稱的后綴。強烈建議您配置前綴和后綴，避免觸發事件嵌套循環觸發引起額外費用。另外，一個Bucket的不同觸發器如果指定了相同的事件類型，則前綴和后綴不能重復。詳細信息，請參見OSS觸發器觸發規則。	mp4
      觸發事件	選擇一個或多個觸發事件。關于對象存儲OSS的事件類型，請參見OSS事件定義。	oss:ObjectCreated:PutObject, oss:ObjectCreated:PostObject, oss:ObjectCreated:CompleteMultipartUpload
      角色名稱	選擇AliyunOSSEventNotificationRole。 說明 如果您第一次創建該類型的觸發器，則需要在單擊確定后，在彈出的對話框中選擇立即授權。	AliyunOSSEventNotificationRole

4. 編寫函數代碼。

   1. 在函數詳情頁面，單擊函數代碼頁簽，在代碼編輯器中編寫代碼。

      函數需要將MP4格式的視頻文件轉為FLV格式，并將FLV格式視頻文件保存在OSS Bucket下的dest目錄。本文以Python語言為例，示例代碼如下。

      # -*- coding: utf-8 -*-
      import logging
      import oss2
      import os
      import json
      import subprocess
      import shutil
      
      logging.getLogger("oss2.api").setLevel(logging.ERROR)
      logging.getLogger("oss2.auth").setLevel(logging.ERROR)
      LOGGER = logging.getLogger()
      
      
      def get_fileNameExt(filename):
          (_, tempfilename) = os.path.split(filename)
          (shortname, extension) = os.path.splitext(tempfilename)
          return shortname, extension
      
      
      def handler(event, context):
          LOGGER.info(event)
          evt = json.loads(event)
          evt = evt["events"]
          oss_bucket_name = evt[0]["oss"]["bucket"]["name"]
          object_key = evt[0]["oss"]["object"]["key"]
          
          output_dir = "dest"
          dst_format = "flv"
          shortname, _ = get_fileNameExt(object_key)
          creds = context.credentials
          auth = oss2.StsAuth(creds.accessKeyId,
                              creds.accessKeySecret, creds.securityToken)
          oss_client = oss2.Bucket(auth, 'oss-%s-internal.aliyuncs.com' %
                                   context.region, oss_bucket_name)
      
          exist = oss_client.object_exists(object_key)
          if not exist:
              raise Exception("object {} is not exist".format(object_key))
      
          input_path = oss_client.sign_url('GET', object_key, 6 * 3600)
          # m3u8 特殊處理
          rid = context.request_id
          if dst_format == "m3u8":
              return handle_m3u8(rid, oss_client, input_path, shortname, output_dir)
          else:
              return handle_common(rid, oss_client, input_path, shortname, output_dir, dst_format)
      
      
      def handle_m3u8(request_id, oss_client, input_path, shortname, output_dir):
          ts_dir = '/tmp/ts'
          if os.path.exists(ts_dir):
              shutil.rmtree(ts_dir)
          os.mkdir(ts_dir)
          transcoded_filepath = os.path.join('/tmp', shortname + '.ts')
          split_transcoded_filepath = os.path.join(
              ts_dir, shortname + '_%03d.ts')
          cmd1 = ['ffmpeg', '-y', '-i', input_path, '-c:v',
                  'libx264', transcoded_filepath]
      
          cmd2 = ['ffmpeg', '-y', '-i', transcoded_filepath, '-c', 'copy', '-map',  '0',  '-f', 'segment',
                  '-segment_list', os.path.join(ts_dir, 'playlist.m3u8'), '-segment_time', '10', split_transcoded_filepath]
      
          try:
              subprocess.run(
                  cmd1, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True)
      
              subprocess.run(
                  cmd2, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True)
      
              for filename in os.listdir(ts_dir):
                  filepath = os.path.join(ts_dir, filename)
                  filekey = os.path.join(output_dir, shortname, filename)
                  oss_client.put_object_from_file(filekey, filepath)
                  os.remove(filepath)
                  print("Uploaded {} to {}".format(filepath, filekey))
      
          except subprocess.CalledProcessError as exc:
              # if transcode fail，trigger invoke dest-fail function
              raise Exception(request_id + " transcode failure, detail: " + str(exc))
      
          finally:
              if os.path.exists(ts_dir):
                  shutil.rmtree(ts_dir)
      
              # remove ts 文件
              if os.path.exists(transcoded_filepath):
                  os.remove(transcoded_filepath)
      
          return {}
      
      
      def handle_common(request_id, oss_client, input_path, shortname, output_dir, dst_format):
          transcoded_filepath = os.path.join('/tmp', shortname + '.' + dst_format)
          if os.path.exists(transcoded_filepath):
              os.remove(transcoded_filepath)
          cmd = ["ffmpeg", "-y", "-i", input_path, transcoded_filepath]
          try:
              subprocess.run(
                  cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True)
      
              oss_client.put_object_from_file(
                  os.path.join(output_dir, shortname + '.' + dst_format), transcoded_filepath)
          except subprocess.CalledProcessError as exc:
              # if transcode fail，trigger invoke dest-fail function
              raise Exception(request_id + " transcode failure, detail: " + str(exc))
          finally:
              if os.path.exists(transcoded_filepath):
                  os.remove(transcoded_filepath)
          return {}

   2. 單擊部署代碼。

5. 測試函數代碼。

   您可以通過配置函數入口參數模擬OSS事件參數進行測試，驗證代碼的正確性。在實際操作過程中，當發生OSS事件時，函數會自動被觸發執行。

   1. 在函數詳情頁面，單擊函數代碼頁簽，然后單擊測試函數右側圖標，從下拉列表中，選擇配置測試參數。

   2. 在配置測試參數面板，選擇創建新測試事件或編輯已有測試事件頁簽，填寫事件名稱和事件內容。然后單擊確定。

      本文配置的事件內容如下，關于event參數具體信息，請參見步驟二：配置函數入口參數。

      {
          "events": [
              {
                  "eventName": "oss:ObjectCreated:CompleteMultipartUpload",
                  "eventSource": "acs:oss",
                  "eventTime": "2022-08-13T06:45:43.000Z",
                  "eventVersion": "1.0",
                  "oss": {
                      "bucket": {
                          "arn": "acs:oss:cn-hangzhou:123456789:testbucket",
                          "name": "testbucket",
                          "ownerIdentity": "164901546557****"
                      },
                      "object": {
                          "deltaSize": 122539,
                          "eTag": "688A7BF4F233DC9C88A80BF985AB****",
                          "key": "source/a.mp4",
                          "size": 122539
                      },
                      "ossSchemaVersion": "1.0",
                      "ruleId": "9adac8e253828f4f7c0466d941fa3db81161****"
                  },
                  "region": "cn-hangzhou",
                  "requestParameters": {
                      "sourceIPAddress": "140.205.XX.XX"
                  },
                  "responseElements": {
                      "requestId": "58F9FF2D3DF792092E12044C"
                  },
                  "userIdentity": {
                      "principalId": "164901546557****"
                  }
              }
          ]
      }

   3. 單擊測試函數，在函數代碼頁簽查看執行結果。

視頻處理工作流系統

前提條件

• 開通服務并創建Bucket

  • 函數計算：開通函數計算服務。

  • 對象存儲OSS：控制臺創建存儲空間。

  • Serverless 工作流：開通Serverless工作流。

  • 文件存儲 NAS：開通文件存儲服務NAS。

  • 專有網絡 VPC：開通專有網絡VPC。

• 配置服務角色

  • AliyunFcDefaultRole：創建服務時需要配置該角色，同時該角色需要配置權限AliyunOSSFullAccess、AliyunFnFFullAccess、AliyunFCInvocationAccess，用于調用函數、管理工作流和管理對象存儲OSS。

  • AliyunOSSEventNotificationRole：OSS默認使用此角色發送事件通知。

  • fnf-execution-default-role：創建及管理工作流所需要的角色，該角色需要配置權限AliyunFCInvocationAccess、AliyunFnFFullAccess。

• Serverless Devs工具安裝及配置

  • 安裝Serverless Devs工具及依賴

  • 配置Serverless Devs

操作步驟