from fastapi import FastAPI, BackgroundTasks, Request from fastapi.responses import HTMLResponse, JSONResponse, Response from fastapi.staticfiles import StaticFiles import pathlib, os, uvicorn, base64, json, uuid, time from typing import Dict, List, Any, Optional import asyncio import logging import threading import concurrent.futures import requests import fitz logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) BASE = pathlib.Path(__file__).parent app = FastAPI() app.mount("/static", StaticFiles(directory=BASE), name="static") CACHE_DIR = BASE / "cache" if not CACHE_DIR.exists(): CACHE_DIR.mkdir(parents=True) EMBEDDING_DIR = BASE / "embeddings" if not EMBEDDING_DIR.exists(): EMBEDDING_DIR.mkdir(parents=True) # Fireworks AI API 설정 (VLM 모델) FIREWORKS_API_KEY = os.getenv("FIREWORKS_API", "").strip() # 줄바꿈/공백 제거 FIREWORKS_API_URL = "https://api.fireworks.ai/inference/v1/chat/completions" FIREWORKS_VLM_MODEL = "accounts/fireworks/models/qwen3-vl-235b-a22b-instruct" HAS_VALID_API_KEY = bool(FIREWORKS_API_KEY) if HAS_VALID_API_KEY: logger.info("Fireworks AI VLM API 키 설정 완료") else: logger.warning("유효한 Fireworks AI API 키가 없습니다. AI 기능이 제한됩니다.") # 고정 PDF 파일 경로 PROMPT_PDF_PATH = BASE / "prompt.pdf" PROMPT_PDF_ID = "prompt_pdf_main" # 다중 PDF 지원 PDF_FILES = { "prompt": { "path": BASE / "prompt.pdf", "id": "prompt_pdf_main", "name": "샘플-프롬프트북" }, "ktx": { "path": BASE / "ktx2512.pdf", "id": "ktx_pdf_main", "name": "샘플-코레일잡지" } } # 현재 선택된 PDF (기본값: prompt) current_pdf_key = "prompt" pdf_cache: Dict[str, Dict[str, Any]] = {} cache_locks = {} pdf_embeddings: Dict[str, Dict[str, Any]] = {} # VLM 분석 상태 추적 (메모리) analysis_status: Dict[str, Dict[str, Any]] = {} def get_cache_path(pdf_name: str): return CACHE_DIR / f"{pdf_name}_cache.json" def get_embedding_path(pdf_id: str): return EMBEDDING_DIR / f"{pdf_id}_embedding.json" def get_analysis_cache_path(pdf_id: str): """VLM 분석 결과 캐시 경로""" return EMBEDDING_DIR / f"{pdf_id}_vlm_analysis.json" def load_analysis_cache(pdf_id: str) -> Optional[Dict[str, Any]]: """VLM 분석 캐시 로드""" cache_path = get_analysis_cache_path(pdf_id) if cache_path.exists(): try: with open(cache_path, "r", encoding="utf-8") as f: data = json.load(f) logger.info(f"VLM 분석 캐시 로드 완료: {pdf_id}") return data except Exception as e: logger.error(f"분석 캐시 로드 오류: {e}") return None def save_analysis_cache(pdf_id: str, analysis_data: Dict[str, Any]): """VLM 분석 결과 캐시 저장""" cache_path = get_analysis_cache_path(pdf_id) try: with open(cache_path, "w", encoding="utf-8") as f: json.dump(analysis_data, f, ensure_ascii=False, indent=2) logger.info(f"VLM 분석 캐시 저장 완료: {pdf_id}") except Exception as e: logger.error(f"분석 캐시 저장 오류: {e}") def get_pdf_page_as_base64(pdf_path: str, page_num: int, scale: float = 1.0) -> str: """PDF 페이지를 base64 이미지로 변환""" try: doc = fitz.open(pdf_path) if page_num >= doc.page_count: doc.close() return None page = doc[page_num] pix = page.get_pixmap(matrix=fitz.Matrix(scale, scale)) img_data = pix.tobytes("jpeg", 85) b64_img = base64.b64encode(img_data).decode('utf-8') doc.close() return b64_img except Exception as e: logger.error(f"PDF 페이지 이미지 변환 오류: {e}") return None def get_pdf_pages_as_base64(pdf_path: str, start_page: int = 0, max_pages: int = 10, scale: float = 0.7) -> List[Dict[str, Any]]: """PDF 여러 페이지를 base64 이미지 리스트로 변환 (배치 처리용)""" try: doc = fitz.open(pdf_path) total_pages = doc.page_count end_page = min(start_page + max_pages, total_pages) images = [] for page_num in range(start_page, end_page): page = doc[page_num] pix = page.get_pixmap(matrix=fitz.Matrix(scale, scale)) img_data = pix.tobytes("jpeg", 75) b64_img = base64.b64encode(img_data).decode('utf-8') images.append({ "page": page_num + 1, "image_base64": b64_img }) doc.close() logger.info(f"PDF {start_page+1}~{end_page}/{total_pages}페이지 이미지 변환 완료") return images, total_pages except Exception as e: logger.error(f"PDF 페이지들 이미지 변환 오류: {e}") return [], 0 def call_fireworks_vlm_api(messages: List[Dict], max_tokens: int = 4096, temperature: float = 0.6) -> str: """Fireworks AI VLM API 호출 (이미지 분석 지원)""" if not HAS_VALID_API_KEY: raise Exception("API 키가 설정되지 않았습니다.") payload = { "model": FIREWORKS_VLM_MODEL, "max_tokens": max_tokens, "top_p": 1, "top_k": 40, "presence_penalty": 0, "frequency_penalty": 0, "temperature": temperature, "messages": messages } headers = { "Accept": "application/json", "Content-Type": "application/json", "Authorization": f"Bearer {FIREWORKS_API_KEY}" } response = requests.post(FIREWORKS_API_URL, headers=headers, data=json.dumps(payload), timeout=180) if response.status_code != 200: raise Exception(f"API 오류: {response.status_code} - {response.text}") result = response.json() return result["choices"][0]["message"]["content"] def analyze_batch_pages_sync(pdf_path: str, start_page: int, batch_size: int = 5) -> str: """배치 페이지 분석 (동기)""" page_images, total_pages = get_pdf_pages_as_base64(pdf_path, start_page, batch_size, scale=0.6) if not page_images: return "" content_parts = [] for img_data in page_images: content_parts.append({ "type": "image_url", "image_url": { "url": f"data:image/jpeg;base64,{img_data['image_base64']}" } }) page_range = f"{start_page + 1}~{start_page + len(page_images)}" content_parts.append({ "type": "text", "text": f"""위 이미지들은 PDF 문서의 {page_range}페이지입니다. 각 페이지의 내용을 상세하게 분석하여 텍스트로 추출해주세요. - 모든 텍스트 내용을 빠짐없이 추출 - 표, 차트, 그래프가 있으면 내용 설명 - 이미지가 있으면 설명 - 페이지별로 구분하여 작성 한국어로 작성해주세요.""" }) messages = [{"role": "user", "content": content_parts}] return call_fireworks_vlm_api(messages, max_tokens=4096, temperature=0.3) async def analyze_pdf_with_vlm_batched(pdf_id: str, force_refresh: bool = False) -> Dict[str, Any]: """VLM으로 PDF 배치 분석 후 캐시에 저장""" global analysis_status # 이미 분석 중인지 확인 if pdf_id in analysis_status and analysis_status[pdf_id].get("status") == "analyzing": logger.info(f"PDF {pdf_id} 이미 분석 중...") return {"status": "analyzing", "progress": analysis_status[pdf_id].get("progress", 0)} # 캐시 확인 if not force_refresh: cached = load_analysis_cache(pdf_id) if cached: analysis_status[pdf_id] = {"status": "completed", "progress": 100} return cached pdf_path = str(PROMPT_PDF_PATH) if not PROMPT_PDF_PATH.exists(): analysis_status[pdf_id] = {"status": "error", "error": "PDF 파일 없음"} return {"error": "PDF 파일을 찾을 수 없습니다."} if not HAS_VALID_API_KEY: analysis_status[pdf_id] = {"status": "error", "error": "API 키 없음"} return {"error": "API 키가 설정되지 않았습니다."} # 분석 시작 analysis_status[pdf_id] = {"status": "analyzing", "progress": 0, "started_at": time.time()} try: # PDF 총 페이지 수 확인 doc = fitz.open(pdf_path) total_pages = doc.page_count doc.close() logger.info(f"PDF 분석 시작: 총 {total_pages}페이지") # 배치로 나눠서 분석 (5페이지씩) batch_size = 5 all_analyses = [] for start_page in range(0, min(total_pages, 25), batch_size): # 최대 25페이지 try: progress = int((start_page / min(total_pages, 25)) * 100) analysis_status[pdf_id]["progress"] = progress logger.info(f"배치 분석 중: {start_page + 1}페이지부터 (진행률: {progress}%)") # 동기 함수를 별도 스레드에서 실행 loop = asyncio.get_event_loop() batch_result = await loop.run_in_executor( None, analyze_batch_pages_sync, pdf_path, start_page, batch_size ) if batch_result: all_analyses.append(f"### 페이지 {start_page + 1}~{min(start_page + batch_size, total_pages)}\n{batch_result}") # API 레이트 리밋 방지 await asyncio.sleep(2) except Exception as batch_error: logger.error(f"배치 {start_page} 분석 오류: {batch_error}") all_analyses.append(f"### 페이지 {start_page + 1}~{min(start_page + batch_size, total_pages)}\n[분석 실패: {str(batch_error)}]") # 전체 분석 결과 합치기 combined_analysis = "\n\n".join(all_analyses) # 성공한 분석이 있는지 확인 successful_analyses = [a for a in all_analyses if "[분석 실패:" not in a] if not successful_analyses: logger.error("모든 배치 분석 실패") analysis_status[pdf_id] = {"status": "error", "error": "모든 페이지 분석 실패"} return {"error": "PDF 분석에 실패했습니다. API 키를 확인해주세요."} # 요약 생성 summary = "" if combined_analysis: try: summary_messages = [ { "role": "system", "content": "다음 PDF 분석 내용을 500자 이내로 요약해주세요. 핵심 내용과 주요 키워드를 포함해주세요." }, { "role": "user", "content": combined_analysis[:8000] # 토큰 제한 } ] summary = call_fireworks_vlm_api(summary_messages, max_tokens=1024, temperature=0.5) except Exception as sum_err: logger.error(f"요약 생성 오류: {sum_err}") summary = combined_analysis[:500] + "..." analysis_data = { "pdf_id": pdf_id, "total_pages": total_pages, "analyzed_pages": min(total_pages, 25), "analysis": combined_analysis, "summary": summary, "created_at": time.time() } # 캐시에 저장 save_analysis_cache(pdf_id, analysis_data) analysis_status[pdf_id] = {"status": "completed", "progress": 100} logger.info(f"PDF 분석 완료: {pdf_id}") return analysis_data except Exception as e: logger.error(f"VLM PDF 분석 오류: {e}") analysis_status[pdf_id] = {"status": "error", "error": str(e)} return {"error": str(e)} async def run_initial_analysis(): """서버 시작 시 초기 분석 실행""" logger.info("초기 PDF 분석 시작...") try: result = await analyze_pdf_with_vlm_batched(PROMPT_PDF_ID) if "error" in result: logger.error(f"초기 분석 실패: {result['error']}") else: logger.info("초기 PDF 분석 완료!") except Exception as e: logger.error(f"초기 분석 예외: {e}") def extract_pdf_text(pdf_path: str) -> List[Dict[str, Any]]: try: doc = fitz.open(pdf_path) chunks = [] for page_num in range(len(doc)): page = doc[page_num] text = page.get_text("text") if not text.strip(): text = page.get_text("blocks") if text: text = "\n".join([block[4] for block in text if len(block) > 4 and isinstance(block[4], str)]) if not text.strip(): text = f"[페이지 {page_num + 1} - 이미지 기반 페이지]" chunks.append({ "page": page_num + 1, "text": text.strip() if text.strip() else f"[페이지 {page_num + 1}]", "chunk_id": f"page_{page_num + 1}" }) doc.close() return chunks except Exception as e: logger.error(f"PDF 텍스트 추출 오류: {e}") return [] async def query_pdf(pdf_id: str, query: str) -> Dict[str, Any]: """캐시된 VLM 분석 결과를 기반으로 질의응답""" try: if not HAS_VALID_API_KEY: return { "error": "Fireworks AI API 키가 설정되지 않았습니다.", "answer": "죄송합니다. 현재 AI 기능이 비활성화되어 있어 질문에 답변할 수 없습니다." } # 캐시된 분석 결과 확인 analysis_data = load_analysis_cache(pdf_id) if not analysis_data: # 분석 상태 확인 if pdf_id in analysis_status: status = analysis_status[pdf_id].get("status") if status == "analyzing": progress = analysis_status[pdf_id].get("progress", 0) return {"error": f"분석 진행 중 ({progress}%)", "answer": f"PDF 분석이 진행 중입니다 ({progress}%). 잠시만 기다려주세요."} elif status == "error": return {"error": "분석 실패", "answer": f"PDF 분석에 실패했습니다: {analysis_status[pdf_id].get('error', '알 수 없는 오류')}"} return {"error": "분석 데이터 없음", "answer": "PDF가 아직 분석되지 않았습니다. 잠시 후 다시 시도해주세요."} analysis_text = analysis_data.get("analysis", "") total_pages = analysis_data.get("total_pages", 0) if not analysis_text: return {"error": "분석 데이터 없음", "answer": "PDF 분석 데이터를 찾을 수 없습니다."} # 캐시된 분석 내용을 기반으로 질문에 답변 messages = [ { "role": "system", "content": f"""당신은 PDF 문서 분석 전문가입니다. 아래는 {total_pages}페이지 PDF 문서를 VLM으로 분석한 결과입니다. 이 분석 내용을 기반으로 사용자의 질문에 정확하고 친절하게 한국어로 답변해주세요. 답변할 때 관련 페이지 번호가 있으면 언급해주세요. 분석 내용에 없는 정보는 "해당 정보를 찾을 수 없습니다"라고 솔직히 답해주세요. === PDF 분석 결과 === {analysis_text[:12000]} ==================""" }, { "role": "user", "content": query } ] try: answer = call_fireworks_vlm_api(messages, max_tokens=4096, temperature=0.6) return { "answer": answer, "pdf_id": pdf_id, "query": query } except Exception as api_error: logger.error(f"Fireworks API 호출 오류: {api_error}") error_message = str(api_error) return {"error": f"AI 오류: {error_message}", "answer": "처리 중 오류가 발생했습니다. 잠시 후 다시 시도해주세요."} except Exception as e: logger.error(f"질의응답 처리 오류: {e}") return {"error": str(e), "answer": "처리 중 오류가 발생했습니다."} async def summarize_pdf(pdf_id: str) -> Dict[str, Any]: """캐시된 VLM 분석 결과에서 요약 추출""" try: if not HAS_VALID_API_KEY: return { "error": "Fireworks AI API 키가 설정되지 않았습니다.", "summary": "API 키가 없어 요약을 생성할 수 없습니다." } # 캐시된 분석 결과 확인 analysis_data = load_analysis_cache(pdf_id) if not analysis_data: # 분석 상태 확인 if pdf_id in analysis_status: status = analysis_status[pdf_id].get("status") if status == "analyzing": progress = analysis_status[pdf_id].get("progress", 0) return {"error": f"분석 진행 중", "summary": f"PDF 분석이 진행 중입니다 ({progress}%). 잠시만 기다려주세요."} elif status == "error": return {"error": "분석 실패", "summary": f"PDF 분석에 실패했습니다."} return {"error": "분석 데이터 없음", "summary": "PDF가 아직 분석되지 않았습니다."} summary = analysis_data.get("summary", "") total_pages = analysis_data.get("total_pages", 0) analyzed_pages = analysis_data.get("analyzed_pages", total_pages) if summary: return { "summary": summary, "pdf_id": pdf_id, "total_pages": total_pages, "analyzed_pages": analyzed_pages } # 요약이 없으면 분석 내용에서 추출 analysis_text = analysis_data.get("analysis", "") if analysis_text: return { "summary": analysis_text[:500] + "...", "pdf_id": pdf_id, "total_pages": total_pages } return {"error": "요약 없음", "summary": "요약을 생성할 수 없습니다."} except Exception as e: logger.error(f"PDF 요약 생성 오류: {e}") return { "error": str(e), "summary": "PDF 요약 중 오류가 발생했습니다." } async def cache_pdf(pdf_path: str): try: pdf_file = pathlib.Path(pdf_path) pdf_name = pdf_file.stem if pdf_name not in cache_locks: cache_locks[pdf_name] = threading.Lock() if pdf_name in pdf_cache and pdf_cache[pdf_name].get("status") in ["processing", "completed"]: logger.info(f"PDF {pdf_name} 이미 캐싱 완료 또는 진행 중") return with cache_locks[pdf_name]: if pdf_name in pdf_cache and pdf_cache[pdf_name].get("status") in ["processing", "completed"]: return pdf_cache[pdf_name] = {"status": "processing", "progress": 0, "pages": []} cache_path = get_cache_path(pdf_name) if cache_path.exists(): try: with open(cache_path, "r") as cache_file: cached_data = json.load(cache_file) if cached_data.get("status") == "completed" and cached_data.get("pages"): pdf_cache[pdf_name] = cached_data pdf_cache[pdf_name]["status"] = "completed" logger.info(f"캐시 파일에서 {pdf_name} 로드 완료") return except Exception as e: logger.error(f"캐시 파일 로드 실패: {e}") doc = fitz.open(pdf_path) total_pages = doc.page_count if total_pages > 0: page = doc[0] pix_thumb = page.get_pixmap(matrix=fitz.Matrix(0.2, 0.2)) thumb_data = pix_thumb.tobytes("png") b64_thumb = base64.b64encode(thumb_data).decode('utf-8') thumb_src = f"data:image/png;base64,{b64_thumb}" pdf_cache[pdf_name]["pages"] = [{"thumb": thumb_src, "src": ""}] pdf_cache[pdf_name]["progress"] = 1 pdf_cache[pdf_name]["total_pages"] = total_pages scale_factor = 1.0 jpeg_quality = 80 def process_page(page_num): try: page = doc[page_num] pix = page.get_pixmap(matrix=fitz.Matrix(scale_factor, scale_factor)) img_data = pix.tobytes("jpeg", jpeg_quality) b64_img = base64.b64encode(img_data).decode('utf-8') img_src = f"data:image/jpeg;base64,{b64_img}" thumb_src = "" if page_num > 0 else pdf_cache[pdf_name]["pages"][0]["thumb"] return { "page_num": page_num, "src": img_src, "thumb": thumb_src } except Exception as e: logger.error(f"페이지 {page_num} 처리 오류: {e}") return { "page_num": page_num, "src": "", "thumb": "", "error": str(e) } pages = [None] * total_pages processed_count = 0 batch_size = 5 for batch_start in range(0, total_pages, batch_size): batch_end = min(batch_start + batch_size, total_pages) current_batch = list(range(batch_start, batch_end)) with concurrent.futures.ThreadPoolExecutor(max_workers=min(5, batch_size)) as executor: batch_results = list(executor.map(process_page, current_batch)) for result in batch_results: page_num = result["page_num"] pages[page_num] = { "src": result["src"], "thumb": result["thumb"] } processed_count += 1 progress = round(processed_count / total_pages * 100) pdf_cache[pdf_name]["progress"] = progress pdf_cache[pdf_name]["pages"] = pages pdf_cache[pdf_name] = { "status": "completed", "progress": 100, "pages": pages, "total_pages": total_pages } try: with open(cache_path, "w") as cache_file: json.dump(pdf_cache[pdf_name], cache_file) logger.info(f"PDF {pdf_name} 캐싱 완료, {total_pages}페이지") except Exception as e: logger.error(f"최종 캐시 저장 실패: {e}") except Exception as e: import traceback logger.error(f"PDF 캐싱 오류: {str(e)}\n{traceback.format_exc()}") if 'pdf_name' in locals() and pdf_name in pdf_cache: pdf_cache[pdf_name]["status"] = "error" pdf_cache[pdf_name]["error"] = str(e) @app.on_event("startup") async def startup_event(): if PROMPT_PDF_PATH.exists(): logger.info(f"prompt.pdf 파일 발견: {PROMPT_PDF_PATH}") # 플립북 캐싱 asyncio.create_task(cache_pdf(str(PROMPT_PDF_PATH))) # VLM 분석 - 에러 핸들링 포함 asyncio.create_task(run_initial_analysis()) else: logger.warning(f"prompt.pdf 파일을 찾을 수 없습니다: {PROMPT_PDF_PATH}") @app.get("/api/pdf-list") async def get_pdf_list(): """사용 가능한 PDF 목록 반환""" global current_pdf_key pdf_list = [] for key, info in PDF_FILES.items(): pdf_list.append({ "key": key, "name": info["name"], "exists": info["path"].exists(), "is_current": key == current_pdf_key }) return {"pdfs": pdf_list, "current": current_pdf_key} @app.post("/api/switch-pdf/{pdf_key}") async def switch_pdf(pdf_key: str, background_tasks: BackgroundTasks): """PDF 전환""" global current_pdf_key if pdf_key not in PDF_FILES: return JSONResponse(content={"error": "유효하지 않은 PDF 키입니다."}, status_code=400) pdf_info = PDF_FILES[pdf_key] if not pdf_info["path"].exists(): return JSONResponse(content={"error": f"PDF 파일을 찾을 수 없습니다: {pdf_info['name']}"}, status_code=404) current_pdf_key = pdf_key logger.info(f"PDF 전환: {pdf_key} - {pdf_info['name']}") # 캐싱 시작 pdf_name = pdf_info["path"].stem if pdf_name not in pdf_cache or pdf_cache[pdf_name].get("status") != "completed": background_tasks.add_task(cache_pdf, str(pdf_info["path"])) # VLM 분석 시작 (캐시가 없는 경우) if not load_analysis_cache(pdf_info["id"]): asyncio.create_task(analyze_pdf_with_vlm_batched_for_key(pdf_key)) return { "success": True, "current": pdf_key, "name": pdf_info["name"], "id": pdf_info["id"] } async def analyze_pdf_with_vlm_batched_for_key(pdf_key: str, force_refresh: bool = False) -> Dict[str, Any]: """특정 PDF에 대한 VLM 분석""" global analysis_status if pdf_key not in PDF_FILES: return {"error": "유효하지 않은 PDF 키"} pdf_info = PDF_FILES[pdf_key] pdf_id = pdf_info["id"] pdf_path = str(pdf_info["path"]) # 이미 분석 중인지 확인 if pdf_id in analysis_status and analysis_status[pdf_id].get("status") == "analyzing": logger.info(f"PDF {pdf_id} 이미 분석 중...") return {"status": "analyzing", "progress": analysis_status[pdf_id].get("progress", 0)} # 캐시 확인 if not force_refresh: cached = load_analysis_cache(pdf_id) if cached: analysis_status[pdf_id] = {"status": "completed", "progress": 100} return cached if not pdf_info["path"].exists(): analysis_status[pdf_id] = {"status": "error", "error": "PDF 파일 없음"} return {"error": "PDF 파일을 찾을 수 없습니다."} if not HAS_VALID_API_KEY: analysis_status[pdf_id] = {"status": "error", "error": "API 키 없음"} return {"error": "API 키가 설정되지 않았습니다."} # 분석 시작 analysis_status[pdf_id] = {"status": "analyzing", "progress": 0, "started_at": time.time()} try: doc = fitz.open(pdf_path) total_pages = doc.page_count doc.close() logger.info(f"PDF 분석 시작 ({pdf_key}): 총 {total_pages}페이지") batch_size = 5 all_analyses = [] for start_page in range(0, min(total_pages, 25), batch_size): try: progress = int((start_page / min(total_pages, 25)) * 100) analysis_status[pdf_id]["progress"] = progress logger.info(f"배치 분석 중 ({pdf_key}): {start_page + 1}페이지부터 (진행률: {progress}%)") loop = asyncio.get_event_loop() batch_result = await loop.run_in_executor( None, analyze_batch_pages_sync, pdf_path, start_page, batch_size ) if batch_result: all_analyses.append(f"### 페이지 {start_page + 1}~{min(start_page + batch_size, total_pages)}\n{batch_result}") await asyncio.sleep(2) except Exception as batch_error: logger.error(f"배치 {start_page} 분석 오류: {batch_error}") all_analyses.append(f"### 페이지 {start_page + 1}~{min(start_page + batch_size, total_pages)}\n[분석 실패: {str(batch_error)}]") combined_analysis = "\n\n".join(all_analyses) successful_analyses = [a for a in all_analyses if "[분석 실패:" not in a] if not successful_analyses: logger.error("모든 배치 분석 실패") analysis_status[pdf_id] = {"status": "error", "error": "모든 페이지 분석 실패"} return {"error": "PDF 분석에 실패했습니다."} summary = "" if combined_analysis: try: summary_messages = [ {"role": "system", "content": "다음 PDF 분석 내용을 500자 이내로 요약해주세요. 핵심 내용과 주요 키워드를 포함해주세요."}, {"role": "user", "content": combined_analysis[:8000]} ] summary = call_fireworks_vlm_api(summary_messages, max_tokens=1024, temperature=0.5) except Exception as sum_err: logger.error(f"요약 생성 오류: {sum_err}") summary = combined_analysis[:500] + "..." analysis_data = { "pdf_id": pdf_id, "pdf_key": pdf_key, "total_pages": total_pages, "analyzed_pages": min(total_pages, 25), "analysis": combined_analysis, "summary": summary, "created_at": time.time() } save_analysis_cache(pdf_id, analysis_data) analysis_status[pdf_id] = {"status": "completed", "progress": 100} logger.info(f"PDF 분석 완료: {pdf_id}") return analysis_data except Exception as e: logger.error(f"VLM PDF 분석 오류: {e}") analysis_status[pdf_id] = {"status": "error", "error": str(e)} return {"error": str(e)} @app.get("/api/pdf-info") async def get_pdf_info(): global current_pdf_key pdf_info = PDF_FILES.get(current_pdf_key, PDF_FILES["prompt"]) pdf_path = pdf_info["path"] pdf_id = pdf_info["id"] if not pdf_path.exists(): return {"exists": False, "error": "PDF 파일을 찾을 수 없습니다"} pdf_name = pdf_path.stem is_cached = pdf_name in pdf_cache and pdf_cache[pdf_name].get("status") == "completed" # VLM 분석 캐시 확인 analysis_cached = load_analysis_cache(pdf_id) is not None return { "path": str(pdf_path), "name": pdf_name, "display_name": pdf_info["name"], "id": pdf_id, "key": current_pdf_key, "exists": True, "cached": is_cached, "analysis_cached": analysis_cached } @app.get("/api/analysis-status") async def get_analysis_status(): """VLM 분석 상태 확인""" global current_pdf_key pdf_info = PDF_FILES.get(current_pdf_key, PDF_FILES["prompt"]) pdf_id = pdf_info["id"] # 먼저 캐시 파일 확인 cached = load_analysis_cache(pdf_id) if cached: return { "status": "completed", "total_pages": cached.get("total_pages", 0), "analyzed_pages": cached.get("analyzed_pages", 0), "created_at": cached.get("created_at", 0) } # 메모리 상태 확인 if pdf_id in analysis_status: status_info = analysis_status[pdf_id] return { "status": status_info.get("status", "unknown"), "progress": status_info.get("progress", 0), "error": status_info.get("error") } return {"status": "not_started"} @app.post("/api/reanalyze-pdf") async def reanalyze_pdf(): """PDF 재분석 (캐시 무시)""" global current_pdf_key try: if not HAS_VALID_API_KEY: return JSONResponse(content={"error": "API 키가 설정되지 않았습니다."}, status_code=400) pdf_info = PDF_FILES.get(current_pdf_key, PDF_FILES["prompt"]) pdf_id = pdf_info["id"] # 기존 캐시 삭제 cache_path = get_analysis_cache_path(pdf_id) if cache_path.exists(): cache_path.unlink() logger.info("기존 VLM 분석 캐시 삭제") # 상태 초기화 if pdf_id in analysis_status: del analysis_status[pdf_id] # 백그라운드에서 재분석 시작 asyncio.create_task(analyze_pdf_with_vlm_batched_for_key(current_pdf_key)) return {"status": "started", "message": "PDF 재분석을 시작합니다."} except Exception as e: logger.error(f"재분석 시작 오류: {e}") return JSONResponse(content={"error": str(e)}, status_code=500) @app.get("/api/pdf-thumbnail") async def get_pdf_thumbnail(): global current_pdf_key try: pdf_info = PDF_FILES.get(current_pdf_key, PDF_FILES["prompt"]) pdf_path = pdf_info["path"] if not pdf_path.exists(): return {"thumbnail": None, "error": "PDF 파일을 찾을 수 없습니다"} pdf_name = pdf_path.stem if pdf_name in pdf_cache and pdf_cache[pdf_name].get("pages"): if pdf_cache[pdf_name]["pages"][0].get("thumb"): return {"thumbnail": pdf_cache[pdf_name]["pages"][0]["thumb"]} doc = fitz.open(str(pdf_path)) if doc.page_count > 0: page = doc[0] pix = page.get_pixmap(matrix=fitz.Matrix(0.2, 0.2)) img_data = pix.tobytes("jpeg", 70) b64_img = base64.b64encode(img_data).decode('utf-8') asyncio.create_task(cache_pdf(str(pdf_path))) return {"thumbnail": f"data:image/jpeg;base64,{b64_img}"} return {"thumbnail": None} except Exception as e: logger.error(f"썸네일 생성 오류: {str(e)}") return {"error": str(e), "thumbnail": None} @app.get("/api/cache-status") async def get_cache_status(): global current_pdf_key pdf_info = PDF_FILES.get(current_pdf_key, PDF_FILES["prompt"]) pdf_name = pdf_info["path"].stem if pdf_name in pdf_cache: return pdf_cache[pdf_name] return {"status": "not_cached"} @app.post("/api/ai/query-pdf") async def api_query_pdf(query: Dict[str, str]): global current_pdf_key try: user_query = query.get("query", "") if not user_query: return JSONResponse(content={"error": "질문이 제공되지 않았습니다"}, status_code=400) pdf_info = PDF_FILES.get(current_pdf_key, PDF_FILES["prompt"]) if not pdf_info["path"].exists(): return JSONResponse(content={"error": "PDF 파일을 찾을 수 없습니다"}, status_code=404) result = await query_pdf(pdf_info["id"], user_query) if "answer" in result: return result if "error" in result: return JSONResponse(content=result, status_code=200) return result except Exception as e: logger.error(f"질의응답 API 오류: {e}") return JSONResponse(content={"error": str(e), "answer": "죄송합니다. 처리 중 오류가 발생했습니다."}, status_code=200) @app.get("/api/ai/summarize-pdf") async def api_summarize_pdf(): global current_pdf_key try: pdf_info = PDF_FILES.get(current_pdf_key, PDF_FILES["prompt"]) if not pdf_info["path"].exists(): return JSONResponse(content={"error": "PDF 파일을 찾을 수 없습니다"}, status_code=404) result = await summarize_pdf(pdf_info["id"]) if "summary" in result: return result if "error" in result: return JSONResponse(content=result, status_code=200) return result except Exception as e: logger.error(f"PDF 요약 API 오류: {e}") return JSONResponse(content={"error": str(e), "summary": "죄송합니다. 요약 생성 중 오류가 발생했습니다."}, status_code=200) @app.get("/api/cached-pdf") async def get_cached_pdf(background_tasks: BackgroundTasks): global current_pdf_key try: pdf_info = PDF_FILES.get(current_pdf_key, PDF_FILES["prompt"]) pdf_path = pdf_info["path"] pdf_name = pdf_path.stem if pdf_name in pdf_cache: status = pdf_cache[pdf_name].get("status", "") if status == "completed": return pdf_cache[pdf_name] elif status == "processing": progress = pdf_cache[pdf_name].get("progress", 0) pages = pdf_cache[pdf_name].get("pages", []) total_pages = pdf_cache[pdf_name].get("total_pages", 0) return { "status": "processing", "progress": progress, "pages": pages, "total_pages": total_pages, "available_pages": len([p for p in pages if p and p.get("src")]) } background_tasks.add_task(cache_pdf, str(pdf_path)) return {"status": "started", "progress": 0} except Exception as e: logger.error(f"캐시된 PDF 제공 오류: {str(e)}") return {"error": str(e), "status": "error"} @app.get("/", response_class=HTMLResponse) async def root(): return get_html_content() def get_html_content(): return HTMLResponse(content=HTML) HTML = """ 🎨 AI 플립북 - Comic Style
📚 AI 플립북
🤖 AI 어시스턴트
샘플-프롬프트북
샘플-코레일잡지

AI 어시스턴트

""" if __name__ == "__main__": uvicorn.run("app:app", host="0.0.0.0", port=int(os.getenv("PORT", 7860)))