利用 HAQM Bedrock Data Automation（BDA）对视频数据进行自动化处理与检索

# invokes bda by async approach with a given input file
def invoke_insight_generation_async(
        input_s3_uri,
        output_s3_uri,
        data_project_arn, blueprints = None):

    # Extract account ID from project ARN
    account_id = data_project_arn.split(':')[4]
    
    # Construct default Profile ARN using fixed us-west-2 region，which currently supports the 2 models required in the following process
    profile_arn = f"arn:aws:bedrock:us-west-2:{account_id}:data-automation-profile/us.data-automation-v1"

    payload = {
        "inputConfiguration": {
            "s3Uri": input_s3_uri
        },
        "outputConfiguration": {
            "s3Uri": output_s3_uri
        },
        "dataAutomationConfiguration": {
            "dataAutomationProjectArn": data_project_arn,
            "stage": "LIVE"
        },
        "dataAutomationProfileArn": profile_arn,
        "notificationConfiguration": {
            "eventBridgeConfiguration": {"eventBridgeEnabled": True},
        }
    }

    try:
        response = bda_client_runtime.invoke_data_automation_async(**payload)
        print(f"Successfully invoked data automation: {response}")
        return response
    except Exception as e:
        print(f"Error invoking data automation: {str(e)}")
        raise

try:
        response = bedrock_client.invoke_model(
            modelId="amazon.titan-embed-text-v2:0",
            contentType="application/json",
            accept="application/json",
            body=json.dumps({
                "inputText": text
            })
        )

def split_transcript_into_chunks(self, transcript_text, max_chunk_size=500, min_chunk_size=100):
        if not transcript_text:
            return []

        # 如果输入是字典，尝试提取文本
        if isinstance(transcript_text, dict):
            if 'text' in transcript_text:
                transcript_text = transcript_text['text']
            elif 'representation' in transcript_text and 'text' in transcript_text['representation']:
                transcript_text = transcript_text['representation']['text']
            else:
                transcript_text = json.dumps(transcript_text)

        # 确保输入是字符串
        if not isinstance(transcript_text, str):
            transcript_text = str(transcript_text)

        # 按句子分割（句号、问号、感叹号后跟空格）
        # 使用更宽松的模式，允许句子结束后有任何数量的空格
        sentences = re.split(r'([.!?])\s*', transcript_text)

        # 处理分割结果，将标点符号重新附加到句子
        processed_sentences = []
        i = 0
        while i < len(sentences):
            sentence = sentences[i]

            # 如果是标点符号，附加到前一个句子
            if i > 0 and sentence in ['.', '!', '?']:
                processed_sentences[-1] += sentence
            else:
                processed_sentences.append(sentence)

            i += 1

        # 将句子组合成块
        chunks = []
        current_chunk = ""

        for sentence in processed_sentences:
            # 跳过空句子
            if not sentence.strip():
                continue

            # 如果当前块加上新句子会超过最大大小，保存当前块
            if current_chunk and len(current_chunk) + len(sentence) + 1 > max_chunk_size:
                if len(current_chunk) >= min_chunk_size:
                    chunks.append(current_chunk.strip())
                current_chunk = sentence
            else:
                if current_chunk:
                    current_chunk += " " + sentence
                else:
                    current_chunk = sentence

        # 添加最后一个块
        if current_chunk and len(current_chunk) >= min_chunk_size:
            chunks.append(current_chunk.strip())

        return chunks

def flatten_video_data(self, video_data, video_name):
        flattened_data = []
        
        # 处理章节
        chapters = video_data.get('chapters', [])
        for chapter in chapters:
            chapter_summary = chapter.get('chapter_summary', {})
            flattened_chapter_summary = {
                "video_name": video_name,
                "source": f"chapter_{chapter.get('chapter_index', 0)}_summary",
                "text": chapter_summary.get('text', ""),
                "embedding": chapter_summary.get('embedding', []),
                "start_timestamp_millis": chapter.get('start_timestamp_millis'),
                "end_timestamp_millis": chapter.get('end_timestamp_millis')
            }
            flattened_data.append(flattened_chapter_summary)

            # 处理章节转录块
            transcript_chunks = chapter.get('transcript_chunks', [])
            for chunk in transcript_chunks:
                flattened_chunk = {
                    "video_name": video_name,
                    "source": f"chapter_{chapter.get('chapter_index', 0)}_transcript_chunk_{chunk.get('chunk_index', 0)}",
                    "text": chunk.get('text', ""),
                    "embedding": chunk.get('embedding', []),
                    "start_timestamp_millis": chapter.get('start_timestamp_millis'),
                    "end_timestamp_millis": chapter.get('end_timestamp_millis')
                }
                flattened_data.append(flattened_chunk)

        return flattened_data

flowchart LR

  A([接收搜索请求])-->B[解析查询参数]
  B-->C{执行混合搜索}
  C-->D[向量搜索]
  C-->E[文本搜索]
  D-->F([合并搜索结果])

  E-->F
  F-->G([使用Cohere重排序])
  G-->H([返回最终结果])

text: "<sample_text>"
video_name: "<sample_name>.mov" 
source: "chapter_6_summary"
start_timestamp_millis: 181666
end_timestamp_millis: 227100

pipeline = [
                {
                    "$search": {
                        "vectorSearch": {
                            "vector": query_embedding,
                            "path": "embedding",
                            "similarity": "cosine",
                            "k": top_k * 3,  # Fetch more results since we'll filter them
                            "efSearch": 64
                        }
                    }
                },
                # Add a $match stage to filter results based on search mode (scene/transcripts)
                {
                    "$match": filter_condition
                },
                # Limit to top_k results after filtering
                {
                    "$limit": top_k
                },
                {
                    "$project": {
                        "text": 1,
                        "video_name": 1,
                        "source": 1,
                        "start_timestamp_millis": 1,
                        "end_timestamp_millis": 1
                    }
                }
            ] 

all_results = processed_vector_results + processed_text_results
            # Remove duplicates based on '_id' and prefer text search results
            unique_results = []
            seen_ids = set()
            for result in reversed(all_results):  # Reverse the list to process text results first
                result_id = result['_id']
                if result_id not in seen_ids:
                    seen_ids.add(result_id)
                    unique_results.append(result)
            unique_results.reverse()  # Reverse the list back to original order
            # Return unique results directly
            return {"results": unique_results}

response = self.bedrock_client.invoke_model(
            modelId="cohere.rerank-v3-5:0",
            contentType="application/json",
            accept="application/json",
            body=json.dumps({
                "api_version": 2,
                "query": query,
                "documents": documents,
                "top_n": len(documents)
            })
        )