""" FastAPI 서버 - 유튜브 영상 분석 API 허깅페이스 스페이스 배포용 """ from fastapi import FastAPI, HTTPException, Query from fastapi.responses import JSONResponse, HTMLResponse from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field from typing import Optional, Literal import os import tempfile import json from pathlib import Path # 로컬 모듈 import from youtube_segmenter import process_youtube_video from youtube_summarizer import summarize_youtube_segments # FastAPI 앱 생성 app = FastAPI( title="YouTube Video Analyzer API", description=""" # 유튜브 영상 분석 API 이 API는 유튜브 영상의 자막을 주제별로 분리하고 AI로 요약하는 서비스입니다. ## 주요 기능 ### 1. 세그먼트 분리 - **ruptures PELT 알고리즘**: 통계적 변화점 탐지로 대주제 분리 - **의미 유사도 분석**: Sentence Transformers로 소주제 정밀 분할 - **타임스탬프 매핑**: 각 세그먼트의 영상 시작 시간 제공 ### 2. AI 요약 - **Gemini 2.0 Flash Lite**: Google의 최신 경량 AI 모델 사용 - **주제 라벨링**: 2-5단어로 핵심 주제 추출 - **스마트 필터링**: 의미 없는 짧은 리액션 자동 제외 ## 기술 스택 - **youtube-transcript-api**: 자막 추출 - **ruptures**: 통계적 변화점 탐지 - **sentence-transformers**: 문장 임베딩 - **google-generativeai**: AI 요약 - **FastAPI**: REST API 서버 ## 파라미터 가이드 ### Penalty (대주제 분리) - **1.0-3.0**: 매우 세밀한 분리 (많은 주제) - **5.0**: 균형잡힌 분리 (기본값) - **7.0-10.0**: 큰 주제 단위 분리 - **15.0-30.0**: 매우 큰 덩어리로 분리 ### Threshold (소주제 분리) - **70-80**: 세밀한 분리 - **85-92**: 균형잡힌 분리 (권장) - **90**: 기본값 - **95-98**: 큰 덩어리 분리 ## 사용 예시 ### GET 요청 ``` /process?youtube_url=https://youtube.com/watch?v=VIDEO_ID&mode=summary&penalty=5.0&threshold=90&gemini_api_key=YOUR_KEY&youtube_api_key=YOUR_KEY ``` ### POST 요청 ```json { "youtube_url": "https://youtube.com/watch?v=VIDEO_ID", "mode": "summary", "penalty": 5.0, "threshold": 90, "gemini_api_key": "your_api_key_here", "youtube_api_key": "your_youtube_api_key_here" } ``` """, version="1.0.0", contact={ "name": "YouTube Analyzer", "url": "https://github.com/yourusername/youtube-analyzer" } ) # CORS 설정 app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Request 모델 class ProcessRequest(BaseModel): youtube_api_key: str = Field(..., description="YouTube API 키 (필수)", example="YOUR_YOUTUBE_API_KEY") youtube_url: str = Field(..., description="유튜브 URL", example="https://www.youtube.com/watch?v=tLwZfLMHjKo") mode: Literal["segment", "summary"] = Field( default="summary", description="처리 모드: 'segment' (분리만) 또는 'summary' (분리+요약)" ) penalty: float = Field( default=5.0, ge=1.0, le=30.0, description="대주제 분리 민감도 (1.0-30.0)\n- **작은 값 (1.0~3.0)**: 변화를 매우 민감하게 감지하여 주제를 잘게 쪼갭니다.\n- **기본 값 (5.0)**: 권장하는 기본 설정입니다.\n- **큰 값 (10.0+)**: 전체적인 큰 흐름 위주로 굵직하게 주제를 나눕니다." ) threshold: int = Field( default=90, ge=50, le=99, description="소주제 분리 민감도 (50-99)\n- **낮은 값 (70-80)**: 문장 간의 미세한 차이도 감지하여 세밀하게 나눕니다.\n- **기본 값 (90)**: 권장하는 기본 설정입니다.\n- **높은 값 (95+)**: 아주 확실한 주제 변화가 있을 때만 나눕니다." ) gemini_api_key: Optional[str] = Field( default=None, description="Gemini API 키 (요약 모드에서 필수)" ) class Config: schema_extra = { "example": { "youtube_url": "https://www.youtube.com/watch?v=tLwZfLMHjKo", "mode": "summary", "penalty": 5.0, "threshold": 90, "gemini_api_key": "your_gemini_api_key_here", "youtube_api_key": "your_youtube_api_key_here" } } # Response 모델 class SegmentItem(BaseModel): url: str chunk_id: str chunk_time: str text: str class SummaryItem(BaseModel): url: str chunk_id: str chunk_time: str text: str summary: Optional[str] topic: Optional[str] @app.get("/", response_class=HTMLResponse, include_in_schema=False) async def root(): """API 홈페이지 - Swagger UI로 리다이렉트""" return """ Redirecting...

Redirecting to API Documentation...

