From 1fa876c49d738b72e79a911f3cac50cc0d3803a4 Mon Sep 17 00:00:00 2001
From: seoyunje <origin6165@naver.com>
Date: Thu, 28 May 2026 13:06:27 +0900
Subject: [PATCH 1/7] =?UTF-8?q?=E2=9C=8F=EF=B8=8F=20Fix=20a=20few=20bug=20?=
 =?UTF-8?q?of=20explain=5Fframe?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 App/routes/explain.py                     | 2 +-
 App/services/explain_svc.py               | 2 +-
 App/services/video_svc.py                 | 2 ++
 explainability/explainer/cam_explainer.py | 2 +-
 4 files changed, 5 insertions(+), 3 deletions(-)

diff --git a/App/routes/explain.py b/App/routes/explain.py
index be238da..adcc763 100644
--- a/App/routes/explain.py
+++ b/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(
diff --git a/App/services/explain_svc.py b/App/services/explain_svc.py
index 1a725ef..dddec6b 100644
--- a/App/services/explain_svc.py
+++ b/App/services/explain_svc.py
@@ -64,7 +64,7 @@ def _run_visualization(explainer: CAMExplainer, image_path: str, category: int,
                                                        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:
     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"])
diff --git a/App/services/video_svc.py b/App/services/video_svc.py
index 3e5b586..0229541 100644
--- a/App/services/video_svc.py
+++ b/App/services/video_svc.py
@@ -554,6 +554,8 @@ async def get_video_frame_by_index(conn: Connection, video_id: int, frame_index:
     except SQLAlchemyError as e:
         print(f"[Frame Query Error] {e}")
         raise HTTPException(status_code=503, detail="데이터베이스 조회 중 문제가 발생했습니다.")
+    except HTTPException:
+        raise
     except Exception as e:
         print(e)
         raise HTTPException(status_code=500, detail="알수없는 이유로 문제가 발생하였습니다.")
diff --git a/explainability/explainer/cam_explainer.py b/explainability/explainer/cam_explainer.py
index c5d9a42..6ea15bb 100644
--- a/explainability/explainer/cam_explainer.py
+++ b/explainability/explainer/cam_explainer.py
@@ -296,7 +296,7 @@ def display_heatmap_bbox_on_image(self, img_path: str, **kwargs) -> np.ndarray:
         show_bbox_on_image(heatmap, cam_recovered, binary_mask, kwargs.get("thickness", 1))
         return heatmap
     
-    def display_heatmap_bbox_on_image_from_array(self, face: np.ndarray, **kwargs) -> np.ndarray:
+    def display_heatmap_bbox_from_array(self, face: np.ndarray, **kwargs) -> np.ndarray:
         cam_recovered, face = self._prepare_cam_from_array(
             face, 
             category = kwargs.get("category", 1), 

From 1eb4a289986fcea07a71efa5d8492560b4db7eee Mon Sep 17 00:00:00 2001
From: seoyunje <origin6165@naver.com>
Date: Thu, 28 May 2026 13:20:28 +0900
Subject: [PATCH 2/7] =?UTF-8?q?=F0=9F=93=9D=20Convert=20auth=5Fsvc=20inlin?=
 =?UTF-8?q?e=20comments=20to=20docstrings?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 App/services/auth_svc.py | 68 ++++++++++++++++++++++++++++++++++------
 1 file changed, 58 insertions(+), 10 deletions(-)

diff --git a/App/services/auth_svc.py b/App/services/auth_svc.py
index 5c28905..139761f 100644
--- a/App/services/auth_svc.py
+++ b/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

From 0aac933497e4e1ede65dc69285fe7c3355d8f4b5 Mon Sep 17 00:00:00 2001
From: seoyunje <origin6165@naver.com>
Date: Thu, 28 May 2026 13:31:13 +0900
Subject: [PATCH 3/7] =?UTF-8?q?=F0=9F=93=9D=20Convert=20explain=5Fsvc=20in?=
 =?UTF-8?q?line=20comments=20to=20docstrings?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 App/services/explain_svc.py | 124 +++++++++++++++++++++++++++++++-----
 1 file changed, 107 insertions(+), 17 deletions(-)

diff --git a/App/services/explain_svc.py b/App/services/explain_svc.py
index dddec6b..5b3cbcf 100644
--- a/App/services/explain_svc.py
+++ b/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,13 +12,11 @@
 )
 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
@@ -27,8 +24,17 @@
 
 _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: 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,7 +120,6 @@ def _run_visualization_from_array(explainer: CAMExplainer, face: np.ndarray, cat
         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,
@@ -83,7 +127,30 @@ def process_explain_image_task(user_email: 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:

From 174913bcbb486217a3ae60faccf1c3543c2643b2 Mon Sep 17 00:00:00 2001
From: seoyunje <origin6165@naver.com>
Date: Thu, 28 May 2026 13:39:01 +0900
Subject: [PATCH 4/7] =?UTF-8?q?=F0=9F=93=9D=20Convert=20inference=5Fsvc=20?=
 =?UTF-8?q?inline=20comments=20to=20docstrings?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 App/services/inference_svc.py | 91 ++++++++++++++++++++++++++++++-----
 1 file changed, 80 insertions(+), 11 deletions(-)

diff --git a/App/services/inference_svc.py b/App/services/inference_svc.py
index a9468da..d96838e 100644
--- a/App/services/inference_svc.py
+++ b/App/services/inference_svc.py
@@ -11,7 +11,6 @@
 from inference.video_predictor_prt import VideoPredictor
 from inference.utils import PredictorError
 
-# 사용자 모델 설정 변수명 
 MODEL_CONFIG = {
     'v1': {
         'fast': {'서양인': ("ms_eff_vit_b0", "ff++")},
@@ -32,12 +31,23 @@
 _image_cache: dict = {}
 _video_cache: dict = {}
 
- # 캐시 확인 및 모델 초기화
 def _get_or_create_predictor(
     model_name: str,
     dataset: str,
     predictor_mode: Literal["image", "video"],
 ) -> ImagePredictor | VideoPredictor:
+    """캐시에서 Predictor를 조회하거나, 없으면 새로 생성해 캐싱 후 반환한다.
+
+    이미지/비디오 캐시를 분리 관리해 모드별 독립적인 인스턴스를 유지한다.
+
+    Args:
+        model_name (str): 추론에 사용할 모델 아키텍처 이름.
+        dataset (str): 모델 학습에 사용된 데이터셋 이름.
+        predictor_mode (Literal["image", "video"]): Predictor 종류 선택.
+
+    Returns:
+        ImagePredictor | VideoPredictor: 캐시된 또는 새로 생성된 Predictor 인스턴스.
+    """
     cache_key = (model_name, dataset)
  
     if predictor_mode == "image":
@@ -59,8 +69,22 @@ def _get_or_create_predictor(
         )
     return _video_cache[cache_key]
 
-# 사용자 이미지 딥페이크 여부 판단 로직
-def predict_image(image_loc: str, version_type: str, model_type: str, domain_type: str):
+def predict_image(image_loc: str, version_type: str, model_type: str, domain_type: str) -> dict:
+    """이미지 경로를 받아 딥페이크 여부를 추론하고 분석 결과를 반환한다.
+
+    Args:
+        image_loc (str): 추론 대상 이미지의 서버 저장 경로 (DB 기준, '/'로 시작).
+        version_type (str): 모델 버전 타입 (MODEL_CONFIG 키값).
+        model_type (str): 추론 모드 ("fast" | "pro").
+        domain_type (str): 탐지 도메인 ("서양인" | "동양인").
+
+    Returns:
+        dict:
+            - analysis (dict): prob, face_conf, face_ratio, face_brightness 포함.
+              얼굴 미감지 또는 오류 발생 시 모든 값 -1.
+            - message (str): 처리 결과 메세지.
+            - status (str): "success" | "warning" | "failed"
+    """
     
     model_name, dataset = MODEL_CONFIG[version_type][model_type][domain_type]
 
@@ -95,10 +119,25 @@ def predict_image(image_loc: str, version_type: str, model_type: str, domain_typ
             "status": "failed"
         }
 
-# Celery 이미지 추론 Task
 @celery_app.task(name="process_image_task")
 def process_image_task(image_id: int, image_loc: str, version_type: str, model_type: str, 
-                             domain_type: str, user_id: int | None):
+                             domain_type: str, user_id: int | None) -> None:
+    """딥페이크 이미지 추론을 처리하는 Celery 비동기 태스크.
+
+    추론 결과를 DB에 저장하며, SQLAlchemy 오류 발생 시 실패 상태로 업데이트한다.
+    Celery 워커(동기) 내에서 asyncio 이벤트 루프를 직접 구동해 비동기 로직을 실행한다.
+
+    Args:
+        image_id (int): 이미지 레코드 PK.
+        image_loc (str): 추론 대상 이미지의 서버 저장 경로 (DB 기준, '/'로 시작).
+        version_type (str): 모델 버전 타입 (MODEL_CONFIG 키값).
+        model_type (str): 추론 모드 ("fast" | "pro").
+        domain_type (str): 탐지 도메인 ("서양인" | "동양인").
+        user_id (int | None): 로그인 사용자 ID. 비회원이면 None.
+
+    Note:
+        비회원(user_id=None) 요청은 추론 완료 60초 후 이미지 파일 및 DB 레코드가 자동 삭제된다.
+    """
     async def run_inference():    
         try:
             # 이미지 추론, DeepFake 결과값 반환
@@ -142,9 +181,23 @@ async def run_inference():
         loop.close()
     
         
-# 사용자 비디오 딥페이크 여부 판단 로직
-def predict_video(video_loc: str, version_type: str, model_type: str, domain_type: str):
-    
+def predict_video(video_loc: str, version_type: str, model_type: str, domain_type: str) -> dict:
+    """비디오 경로를 받아 딥페이크 여부를 추론하고 분석 결과를 반환한다.
+
+    Args:
+        video_loc (str): 추론 대상 비디오의 서버 저장 경로 (DB 기준, '/'로 시작).
+        version_type (str): 모델 버전 타입 (MODEL_CONFIG 키값).
+        model_type (str): 추론 모드 ("fast" | "pro").
+        domain_type (str): 탐지 도메인 ("서양인" | "동양인").
+
+    Returns:
+        dict:
+            - analysis (dict): prob, face_conf, face_ratio, face_brightness,
+              frame_results, fps, total_frames, num_sampled, num_extracted, num_detected 포함.
+              오류 발생 시 주요 수치 값 -1.
+            - message (str): 처리 결과 메세지.
+            - status (str): "success" | "warning" | "failed"
+    """
     model_name, dataset = MODEL_CONFIG[version_type][model_type][domain_type]
 
     predictor = _get_or_create_predictor(model_name, dataset, "video")
@@ -180,10 +233,26 @@ def predict_video(video_loc: str, version_type: str, model_type: str, domain_typ
             "status": "failed"
         }
         
-# Celery 비디오 추론 Task
 @celery_app.task(name="process_video_task")
 def process_video_task(video_id: int, video_loc: str, version_type: str, model_type: str, 
-                             domain_type: str, user_id: int | None):
+                             domain_type: str, user_id: int | None) -> None:
+    """딥페이크 비디오 추론을 처리하는 Celery 비동기 태스크.
+
+    추론 결과를 DB에 저장하며, 성공 시 메타 데이터와 프레임별 결과를 추가로 저장한다.
+    메타/프레임 저장 실패는 독립적으로 처리되어 메인 추론 결과에 영향을 주지 않는다.
+    Celery 워커(동기) 내에서 asyncio 이벤트 루프를 직접 구동해 비동기 로직을 실행한다.
+
+    Args:
+        video_id (int): 비디오 레코드 PK.
+        video_loc (str): 추론 대상 비디오의 서버 저장 경로 (DB 기준, '/'로 시작).
+        version_type (str): 모델 버전 타입 (MODEL_CONFIG 키값).
+        model_type (str): 추론 모드 ("fast" | "pro").
+        domain_type (str): 탐지 도메인 ("서양인" | "동양인").
+        user_id (int | None): 로그인 사용자 ID. 비회원이면 None.
+
+    Note:
+        비회원(user_id=None) 요청은 추론 완료 60초 후 비디오 파일 및 DB 레코드가 자동 삭제된다.
+    """
     async def run_inference():
         try:
             # 비디오 추론, DeepFake 결과값 반환

From 3d358868578d6d3eb70172c1b7db9f16c639c9ad Mon Sep 17 00:00:00 2001
From: seoyunje <origin6165@naver.com>
Date: Thu, 28 May 2026 13:45:46 +0900
Subject: [PATCH 5/7] =?UTF-8?q?=F0=9F=93=9D=20Convert=20session=5Fsvc=20in?=
 =?UTF-8?q?line=20comments=20to=20docstrings?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 App/services/session_svc.py | 30 +++++++++++++++++++++++++++---
 1 file changed, 27 insertions(+), 3 deletions(-)

diff --git a/App/services/session_svc.py b/App/services/session_svc.py
index c9845e1..a144196 100644
--- a/App/services/session_svc.py
+++ b/App/services/session_svc.py
@@ -1,12 +1,36 @@
-from fastapi import APIRouter, Depends, status, Request
+from fastapi import status, Request
 from fastapi.exceptions import HTTPException
 
-def get_session_user_opt(request:Request):
+def get_session_user_opt(request:Request) -> dict | None:
+    """세션에서 로그인 사용자 정보를 조회한다. 비로그인 상태면 None을 반환한다.
+
+    로그인 여부와 관계없이 접근 가능한 엔드포인트에 사용한다.
+
+    Args:
+        request (Request): FastAPI 요청 객체.
+
+    Returns:
+        dict | None: 로그인 상태면 {"id": int, "name": str, "email": str} 반환,
+            비로그인 상태면 None.
+    """
     if "session_user" in request.state.session:
         return request.state.session["session_user"]
     
 
-def get_session_user_prt(request:Request):
+def get_session_user_prt(request:Request) -> dict:
+    """세션에서 로그인 사용자 정보를 조회한다. 비로그인 상태면 401을 raise한다.
+
+    로그인이 필수인 엔드포인트의 Depends 의존성으로 사용한다.
+
+    Args:
+        request (Request): FastAPI 요청 객체.
+
+    Returns:
+        dict: {"id": int, "name": str, "email": str}
+
+    Raises:
+        HTTPException 401: 세션에 사용자 정보가 없을 때.
+    """
     if "session_user" not in request.state.session:
         raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
                             detail="해당 서비스는 로그인이 필요합니다.")

From d83d7f45ca109f28319b69df42d95a9e80d9bcae Mon Sep 17 00:00:00 2001
From: seoyunje <origin6165@naver.com>
Date: Thu, 28 May 2026 14:01:09 +0900
Subject: [PATCH 6/7] =?UTF-8?q?=F0=9F=93=9D=20Convert=20image=5Fsvc=20inli?=
 =?UTF-8?q?ne=20comments=20to=20docstrings?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 App/services/image_svc.py | 365 +++++++++++++++++++++++++-------------
 1 file changed, 241 insertions(+), 124 deletions(-)

diff --git a/App/services/image_svc.py b/App/services/image_svc.py
index 950f94c..da1b00e 100644
--- a/App/services/image_svc.py
+++ b/App/services/image_svc.py
@@ -21,10 +21,19 @@
 os.makedirs(EXPLAIN_UPLOAD_DIR, exist_ok=True)
 os.makedirs(UPLOAD_DIR, exist_ok=True)
 
-# 사용자 업로드 이미지 서버 내 저장 (회원/비회원 공통)
-# 호출 : inference.py 
-# image_loc = await image_svc.upload_image(user_email, imagefile) -> 이미지 업로드 이후, 이미지 저장 경로 반환
 async def upload_image(user_email: str | None, imagefile: UploadFile) -> str:
+    """업로드된 이미지를 서버에 저장하고 DB 저장용 경로를 반환한다. 회원/비회원 공통.
+
+    Args:
+        user_email (str | None): 로그인 사용자 이메일. 비회원이면 None.
+        imagefile (UploadFile): FastAPI 업로드 파일 객체.
+
+    Returns:
+        str: 저장된 이미지의 DB 기준 경로 ('/'로 시작, '\\' 정규화됨).
+
+    Raises:
+        HTTPException 500: 파일 쓰기 또는 예기치 못한 오류 발생 시.
+    """
     try:
         # 사용자별 하위 디렉토리 결정
         sub_dir = user_email if user_email else "anonymous"
@@ -64,8 +73,20 @@ async def upload_image(user_email: str | None, imagefile: UploadFile) -> str:
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail="이미지 업로드 과정에서 예상치 못한 오류가 발생했습니다.")
 
-# 딥페이크 이미지 위조 흔적 시각화 파일 서버 내 저장 (회원 전용)
 async def upload_image_cam(user_email: str, image_id: int, image: np.ndarray) -> str:
+    """CAM 시각화 이미지를 서버에 저장하고 DB 저장용 경로를 반환한다. 회원 전용.
+
+    Args:
+        user_email (str): 요청 사용자 이메일 (저장 경로 구분용).
+        image_id (int): 이미지 레코드 PK (파일명 구성에 사용).
+        image (np.ndarray): 저장할 시각화 이미지 배열 (RGB).
+
+    Returns:
+        str: 저장된 CAM 이미지의 DB 기준 경로 ('/'로 시작).
+
+    Raises:
+        Exception: 파일 쓰기 실패 시 그대로 re-raise.
+    """
     user_dir = os.path.join(EXPLAIN_UPLOAD_DIR, user_email)
     os.makedirs(user_dir, exist_ok=True)
     
@@ -83,8 +104,21 @@ async def upload_image_cam(user_email: str, image_id: int, image: np.ndarray) ->
 
     return cam_loc[1:].replace("\\", "/")
 
-# 딥페이크 비디오 프레임 위조 흔적 시각화 파일 서버 내 저장 (회원 전용)
 async def upload_frame_cam(user_email: str, video_id: int, frame_time: float, image: np.ndarray) -> str:
+    """비디오 프레임 CAM 시각화 이미지를 서버에 저장하고 DB 저장용 경로를 반환한다. 회원 전용.
+
+    Args:
+        user_email (str): 요청 사용자 이메일 (저장 경로 구분용).
+        video_id (int): 비디오 레코드 PK (파일명 구성에 사용).
+        frame_time (float): 시각화한 프레임의 타임스탬프 (초 단위, 파일명 구성에 사용).
+        image (np.ndarray): 저장할 시각화 이미지 배열 (RGB).
+
+    Returns:
+        str: 저장된 CAM 이미지의 DB 기준 경로 ('/'로 시작).
+
+    Raises:
+        Exception: 파일 쓰기 실패 시 그대로 re-raise.
+    """
     user_dir = os.path.join(EXPLAIN_UPLOAD_DIR, user_email)
     os.makedirs(user_dir, exist_ok=True)
     
@@ -102,10 +136,17 @@ async def upload_frame_cam(user_email: str, video_id: int, frame_time: float, im
 
     return cam_loc[1:].replace("\\", "/")
 
-# 사용자 업로드 이미지 서버 내 삭제
-# 호출 : image.py : history 삭제 할 때 db 와 실제 파일 삭제
-# 호출 : inference.py : 추론 FAIL일 때 delete_video and delete_video_db 실행
 async def delete_image(image_loc: str):
+    """서버에서 이미지 물리 파일을 삭제한다.
+
+    파일이 존재하지 않으면 경고 로그만 출력하고 정상 종료한다.
+
+    Args:
+        image_loc (str): 삭제할 이미지의 DB 기준 경로 ('/'로 시작).
+
+    Raises:
+        HTTPException 500: 파일 삭제 중 예기치 못한 오류 발생 시.
+    """
     try:
         
         file_path = "." + image_loc 
@@ -120,99 +161,49 @@ async def delete_image(image_loc: str):
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail="알수없는 이유로 문제가 발생하였습니다."
         )
-    
 
-# 사용자 전체 히스토리 조회 (image_result 테이블 반영)
-# 호출 : image.py / video.py
-async def get_user_histories(conn: Connection, user_id: int):
-    try:
-        # SQL에 맞춰 테이블명과 컬럼 변경
-        query = ("""
-            SELECT id, user_id, image_loc, label, version_type, model_type, domain_type, created_at
-            FROM image_result
-            WHERE user_id = :user_id
-            ORDER BY created_at DESC;
-        """)
-        stmt = text(query)
-        result = await conn.execute(stmt, {"user_id": user_id})
+async def delete_image_db(conn: Connection, image_id: int):
+    """DB에서 이미지 레코드를 삭제한다.
 
-        user_histories = [UserHistory(
-            image_id = row.id,
-            user_id = row.user_id,
-            image_loc = row.image_loc,
-            label = row.label,
-            version_type = row.version_type,
-            model_type = row.model_type,
-            domain_type = row.domain_type,
-            created_at = row.created_at
-        )
-            for row in result]
-        
-        result.close()
+    Args:
+        conn (Connection): SQLAlchemy 비동기 DB 커넥션.
+        image_id (int): 삭제할 이미지 레코드 PK.
+
+    Raises:
+        HTTPException 404: 해당 image_id 레코드가 없을 때.
+        HTTPException 503: DB 쿼리 중 SQLAlchemy 오류 발생 시.
+        HTTPException 500: 예기치 못한 오류 발생 시.
+    """
+    try:
+        delete_query = text("DELETE FROM image_result WHERE id = :image_id")
+        result = await conn.execute(delete_query, {"image_id": image_id})
 
+        if result.rowcount == 0:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"해당 이미지 id {image_id}는(은) 존재하지 않아 삭제할 수 없습니다.")
             
-        return user_histories
-    
-    except SQLAlchemyError as e:
-        print(f"히스토리 조회 실패: {e}")
-        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="요청하신 서비스가 잠시 내부적으로 문제가 발생하였습니다.")
+        await conn.commit()
 
-    except Exception as e:
-        print(e)
-        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="알수없는 이유로 문제가 발생하였습니다.")
-    
-# 사용자 개별 히스토리 조회
-# 호출 : image.py / video.py
-async def get_user_history(conn: Connection, image_id: int):
-    try:
-        query = """
-            SELECT id, user_id, image_loc, status, label, score, face_conf, face_ratio, face_brightness, version_type, model_type, domain_type, result_msg, created_at
-            FROM image_result
-            WHERE id = :image_id;
-        """
-        stmt = text(query)
-        result = await conn.execute(stmt, {"image_id": image_id})
-        
-        row = result.fetchone()
-        if row is None:
-            return None
-        
-        user_history = UserHistory_indi(
-            image_id = row.id,
-            user_id = row.user_id,
-            image_loc = row.image_loc,
-            status = row.status,
-            label = row.label,
-            score = row.score,
-            face_conf = row.face_conf,
-            face_ratio = row.face_ratio,
-            face_brightness = row.face_brightness,
-            version_type = row.version_type,
-            model_type = row.model_type,
-            domain_type = row.domain_type,
-            result_msg = row.result_msg,
-            created_at = row.created_at
-        )
-        
-        result.close()
-        
-        return user_history
-    
     except SQLAlchemyError as e:
