Amazon Bedrock助力飞书深诺电商广告分类

Python
import ffmpeg
import os

def transform_video(file_name):

    # 获取文件大小（以字节为单位）
    file_size_bytes = os.path.getsize(file_name)
    # 将文件大小转换为合适的单位（例如 MB）
    file_size_mb = file_size_bytes / (1024 * 1024)
    print(f'视频文件大小为: {file_size_mb:.2f} MB')
    
    # 指定新的输出文件路径和名称
    output_filename = os.path.splitext(os.path.basename(file_name))[0] + '_proccessed.mp4'
    output_path = "media/" + output_filename

    # 检查输出文件是否已存在
    if os.path.isfile(output_path):
        # 如果文件存在,则删除旧文件
        os.remove(output_path)

    # 如果视频文件过大，超过25MB，则压缩一下
    if file_size_mb > 25:
        print(f'视频文件大于25MB，需要压缩\n')
        try:
            probe = ffmpeg.probe(file_name)

            # 检查视频流是否存在
            video_stream = next((stream for stream in probe['streams'] if stream['codec_type'] == 'video'), None)
            if video_stream is None:
                print(f"{file_name} does not contain a video stream.")
            else:
                # 获取视频分辨率
                video_width = video_stream['width']
                video_height = video_stream['height']
                print(f"Original Video resolution: {video_width}x{video_height}")

                # 计算新的分辨率(长宽各为原来的一半)
                new_width = video_width // 2
                new_height = video_height // 2

                # 设置ffmpeg流
                stream = ffmpeg.input(file_name)

                # 设置视频编码器和分辨率
                stream = ffmpeg.filter(stream, 'scale', new_width, new_height)
                stream = ffmpeg.output(stream, output_path, vcodec='libx264', acodec='aac', f='mp4', crf=23, preset='veryfast')

                # 运行 FFmpeg 命令
                ffmpeg.run(stream, overwrite_output=True)
                print(f'视频压缩成功,输出文件: {output_path}')

        except ffmpeg.Error as e:
            print(f"Error: {e.stderr}", e.stdout)

    # 如果视频格文件编码为 Nova不支持的类型，尝试转为 Nova 支持的编码H264
    else:
        print(f'视频文件大小OK，直接转成H264')

        # 设置ffmpeg流
        stream = ffmpeg.input(file_name)
        stream = ffmpeg.output(stream, output_path, vcodec='libx264', acodec='aac', f='mp4')

        # 运行ffmpeg命令
        ffmpeg.run(stream)
        print(f'Video converted to H.264 and saved as {output_path}')
    
    # 返回新的处理后的文件全路径
    return output_path

Python
import base64
import boto3
import json

def video_understanding(file_name):
    # Create a Bedrock Runtime client in the AWS Region of your choice.
    client = boto3.client(
        "bedrock-runtime",
        region_name="us-east-1",
    )

    MODEL_ID = "us.amazon.nova-lite-v1:0"

    # Open the image you'd like to use and encode it as a Base64 string.
    with open(file_name, "rb") as video_file:
        binary_data = video_file.read()
        base_64_encoded_data = base64.b64encode(binary_data)
        base64_string = base_64_encoded_data.decode("utf-8")
    # Define your system prompt(s).
    system_list= [
        {
            "text": "You are a expert to understand advertising video materials."
        }
    ]
    # Define a "user" message including both the image and a text prompt.
    message_list = [
        {
            "role": "user",
            "content": [
                {
                    "video": {
                        "format": "mp4",
                        "source": {"bytes": base64_string},
                    }
                },
                {
                    "text": "Please understand and describe the information in the video, focusing on the product info to classify the product. Within 100 words."
                },
            ],
        }
    ]
    # Configure the inference parameters.
    inf_params = {"maxTokens": 3000, "topP": 0.1, "topK": 20, "temperature": 0.3}

    native_request = {
        "schemaVersion": "messages-v1",
        "messages": message_list,
        "system": system_list,
        "inferenceConfig": inf_params,
    }
    # Invoke the model and extract the response body.
    response = client.invoke_model(modelId=MODEL_ID, body=json.dumps(native_request))
    model_response = json.loads(response["body"].read())
    # Pretty print the response JSON.
    print("[Full Response]")
    print(json.dumps(model_response, indent=2))
    # Print the text content for easy readability.
    content_text = model_response["output"]["message"]["content"][0]["text"]
    print("\n[Response Content Text]")
    print(content_text)
    
    return content_text

Python
import base64
import boto3
import json