""" @app.get("/health") async def health_check(): """헬스 체크 엔드포인트""" return {"status": "healthy", "message": "API is running"} @app.get("/process", tags=["Processing"]) async def process_get( youtube_api_key: str = Query(..., description="YouTube API 키 (필수)"), youtube_url: str = Query(..., description="유튜브 URL", example="https://www.youtube.com/watch?v=tLwZfLMHjKo"), mode: Literal["segment", "summary"] = Query( default="summary", description="처리 모드: 'segment' (분리만) 또는 'summary' (분리+요약)" ), penalty: float = Query( default=5.0, ge=1.0, le=30.0, description="대주제 분리 민감도 (1.0-30.0)\n- **작은 값 (1.0~3.0)**: 변화를 매우 민감하게 감지하여 주제를 잘게 쪼갭니다.\n- **기본 값 (5.0)**: 권장하는 기본 설정입니다.\n- **큰 값 (10.0+)**: 전체적인 큰 흐름 위주로 굵직하게 주제를 나눕니다." ), threshold: int = Query( default=90, ge=50, le=99, description="소주제 분리 민감도 (50-99)\n- **낮은 값 (70-80)**: 문장 간의 미세한 차이도 감지하여 세밀하게 나눕니다.\n- **기본 값 (90)**: 권장하는 기본 설정입니다.\n- **높은 값 (95+)**: 아주 확실한 주제 변화가 있을 때만 나눕니다." ), gemini_api_key: Optional[str] = Query( default=None, description="Gemini API 키 (요약 모드에서 필수)" ) ): """ GET 방식으로 유튜브 영상 처리 - **youtube_api_key**: YouTube API 키 (필수) - **youtube_url**: 유튜브 URL (필수) - **mode**: segment (분리만) 또는 summary (분리+요약) - **penalty**: 대주제 분리 민감도 (기본값: 5.0) - **threshold**: 소주제 분리 민감도 (기본값: 90) - **gemini_api_key**: Gemini API 키 (요약 모드에서 필요) """ request = ProcessRequest( youtube_api_key=youtube_api_key, youtube_url=youtube_url, mode=mode, penalty=penalty, threshold=threshold, gemini_api_key=gemini_api_key ) return await process_video(request) @app.post("/process", tags=["Processing"]) async def process_post(request: ProcessRequest): """ POST 방식으로 유튜브 영상 처리 Request Body: ```json { "youtube_api_key": "your_youtube_api_key", "youtube_url": "https://youtube.com/watch?v=VIDEO_ID", "mode": "summary", "penalty": 5.0, "threshold": 90, "gemini_api_key": "your_gemini_api_key" } ``` Response: - mode가 "segment"인 경우: 세그먼트 배열 반환 - mode가 "summary"인 경우: 요약 포함 세그먼트 배열 반환 """ return await process_video(request) async def process_video(request: ProcessRequest): """실제 처리 로직""" try: # YouTube API 키 설정 (필수값) os.environ['YOUTUBE_API_KEY'] = request.youtube_api_key # Gemini API 키 설정 (요약 모드인 경우) if request.mode == "summary": if request.gemini_api_key: os.environ['GEMINI_KEY'] = request.gemini_api_key elif not os.getenv('GEMINI_KEY'): raise HTTPException( status_code=400, detail="요약 모드에서는 gemini_api_key가 필요합니다. (환경변수 또는 파라미터로 제공)" ) # 임시 디렉토리 생성 with tempfile.TemporaryDirectory() as temp_dir: # 1단계: 세그먼트 분리 segment_file = process_youtube_video( request.youtube_url, penalty=request.penalty, threshold=request.threshold, output_dir=temp_dir ) # 세그먼트 파일 읽기 with open(segment_file, 'r', encoding='utf-8') as f: segment_data = json.load(f) # 분리만 모드인 경우 if request.mode == "segment": return JSONResponse(content={ "status": "success", "mode": "segment", "youtube_url": request.youtube_url, "total_segments": len(segment_data), "segments": segment_data }) # 2단계: 요약 생성 summary_file = summarize_youtube_segments( segment_file, output_json_path=os.path.join(temp_dir, "summary.json"), gemini_api_key=request.gemini_api_key ) # 요약 파일 읽기 with open(summary_file, 'r', encoding='utf-8') as f: summary_data = json.load(f) return JSONResponse(content={ "status": "success", "mode": "summary", "youtube_url": request.youtube_url, "total_segments": len(summary_data), "segments": summary_data }) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/info", tags=["Information"]) async def api_info(): """ API 정보 조회 프로젝트의 목적, 기술 스택, 사용 방법 등을 제공합니다. """ return { "project_name": "YouTube Video Analyzer", "version": "1.0.0", "description": "유튜브 영상의 자막을 주제별로 분리하고 AI로 요약하는 서비스", "purpose": { "ko": "긴 유튜브 영상의 내용을 빠르게 파악하고, 원하는 부분으로 이동할 수 있도록 돕습니다.", "en": "Helps quickly understand long YouTube videos and navigate to desired sections." }, "features": [ "통계적 변화점 탐지 (ruptures PELT)", "의미 기반 세그먼트 분할 (Sentence Transformers)", "타임스탬프 매핑", "AI 요약 및 주제 라벨링 (Gemini 2.0 Flash Lite)" ], "tech_stack": { "backend": "FastAPI", "ai_models": ["sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2", "gemini-2.0-flash-lite"], "algorithms": ["ruptures PELT", "Semantic Chunking"], "deployment": "Hugging Face Spaces" }, "endpoints": { "GET /process": "유튜브 영상 처리 (쿼리 파라미터 사용)", "POST /process": "유튜브 영상 처리 (JSON body 사용)", "GET /health": "헬스 체크", "GET /info": "API 정보" } } if __name__ == "__main__": import uvicorn # 타임아웃 설정을 300초(5분)로 증가 uvicorn.run(app, host="0.0.0.0", port=7860, timeout_keep_alive=300)