Feature/video2 cam

App/routes/explain.py

@@ -127,7 +127,7 @@ async def explain_frame(
         )
 
     # 비디오 내 해당 frame이 몇초에 위치한 frame인지 확인
-    frame_time = video_svc.get_video_frame_by_index(conn, video_id, frame_index)
+    frame_time = await video_svc.get_video_frame_by_index(conn, video_id, frame_index)
 
     # Celery Task 호출(Redis Broker 활용)
     task = explain_svc.process_explain_frame_task.delay(

App/services/auth_svc.py

@@ -1,4 +1,3 @@
-from datetime import datetime, timedelta, timezone
 from fastapi import status
 from fastapi.exceptions import HTTPException
 from passlib.context import CryptContext
@@ -9,14 +8,43 @@
 # 비밀번호 암호화 및 검증
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
-def get_hashed_password(password: str):
+def get_hashed_password(password: str) -> str:
+    """평문 비밀번호를 bcrypt 알고리즘으로 해싱한다.
+
+    Args:
+        password (str): 사용자가 입력한 평문 비밀번호.
+
+    Returns:
+        str: bcrypt 해싱된 비밀번호 문자열.
+    """
     return pwd_context.hash(password)
 
-def verify_password(plain_password: str, hashed_password: str):
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    """평문 비밀번호와 해시된 비밀번호의 일치 여부를 검증한다.
+
+    Args:
+        plain_password (str): 사용자가 입력한 평문 비밀번호.
+        hashed_password (str): DB에 저장된 bcrypt 해시 비밀번호.
+
+    Returns:
+        bool: 비밀번호 일치 여부.
+    """
     return pwd_context.verify(plain_password, hashed_password)
 
-# 회원가입 시, 이미 존재하는 이메일과 중복 확인
-async def get_user_by_email(conn: Connection, email: str):
+async def get_user_by_email(conn: Connection, email: str) -> UserData | None:
+    """이메일로 사용자를 조회해 중복 가입 여부를 확인한다.
+
+    Args:
+        conn (Connection): SQLAlchemy 비동기 DB 커넥션.
+        email (str): 조회할 이메일 주소.
+
+    Returns:
+        UserData | None: 이메일이 존재하면 UserData 반환, 없으면 None.
+
+    Raises:
+        HTTPException 503: DB 쿼리 중 SQLAlchemy 오류 발생 시.
+        HTTPException 500: 예기치 못한 예외 발생 시.
+    """
     try:
         query = f"""
         SELECT id, name, email from user
@@ -45,8 +73,18 @@ async def get_user_by_email(conn: Connection, email: str):
         raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                             detail="알수없는 이유로 서비스 오류가 발생하였습니다.")
 
-# 실제 DB에 회원 정보(이름, 이메일, 해싱된 패스워드)
-async def register_user(conn: Connection, name: str, email:str, hashed_password: str):
+async def register_user(conn: Connection, name: str, email:str, hashed_password: str) -> None:
+    """신규 사용자 정보를 DB에 저장한다.
+
+    Args:
+        conn (Connection): SQLAlchemy 비동기 DB 커넥션.
+        name (str): 사용자 이름.
+        email (str): 사용자 이메일 주소.
+        hashed_password (str): bcrypt 해시 처리된 비밀번호.
+
+    Raises:
+        HTTPException 400: DB INSERT 실패 또는 잘못된 요청 데이터.
+    """
     try:
         query = f"""
         INSERT INTO user(name, email, hashed_password)
@@ -64,9 +102,19 @@ async def register_user(conn: Connection, name: str, email:str, hashed_password:
         raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
                             detail="요청데이터가 제대로 전달되지 않았습니다")
 
-# 해당 이메일 가진 회원정보 가져오기
-async def authenticate_user(conn: Connection, email: str):
-    """로그인 시 이메일을 검증하여 일치하면 유저 정보를 반환합니다."""
+async def authenticate_user(conn: Connection, email: str) -> UserDataPASS | bool:
+    """로그인 시 이메일로 사용자를 조회해 인증 정보를 반환한다.
+
+    Args:
+        conn (Connection): SQLAlchemy 비동기 DB 커넥션.
+        email (str): 로그인 요청 이메일.
+
+    Returns:
+        UserDataPASS | bool: 사용자가 존재하면 UserDataPASS 반환, 없으면 False.
+
+    Raises:
+        HTTPException 400: DB 쿼리 중 SQLAlchemy 오류 발생 시.
+    """
     try:
         query = """
         SELECT id, name, email, hashed_password FROM user

App/services/explain_svc.py

@@ -1,7 +1,6 @@
 import cv2
 import asyncio
 import numpy as np
-from functools import partial
 from fastapi import status
 from fastapi.exceptions import HTTPException
 from sqlalchemy import Connection
@@ -13,22 +12,29 @@
 )
 from celery_app import celery_app
 
-# --- CAM Explainer Registry ---
-# LOW branch: 국소 위조 흔적 포착
-# HIGH branch: 전역 위조 흔적 포착 
+# LOW branch: 국소 위조 흔적 포착 / HIGH branch: 전역 위조 흔적 포착 
 EXPLAINER_REGISTRY = {
-    "hirescam":   HiResCAMExplainer, #LOW branch
-    "gradcamelementwise": GradCAMElementWiseExplainer, #LOW branch
-    "layercam":   LayerCAMExplainer, #LOW branch
+    "hirescam":   HiResCAMExplainer, # LOW branch
+    "gradcamelementwise": GradCAMElementWiseExplainer, # LOW branch
+    "layercam":   LayerCAMExplainer, # LOW branch
     "eigengradcam": EigenGradCAMExplainer, # HIGH branch
     "gradcamplusplus":   GradCAMPlusPlusExplainer, # HIGH branch
     "xgradcam":     XGradCAMExplainer, # HIGH branch
 }
 
 _explainer_cache: dict = {}
 
-# 비디오 내 특정 Frame 추출 + Face Cropping
-def _extract_face_from_frame(video_path: str, frame_time: float, explainer: CAMExplainer):
+def _extract_face_from_frame(video_path: str, frame_time: float, explainer: CAMExplainer) -> np.ndarray:
+    """비디오의 특정 시간대 프레임에서 얼굴 영역을 추출해 크롭된 배열을 반환한다.
+
+    Args:
+        video_path (str): 비디오 파일 경로.
+        frame_time (float): 프레임을 추출할 시간 (초 단위).
+        explainer (CAMExplainer): 얼굴 감지 및 크롭에 사용할 CAMExplainer 인스턴스.
+
+    Returns:
+        np.ndarray: 크롭된 얼굴 이미지 배열 (RGB).
+    """
     cap = cv2.VideoCapture(video_path)
     cap.set(cv2.CAP_PROP_POS_MSEC, frame_time * 1000)
     ret, frame = cap.read()
@@ -39,8 +45,22 @@ def _extract_face_from_frame(video_path: str, frame_time: float, explainer: CAME
 
     return explainer._crop_face(frame_rgb, bbox[:4])
 
-# 캐시된 CAMExplainer 객체 반환하거나 새로 생성 
-def _get_or_create_explainer(model_name: str, dataset: str, explain_req_dict: dict):
+def _get_or_create_explainer(model_name: str, dataset: str, explain_req_dict: dict) -> CAMExplainer:
+    """캐시에서 CAMExplainer를 조회하거나, 없으면 새로 생성해 캐싱 후 반환한다.
+
+    캐시 키는 (model_name, dataset, explainer_type, branch_level) 4개 파라미터로 구성된다.
+    추론 시점 파라미터(aug_smooth, threshold 등)는 키에서 제외해 메모리 사용을 최소화한다.
+
+    Args:
+        model_name (str): 추론에 사용할 모델 이름.
+        dataset (str): 모델 학습에 사용된 데이터셋 이름.
+        explain_req_dict (dict): 시각화 요청 파라미터 딕셔너리.
+            - explainer_type (str): EXPLAINER_REGISTRY 키값.
+            - branch_level (str): 모델 브랜치 레벨 ("LOW" | "HIGH").
+
+    Returns:
+        CAMExplainer: 캐시된 또는 새로 생성된 CAMExplainer 인스턴스.
+    """
     cache_key = (model_name, dataset, explain_req_dict["explainer_type"], explain_req_dict["branch_level"])
 
     if cache_key not in _explainer_cache:
@@ -51,8 +71,23 @@ def _get_or_create_explainer(model_name: str, dataset: str, explain_req_dict: di
         )
     return _explainer_cache[cache_key]
 
-# 시각화 이미지 생성 (heatmap, contour, bbox 선택)
 def _run_visualization(explainer: CAMExplainer, image_path: str, category: int, explain_req_dict: dict) -> np.ndarray:
+    """이미지 경로를 받아 display_type에 따라 시각화 이미지를 생성한다.
+
+    Args:
+        explainer (CAMExplainer): 시각화에 사용할 CAMExplainer 인스턴스.
+        image_path (str): 입력 이미지 파일 경로.
+        category (int): 타겟 클래스 인덱스 (위조: 1, 정상: 0).
+        explain_req_dict (dict): 시각화 요청 파라미터 딕셔너리.
+            - display_type (str): "heatmap" | "bbox" | "heatmap_bbox"
+            - overlay_ratio (float): 히트맵 오버레이 시 원본 이미지 가중치.
+            - threshold (float | str): CAM 이진화 임계값.
+            - aug_smooth (bool): Test-time augmentation 가동 여부.
+            - eigen_smooth (bool): PCA 기반 노이즈 제거 가동 여부.
+
+    Returns:
+        np.ndarray: 시각화가 적용된 얼굴 이미지 (RGB, np.uint8).
+    """
     if explain_req_dict["display_type"] == "heatmap":
         return explainer.display_heatmap_on_image(image_path, image_weight=explain_req_dict["overlay_ratio"], threshold=explain_req_dict["threshold"],
                                                  category=category, aug_smooth=explain_req_dict["aug_smooth"], eigen_smooth=explain_req_dict["eigen_smooth"])
@@ -63,8 +98,18 @@ def _run_visualization(explainer: CAMExplainer, image_path: str, category: int,
         return explainer.display_heatmap_bbox_on_image(image_path, image_weight=explain_req_dict["overlay_ratio"], threshold=explain_req_dict["threshold"], 
                                                        category=category, aug_smooth=explain_req_dict["aug_smooth"], eigen_smooth=explain_req_dict["eigen_smooth"])
 
-# 비디오 프레임 시각화 생성 (heatmap, contour, bbox 선택)
-def _run_visualization_from_array(explainer: CAMExplainer, face: str, category: int, explain_req_dict: dict) -> np.ndarray:
+def _run_visualization_from_array(explainer: CAMExplainer, face: np.ndarray, category: int, explain_req_dict: dict) -> np.ndarray:
+    """크롭된 얼굴 배열을 받아 display_type에 따라 시각화 이미지를 생성한다.
+
+    Args:
+        explainer (CAMExplainer): 시각화에 사용할 CAMExplainer 인스턴스.
+        face (np.ndarray): 전처리(크롭)가 완료된 얼굴 이미지 배열 (RGB).
+        category (int): 타겟 클래스 인덱스 (위조: 1, 정상: 0).
+        explain_req_dict (dict): 시각화 요청 파라미터 딕셔너리. (_run_visualization과 동일)
+
+    Returns:
+        np.ndarray: 시각화가 적용된 얼굴 이미지 (RGB, np.uint8).
+    """
     if explain_req_dict["display_type"] == "heatmap":
         return explainer.display_heatmap_from_array(face, image_weight=explain_req_dict["overlay_ratio"], threshold=explain_req_dict["threshold"],
                                                  category=category, aug_smooth=explain_req_dict["aug_smooth"], eigen_smooth=explain_req_dict["eigen_smooth"])
@@ -75,15 +120,37 @@ def _run_visualization_from_array(explainer: CAMExplainer, face: str, category:
         return explainer.display_heatmap_bbox_from_array(face, image_weight=explain_req_dict["overlay_ratio"], threshold=explain_req_dict["threshold"], 
                                                        category=category, aug_smooth=explain_req_dict["aug_smooth"], eigen_smooth=explain_req_dict["eigen_smooth"])
 
-# 딥페이크 이미지 위조 흔적 시각화 처리
 @celery_app.task(name="process_explain_image_task")
 def process_explain_image_task(user_email: str, 
                                version_type: str,
                                domain_type: str,
                                image_loc: str,
                                image_id: int,
                                category: int,
-                               explain_req_dict: dict):
+                               explain_req_dict: dict) -> dict:
+    """딥페이크 이미지 위조 흔적 시각화를 처리하는 Celery 비동기 태스크.
+
+    CAMExplainer로 시각화 이미지를 생성하고 서버에 저장한다.
+    Celery 워커(동기) 내에서 asyncio 이벤트 루프를 직접 구동해 비동기 로직을 실행한다.
+
+    Args:
+        user_email (str): 요청 사용자 이메일 (저장 경로 구분용).
+        version_type (str): 모델 버전 타입 (MODEL_CONFIG 키값).
+        domain_type (str): 탐지 도메인 타입 (MODEL_CONFIG 키값).
+        image_loc (str): 분석 대상 이미지의 서버 저장 경로 (DB 기준, '/'로 시작).
+        image_id (int): 이미지 레코드 PK.
+        category (int): 타겟 클래스 인덱스 (위조: 1, 정상: 0).
+        explain_req_dict (dict): 시각화 요청 파라미터 딕셔너리.
+
+    Returns:
+        dict:
+            - status (str): "SUCCESS" | "FAILED"
+            - message (str): 처리 결과 메세지.
+            - cam_loc (str | None): 저장된 시각화 이미지 경로. 실패 시 None.
+
+    Note:
+        생성된 시각화 파일은 태스크 완료 60초 후 자동 삭제된다.
+    """
     async def run_explain():
         cam_loc = None
         try:
@@ -135,7 +202,6 @@ async def run_explain():
     finally:
         loop.close()
 
-# 딥페이크 비디오 프레임 위조 흔적 시각화 처리
 @celery_app.task(name="process_explain_frame_task")
 def process_explain_frame_task(user_email: str, 
                                version_type: str,
@@ -144,7 +210,31 @@ def process_explain_frame_task(user_email: str,
                                video_id: int,
                                category: int,
                                frame_time: float, 
-                               explain_req_dict: dict):
+                               explain_req_dict: dict) -> dict:
+    """딥페이크 비디오 특정 프레임의 위조 흔적 시각화를 처리하는 Celery 비동기 태스크.
+
+    비디오에서 프레임을 추출하고 얼굴을 크롭한 뒤 CAMExplainer로 시각화 이미지를 생성해 저장한다.
+    Celery 워커(동기) 내에서 asyncio 이벤트 루프를 직접 구동해 비동기 로직을 실행한다.
+
+    Args:
+        user_email (str): 요청 사용자 이메일 (저장 경로 구분용).
+        version_type (str): 모델 버전 타입 (MODEL_CONFIG 키값).
+        domain_type (str): 탐지 도메인 타입 (MODEL_CONFIG 키값).
+        video_loc (str): 분석 대상 비디오의 서버 저장 경로 (DB 기준, '/'로 시작).
+        video_id (int): 비디오 레코드 PK.
+        category (int): 타겟 클래스 인덱스 (위조: 1, 정상: 0).
+        frame_time (float): 시각화할 프레임의 타임스탬프 (초 단위).
+        explain_req_dict (dict): 시각화 요청 파라미터 딕셔너리.
+
+    Returns:
+        dict:
+            - status (str): "SUCCESS" | "FAILED"
+            - message (str): 처리 결과 메세지.
+            - cam_loc (str | None): 저장된 시각화 이미지 경로. 실패 시 None.
+
+    Note:
+        생성된 시각화 파일은 태스크 완료 60초 후 자동 삭제된다.
+    """
     async def run_explain():
         cam_loc = None
         try: