app.py

import gradio as gr
import subprocess
import os
from datetime import datetime
from PIL import Image
import tempfile
import logging

# 로깅 설정
logging.basicConfig(
    filename='app.log',
    filemode='a',
    format='%(asctime)s - %(levelname)s - %(message)s',
    level=logging.DEBUG
)

def on_video_upload(video):
    logging.info("on_video_upload 함수 호출")
    try:
        # 비디오 입력 처리
        if isinstance(video, str):
            video_path = video
            logging.debug(f"업로드된 비디오 경로 (str): {video_path}")
        elif isinstance(video, dict) and 'name' in video:
            video_path = video['name']
            logging.debug(f"업로드된 비디오 경로 (dict): {video_path}")
        else:
            logging.error("비디오 파일 경로를 인식할 수 없습니다.")
            raise Exception("비디오 파일 경로를 인식할 수 없습니다.")

        if not os.path.exists(video_path):
            logging.error(f"비디오 파일이 존재하지 않습니다: {video_path}")
            raise Exception("비디오 파일이 존재하지 않습니다.")

        # 영상 길이 추출
        probe_duration = subprocess.run(
            ["ffprobe", "-v", "error", "-show_entries",
             "format=duration", "-of",
             "default=noprint_wrappers=1:nokey=1", video_path],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True
        )
        if probe_duration.returncode != 0:
            logging.error(f"ffprobe duration 에러: {probe_duration.stderr}")
            raise Exception("영상 길이 추출 실패")
        duration = float(probe_duration.stdout.strip())
        duration_str = str(datetime.utcfromtimestamp(duration).strftime('%H:%M:%S'))
        logging.debug(f"영상 길이: {duration_str}")

        # 해상도 추출
        probe_resolution = subprocess.run(
            ["ffprobe", "-v", "error", "-select_streams", "v:0",
             "-show_entries", "stream=width,height",
             "-of", "csv=s=x:p=0", video_path],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True
        )
        if probe_resolution.returncode != 0:
            logging.error(f"ffprobe resolution 에러: {probe_resolution.stderr}")
            raise Exception("영상 해상도 추출 실패")
        resolution = probe_resolution.stdout.strip()
        logging.debug(f"영상 해상도: {resolution}")

        # 시작 시간과 끝 시간 초기화
        start_time = "00:00:00"
        end_time = duration_str

        # 캡쳐본 생성
        start_capture = capture_frame(video_path, start_time)
        end_capture = capture_frame(video_path, end_time)

        return duration_str, resolution, start_time, end_time, start_capture, end_capture
    except Exception as e:
        logging.exception("on_video_upload 함수에서 에러 발생")
        return "에러 발생", "에러 발생", "에러", "에러", None, None

def on_any_change(video, start_time, end_time):
    logging.info("on_any_change 함수 호출")
    try:
        # 비디오 입력 처리
        if isinstance(video, str):
            video_path = video
            logging.debug(f"비디오 경로 (str): {video_path}")
        elif isinstance(video, dict) and 'name' in video:
            video_path = video['name']
            logging.debug(f"비디오 경로 (dict): {video_path}")
        else:
            logging.error("비디오 파일 경로를 인식할 수 없습니다.")
            raise Exception("비디오 파일 경로를 인식할 수 없습니다.")

        if not os.path.exists(video_path):
            logging.error(f"비디오 파일이 존재하지 않습니다: {video_path}")
            raise Exception("비디오 파일이 존재하지 않습니다.")

        logging.debug(f"시작 시간: {start_time}, 끝 시간: {end_time}")

        start_capture = capture_frame(video_path, start_time)
        end_capture = capture_frame(video_path, end_time)
        return start_capture, end_capture
    except Exception as e:
        logging.exception("on_any_change 함수에서 에러 발생")
        return None, None