-        print(f"히스토리 조회 실패: {e}")
+        await conn.rollback()
         raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="요청하신 서비스가 잠시 내부적으로 문제가 발생하였습니다.")
 
     except HTTPException:
         raise
-
+    
     except Exception as e:
-        print(e)
+        await conn.rollback()
         raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="알수없는 이유로 문제가 발생하였습니다.")
 
-# 비회원 데이터 1분 후 자동 삭제 태스크
-# inference_svc.py : def process_image_task에서 추론이 성공 했을 때, 비회원일 경우 1분 후 자동 삭제
 @celery_app.task(name="cleanup_anonymous_image")
 def cleanup_anonymous_image(image_id: int, image_loc: str):
+    """비회원 이미지 DB 레코드 및 물리 파일을 삭제하는 Celery 태스크.
+
+    DB 삭제와 파일 삭제는 독립적으로 처리되어 한쪽 실패가 다른쪽에 영향을 주지 않는다.
+
+    Args:
+        image_id (int): 삭제할 이미지 레코드 PK.
+        image_loc (str): 삭제할 이미지의 DB 기준 경로.
+    """
     loop = asyncio.new_event_loop()
     asyncio.set_event_loop(loop)
     try:
@@ -240,6 +231,13 @@ async def _delete():
 
 @celery_app.task(name="cleanup_image_cam")
 def cleanup_image_cam(cam_loc: str):
