From c59e4a04e5667eb82aece0869dfb500d8b5d24b2 Mon Sep 17 00:00:00 2001 From: RyuSeongHo Date: Tue, 2 Dec 2025 16:39:56 +0900 Subject: [PATCH 1/3] feat: custom_retrievals row added --- app/crud.py | 12 +++++++++--- app/models.py | 1 + 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/app/crud.py b/app/crud.py index 70af968..8be1f89 100644 --- a/app/crud.py +++ b/app/crud.py @@ -120,6 +120,12 @@ def get_summaries_up_to(db: Session, movie_id: int, summary_id: int) -> List[Mov def get_custom_prompts(db: Session, movie_id: int) -> Optional[List[str]]: """영화의 커스텀 프롬프트들 조회""" - summaries = db.query(Movie).filter(Movie.id == movie_id).first() - if summaries and summaries.custom_prompts: - return summaries.custom_prompts \ No newline at end of file + movie = db.query(Movie).filter(Movie.id == movie_id).first() + if movie and movie.custom_prompts: + return movie.custom_prompts + +def get_custom_retrievals(db: Session, movie_id: int) -> Optional[List[str]]: + """영화의 커스텀 검색어들 조회""" + movie = db.query(Movie).filter(Movie.id == movie_id).first() + if movie and movie.custom_retrievals: + return movie.custom_retrievals \ No newline at end of file diff --git a/app/models.py b/app/models.py index 7b5577e..761d69e 100644 --- a/app/models.py +++ b/app/models.py @@ -14,6 +14,7 @@ class Movie(Base): updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) custom_prompts = Column(ARRAY(String), nullable=True) + custom_retrievals = Column(ARRAY(String), nullable=True) # 관계 설정 summaries = relationship("MovieManagerSummary", back_populates="movie") From 152b5a8d4a846e5de0c0424584c4ad693cc8c985 Mon Sep 17 00:00:00 2001 From: RyuSeongHo Date: Tue, 2 Dec 2025 19:56:21 +0900 Subject: [PATCH 2/3] feat: vector embedding and store to s3 --- app/routers/marengo.py | 12 ++- app/routers/pipeline.py | 4 +- app/routers/scene.py | 4 +- app/services/marengo_service.py | 36 ++++++-- app/services/moviemanager_service.py | 50 ++++++++++-- app/services/scene_service.py | 118 +++++++++++++++++++++++++-- 6 files changed, 192 insertions(+), 32 deletions(-) diff --git a/app/routers/marengo.py b/app/routers/marengo.py index 3b9c556..1cff7c3 100644 --- a/app/routers/marengo.py +++ b/app/routers/marengo.py @@ -1,7 +1,7 @@ # app/routers/marengo.py from fastapi import APIRouter, HTTPException -from app.services.marengo_service import init_marengo_client, get_marengo_response +from app.services.marengo_service import init_marengo_client, embed_marengo from app.schemas import ChatRequest, ChatResponse router = APIRouter(prefix="/marengo", tags=["marengo"]) @@ -15,11 +15,9 @@ def chat_endpoint(req: ChatRequest): 사용자가 보낸 메시지를 Marengo API에 전달 후, 임베딩 벡터를 반환합니다. """ try: - result = get_marengo_response(req.message) - print("Raw response from Marengo:", result) - # If the result is a list of TextBlock, concatenate their text fields - if isinstance(result, list): - result = "".join(block.text for block in result) - return ChatResponse(response=result) + result = embed_marengo("text", req.message) + # convert List[float] to ChatResponse + result_str = ','.join(map(str, result)) + return ChatResponse(response=result_str) except Exception as e: raise HTTPException(status_code=500, detail=f"Marengo API 호출 오류: {e}") \ No newline at end of file diff --git a/app/routers/pipeline.py b/app/routers/pipeline.py index 50eae99..aa23301 100644 --- a/app/routers/pipeline.py +++ b/app/routers/pipeline.py @@ -1,6 +1,6 @@ from fastapi import APIRouter, HTTPException from app.services.transcribe_service import transcribe_video -from app.services.scene_service import get_video_scenes +from app.services.scene_service import scene_process from app.services.summarize_service import summarize_content from app.schemas import PipelineRequest, SummarizeResponse import asyncio @@ -18,7 +18,7 @@ async def pipeline_endpoint(req: PipelineRequest): try: # 병렬 실행 transcribe_task = asyncio.to_thread(transcribe_video, req.s3_video_uri, req.language_code) - scene_task = asyncio.to_thread(get_video_scenes, req.s3_video_uri, req.threshold) + scene_task = asyncio.to_thread(scene_process, req.s3_video_uri, req.threshold) utterances, scenes = await asyncio.gather(transcribe_task, scene_task) # scene의 base64 이미지와 start_time만 추출 diff --git a/app/routers/scene.py b/app/routers/scene.py index 078e9f4..6fd8b98 100644 --- a/app/routers/scene.py +++ b/app/routers/scene.py @@ -1,5 +1,5 @@ from fastapi import APIRouter, HTTPException -from app.services.scene_service import get_video_scenes +from app.services.scene_service import scene_process from app.schemas import SceneRequest, SceneResponse from typing import List @@ -14,7 +14,7 @@ def detect_scenes_endpoint(req: SceneRequest): raise HTTPException(status_code=400, detail="s3_video_uri는 's3://'로 시작해야 합니다.") try: - scenes = get_video_scenes( + scenes = scene_process( s3_uri=req.s3_video_uri, threshold=req.threshold ) diff --git a/app/services/marengo_service.py b/app/services/marengo_service.py index cad7a7d..058d607 100644 --- a/app/services/marengo_service.py +++ b/app/services/marengo_service.py @@ -1,6 +1,7 @@ # app/services/marengo_service.py from dotenv import load_dotenv +from typing import List import boto3 import os import json @@ -33,15 +34,27 @@ def init_marengo_client(): aws_access_key_id=aws_key, aws_secret_access_key=aws_secret) -def get_marengo_response(user_message: str) -> str: +def embed_marengo(input_type: str, input: str) -> List[float]: """ - Bedrock Marengo API를 호출하여 텍스트 응답을 반환합니다. + Bedrock Marengo API를 호출하여 임베딩 벡터를 반환합니다. """ - message = { - "inputType": "text", - "inputText": user_message - } + if input_type not in ["text", "image"]: + raise ValueError("input_type은 'text' 또는 'image'여야 합니다.") + + if input_type == "text": + message = { + "inputType": input_type, + "inputText": input + } + + if input_type == "image": + message = { + "inputType": input_type, + "mediaSource": { + "base64String": input + } + } if marengo_client is None: raise RuntimeError("Marengo Bedrock 클라이언트가 초기화되지 않았습니다.") @@ -50,4 +63,13 @@ def get_marengo_response(user_message: str) -> str: body=json.dumps(message) ) - return response.content \ No newline at end of file + + + # response["body"]는 StreamingBody → .read() 필요 + result = json.loads(response["body"].read()) + + embedding = result['data'][0]['embedding'] + + print(type(embedding), len(embedding)) + + return embedding \ No newline at end of file diff --git a/app/services/moviemanager_service.py b/app/services/moviemanager_service.py index 117c7f6..077e9f0 100644 --- a/app/services/moviemanager_service.py +++ b/app/services/moviemanager_service.py @@ -4,9 +4,9 @@ import re from typing import List, Dict from app.services.transcribe_service import transcribe_video -from app.services.scene_service import get_video_scenes +from app.services.scene_service import scene_process, download_json_from_s3 from app.services.video_chunk_service import generate_video_chunks_info, extract_chunk_for_processing, cleanup_chunk_file -from app.crud import create_or_update_summary, get_summaries_up_to, delete_summaries_from, update_movie_status, mark_movie_failed, get_resume_info, get_movie, get_custom_prompts +from app.crud import create_or_update_summary, get_summaries_up_to, delete_summaries_from, update_movie_status, mark_movie_failed, get_resume_info, get_movie, get_custom_prompts, get_custom_retrievals from app.database import SessionLocal import asyncio @@ -320,6 +320,34 @@ def collect_thumbnail_info(video_summaries: List[Dict], s3_video_uri: str = None "urls": [] } +async def get_final_scenes(custom_retrievals: List[str]) -> List[Dict]: + """ + 커스텀 검색어들을 사용하여 최종 장면 검색 결과를 생성합니다. + + 1. 데이터베이스에서 movie에 임베딩 존재하는지 확인하기. + 2. 임베딩 존재하면 s3 접근하여 임베딩 벡터 리스트 가져오기. + 3. 가져온 벡터들 중 일정 이상 유사한 벡터 여러 개 선택. + 4. 선택한 벡터들에 해당하는 S3 이미지(장면) URI 반환. + """ + + pass + + # db = SessionLocal() + # embedding_uri = get_embedding_uri(db, movie_id, 1) # movie에 저장된 임베딩 존재하는지 확인 + # db.close() + # if not embedding_uri: + # raise ValueError("해당 영화에 대한 임베딩이 존재하지 않습니다. 먼저 영화를 처리해야 합니다.") + + # print(f"📊 임베딩 URI: {embedding_uri}") + + # # S3에서 임베딩 벡터 리스트 가져오기 + # embeddings = download_json_from_s3(embedding_uri) + + + + + + async def create_final_results(video_summaries: List[str], custom_prompts: List[str], characters_info: str) -> List[tuple]: """ 모든 비디오 요약을 종합하여 최종 요약을 생성합니다. @@ -528,9 +556,9 @@ async def process_single_video(s3_video_uri: str, characters_info: str, movie_id # 청크를 임시 S3에 업로드하지 않고 로컬 파일 URI로 처리 chunk_uri = f"file://{chunk_file_path}" - # transcribe와 scene 병렬 처리 (movie_id와 original_uri 전달) + # transcribe process와 scene process 병렬 처리 transcribe_task = asyncio.to_thread(transcribe_video, chunk_uri, language_code) - scene_task = asyncio.to_thread(get_video_scenes, chunk_uri, threshold, movie_id, s3_video_uri) + scene_task = asyncio.to_thread(scene_process, chunk_uri, threshold, movie_id, s3_video_uri) utterances, scenes = await asyncio.gather(transcribe_task, scene_task) print(f"✅ STT 결과: {len(utterances) if utterances else 0}개의 발화") @@ -600,14 +628,20 @@ async def process_single_video(s3_video_uri: str, characters_info: str, movie_id # 커스텀 프롬프트 가져오기 db = SessionLocal() custom_prompts = get_custom_prompts(db, movie_id) + custom_retrievals = get_custom_retrievals(db, movie_id) db.close() - print(f"프롬프트 {len(custom_prompts)}개 로드 완료 for 최종 요약 생성") + print(f"프롬프트 {len(custom_prompts)}개, 검색어 {len(custom_retrievals)}개 로드 완료") - print("🎭 최종 프롬프트 응답 결과 생성 중...") + print("🎭 최종 프롬프트 응답 결과 생성 중...") # 최종 프롬프트 응답 결과 생성 final_summary = await create_final_results([vs["summary"] for vs in video_summaries], custom_prompts, characters_info) print(f"✅ 최종 요약 생성 완료 (길이: {len(final_summary)} 문자)") - + + # 최종 장면 검색 결과 생성 (아래 함수는 위와 다르게 직접 db 조회를 통해 정보에 접근한다.) + #final_scenes = await get_final_scenes(custom_retrievals) + # s3 uri들의 리스트의 딕셔너리 형태가 되어야 할 것. + print(f"✅ 최종 장면 검색 결과 생성 완료 문자)") + # 최종 요약도 데이터베이스에 저장 (모든 청크 다음 순서) print(f"💾 최종 요약 데이터베이스 저장 시작...") final_summary_id = total_chunks + 1 # 마지막 청크 다음 순서 @@ -835,7 +869,7 @@ async def process_videos_from_folder(s3_folder_path: str, characters_info: str, # transcribe와 scene 병렬 처리 (movie_id 전달) transcribe_task = asyncio.to_thread(transcribe_video, video_uri, language_code) - scene_task = asyncio.to_thread(get_video_scenes, video_uri, threshold, movie_id) + scene_task = asyncio.to_thread(scene_process, video_uri, threshold, movie_id) utterances, scenes = await asyncio.gather(transcribe_task, scene_task) print(f"✅ STT 결과: {len(utterances) if utterances else 0}개의 발화") diff --git a/app/services/scene_service.py b/app/services/scene_service.py index 8959d8c..120f30e 100644 --- a/app/services/scene_service.py +++ b/app/services/scene_service.py @@ -4,9 +4,11 @@ from typing import List, Dict import cv2 from scenedetect import detect, ContentDetector +from app.services.marengo_service import embed_marengo import numpy as np import base64 import uuid +import json def get_output_bucket() -> str: """ @@ -42,6 +44,28 @@ def download_video_from_s3(s3_uri: str) -> str: # 임시 파일 삭제 os.unlink(temp_file.name) raise e + +def download_json_from_s3(s3_uri: str) -> Dict: + """ + S3에서 JSON 파일을 다운로드하여 파싱합니다. + """ + if not s3_uri.startswith("s3://"): + raise ValueError("s3_uri는 's3://'로 시작해야 합니다.") + + # S3 URI 파싱 + bucket = s3_uri.split('/')[2] + key = '/'.join(s3_uri.split('/')[3:]) + + # S3 클라이언트 생성 + s3 = boto3.client('s3') + + try: + # S3에서 JSON 파일 다운로드 + response = s3.get_object(Bucket=bucket, Key=key) + json_data = response['Body'].read().decode('utf-8') + return json.loads(json_data) + except Exception as e: + raise e def frame_to_base64(frame: np.ndarray) -> str: """ @@ -83,7 +107,7 @@ def save_frame_to_s3(frame: np.ndarray, prefix: str = "scenes") -> str: # 임시 파일 삭제 os.unlink(temp_file.name) -def detect_scenes(video_path: str, threshold: float = 30.0, max_scenes_count: int = 20, movie_id: int = None, original_uri: str = None) -> List[Dict]: +def detect_and_embed_scenes(video_path: str, threshold: float = 30.0, max_scenes_count: int = 20, movie_id: int = None, original_uri: str = None) -> List[Dict]: """ 비디오에서 주요 장면을 감지하고 각 장면의 대표 프레임을 base64로 반환합니다. 품질이 좋은 프레임은 S3 thumbnails/ 경로에도 저장합니다. @@ -132,6 +156,8 @@ def detect_scenes(video_path: str, threshold: float = 30.0, max_scenes_count: in selected_scenes.append(closest_scene) scenes = selected_scenes + embed_uri_pairs = {} + # 품질 검사 및 S3 저장 (최대 20개 장면에 대해서만 수행) if movie_id is not None: print(f"🔍 최대 {len(scenes)}개 장면에 대해 품질 검사 수행...") @@ -149,8 +175,16 @@ def detect_scenes(video_path: str, threshold: float = 30.0, max_scenes_count: in print(f" 선명도: {quality_check['sharpness']:.1f} ({'✅' if quality_check['sharpness_ok'] else '❌'})") if quality_check['is_good_quality']: + # scene retrieval 과정 수행 필요 + # marengo_service에서 aws bedrock marengo embed model 호출하여 임베딩을 받아오는 함수 사용 + # 임베딩을 썸네일과 함께 S3에 저장, DB에 메타데이터 저장. + thumbnail_url = save_thumbnail_to_s3(frame, movie_id, video_name, scene_index + 1, original_uri) scene_data['thumbnail_url'] = thumbnail_url + + embedded_vector = embed_marengo("image", scene_data["frame_image"]) + embed_uri_pairs[thumbnail_url] = embedded_vector + print(f"✅ Scene {scene_index + 1}: 품질 양호 → S3 저장 완료") else: print(f"⚠️ Scene {scene_index + 1}: 품질 부족 → S3 저장 생략") @@ -158,12 +192,18 @@ def detect_scenes(video_path: str, threshold: float = 30.0, max_scenes_count: in except Exception as e: print(f"❌ Scene {scene_index + 1} 처리 중 오류: {str(e)}") + if embed_uri_pairs: + save_json_to_s3(embed_uri_pairs, movie_id, video_name, original_uri=original_uri) + print(f"✅ 총 {len(embed_uri_pairs)}개 장면 임베딩 완료 및 S3 저장 완료.") + return scenes -def get_video_scenes(uri: str, threshold: float = 30.0, movie_id: int = None, original_uri: str = None) -> List[Dict]: +def scene_process(uri: str, threshold: float = 30.0, movie_id: int = None, original_uri: str = None) -> List[Dict]: """ - 비디오의 주요 장면을 감지하고 각 장면의 대표 프레임을 base64로 반환합니다. - 품질이 좋은 프레임은 S3 thumbnails/ 경로에도 저장합니다. + 전체 장면 처리 프로세스입니다. 다음과 같은 과정을 거칩니다. + 1. 해당 비디오를 청크로 분할합니다. + 2. 분할한 비디오 청크에서 pySceneDetect를 사용하여 장면을 감지합니다. + 3. 우수한 장면은 S3에 저장되며, marengo를 통한 임베딩 역시 수행 후 저장됩니다. (임베딩 메타데이터는 DB에 저장) Args: uri: S3 URI (s3://) 또는 로컬 파일 URI (file://) @@ -198,8 +238,8 @@ def get_video_scenes(uri: str, threshold: float = 30.0, movie_id: int = None, or raise ValueError("URI는 's3://' 또는 'file://'로 시작해야 합니다.") try: - # 장면 감지 (movie_id와 original_uri 전달) - scenes = detect_scenes(video_path, threshold, movie_id=movie_id, original_uri=original_uri) + # 다운로드받은 영상 장면 감지 직후 임베딩 + scenes = detect_and_embed_scenes(video_path, threshold, movie_id=movie_id, original_uri=original_uri) return scenes finally: # 임시 파일 삭제 (S3에서 다운로드한 경우만) @@ -244,6 +284,72 @@ def check_frame_quality(frame: np.ndarray) -> Dict[str, float]: "is_good_quality": is_good_quality } +def save_json_to_s3(dict_data: dict, movie_id: int, video_name: str, original_uri: str = None) -> str: + """ + uri-임베딩 쌍을 원본 비디오와 같은 디렉토리의 embeddings/ 폴더에 저장합니다. + + Args: + dict_data: 저장할 JSON 데이터 + movie_id: 영화 ID + video_name: 비디오 파일명 + original_uri: 원본 비디오 URI (디렉토리 구조 유지용) + + Returns: + str: S3 URL + """ + # S3 클라이언트 생성 + s3 = boto3.client('s3') + + # 출력 버킷 가져오기 + output_bucket = get_output_bucket() + + # JSON 데이터를 문자열로 변환 + json_data = json.dumps(dict_data) + + try: + # 썸네일 저장 경로 결정 + if original_uri and original_uri.startswith("s3://"): + # 원본 비디오 URI에서 디렉토리 구조 추출 + # 예: s3://bucket/movies/series1/episode1.mp4 → movies/series1/embeddings/ + uri_parts = original_uri.replace("s3://", "").split("/") + bucket_from_uri = uri_parts[0] + + if len(uri_parts) > 1: + # 디렉토리 부분 추출 (파일명 제외) + directory_path = "/".join(uri_parts[1:-1]) + if directory_path: + # 같은 디렉토리에 embeddings 폴더 생성 + embeddings_dir = f"{directory_path}/embeddings" + else: + # 루트 디렉토리인 경우 + embeddings_dir = "embeddings" + else: + # 버킷 루트인 경우 + embeddings_dir = "embeddings" + else: + # original_uri가 없거나 S3 URI가 아닌 경우 기본 경로 사용 + embeddings_dir = f"embeddings/{movie_id}" + + # 파일명 생성 + filename = "embeddings.json" + + # 최종 S3 키 생성 + key = f"{embeddings_dir}/{filename}" + + # S3에 업로드 + s3.put_object(Body=json_data, Bucket=output_bucket, Key=key, ContentType='application/json') + + # 공개 URL 생성 (또는 presigned URL) + url = f"https://{output_bucket}.s3.amazonaws.com/{key}" + + print(f"✅ 임베딩 저장 완료: {url}") + print(f" 경로: {key}") + return url + + except Exception as e: + print(f"❌ 임베딩 저장 실패: {str(e)}") + raise e + def save_thumbnail_to_s3(frame: np.ndarray, movie_id: int, video_name: str, scene_index: int, original_uri: str = None) -> str: """ 썸네일 후보 프레임을 원본 비디오와 같은 디렉토리의 thumbnails/ 폴더에 저장합니다. From c07a126ba7d3f13c8fc5ff44f8623ab0126df72e Mon Sep 17 00:00:00 2001 From: RyuSeongHo Date: Wed, 3 Dec 2025 12:27:13 +0900 Subject: [PATCH 3/3] feat: scene retrieval for moviemanager --- app/crud.py | 16 ++++- app/models.py | 1 + app/schemas.py | 1 + app/services/marengo_service.py | 2 - app/services/moviemanager_service.py | 100 ++++++++++++++++++++++----- 5 files changed, 99 insertions(+), 21 deletions(-) diff --git a/app/crud.py b/app/crud.py index 8be1f89..7affdef 100644 --- a/app/crud.py +++ b/app/crud.py @@ -128,4 +128,18 @@ def get_custom_retrievals(db: Session, movie_id: int) -> Optional[List[str]]: """영화의 커스텀 검색어들 조회""" movie = db.query(Movie).filter(Movie.id == movie_id).first() if movie and movie.custom_retrievals: - return movie.custom_retrievals \ No newline at end of file + return movie.custom_retrievals + +def get_embedding_uri(db: Session, movie_id: int) -> Optional[str]: + """영화의 임베딩 S3 URI 조회""" + movie = db.query(Movie).filter(Movie.id == movie_id).first() + if movie: + return movie.embedding_uri + +def set_embedding_uri(db: Session, movie_id: int, embedding_uri: str) -> Optional[str]: + """영화의 임베딩 S3 URI 설정""" + movie = db.query(Movie).filter(Movie.id == movie_id).first() + if movie: + movie.embedding_uri = embedding_uri + db.commit() + return movie.embedding_uri \ No newline at end of file diff --git a/app/models.py b/app/models.py index 761d69e..765e913 100644 --- a/app/models.py +++ b/app/models.py @@ -12,6 +12,7 @@ class Movie(Base): status = Column(String(50), default="PENDING") # PENDING, PROCEEDING[N/M], ORGANIZING, COMPLETE, FAILED_* created_at = Column(DateTime(timezone=True), server_default=func.now()) updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) + embedding_uri = Column(String(255), nullable=True) # S3에 저장된 임베딩 벡터 URI custom_prompts = Column(ARRAY(String), nullable=True) custom_retrievals = Column(ARRAY(String), nullable=True) diff --git a/app/schemas.py b/app/schemas.py index d84246f..f19fafb 100644 --- a/app/schemas.py +++ b/app/schemas.py @@ -91,4 +91,5 @@ class MovieManagerRequest(BaseModel): class MovieManagerResponse(BaseModel): prompt2results: List[tuple] + retrieval2uris: dict[str, List[str]] thumbnail_folder_uri: str = None # 썸네일 후보 폴더 URI \ No newline at end of file diff --git a/app/services/marengo_service.py b/app/services/marengo_service.py index 058d607..08164e4 100644 --- a/app/services/marengo_service.py +++ b/app/services/marengo_service.py @@ -69,7 +69,5 @@ def embed_marengo(input_type: str, input: str) -> List[float]: result = json.loads(response["body"].read()) embedding = result['data'][0]['embedding'] - - print(type(embedding), len(embedding)) return embedding \ No newline at end of file diff --git a/app/services/moviemanager_service.py b/app/services/moviemanager_service.py index 077e9f0..a14bb1b 100644 --- a/app/services/moviemanager_service.py +++ b/app/services/moviemanager_service.py @@ -6,9 +6,23 @@ from app.services.transcribe_service import transcribe_video from app.services.scene_service import scene_process, download_json_from_s3 from app.services.video_chunk_service import generate_video_chunks_info, extract_chunk_for_processing, cleanup_chunk_file -from app.crud import create_or_update_summary, get_summaries_up_to, delete_summaries_from, update_movie_status, mark_movie_failed, get_resume_info, get_movie, get_custom_prompts, get_custom_retrievals +from app.services.marengo_service import embed_marengo +from app.crud import ( + create_or_update_summary, + get_summaries_up_to, + delete_summaries_from, + update_movie_status, + mark_movie_failed, + get_resume_info, + get_movie, + get_custom_prompts, + get_custom_retrievals, + get_embedding_uri, + set_embedding_uri +) from app.database import SessionLocal import asyncio +import numpy as np def load_prompts() -> Dict[str, str]: """ @@ -320,7 +334,7 @@ def collect_thumbnail_info(video_summaries: List[Dict], s3_video_uri: str = None "urls": [] } -async def get_final_scenes(custom_retrievals: List[str]) -> List[Dict]: +async def get_final_scenes(custom_retrievals: List[str], movie_id: int) -> Dict[str, List[str]]: """ 커스텀 검색어들을 사용하여 최종 장면 검색 결과를 생성합니다. @@ -328,22 +342,65 @@ async def get_final_scenes(custom_retrievals: List[str]) -> List[Dict]: 2. 임베딩 존재하면 s3 접근하여 임베딩 벡터 리스트 가져오기. 3. 가져온 벡터들 중 일정 이상 유사한 벡터 여러 개 선택. 4. 선택한 벡터들에 해당하는 S3 이미지(장면) URI 반환. - """ - pass + 입력: + custom_retrievals: 커스텀 검색어 리스트 + 출력: + List[Dict]: 해당 검색어들에 대응되는 장면 이미지 URI 리스트 + """ - # db = SessionLocal() - # embedding_uri = get_embedding_uri(db, movie_id, 1) # movie에 저장된 임베딩 존재하는지 확인 - # db.close() - # if not embedding_uri: - # raise ValueError("해당 영화에 대한 임베딩이 존재하지 않습니다. 먼저 영화를 처리해야 합니다.") + db = SessionLocal() + embedding_uri = get_embedding_uri(db, movie_id) # movie에 저장된 임베딩 존재하는지 확인 + db.close() + if not embedding_uri: + raise ValueError("해당 영화에 대한 임베딩이 존재하지 않습니다. 먼저 영화를 처리해야 합니다.") - # print(f"📊 임베딩 URI: {embedding_uri}") + print(f"📊 임베딩 URI: {embedding_uri}") + + # S3에서 임베딩 벡터 딕셔너리(JSON) 다운로드 + uri2embedding_dict = download_json_from_s3(embedding_uri) + print(f"✅ S3에서 임베딩 벡터 데이터 다운로드 완료 (총 {len(uri2embedding_dict)}개 항목)") + + # 순서를 유지하며 uri와 embedding을 각각의 리스트로 분리 (Python 3.7+ dict는 유지 보장) + uri = list(uri2embedding_dict.keys()) + scene_feat = list(uri2embedding_dict.values()) + + # 임베딩은 행렬로 변환 + scene_feat_matrix = np.array(scene_feat) + print(f"📊 임베딩 매트릭스 형태: {scene_feat_matrix.shape} (샘플 개수 x 임베딩 차원)") + + retrieval_feat_matrix = np.empty((0, scene_feat_matrix.shape[1])) # 빈 행렬 초기화 + + # 각 입력 retrieval에 대해, + for retrieval in custom_retrievals: + # Bedrock Claude Embedding API 호출하여 임베딩 벡터 획득 + text_vector = embed_marengo("text", retrieval) + print(f"✅ 검색어 임베딩 벡터 획득 완료: '{retrieval}'") + retrieval_feat_matrix = np.vstack([retrieval_feat_matrix, np.array(text_vector)]) - # # S3에서 임베딩 벡터 리스트 가져오기 - # embeddings = download_json_from_s3(embedding_uri) + # marengo 자체는 거의 유사한 크기의 정규화된 벡터를 반환하지만, + # 이후 파이프라인 안전성을 위하여 명시적으로 정규화를 수행한다. + print("📊 임베딩 벡터 정규화 중...") + print(scene_feat_matrix.shape, retrieval_feat_matrix.shape) + scene_feat_matrix = scene_feat_matrix / np.linalg.norm(scene_feat_matrix, axis=1, keepdims=True) + retrieval_feat_matrix = retrieval_feat_matrix / np.linalg.norm(retrieval_feat_matrix, axis=1, keepdims=True) + # 하나만 뽑아서 크기 출력 + print(f"📊 정규화된 임베딩 벡터의 l2 norm: {np.linalg.norm(scene_feat_matrix[0])}, {np.linalg.norm(retrieval_feat_matrix[0])}") + # 코사인 유사도 계산 + similarity_matrix = np.dot(retrieval_feat_matrix, scene_feat_matrix.T) + print(f"📊 유사도 매트릭스 형태: {similarity_matrix.shape}") + print(similarity_matrix) + + # top-k 유사한 장면 선택 + top_k = 3 + result = {} + for i, retrieval in enumerate(custom_retrievals): + idx_k = np.argsort(-similarity_matrix[i])[:top_k] + result[retrieval] = [uri[idx] for idx in idx_k] + + return result @@ -631,16 +688,25 @@ async def process_single_video(s3_video_uri: str, characters_info: str, movie_id custom_retrievals = get_custom_retrievals(db, movie_id) db.close() print(f"프롬프트 {len(custom_prompts)}개, 검색어 {len(custom_retrievals)}개 로드 완료") + + # 프롬프트가 너무 많다면 10개로 제한 + if len(custom_prompts) > 10: + custom_prompts = custom_prompts[:10] + print(f"⚠️ 프롬프트 개수가 너무 많아 10개로 제한합니다.") + if len(custom_retrievals) > 10: + custom_retrievals = custom_retrievals[:10] + print(f"⚠️ 검색어 개수가 너무 많아 10개로 제한합니다.") print("🎭 최종 프롬프트 응답 결과 생성 중...") # 최종 프롬프트 응답 결과 생성 final_summary = await create_final_results([vs["summary"] for vs in video_summaries], custom_prompts, characters_info) - print(f"✅ 최종 요약 생성 완료 (길이: {len(final_summary)} 문자)") + print(f"✅ 최종 요약 생성 완료") # 최종 장면 검색 결과 생성 (아래 함수는 위와 다르게 직접 db 조회를 통해 정보에 접근한다.) - #final_scenes = await get_final_scenes(custom_retrievals) + final_scenes = await get_final_scenes(custom_retrievals, movie_id) # s3 uri들의 리스트의 딕셔너리 형태가 되어야 할 것. - print(f"✅ 최종 장면 검색 결과 생성 완료 문자)") + print(f"✅ 최종 장면 검색 결과 생성 완료") + print(f"{final_scenes}") # 최종 요약도 데이터베이스에 저장 (모든 청크 다음 순서) print(f"💾 최종 요약 데이터베이스 저장 시작...") @@ -662,14 +728,12 @@ async def process_single_video(s3_video_uri: str, characters_info: str, movie_id print("🎉 모든 청크 처리 완료!") print("=" * 80) - # 최종 요약을 줄거리와 평론으로 분리 (이제 필요 없다.) - # parsed_summary = parse_final_summary(final_summary) - # 썸네일 정보 수집 thumbnail_info = collect_thumbnail_info(video_summaries, s3_video_uri) return { "prompt2results": final_summary, + "retrieval2uris": final_scenes, "thumbnail_folder_uri": thumbnail_info["folder_uri"] }