from fastapi import FastAPI, BackgroundTasks from fastapi.responses import HTMLResponse, JSONResponse, Response from fastapi.staticfiles import StaticFiles import pathlib, os, uvicorn, base64, json from typing import Dict, List, Any import asyncio import logging import threading import concurrent.futures # 로깅 설정 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") # PDF 디렉토리 설정 PDF_DIR = BASE / "pdf" if not PDF_DIR.exists(): PDF_DIR.mkdir(parents=True) # 캐시 디렉토리 설정 CACHE_DIR = BASE / "cache" if not CACHE_DIR.exists(): CACHE_DIR.mkdir(parents=True) # 전역 캐시 객체 pdf_cache: Dict[str, Dict[str, Any]] = {} # 캐싱 락 cache_locks = {} # PDF 파일 목록 가져오기 def get_pdf_files(): pdf_files = [] if PDF_DIR.exists(): pdf_files = [f for f in PDF_DIR.glob("*.pdf")] return pdf_files # PDF 썸네일 생성 및 프로젝트 데이터 준비 def generate_pdf_projects(): projects_data = [] pdf_files = get_pdf_files() for pdf_file in pdf_files: projects_data.append({ "path": str(pdf_file), "name": pdf_file.stem, "cached": pdf_file.stem in pdf_cache and pdf_cache[pdf_file.stem].get("status") == "completed" }) return projects_data # 캐시 파일 경로 생성 def get_cache_path(pdf_name: str): return CACHE_DIR / f"{pdf_name}_cache.json" # 최적화된 PDF 페이지 캐싱 함수 async def cache_pdf(pdf_path: str): try: import fitz # PyMuPDF pdf_file = pathlib.Path(pdf_path) pdf_name = pdf_file.stem # 락 생성 - 동일한 PDF에 대해 동시 캐싱 방지 if pdf_name not in cache_locks: cache_locks[pdf_name] = threading.Lock() # 이미 캐싱 중이거나 캐싱 완료된 PDF는 건너뛰기 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}") # PDF 파일 열기 doc = fitz.open(pdf_path) total_pages = doc.page_count # 미리 썸네일만 먼저 생성 (빠른 UI 로딩용) 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 # JPEG 품질 (낮출수록 용량 작아짐) # 페이지 처리 작업자 함수 (병렬 처리용) def process_page(page_num): try: page = doc[page_num] # 이미지로 변환 시 매트릭스 스케일링 적용 (성능 최적화) pix = page.get_pixmap(matrix=fitz.Matrix(scale_factor, scale_factor)) # JPEG 형식으로 인코딩 (PNG보다 크기 작음) 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 try: with open(cache_path, "w") as cache_file: json.dump({ "status": "processing", "progress": pdf_cache[pdf_name]["progress"], "pages": pdf_cache[pdf_name]["pages"], "total_pages": total_pages }, cache_file) except Exception as e: logger.error(f"중간 캐시 저장 실패: {e}") # 캐싱 완료 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 pdf_cache: pdf_cache[pdf_name]["status"] = "error" pdf_cache[pdf_name]["error"] = str(e) # 시작 시 모든 PDF 파일 캐싱 async def init_cache_all_pdfs(): logger.info("PDF 캐싱 작업 시작") pdf_files = get_pdf_files() # 이미 캐시된 PDF 파일 로드 (빠른 시작을 위해 먼저 수행) for cache_file in CACHE_DIR.glob("*_cache.json"): try: pdf_name = cache_file.stem.replace("_cache", "") with open(cache_file, "r") as f: cached_data = json.load(f) 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}") except Exception as e: logger.error(f"캐시 파일 로드 오류: {str(e)}") # 캐싱되지 않은 PDF 파일 병렬 처리 await asyncio.gather(*[asyncio.create_task(cache_pdf(str(pdf_file))) for pdf_file in pdf_files if pdf_file.stem not in pdf_cache or pdf_cache[pdf_file.stem].get("status") != "completed"]) # 백그라운드 작업 시작 함수 @app.on_event("startup") async def startup_event(): # 백그라운드 태스크로 캐싱 실행 asyncio.create_task(init_cache_all_pdfs()) # API 엔드포인트: PDF 프로젝트 목록 @app.get("/api/pdf-projects") async def get_pdf_projects_api(): return generate_pdf_projects() # API 엔드포인트: PDF 썸네일 제공 (최적화) @app.get("/api/pdf-thumbnail") async def get_pdf_thumbnail(path: str): try: pdf_file = pathlib.Path(path) pdf_name = pdf_file.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"]} # 캐시에 없으면 생성 (더 작고 빠른 썸네일) import fitz doc = fitz.open(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) # JPEG 압축 사용 b64_img = base64.b64encode(img_data).decode('utf-8') # 백그라운드에서 캐싱 시작 asyncio.create_task(cache_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} # API 엔드포인트: 캐시 상태 확인 @app.get("/api/cache-status") async def get_cache_status(path: str = None): if path: pdf_file = pathlib.Path(path) pdf_name = pdf_file.stem if pdf_name in pdf_cache: return pdf_cache[pdf_name] return {"status": "not_cached"} else: return {name: {"status": info["status"], "progress": info.get("progress", 0)} for name, info in pdf_cache.items()} # API 엔드포인트: 캐시된 PDF 콘텐츠 제공 (점진적 로딩 지원) @app.get("/api/cached-pdf") async def get_cached_pdf(path: str, background_tasks: BackgroundTasks): try: pdf_file = pathlib.Path(path) pdf_name = pdf_file.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, path) return {"status": "started", "progress": 0} except Exception as e: logger.error(f"캐시된 PDF 제공 오류: {str(e)}") return {"error": str(e), "status": "error"} # API 엔드포인트: PDF 원본 콘텐츠 제공(캐시가 없는 경우) @app.get("/api/pdf-content") async def get_pdf_content(path: str, background_tasks: BackgroundTasks): try: # 캐싱 상태 확인 pdf_file = pathlib.Path(path) if not pdf_file.exists(): return JSONResponse(content={"error": f"파일을 찾을 수 없습니다: {path}"}, status_code=404) pdf_name = pdf_file.stem # 캐시된 경우 리다이렉트 if pdf_name in pdf_cache and (pdf_cache[pdf_name].get("status") == "completed" or (pdf_cache[pdf_name].get("status") == "processing" and pdf_cache[pdf_name].get("progress", 0) > 10)): return JSONResponse(content={"redirect": f"/api/cached-pdf?path={path}"}) # 파일 읽기 with open(path, "rb") as pdf_file: content = pdf_file.read() # 파일명 처리 import urllib.parse filename = pdf_file.name encoded_filename = urllib.parse.quote(filename) # 백그라운드에서 캐싱 시작 background_tasks.add_task(cache_pdf, path) # 응답 헤더 설정 headers = { "Content-Type": "application/pdf", "Content-Disposition": f"inline; filename=\"{encoded_filename}\"; filename*=UTF-8''{encoded_filename}" } return Response(content=content, media_type="application/pdf", headers=headers) except Exception as e: import traceback error_details = traceback.format_exc() logger.error(f"PDF 콘텐츠 로드 오류: {str(e)}\n{error_details}") return JSONResponse(content={"error": str(e)}, status_code=500) # HTML 파일 읽기 함수 def get_html_content(): html_path = BASE / "flipbook_template.html" if html_path.exists(): with open(html_path, "r", encoding="utf-8") as f: return f.read() return HTML # 기본 HTML 사용 # HTML 문자열 (UI 수정 버전) HTML = """