+    """CAM 시각화 파일을 삭제하는 Celery 태스크.
+
+    이미지/비디오 프레임 시각화 파일 모두 이 태스크로 정리한다.
+
+    Args:
+        cam_loc (str): 삭제할 CAM 시각화 파일의 DB 기준 경로.
+    """
     loop = asyncio.new_event_loop()
     asyncio.set_event_loop(loop)
     try:
@@ -253,37 +251,26 @@ async def _delete():
         loop.run_until_complete(_delete())
     
     finally:
-        loop.close()
-        
-# 이미지 DB 레코드 및 물리 파일 완전 삭제
-# 호출 : image.py : history 삭제 시
-# 호출 : inference.py : get_image_result 에서 FALIED일 시
-async def delete_image_db(conn: Connection, image_id: int):
-    try:
-        delete_query = text("DELETE FROM image_result WHERE id = :image_id")
-        result = await conn.execute(delete_query, {"image_id": image_id})
-
-        if result.rowcount == 0:
-            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"해당 이미지 id {id}는(은) 존재하지 않아 삭제할 수 없습니다.")
-            
-        await conn.commit()
-
-    except SQLAlchemyError as e:
-        await conn.rollback()
-        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="요청하신 서비스가 잠시 내부적으로 문제가 발생하였습니다.")
-
-    except HTTPException:
-        raise
+        loop.close()    
     
