Spaces:
Sleeping
Sleeping
| from fastapi import FastAPI, UploadFile, HTTPException | |
| from fastapi.responses import FileResponse | |
| import subprocess | |
| import os | |
| import shutil | |
| from pathlib import Path | |
| import uuid | |
| from fastapi.middleware.cors import CORSMiddleware | |
| app = FastAPI(title="Document Conversion API") | |
| # Create directories for file handling | |
| UPLOAD_DIR = Path("uploads") | |
| OUTPUT_DIR = Path("outputs") | |
| UPLOAD_DIR.mkdir(exist_ok=True) | |
| OUTPUT_DIR.mkdir(exist_ok=True) | |
| async def convert_document( | |
| file: UploadFile, | |
| output_format: str, | |
| filter_options: str = None | |
| ): | |
| try: | |
| # Generate unique filenames | |
| input_filename = f"{uuid.uuid4()}_{file.filename}" | |
| output_filename = f"{Path(input_filename).stem}.{output_format}" | |
| input_path = UPLOAD_DIR / input_filename | |
| output_path = OUTPUT_DIR / output_filename | |
| # Save uploaded file | |
| with open(input_path, "wb") as buffer: | |
| shutil.copyfileobj(file.file, buffer) | |
| # Prepare conversion command | |
| command = ["unoconvert"] | |
| if filter_options: | |
| command.extend(["--filter-options", filter_options]) | |
| command.extend([str(input_path), str(output_path)]) | |
| # Execute conversion | |
| process = subprocess.run( | |
| command, | |
| capture_output=True, | |
| text=True | |
| ) | |
| if process.returncode != 0: | |
| raise HTTPException( | |
| status_code=500, | |
| detail=f"Conversion failed: {process.stderr}" | |
| ) | |
| # Return converted file | |
| return FileResponse( | |
| path=output_path, | |
| filename=output_filename, | |
| media_type="application/octet-stream" | |
| ) | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| finally: | |
| # Cleanup | |
| if input_path.exists(): | |
| input_path.unlink() | |
| if output_path.exists(): | |
| output_path.unlink() | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=["*"], | |
| allow_credentials=True, | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| ) | |
| from fastapi.staticfiles import StaticFiles | |
| app.mount("/", StaticFiles(directory="static/ui", html=True), name="index") | |
| async def health_check(): | |
| return {"status": "healthy"} |