def video_understanding_for_s3(s3_uri):
    
    # 创建 STS 客户端
    sts_client = boto3.client('sts')

    # 获取账户 ID
    account_id = sts_client.get_caller_identity()['Account']
    
    
    # Create a Bedrock Runtime client in the AWS Region of your choice.
    client = boto3.client(
        "bedrock-runtime",
        region_name="us-east-1",
    )

    MODEL_ID = "us.amazon.nova-lite-v1:0"
    # Define your system prompt(s).
    system_list = [
        {
            "text": "You are a expert to understand advertising video materials."
        }
    ]
    # Define a "user" message including both the image and a text prompt.
    message_list = [
        {
            "role": "user",
            "content": [
                {
                    "video": {
                        "format": "mp4",
                        "source": {
                            "s3Location": {
                                "uri": f"{s3_uri}", 
                                "bucketOwner": f"{account_id}"
                            }
                        }
                    }
                },
                {
                    "text": "Please understand and describe the information in the video, focusing on the product info to classify the product. Within 100 words."
                }
            ]
        }
    ]
    # Configure the inference parameters.
    inf_params = {"maxTokens": 300, "topP": 0.1, "topK": 20, "temperature": 0.3}

    native_request = {
        "schemaVersion": "messages-v1",
        "messages": message_list,
        "system": system_list,
        "inferenceConfig": inf_params,
    }
    # Invoke the model and extract the response body.
    response = client.invoke_model(modelId=MODEL_ID, body=json.dumps(native_request))
    model_response = json.loads(response["body"].read())
    # Pretty print the response JSON.
    print("[Full Response]")
    print(json.dumps(model_response, indent=2))
    # Print the text content for easy readability.
    content_text = model_response["output"]["message"]["content"][0]["text"]
    print("\n[Response Content Text]")
    print(content_text)
    return content_text

JSON
[
  {
    "category_id": "1",
    "name": "服装",
    "children": [
      {
        "category_id": "1-1",
        "name": "女装",
        "children": [
          {
            "category_id": "1-1-1",
            "name": "女装裙装"
          },
          {
            "category_id": "1-1-2",
            "name": "女装T恤"
          },
          {
            "category_id": "1-1-3",
            "name": "女装外套/大衣"
          },
          {
            "category_id": "1-1-4",
            "name": "女装裤装/牛仔裤"
          },
          {
            "category_id": "1-1-5",
            "name": "女装西服/套装"
          },
          {
            "category_id": "1-1-6",
            "name": "女装衬衫"
          },
          {
            "category_id": "1-1-7",
            "name": "女装毛衣/卫衣"
          },
          {
            "category_id": "1-1-8",
            "name": "女装睡衣/家居服"
          },
          {
            "category_id": "1-1-9",
            "name": "女装丝袜/袜子"
          },
          {
            "category_id": "1-1-10",
            "name": "女装-其他"
          }
        ]
      },
      ...

JSON
[
  {
    "name": "服装",
    "children": [
      {
        "name": "女装",
        "children": [
          {
            "name": "女装裙装"
          },
          {
            "name": "女装T恤"
          },
          {
            "name": "女装外套/大衣"
          },
          {
            "name": "女装裤装/牛仔裤"
          },
          {
            "name": "女装西服/套装"
          },
          {
            "name": "女装衬衫"
          },
          {
            "name": "女装毛衣/卫衣"
          },
          {
            "name": "女装睡衣/家居服"
          },
          {
            "name": "女装丝袜/袜子"
          },
          {
            "name": "女装-其他"
          }
        ]
      },
      ...

Python
from botocore.config import Config
import json
import re

#role 方式初始化bedrock
config = Config(read_timeout=1000) # second
bedrock_runtime = boto3.client('bedrock-runtime', config=config)

def generate_message(bedrock_runtime, model_id, system_prompt, messages, max_tokens):

    body=json.dumps(
        {
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": max_tokens,
            "temperature": 0.1,
            "top_k": 20,
            "top_p": 0.9,
            "system": system_prompt,
            "messages": messages
        }
    )  
    
#     print(body)
    response = bedrock_runtime.invoke_model(body=body, modelId=model_id)
    response_body = json.loads(response.get('body').read())
   
    return response_body


def video_classify(content):

    # 打开文件并读取4级分类格式化好的JSON数据
    with open('category_no_id.json', 'r', encoding='utf-8') as f:
        data = json.load(f)

    # 将JSON数据转换为字符串,并去除所有空白字符
    category_json = re.sub(r'\s+', '', json.dumps(data, ensure_ascii=False,separators=(',', ':')))
   
    prompt=f'''
        <class>
        {category_json}
        </class>
        <instruction>
        1. 你的任务是将给定的广告视频描述<video_description>，准确分类到<class>预定义的1级，2级和3级类别中。
        2. 从上述<class>类别中选择最匹配的1级，2级和3级子类别，需要1，2，3级分类都需要给出，如果无法确定分类，回答"0"。
        3. 请回复完整分类，包括ID，不需要其他解释。
        4. 直接给出答案，无需前言。
        </instruction>
        <output_format>
        服装|女装|女装裙装
        </output_format>'''
    
    text_video = f''' 
        <video_description>
        {content}
        </video_description>
    '''

    # 使用prompt caching功能，将不变的分类列表caching住
    prompt_final = [{"type": "text","text": prompt, "cache_control": {"type": "ephemeral"}}, {"type": "text","text": text_video}]

    # model_id='anthropic.claude-3-5-sonnet-20240620-v1:0'
    model_id = "us.anthropic.claude-3-7-sonnet-20250219-v1:0"
    system_prompt = "你是一个专业的广告视频内容分类机器人,专门负责分析和分类广告视频。"
    max_tokens = 1000
    user_message =  {"role": "user", "content": prompt_final}
    messages = [user_message]
    # print(messages)
    response = generate_message(bedrock_runtime, model_id, system_prompt, messages, max_tokens)
    print(response)
    return response['content'][0]['text']