-    except Exception as e:
-        await conn.rollback()
-        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="알수없는 이유로 문제가 발생하였습니다.")
-
-# 빈 이미지 DB 생성 이후, image_id 반환(접수 완료)
-# 호출 : inference.py : 빈 이미지 DB 생성 후, image_id 받기
-# image_id = (conn, user_id, image_loc, version_type, model_type, domain_type)
 async def register_image_result(conn: Connection, user_id: int | None, image_loc: str, 
                                 version_type: str, model_type: str, domain_type: str):
+    """PENDING 상태의 빈 이미지 레코드를 DB에 생성하고 image_id를 반환한다.
+
+    Args:
+        conn (Connection): SQLAlchemy 비동기 DB 커넥션.
+        user_id (int | None): 로그인 사용자 ID. 비회원이면 None.
+        image_loc (str): 업로드된 이미지의 DB 기준 경로.
+        version_type (str): 모델 버전 타입.
+        model_type (str): 추론 모드 ("fast" | "pro").
+        domain_type (str): 탐지 도메인.
+
+    Returns:
+        int: 생성된 이미지 레코드의 PK (auto_increment).
+
+    Raises:
+        HTTPException 400: DB INSERT 실패 시. 업로드된 물리 파일도 함께 롤백 삭제된다.
+    """
     try:
         query = """
             INSERT INTO image_result (
@@ -315,15 +302,24 @@ async def register_image_result(conn: Connection, user_id: int | None, image_loc
         raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
                             detail="요청데이터가 제대로 전달되지 않았습니다")
         
-# 이미지 메타데이터 + 추론 결과값 DB에 저장
-# 호출 : inference_svc.py : process_image_task 추론 종료 시점
 async def update_image_result(conn: Connection, image_id: int, analysis: dict,
                               result_msg: str, status: str): 
-    
+    """추론 완료 후 이미지 레코드에 분석 결과를 업데이트한다.
+
+    Args:
+        conn (Connection): SQLAlchemy 비동기 DB 커넥션.
+        image_id (int): 업데이트할 이미지 레코드 PK.
+        analysis (dict): 추론 결과값. prob, face_conf, face_ratio, face_brightness 포함.
+        result_msg (str): 처리 결과 메세지.
+        status (str): 추론 상태 ("success" | "warning" | "failed").
+
+    Raises:
+        SQLAlchemyError: DB 업데이트 실패 시 그대로 re-raise.
+    """
     # status 종류: success, warning, failed
     # success: 이미지 추론 성공
     # warning: 이미지 추론 과정 PredictError 발생
-    # failed: 이미지 추로 과정 알수 없는 오류 발생 
+    # failed: 이미지 추론 과정 알수 없는 오류 발생 
     
     if status == 'success':
         label = "FAKE" if analysis["prob"] > 0.5 else "REAL"
@@ -360,12 +356,22 @@ async def update_image_result(conn: Connection, image_id: int, analysis: dict,
     except SQLAlchemyError as e:
         await conn.rollback()
         raise e
-    
-# 이미지 결과 값 가져오기
-# 프론트엔드가 특정 이미지의 분석 진행 상태 및 최종 결과를 확인하고자 할 때 데이터 반환
-# 호출 위치: routers/inference.py - get_image_result() API
-async def get_image_result(conn: Connection, 
-                           image_id: int):
+
+async def get_image_result(conn: Connection, image_id: int):
+    """image_id로 이미지 분석 결과를 조회해 반환한다.
+
+    Args:
+        conn (Connection): SQLAlchemy 비동기 DB 커넥션.
+        image_id (int): 조회할 이미지 레코드 PK.
+
+    Returns:
+        ImageData_indi: 이미지 분석 결과 스키마 객체.
+
+    Raises:
+        HTTPException 404: 해당 image_id 레코드가 없을 때.
+        HTTPException 503: DB 쿼리 중 SQLAlchemy 오류 발생 시.
+        HTTPException 500: 예기치 못한 오류 발생 시.
+    """
     try:
         query = text("SELECT * FROM image_result WHERE id = :image_id")
         result = await conn.execute(query, {"image_id": image_id})
@@ -405,3 +411,114 @@ async def get_image_result(conn: Connection,
         print(e)
         raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, 
                             detail="알수없는 이유로 문제가 발생하였습니다.")
+
+async def get_user_histories(conn: Connection, user_id: int):
+    """사용자의 전체 이미지 분석 히스토리를 최신순으로 조회한다.
+
+    Args:
+        conn (Connection): SQLAlchemy 비동기 DB 커넥션.
+        user_id (int): 조회할 사용자 PK.
+
+    Returns:
+        list[UserHistory]: 이미지 히스토리 스키마 리스트. 결과 없으면 빈 리스트.
+
+    Raises:
+        HTTPException 503: DB 쿼리 중 SQLAlchemy 오류 발생 시.
+        HTTPException 500: 예기치 못한 오류 발생 시.
+    """
+    try:
+        # SQL에 맞춰 테이블명과 컬럼 변경
+        query = ("""
+            SELECT id, user_id, image_loc, label, version_type, model_type, domain_type, created_at
+            FROM image_result
+            WHERE user_id = :user_id
+            ORDER BY created_at DESC;
+        """)
+        stmt = text(query)
+        result = await conn.execute(stmt, {"user_id": user_id})
+
+        user_histories = [UserHistory(
+            image_id = row.id,
+            user_id = row.user_id,
+            image_loc = row.image_loc,
+            label = row.label,
+            version_type = row.version_type,
+            model_type = row.model_type,
+            domain_type = row.domain_type,
+            created_at = row.created_at
+        )
+            for row in result]
+        
+        result.close()
+
+            
+        return user_histories
+    
+    except SQLAlchemyError as e:
+        print(f"히스토리 조회 실패: {e}")
+        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="요청하신 서비스가 잠시 내부적으로 문제가 발생하였습니다.")
+
+    except Exception as e:
+        print(e)
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="알수없는 이유로 문제가 발생하였습니다.")
+
+async def get_user_history(conn: Connection, image_id: int):
+    """image_id로 이미지 개별 상세 히스토리를 조회한다.
+
+    Args:
+        conn (Connection): SQLAlchemy 비동기 DB 커넥션.
+        image_id (int): 조회할 이미지 레코드 PK.
+
+    Returns:
+        UserHistory_indi | None: 상세 히스토리 스키마 객체. 레코드 없으면 None.
+
+    Raises:
+        HTTPException 503: DB 쿼리 중 SQLAlchemy 오류 발생 시.
+        HTTPException 500: 예기치 못한 오류 발생 시.
+    """
+    try:
+        query = """
+            SELECT id, user_id, image_loc, status, label, score, face_conf, face_ratio, face_brightness, version_type, model_type, domain_type, result_msg, created_at
+            FROM image_result
+            WHERE id = :image_id;
+        """
+        stmt = text(query)
+        result = await conn.execute(stmt, {"image_id": image_id})
+        
+        row = result.fetchone()
+        if row is None:
+            return None
+        
+        user_history = UserHistory_indi(
+            image_id = row.id,
+            user_id = row.user_id,
+            image_loc = row.image_loc,
+            status = row.status,
+            label = row.label,
+            score = row.score,
+            face_conf = row.face_conf,
+            face_ratio = row.face_ratio,
+            face_brightness = row.face_brightness,
+            version_type = row.version_type,
+            model_type = row.model_type,
+            domain_type = row.domain_type,
+            result_msg = row.result_msg,
+            created_at = row.created_at
+        )
+        
+        result.close()
+        
+        return user_history
+    
+    except SQLAlchemyError as e:
+        print(f"히스토리 조회 실패: {e}")
+        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="요청하신 서비스가 잠시 내부적으로 문제가 발생하였습니다.")
+
+    except HTTPException:
+        raise
+
+    except Exception as e:
+        print(e)
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="알수없는 이유로 문제가 발생하였습니다.")
+
+        
\ No newline at end of file

From 2092e9e88308e4b1ea2a8464cabca01b795b4926 Mon Sep 17 00:00:00 2001
From: seoyunje <origin6165@naver.com>
Date: Thu, 28 May 2026 14:07:49 +0900
Subject: [PATCH 7/7] =?UTF-8?q?=F0=9F=93=9D=20Convert=20video=5Fsvc=20inli?=
 =?UTF-8?q?ne=20comments=20to=20docstrings?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 App/services/video_svc.py | 433 ++++++++++++++++++++++++++------------
 1 file changed, 301 insertions(+), 132 deletions(-)

diff --git a/App/services/video_svc.py b/App/services/video_svc.py
index 0229541..14fbb03 100644
--- a/App/services/video_svc.py
+++ b/App/services/video_svc.py
@@ -1,24 +1,34 @@
 import os
-import asyncio
 import time
-import aiofiles as aio
 from dotenv import load_dotenv
-
+import asyncio
+import aiofiles as aio
 from fastapi import UploadFile, status
 from fastapi.exceptions import HTTPException
 from sqlalchemy import text, Connection
-from sqlalchemy.exc import SQLAlchemyError, DBAPIError
+from sqlalchemy.exc import SQLAlchemyError
 from schemas.video_schema import (
-    VideoData, VideoDetailData, VideoFrameData, VideoMetaData )
+    VideoData, VideoDetailData, VideoFrameData, VideoMetaData)
 from db.database import celery_db_conn
 from celery_app import celery_app
+
 load_dotenv()
 UPLOAD_DIR = os.getenv("UPLOAD_DIR")
 
-# 사용자 업로드 동영상 서버 내 저장 (회원/비회원 공통)
-# 호출 : inference.py : video_loc = await video_svc.upload_video(user_email, videofile)
-#  -> 비디오 업로드 이후, 비디오 저장 경로 반환
+
 async def upload_video(user_email: str | None, videofile: UploadFile) -> str:
+    """업로드된 비디오를 서버에 저장하고 DB 저장용 경로를 반환한다. 회원/비회원 공통.
+
+    Args:
+        user_email (str | None): 로그인 사용자 이메일. 비회원이면 None.
+        videofile (UploadFile): FastAPI 업로드 파일 객체.
+
+    Returns:
+        str: 저장된 비디오의 DB 기준 경로 ('/'로 시작, '\\' 정규화됨).
+
+    Raises:
+        HTTPException 500: 디렉토리 생성, 파일 쓰기, 또는 예기치 못한 오류 발생 시.
+    """
     try:
         # 1. 사용자별 하위 디렉토리 결정
         sub_dir = user_email if user_email else "anonymous"
@@ -66,9 +76,18 @@ async def upload_video(user_email: str | None, videofile: UploadFile) -> str:
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail="비디오 업로드 과정에서 예상치 못한 오류가 발생했습니다.")
 
-# 사용자 업로드 비디오 서버 내 삭제
-# 호출 : inference.py : 추론 FAIL일 때 delete_video and delete_video_db 실행
-async def delete_video(video_loc: str):
+
+async def delete_video(video_loc: str) -> None:
+    """서버에서 비디오 물리 파일을 삭제한다.
+
+    파일이 존재하지 않으면 경고 로그만 출력하고 정상 종료한다.
+
+    Args:
+        video_loc (str): 삭제할 비디오의 DB 기준 경로 ('/'로 시작).
+
+    Raises:
+        HTTPException 500: 파일 삭제 중 오류 발생 시.
+    """
     try:
         file_path = "." + video_loc 
 
@@ -84,9 +103,21 @@ async def delete_video(video_loc: str):
             detail="비디오 파일 삭제 중 오류가 발생했습니다."
         )
 
-# 사용자 전체 비디오 히스토리 조회
-# 호출 : image.py / video.py
-async def get_user_histories(conn: Connection, user_id: int):
+
+async def get_user_histories(conn: Connection, user_id: int) -> list[VideoData]:
+    """사용자의 전체 비디오 분석 히스토리를 최신순으로 조회한다.
+
+    Args:
+        conn (Connection): SQLAlchemy 비동기 DB 커넥션.
+        user_id (int): 조회할 사용자 PK.
+
+    Returns:
+        list[VideoData]: 비디오 히스토리 스키마 리스트. 결과 없으면 빈 리스트.
+
+    Raises:
+        HTTPException 503: DB 쿼리 중 SQLAlchemy 오류 발생 시.
+        HTTPException 500: 예기치 못한 오류 발생 시.
+    """
     try:
         query = """
             SELECT id, user_id, video_loc, status, label, 
