Spaces:
Sleeping
Sleeping
File size: 3,666 Bytes
47a81c7 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 |
from fastapi import FastAPI, File, UploadFile, Form, HTTPException
from fastapi.staticfiles import StaticFiles
from fastapi.responses import JSONResponse
import uvicorn
import os
from typing import List, Optional
import shutil
from pathlib import Path
import uuid
# Import AI functionality modules
from modules.document_processor import process_document
from modules.image_processor import process_image
# Create FastAPI app
app = FastAPI(title="AI Document Analysis API")
# Create upload directory if it doesn't exist
UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)
# Mount static files - updated path to be relative to the current directory
app.mount("/static", StaticFiles(directory="frontend"), name="static")
@app.get("/")
def read_root():
return {"message": "AI Document Analysis API is running"}
@app.post("/api/analyze-document")
async def analyze_document(
file: UploadFile = File(...),
analysis_type: str = Form(...)
):
# Validate file type
allowed_extensions = [".pdf", ".docx", ".pptx", ".xlsx", ".xls"]
file_ext = os.path.splitext(file.filename)[1].lower()
if file_ext not in allowed_extensions:
raise HTTPException(status_code=400, detail=f"File type not supported. Allowed types: {', '.join(allowed_extensions)}")
# Create unique filename
unique_filename = f"{uuid.uuid4()}{file_ext}"
file_path = UPLOAD_DIR / unique_filename
# Save uploaded file
with open(file_path, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
try:
# Process document based on analysis type
if analysis_type == "summarize":
result = process_document(str(file_path), "summarize")
else:
raise HTTPException(status_code=400, detail="Invalid analysis type")
# Return results
return {
"filename": file.filename,
"analysis_type": analysis_type,
"result": result
}
except Exception as e:
# Clean up file on error
if file_path.exists():
os.remove(file_path)
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/analyze-image")
async def analyze_image(
file: UploadFile = File(...),
analysis_type: str = Form(...)
):
# Validate file type
allowed_extensions = [".jpg", ".jpeg", ".png"]
file_ext = os.path.splitext(file.filename)[1].lower()
if file_ext not in allowed_extensions:
raise HTTPException(status_code=400, detail=f"File type not supported. Allowed types: {', '.join(allowed_extensions)}")
# Create unique filename
unique_filename = f"{uuid.uuid4()}{file_ext}"
file_path = UPLOAD_DIR / unique_filename
# Save uploaded file
with open(file_path, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
try:
# Process image based on analysis type
if analysis_type == "caption":
result = process_image(str(file_path), "caption")
else:
raise HTTPException(status_code=400, detail="Invalid analysis type")
# Return results
return {
"filename": file.filename,
"analysis_type": analysis_type,
"result": result
}
except Exception as e:
# Clean up file on error
if file_path.exists():
os.remove(file_path)
raise HTTPException(status_code=500, detail=str(e))
# Add health check endpoint
@app.get("/health")
def health_check():
return {"status": "healthy"}
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=7860, reload=True) |