Service dev4

app/crud.py

@@ -120,6 +120,26 @@ 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
+    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
+
+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

app/models.py

@@ -12,8 +12,10 @@ 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)
 
     # 관계 설정
     summaries = relationship("MovieManagerSummary", back_populates="movie")

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}")

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만 추출

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
         )

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

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,11 @@ def get_marengo_response(user_message: str) -> str:
         body=json.dumps(message)
     )
 
-    return response.content
+
+
+    # response["body"]는 StreamingBody → .read() 필요
+    result = json.loads(response["body"].read())
+
+    embedding = result['data'][0]['embedding']
+
+    return embedding

app/services/moviemanager_service.py

@@ -4,11 +4,25 @@
 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.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,6 +334,77 @@ def collect_thumbnail_info(video_summaries: List[Dict], s3_video_uri: str = None
             "urls": []
         }
 
+async def get_final_scenes(custom_retrievals: List[str], movie_id: int) -> Dict[str, List[str]]:
+    """
+    커스텀 검색어들을 사용하여 최종 장면 검색 결과를 생성합니다.
+
+    1. 데이터베이스에서 movie에 임베딩 존재하는지 확인하기.
+    2. 임베딩 존재하면 s3 접근하여 임베딩 벡터 리스트 가져오기.
+    3. 가져온 벡터들 중 일정 이상 유사한 벡터 여러 개 선택.
+    4. 선택한 벡터들에 해당하는 S3 이미지(장면) URI 반환.
+
+    입력:
+        custom_retrievals: 커스텀 검색어 리스트
+    출력:
+        List[Dict]: 해당 검색어들에 대응되는 장면 이미지 URI 리스트
+    """
+
+    db = SessionLocal()
+    embedding_uri = get_embedding_uri(db, movie_id)  # movie에 저장된 임베딩 존재하는지 확인
+    db.close()
+    if not embedding_uri:
+        raise ValueError("해당 영화에 대한 임베딩이 존재하지 않습니다. 먼저 영화를 처리해야 합니다.")
+
+    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)])
+
+    # 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
+
+
+
+
 async def create_final_results(video_summaries: List[str], custom_prompts: List[str], characters_info: str) -> List[tuple]:
     """
     모든 비디오 요약을 종합하여 최종 요약을 생성합니다.
@@ -528,9 +613,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 +685,29 @@ 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)}개 로드 완료")
+
+        # 프롬프트가 너무 많다면 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("🎭 최종 프롬프트 응답 결과 생성 중...")   
+        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, movie_id)
+        # s3 uri들의 리스트의 딕셔너리 형태가 되어야 할 것.
+        print(f"✅ 최종 장면 검색 결과 생성 완료")
+        print(f"{final_scenes}")
+
         # 최종 요약도 데이터베이스에 저장 (모든 청크 다음 순서)
         print(f"💾 최종 요약 데이터베이스 저장 시작...")
         final_summary_id = total_chunks + 1  # 마지막 청크 다음 순서
@@ -628,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"]
         }
 
@@ -835,7 +933,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}개의 발화")