@@ -120,11 +151,25 @@ async def get_user_histories(conn: Connection, user_id: int):
     except Exception as e:
         print(e)
         raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="알수없는 이유로 문제가 발생하였습니다.")
- 
 
-# 사용자 개별 비디오 히스토리 조회
-# 호출 : image.py / video.py
-async def get_user_history(conn: Connection, user_id: int, video_id: int):
+
+async def get_user_history(conn: Connection, user_id: int, video_id: int) -> VideoDetailData | None:
+    """video_id와 user_id로 비디오 개별 상세 히스토리를 조회한다.
+
+    본인 소유 데이터만 조회되도록 user_id를 함께 필터링한다.
+
+    Args:
+        conn (Connection): SQLAlchemy 비동기 DB 커넥션.
+        user_id (int): 요청 사용자 PK.
+        video_id (int): 조회할 비디오 레코드 PK.
+
+    Returns:
+        VideoDetailData | None: 상세 히스토리 스키마 객체. 레코드 없으면 None.
+
+    Raises:
+        HTTPException 503: DB 쿼리 중 SQLAlchemy 오류 발생 시.
+        HTTPException 500: 예기치 못한 오류 발생 시.
+    """
     try:
         stmt = text("""
             SELECT id, user_id, video_loc, status, label, score, face_conf, face_ratio,
@@ -164,78 +209,26 @@ async def get_user_history(conn: Connection, user_id: int, video_id: int):
     except Exception as e:
         print(e)
         raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="알수없는 이유로 문제가 발생하였습니다.")
-    
-# 비회원 데이터 1분 후 자동 삭제 태스크
-# inference_svc.py : def process_video_task에서 추론이 성공 했을 때, 비회원일 경우 1분 후 자동 삭제
-@celery_app.task(name="cleanup_anonymous_video")
-def cleanup_anonymous_video(video_id: int, video_loc: str, is_success: bool):
-    loop = asyncio.new_event_loop()
-    asyncio.set_event_loop(loop)
-    try:
-        async def _delete():
-            success = True
-            async with celery_db_conn() as conn:
-                if is_success:
-                    try:
-                        await delete_video_meta_result(conn, video_id)
-                    except Exception as e :
-                        print(f"[Cleanup] video meta 삭제 실패 - video_id: {video_id}, error: {e}")
-                        success=False
-
-                    try:  
-                        await delete_video_frame_result(conn, video_id)   
-                    except Exception as e :
-                        print (f"[Cleanup] video frame 삭제 실패 - video_id: {video_id}, error: {e}")
-                        success=False
-                try:
-                    await delete_video_db(conn, video_id)
-                except Exception as e:
-                    print(f"[Cleanup] video_result 삭제 실패 - video_id: {video_id}, error:{e}")
-                    success=False
-            try:
-                await delete_video(video_loc)
-            except Exception as e:
-                print(f"[Cleanup] 비디오 파일 삭제 실패 - video_loc: {video_loc}, error: {e}")
-                success=False
-
-            if success:        
-                print(f"[Cleanup] 비회원 비디오 삭제 완료 - video_id: {video_id}, video_loc: {video_loc}")
-
-        loop.run_until_complete(_delete())
-    finally:
-        loop.close()
-
-# 비디오 DB 레코드 및 물리 파일 완전 삭제
-# 호출 : inference.py : 추론 FAIL일 때 delete_video and delete_video_db 실행
-async def delete_video_db(conn: Connection, video_id: int):
-    try:
-        delete_query = text("DELETE FROM video_result WHERE id = :video_id")
-        result = await conn.execute(delete_query, {"video_id": video_id})
-
-        if result.rowcount == 0:
-            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"해당 비디오 id {video_id}는(은) 존재하지 않아 삭제할 수 없습니다.")
-            
-        await conn.commit()
-
-    
-    except SQLAlchemyError as e:
-        print(e)
-        await conn.rollback()
-        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="요청하신 서비스가 잠시 내부적으로 문제가 발생하였습니다.")
-
-    except HTTPException:
-        raise
-    
-    except Exception as e:
-        print(e)
-        await conn.rollback()
-        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="알수없는 이유로 문제가 발생하였습니다.")
 
 
-# 빈 비디오 DB 생성 이후, video_id 반환(접수 완료)
-# 호출 : inference.py : video_id = await video_svc.register_video_result(conn, user_id, video_loc, version_type, model_type, domain_type)
 async def register_video_result(conn: Connection, user_id: int | None, video_loc: str,
-                                version_type: str, model_type: str, domain_type: str):
+                                version_type: str, model_type: str, domain_type: str) -> int:
+    """PENDING 상태의 빈 비디오 레코드를 DB에 생성하고 video_id를 반환한다.
+
+    Args:
+        conn (Connection): SQLAlchemy 비동기 DB 커넥션.
+        user_id (int | None): 로그인 사용자 ID. 비회원이면 None.
+        video_loc (str): 업로드된 비디오의 DB 기준 경로.
+        version_type (str): 모델 버전 타입.
+        model_type (str): 추론 모드 ("fast" | "pro").
+        domain_type (str): 탐지 도메인.
+
+    Returns:
+        int: 생성된 비디오 레코드의 PK (auto_increment).
+
+    Raises:
+        HTTPException 400: DB INSERT 실패 시. 업로드된 물리 파일도 함께 롤백 삭제된다.
+    """
     try:
         query = """
             INSERT INTO video_result (
@@ -266,12 +259,22 @@ async def register_video_result(conn: Connection, user_id: int | None, video_loc
         await delete_video(video_loc)
         raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
                             detail="요청데이터가 제대로 전달되지 않았습니다")
-        
-# 비디오 메타데이터 + 추론 결과값 DB에 저장
-# 호출 : inference_svc.py : 비디오 메타데이터 + 추론 결과값 DB에 저장 / 더미값으로 오류처리 (무한로딩 방지)
+
+
 async def update_video_result(conn: Connection, video_id: int, analysis: dict,
-                              result_msg: str, status: str):
-    
+                              result_msg: str, status: str) -> None:
+    """추론 완료 후 비디오 레코드에 분석 결과를 업데이트한다.
+
+    Args:
+        conn (Connection): SQLAlchemy 비동기 DB 커넥션.
+        video_id (int): 업데이트할 비디오 레코드 PK.
+        analysis (dict): 추론 결과값. prob, face_conf, face_ratio, face_brightness 포함.
+        result_msg (str): 처리 결과 메세지.
+        status (str): 추론 상태 ("success" | "warning" | "failed").
+
+    Raises:
+        SQLAlchemyError: DB 업데이트 실패 시 그대로 re-raise.
+    """
     if status == 'success':
         label = "FAKE" if analysis["prob"] > 0.5 else "REAL"
     else:
@@ -291,7 +294,7 @@ async def update_video_result(conn: Connection, video_id: int, analysis: dict,
         """
         
         db_status = status.upper()
-        
+  
         stmt = text(query)
         await conn.execute(stmt, {
             "status": db_status,
@@ -308,13 +311,23 @@ async def update_video_result(conn: Connection, video_id: int, analysis: dict,
     except SQLAlchemyError as e:
         await conn.rollback()
         raise e
-    
-# 비디오 결과 값 가져오기
-# 프론트엔드가 특정 비디오의 분석 상태 및 최종 결과를 요청할 때 사용
-# 호출 : inference.py - get_video_result() API 
-# video_data = await video_svc.get_video_result(conn, video_id) video_data에 저장
-async def get_video_result(conn: Connection, 
-                           video_id: int):
+
+
+async def get_video_result(conn: Connection, video_id: int) -> VideoDetailData:
+    """video_id로 비디오 분석 결과를 조회해 반환한다.
+
+    Args:
+        conn (Connection): SQLAlchemy 비동기 DB 커넥션.
+        video_id (int): 조회할 비디오 레코드 PK.
+
+    Returns:
+        VideoDetailData: 비디오 분석 결과 스키마 객체.
+
+    Raises:
+        HTTPException 404: 해당 video_id 레코드가 없을 때.
+        HTTPException 503: DB 쿼리 중 SQLAlchemy 오류 발생 시.
+        HTTPException 500: 예기치 못한 오류 발생 시.
+    """
     try:
         query = text("SELECT * FROM video_result WHERE id = :video_id")
         result = await conn.execute(query, {"video_id": video_id})
@@ -354,10 +367,19 @@ async def get_video_result(conn: Connection,
         raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, 
                             detail="알수없는 이유로 문제가 발생하였습니다.")
 
-# 비디오 메타 데이터 정보 저장하기
-# 비디오 1개에 대한 프레임 요약 통계를 video_meta_result 테이블에 삽입
-# 호출 위치: services/inference_svc.py - process_video_task() → run_inference()
-async def save_video_meta_result(conn: Connection, video_id: int, analysis: dict):
+
+async def save_video_meta_result(conn: Connection, video_id: int, analysis: dict) -> None:
+    """비디오 1개의 프레임 요약 통계를 video_meta_result 테이블에 저장한다.
+
+    Args:
+        conn (Connection): SQLAlchemy 비동기 DB 커넥션.
+        video_id (int): 비디오 레코드 PK.
+        analysis (dict): 추론 결과 딕셔너리. fps, total_frames, num_sampled,
+            num_extracted, num_detected 키를 포함해야 한다.
+
+    Raises:
+        SQLAlchemyError: DB INSERT 실패 시 그대로 re-raise.
+    """
     try:
         query = """
             INSERT INTO video_meta_result (video_id, fps, total_frames, num_sampled, num_extracted, num_detected)
@@ -376,12 +398,21 @@ async def save_video_meta_result(conn: Connection, video_id: int, analysis: dict
     except SQLAlchemyError as e:
         await conn.rollback()
         raise e
-    
-# 비디오 상세 결과 값 저장하기 
-# 비디오 프레임 별 딥페이크 점수, 얼굴 신뢰도, 얼굴 비율, 얼굴 조도 반환
-# 비디오 1개에 속한 여러 프레임의 분석 결과를 한 번에 INSERT.
-# 호출 위치: services/inference_svc.py - process_video_task() → run_inference()
-async def save_video_frame_result(conn: Connection, video_id: int, frame_results: list):
+
+
+async def save_video_frame_result(conn: Connection, video_id: int, frame_results: list) -> None:
+    """비디오에 속한 모든 프레임의 분석 결과를 video_frame_result 테이블에 일괄 저장한다.
+
+    Args:
+        conn (Connection): SQLAlchemy 비동기 DB 커넥션.
+        video_id (int): 비디오 레코드 PK.
+        frame_results (list): 프레임별 분석 결과 딕셔너리 리스트.
+            각 항목은 frame_index, frame_time, score, face_conf,
+            face_ratio, face_brightness 키를 포함해야 한다.
+
+    Raises:
+        SQLAlchemyError: DB INSERT 실패 시 그대로 re-raise.
+    """
     try:
         query = """
             INSERT INTO video_frame_result
@@ -407,9 +438,20 @@ async def save_video_frame_result(conn: Connection, video_id: int, frame_results
     except SQLAlchemyError as e:
         await conn.rollback()
         raise e
-        
-    
-async def delete_video_meta_result(conn: Connection, video_id: int):
+
+
+async def delete_video_meta_result(conn: Connection, video_id: int) -> None:
+    """video_meta_result 테이블에서 해당 비디오의 메타 데이터를 삭제한다.
+
+    Args:
+        conn (Connection): SQLAlchemy 비동기 DB 커넥션.
+        video_id (int): 삭제할 비디오 레코드 PK.
+
+    Raises:
+        HTTPException 404: 해당 video_id 레코드가 없을 때.
+        HTTPException 503: DB 쿼리 중 SQLAlchemy 오류 발생 시.
+        HTTPException 500: 예기치 못한 오류 발생 시.
+    """
     try:
         delete_query = text("""
             DELETE FROM video_meta_result
@@ -418,7 +460,7 @@ async def delete_video_meta_result(conn: Connection, video_id: int):
         result = await conn.execute(delete_query, {"video_id": video_id})
         if result.rowcount == 0:
             raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"해당 비디오 id {video_id}는(은) 존재하지 않아 삭제할 수 없습니다.")
-            
+
         await conn.commit()
 
     except SQLAlchemyError as e:
@@ -435,7 +477,18 @@ async def delete_video_meta_result(conn: Connection, video_id: int):
         raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="알수없는 이유로 문제가 발생하였습니다.")
 
 
-async def delete_video_frame_result(conn: Connection, video_id: int):
+async def delete_video_frame_result(conn: Connection, video_id: int) -> None:
+    """video_frame_result 테이블에서 해당 비디오의 프레임 결과를 삭제한다.
+
+    Args:
+        conn (Connection): SQLAlchemy 비동기 DB 커넥션.
+        video_id (int): 삭제할 비디오 레코드 PK.
+
+    Raises:
+        HTTPException 404: 해당 video_id 레코드가 없을 때.
+        HTTPException 503: DB 쿼리 중 SQLAlchemy 오류 발생 시.
+        HTTPException 500: 예기치 못한 오류 발생 시.
+    """
     try:
         delete_query = text("""
             DELETE FROM video_frame_result
@@ -461,12 +514,23 @@ async def delete_video_frame_result(conn: Connection, video_id: int):
         raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="알수없는 이유로 문제가 발생하였습니다.")
 
 
+async def get_video_meta_result(conn: Connection, video_id: int) -> VideoMetaData:
+    """video_id로 비디오 메타 데이터를 조회한다.
+
+    fps, total_frames, num_sampled, num_extracted, num_detected를 반환한다.
 
+    Args:
+        conn (Connection): SQLAlchemy 비동기 DB 커넥션.
+        video_id (int): 조회할 비디오 레코드 PK.
 
-# 비디오 메타 데이터 정보 가져오기
-# 호출 위치: routers/inference.py - get_video_detail()
-# 초당프레임수, 영상 전체 프레임 수, 샘플링한 프레임 수, 얼굴 추출에 성공한 프레임 수, score산출 성공 프레임 수
-async def get_video_meta_result(conn: Connection, video_id: int):         
+    Returns:
+        VideoMetaData: 비디오 메타 데이터 스키마 객체.
+
+    Raises:
+        HTTPException 404: 해당 video_id 메타 레코드가 없을 때.
+        HTTPException 503: DB 쿼리 중 SQLAlchemy 오류 발생 시.
+        HTTPException 500: 예기치 못한 오류 발생 시.
+    """
     try:
         query = text("""
             SELECT fps, total_frames, num_sampled, num_extracted, num_detected
@@ -496,11 +560,24 @@ async def get_video_meta_result(conn: Connection, video_id: int):
     except Exception as e:
         print(e)
         raise HTTPException(status_code=500, detail="알수없는 이유로 문제가 발생하였습니다.")
- 
-# 비디오 프레임별 상세 결과 조회
-# 호출 위치: routers/inference.py - get_video_detail()
-# video_id에 속한 모든 프레임 결과를 frame_index 반환 하여, 프레임별 점수 그래프, 의심 구간 표시 등 시각화에 활용.
-async def get_video_frame_result(conn: Connection, video_id: int):
+
+
+async def get_video_frame_result(conn: Connection, video_id: int) -> list[VideoFrameData]:
+    """video_id에 속한 모든 프레임 분석 결과를 frame_index 오름차순으로 조회한다.
+
+    프레임별 점수 그래프, 의심 구간 표시 등 시각화에 활용된다.
+
+    Args:
+        conn (Connection): SQLAlchemy 비동기 DB 커넥션.
+        video_id (int): 조회할 비디오 레코드 PK.
+
+    Returns:
+        list[VideoFrameData]: 프레임별 분석 결과 스키마 리스트.
+
+    Raises:
+        HTTPException 503: DB 쿼리 중 SQLAlchemy 오류 발생 시.
+        HTTPException 500: 예기치 못한 오류 발생 시.
+    """
     try:
         query = text("""
             SELECT frame_index, frame_time, score, face_conf, face_ratio, face_brightness
@@ -522,8 +599,6 @@ async def get_video_frame_result(conn: Connection, video_id: int):
             for r in result
         ]
 
-        
-        
     except SQLAlchemyError as e:
         print(f"[Frame Query Error] {e}")
         raise HTTPException(status_code=503, detail="데이터베이스 조회 중 문제가 발생했습니다.")
@@ -531,10 +606,23 @@ async def get_video_frame_result(conn: Connection, video_id: int):
         print(e)
         raise HTTPException(status_code=500, detail="알수없는 이유로 문제가 발생하였습니다.")
 
-# 비디오 특정 프레임의 Frame Time 조회
-# 호출 위치: routers/explain.py - explain_frame()
-# 비디오 내 frame_index에 해당하는 특정 frame의 frame_time만 추출한다
-async def get_video_frame_by_index(conn: Connection, video_id: int, frame_index: int):
+
+async def get_video_frame_by_index(conn: Connection, video_id: int, frame_index: int) -> float:
+    """video_id와 frame_index로 특정 프레임의 frame_time을 조회한다.
+
+    Args:
+        conn (Connection): SQLAlchemy 비동기 DB 커넥션.
+        video_id (int): 비디오 레코드 PK.
+        frame_index (int): 조회할 프레임 인덱스.
+
+    Returns:
+        float: 해당 프레임의 타임스탬프 (초 단위).
+
+    Raises:
+        HTTPException 404: 해당 frame_index가 없을 때.
+        HTTPException 503: DB 쿼리 중 SQLAlchemy 오류 발생 시.
+        HTTPException 500: 예기치 못한 오류 발생 시.
+    """
     try:
         query = text("""
             SELECT frame_time
@@ -559,6 +647,87 @@ async def get_video_frame_by_index(conn: Connection, video_id: int, frame_index:
     except Exception as e:
         print(e)
         raise HTTPException(status_code=500, detail="알수없는 이유로 문제가 발생하였습니다.")
-    
 
-    
\ No newline at end of file
+
+@celery_app.task(name="cleanup_anonymous_video")
+def cleanup_anonymous_video(video_id: int, video_loc: str, is_success: bool) -> None:
+    """비회원 비디오 DB 레코드 및 물리 파일을 삭제하는 Celery 태스크.
+
+    추론 성공 시 meta/frame 결과도 함께 삭제한다.
+    각 삭제 단계는 독립적으로 처리되어 한쪽 실패가 다른쪽에 영향을 주지 않는다.
+
+    Args:
+        video_id (int): 삭제할 비디오 레코드 PK.
+        video_loc (str): 삭제할 비디오의 DB 기준 경로.
+        is_success (bool): 추론 성공 여부. True면 meta/frame 결과도 함께 삭제.
+    """
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    try:
+        async def _delete():
+            success = True
+            async with celery_db_conn() as conn:
+                if is_success:
+                    try:
+                        await delete_video_meta_result(conn, video_id)
+                    except Exception as e:
+                        print(f"[Cleanup] video meta 삭제 실패 - video_id: {video_id}, error: {e}")
+                        success=False
+
+                    try:  
+                        await delete_video_frame_result(conn, video_id)   
+                    except Exception as e:
+                        print(f"[Cleanup] video frame 삭제 실패 - video_id: {video_id}, error: {e}")
+                        success=False
+                try:
+                    await delete_video_db(conn, video_id)
+                except Exception as e:
+                    print(f"[Cleanup] video_result 삭제 실패 - video_id: {video_id}, error:{e}")
+                    success=False
+            try:
+                await delete_video(video_loc)
+            except Exception as e:
+                print(f"[Cleanup] 비디오 파일 삭제 실패 - video_loc: {video_loc}, error: {e}")
+                success=False
+
+            if success:        
+                print(f"[Cleanup] 비회원 비디오 삭제 완료 - video_id: {video_id}, video_loc: {video_loc}")
+
+        loop.run_until_complete(_delete())
+    finally:
+        loop.close()
+
+
+async def delete_video_db(conn: Connection, video_id: int) -> None:
+    """DB에서 비디오 레코드를 삭제한다.
+
+    Args:
+        conn (Connection): SQLAlchemy 비동기 DB 커넥션.
+        video_id (int): 삭제할 비디오 레코드 PK.
+
+    Raises:
+        HTTPException 404: 해당 video_id 레코드가 없을 때.
+        HTTPException 503: DB 쿼리 중 SQLAlchemy 오류 발생 시.
+        HTTPException 500: 예기치 못한 오류 발생 시.
+    """
+    try:
+        delete_query = text("DELETE FROM video_result WHERE id = :video_id")
+        result = await conn.execute(delete_query, {"video_id": video_id})
+
+        if result.rowcount == 0:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"해당 비디오 id {video_id}는(은) 존재하지 않아 삭제할 수 없습니다.")
+            
+        await conn.commit()
+
+    except SQLAlchemyError as e:
+        print(e)
+        await conn.rollback()
+        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="요청하신 서비스가 잠시 내부적으로 문제가 발생하였습니다.")
+
+    except HTTPException:
+        raise
+    
+    except Exception as e:
+        print(e)
+        await conn.rollback()
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="알수없는 이유로 문제가 발생하였습니다.")
\ No newline at end of file