""" FastAPI File Uploader/Downloader Service """ import os import json import uuid from datetime import datetime from pathlib import Path from typing import Optional from fastapi import FastAPI, UploadFile, File, HTTPException, Request from fastapi.responses import FileResponse, HTMLResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates import uvicorn # 애플리케이션 설정 app = FastAPI(title="File Uploader", description="파일 업로드/다운로드 서비스", version="1.0.0") # 경로 설정 BASE_DIR = Path(__file__).parent ASSETS_DIR = BASE_DIR / "assets" IMAGES_DIR = ASSETS_DIR / "images" VIDEOS_DIR = ASSETS_DIR / "videos" FILES_DIR = ASSETS_DIR / "files" METADATA_FILE = ASSETS_DIR / "metadata.json" STATIC_DIR = BASE_DIR / "static" # 디렉토리 생성 for directory in [IMAGES_DIR, VIDEOS_DIR, FILES_DIR]: directory.mkdir(parents=True, exist_ok=True) # 파일 크기 제한 (100MB) MAX_FILE_SIZE = 100 * 1024 * 1024 # 허용된 파일 확장자 ALLOWED_EXTENSIONS = { "image": {".jpg", ".jpeg", ".png", ".gif", ".bmp", ".webp", ".svg"}, "video": {".mp4", ".avi", ".mov", ".wmv", ".flv", ".mkv", ".webm"}, } def get_file_category(filename: str) -> str: """파일 확장자에 따른 카테고리 분류""" ext = Path(filename).suffix.lower() for category, extensions in ALLOWED_EXTENSIONS.items(): if ext in extensions: return category return "file" def get_save_directory(category: str) -> Path: """카테고리에 따른 저장 디렉토리 반환""" directories = { "image": IMAGES_DIR, "video": VIDEOS_DIR, "file": FILES_DIR, } return directories.get(category, FILES_DIR) def generate_unique_filename(original_filename: str) -> str: """고유한 파일명 생성""" ext = Path(original_filename).suffix return f"{uuid.uuid4()}{ext}" def load_metadata() -> dict: """메타데이터 로드""" if not METADATA_FILE.exists(): return {"files": []} try: with open(METADATA_FILE, "r", encoding="utf-8") as f: return json.load(f) except (json.JSONDecodeError, IOError): return {"files": []} def save_metadata(metadata: dict) -> None: """메타데이터 저장""" with open(METADATA_FILE, "w", encoding="utf-8") as f: json.dump(metadata, f, ensure_ascii=False, indent=2) def add_file_metadata( original_filename: str, saved_filename: str, size: int, mime_type: str, category: str, ) -> None: """파일 메타데이터 추가""" metadata = load_metadata() file_info = { "original_filename": original_filename, "saved_filename": saved_filename, "size": size, "mime_type": mime_type, "category": category, "upload_date": datetime.now().isoformat(), } metadata["files"].append(file_info) save_metadata(metadata) def find_file_by_saved_name(saved_filename: str) -> Optional[dict]: """저장된 파일명으로 파일 정보 찾기""" metadata = load_metadata() for file_info in metadata["files"]: if file_info["saved_filename"] == saved_filename: return file_info return None # 정적 파일 서빙 app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static") @app.get("/web", response_class=HTMLResponse) async def web_interface(): """웹 인터페이스 제공""" html_file = STATIC_DIR / "index.html" if not html_file.exists(): raise HTTPException(status_code=404, detail="Web interface not found") return FileResponse(html_file) @app.post("/api/files/upload") async def upload_file(file: UploadFile = File(...)): """파일 업로드 API""" # 파일 크기 확인 contents = await file.read() file_size = len(contents) if file_size > MAX_FILE_SIZE: raise HTTPException( status_code=413, detail=f"파일 크기가 너무 큽니다. 최대 {MAX_FILE_SIZE // (1024 * 1024)}MB까지 업로드 가능합니다.", ) # 파일 카테고리 결정 category = get_file_category(file.filename) # 고유 파일명 생성 saved_filename = generate_unique_filename(file.filename) # 저장 경로 결정 save_dir = get_save_directory(category) file_path = save_dir / saved_filename # 파일 저장 try: with open(file_path, "wb") as f: f.write(contents) except IOError as e: raise HTTPException(status_code=500, detail=f"파일 저장 실패: {str(e)}") # 메타데이터 저장 add_file_metadata( original_filename=file.filename, saved_filename=saved_filename, size=file_size, mime_type=file.content_type or "application/octet-stream", category=category, ) return { "success": True, "filename": saved_filename, "original_filename": file.filename, "size": file_size, "category": category, } @app.get("/api/files/download/{filename}") async def download_file(filename: str): """파일 다운로드 API""" # 경로 순회 공격 방지 if ".." in filename or "/" in filename or "\\" in filename: raise HTTPException(status_code=400, detail="잘못된 파일명입니다.") # 파일 메타데이터 조회 file_info = find_file_by_saved_name(filename) if not file_info: raise HTTPException(status_code=404, detail="파일을 찾을 수 없습니다.") # 파일 경로 확인 save_dir = get_save_directory(file_info["category"]) file_path = save_dir / filename if not file_path.exists(): raise HTTPException(status_code=404, detail="파일을 찾을 수 없습니다.") return FileResponse( path=file_path, filename=file_info["original_filename"], media_type=file_info.get("mime_type", "application/octet-stream"), ) @app.get("/api/files/list") async def list_files(): """파일 목록 조회 API""" metadata = load_metadata() return {"files": metadata["files"]} @app.delete("/api/files/{filename}") async def delete_file(filename: str): """파일 삭제 API""" # 경로 순회 공격 방지 if ".." in filename or "/" in filename or "\\" in filename: raise HTTPException(status_code=400, detail="잘못된 파일명입니다.") # 파일 메타데이터 조회 file_info = find_file_by_saved_name(filename) if not file_info: raise HTTPException(status_code=404, detail="파일을 찾을 수 없습니다.") # 파일 삭제 save_dir = get_save_directory(file_info["category"]) file_path = save_dir / filename try: if file_path.exists(): file_path.unlink() except IOError as e: raise HTTPException(status_code=500, detail=f"파일 삭제 실패: {str(e)}") # 메타데이터에서 제거 metadata = load_metadata() metadata["files"] = [f for f in metadata["files"] if f["saved_filename"] != filename] save_metadata(metadata) return {"success": True, "message": "파일이 삭제되었습니다."} if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)