pdf_merger / app /routers /pdf.py
muhammadnoman76's picture
Initial deploy
a317be6
from typing import List
from fastapi import APIRouter, Depends, File, UploadFile, HTTPException
from fastapi.responses import FileResponse as FastAPIFileResponse
from ..schemas.file import FileResponse, MergeRequest, MergeResponse
from ..services.pdf_merger import save_upload_file, get_user_files, merge_pdfs
from ..services.auth import get_current_user
from ..models.user import User
router = APIRouter(
prefix="/pdf",
tags=["PDF Operations"]
)
@router.post("/upload", response_model=FileResponse)
async def upload_pdf(
file: UploadFile = File(...),
current_user: User = Depends(get_current_user)
):
"""Upload a PDF file."""
if not file.filename.lower().endswith('.pdf'):
raise HTTPException(status_code=400, detail="File must be a PDF")
return await save_upload_file(file, current_user)
@router.get("/files", response_model=List[FileResponse])
async def list_files(current_user: User = Depends(get_current_user)):
"""List all files uploaded by the current user."""
return await get_user_files(current_user)
@router.get("/merged", response_model=List[MergeResponse])
async def list_merged_files(current_user: User = Depends(get_current_user)):
"""List all merged files created by the user."""
from ..services.pdf_merger import get_user_merged_files
return await get_user_merged_files(current_user)
@router.post("/merge", response_model=MergeResponse)
async def merge_files(
merge_req: MergeRequest,
current_user: User = Depends(get_current_user)
):
"""Merge multiple PDF files into one."""
return await merge_pdfs(merge_req, current_user)
@router.get("/download/{file_path:path}")
async def download_file(
file_path: str,
current_user: User = Depends(get_current_user)
# Note: In a real app, you'd want to verify the user owns the file or the merged file
# For this demo, we'll assume if they have the path (ID based in real world), they can access
# But since we store full paths in DB, we should be careful.
# Better approach: download by ID and look up path.
# For now, let's just serve it if it exists in our logical directories.
):
"""Download a processed PDF file."""
import os
from ..config import get_settings
settings = get_settings()
# Simple security check to prevent directory traversal
real_path = os.path.realpath(file_path)
if not (real_path.startswith(os.path.realpath(settings.upload_dir)) or
real_path.startswith(os.path.realpath(settings.merged_dir))):
raise HTTPException(status_code=403, detail="Access denied")
if not os.path.exists(real_path):
raise HTTPException(status_code=404, detail="File not found")
return FastAPIFileResponse(
real_path,
media_type="application/pdf",
filename=os.path.basename(real_path),
content_disposition_type="attachment"
)