def on_generate_click(video, start_time, end_time, fps, resize_factor, speed_factor):
    logging.info("on_generate_click 함수 호출")
    try:
        # 비디오 입력 처리
        if isinstance(video, str):
            video_path = video
            logging.debug(f"비디오 경로 (str): {video_path}")
        elif isinstance(video, dict) and 'name' in video:
            video_path = video['name']
            logging.debug(f"비디오 경로 (dict): {video_path}")
        else:
            logging.error("비디오 파일 경로를 인식할 수 없습니다.")
            raise Exception("비디오 파일 경로를 인식할 수 없습니다.")

        if not os.path.exists(video_path):
            logging.error(f"비디오 파일이 존재하지 않습니다: {video_path}")
            raise Exception("비디오 파일이 존재하지 않습니다.")

        logging.debug(f"시작 시간: {start_time}, 끝 시간: {end_time}")
        logging.debug(f"FPS: {fps}, 해상도 배율: {resize_factor}, 배속: {speed_factor}")

        # 시작 시간과 끝 시간의 유효성 검사
        start_seconds = time_str_to_seconds(start_time)
        end_seconds = time_str_to_seconds(end_time)
        if start_seconds >= end_seconds:
            logging.error("시작 시간이 끝 시간보다 크거나 같습니다.")
            return None, "시작 시간이 끝 시간보다 크거나 같습니다.", None

        # 영상 길이와 일치 여부 확인
        probe_duration = subprocess.run(
            ["ffprobe", "-v", "error", "-show_entries",
             "format=duration", "-of",
             "default=noprint_wrappers=1:nokey=1", video_path],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True
        )
        if probe_duration.returncode != 0:
            logging.error(f"ffprobe duration 에러: {probe_duration.stderr}")
            raise Exception("영상 길이 추출 실패")
        duration_video = float(probe_duration.stdout.strip())
        if end_seconds > duration_video:
            logging.error("끝 시간이 영상 길이를 초과합니다.")
            return None, "끝 시간이 영상 길이를 초과합니다.", None

        # GIF의 지속 시간 계산 (end_time - start_time)
        gif_duration_seconds = end_seconds - start_seconds
        gif_duration_str = str(datetime.utcfromtimestamp(gif_duration_seconds).strftime('%H:%M:%S'))
        logging.debug(f"GIF 지속 시간: {gif_duration_str} (초: {gif_duration_seconds})")

        # 임시 파일 생성
        with tempfile.NamedTemporaryFile(suffix='.gif', delete=False) as tmp_gif:
            gif_path = tmp_gif.name
        logging.debug(f"생성된 GIF 임시 경로: {gif_path}")

        # GIF 생성 명령어
        cmd = [
            "ffmpeg",
            "-y",  # 기존 파일 덮어쓰기
            "-ss", start_time,
            "-to", end_time,
            "-i", video_path,
            "-vf", f"fps={fps},scale=iw*{resize_factor}:ih*{resize_factor}",
            gif_path
        ]
        logging.debug(f"실행할 ffmpeg 명령어: {' '.join(cmd)}")
        result = subprocess.run(cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
        logging.debug(f"ffmpeg stdout: {result.stdout}")
        logging.debug(f"ffmpeg stderr: {result.stderr}")
        logging.info("GIF 생성 완료")

        # GIF 속도 조절 (speed_factor)
        if speed_factor != 1:
            gif_fast = gif_path.replace('.gif', '_fast.gif')
            cmd_speed = [
                "ffmpeg",
                "-y",  # 기존 파일 덮어쓰기
                "-i", gif_path,
                "-filter_complex", f"[0:v]setpts={1/speed_factor}*PTS[v]",
                "-map", "[v]",
                gif_fast
            ]
            logging.debug(f"실행할 속도 조절 ffmpeg 명령어: {' '.join(cmd_speed)}")
            result_speed = subprocess.run(cmd_speed, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
            logging.debug(f"ffmpeg speed 조절 stdout: {result_speed.stdout}")
            logging.debug(f"ffmpeg speed 조절 stderr: {result_speed.stderr}")
            os.remove(gif_path)
            gif_path = gif_fast
            logging.info(f"GIF 속도 조절 완료: {gif_path}")

        # 파일 크기 측정
        file_size = os.path.getsize(gif_path)
        logging.debug(f"생성된 GIF 파일 크기: {file_size} bytes")

        # 다운로드를 위한 파일 경로
        gif_download = gif_path

        return gif_path, str(file_size), gif_download
    except subprocess.CalledProcessError as e:
        logging.error(f"subprocess 에러: {e.stderr}")
        return None, f"GIF 생성 중 subprocess 에러 발생: {e.stderr}", None
    except Exception as e:
        logging.exception("on_generate_click 함수에서 에러 발생")
        return None, f"에러 발생: {str(e)}", None

def capture_frame(video_path, timestamp):
    logging.info(f"capture_frame 함수 호출: {video_path} at {timestamp}")
    try:
        with tempfile.NamedTemporaryFile(suffix='.png', delete=False) as tmp_img:
            img_path = tmp_img.name
        logging.debug(f"생성된 이미지 임시 경로: {img_path}")

        cmd = [
            "ffmpeg",
            "-y",  # 기존 파일 덮어쓰기
            "-ss", timestamp,
            "-i", video_path,
            "-vframes", "1",
            "-q:v", "2",
            img_path
        ]
        logging.debug(f"실행할 ffmpeg 캡쳐 명령어: {' '.join(cmd)}")
        result = subprocess.run(cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
        logging.debug(f"ffmpeg 캡쳐 stdout: {result.stdout}")
        logging.debug(f"ffmpeg 캡쳐 stderr: {result.stderr}")
        logging.info(f"프레임 캡쳐 완료: {img_path}")
        return img_path
    except subprocess.CalledProcessError as e:
        logging.error(f"ffmpeg 캡쳐 에러: {e.stderr}")
        return None
    except Exception as e:
        logging.exception("capture_frame 함수에서 에러 발생")
        return None

def time_str_to_seconds(time_str):
    """HH:MM:SS 포맷의 시간을 초 단위로 변환"""
    try:
        parts = time_str.split(':')
        if len(parts) != 3:
            raise ValueError("시간 포맷이 올바르지 않습니다.")
        hours, minutes, seconds = map(int, parts)
        return hours * 3600 + minutes * 60 + seconds
    except Exception as e:
        logging.error(f"time_str_to_seconds 에러: {e}")
        raise

with gr.Blocks() as demo:
    gr.Markdown("# 영상 처리 컨트롤 타워")

    with gr.Row():
        video_input = gr.Video(label="영상 업로드")

    with gr.Row():
        duration_text = gr.Textbox(label="영상 길이", interactive=False)
        resolution_text = gr.Textbox(label="해상도", interactive=False)

    with gr.Row():
        start_time = gr.Textbox(label="시작 시간 (HH:MM:SS)", value="00:00:00")
        end_time = gr.Textbox(label="끝 시간 (HH:MM:SS)", value="00:00:00")

    with gr.Row():
        start_image = gr.Image(label="시작 지점 캡쳐본")
        end_image = gr.Image(label="끝 지점 캡쳐본")

    with gr.Row():
        fps_slider = gr.Slider(label="FPS", minimum=1, maximum=60, step=1, value=10)
        resize_slider = gr.Slider(label="해상도 배율", minimum=0.5, maximum=2.0, step=0.1, value=1.0)
        speed_slider = gr.Slider(label="배속", minimum=0.5, maximum=2.0, step=0.1, value=1.0)

    with gr.Row():
        generate_button = gr.Button("GIF 생성")

    with gr.Row():
        gif_output = gr.Image(label="결과 GIF")
        file_size_output = gr.Textbox(label="파일 용량")
        download_output = gr.File(label="GIF 다운로드")

    # 이벤트 연결
    video_input.upload(on_video_upload, inputs=video_input, outputs=[duration_text, resolution_text, start_time, end_time, start_image, end_image])
    start_time.change(on_any_change, inputs=[video_input, start_time, end_time], outputs=[start_image, end_image])
    end_time.change(on_any_change, inputs=[video_input, start_time, end_time], outputs=[start_image, end_image])
    generate_button.click(on_generate_click, inputs=[video_input, start_time, end_time, fps_slider, resize_slider, speed_slider], outputs=[gif_output, file_size_output, download_output])

demo.launch()