Spaces:
Sleeping
Sleeping
File size: 2,180 Bytes
44a025a 11aa943 44a025a 11aa943 44a025a 11aa943 44a025a 11aa943 |
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 |
import os
import shutil
from fastapi import APIRouter, UploadFile, File, HTTPException
from typing import Dict, Any
from app.services.embedding_service import process_and_create_embeddings
router = APIRouter()
@router.post("/create-embeddings")
async def create_embeddings(file: UploadFile = File(...)):
"""
Create embeddings from an uploaded Excel or CSV file containing question-answer pairs.
- **file**: The Excel or CSV file containing questions and answers
Returns a success message if embeddings are created successfully.
"""
# Get file extension
file_extension = os.path.splitext(file.filename)[1].lower()
# Determine file type based on extension
if file_extension == ".xlsx" or file_extension == ".xls":
file_type = "excel"
elif file_extension == ".csv":
file_type = "csv"
else:
raise HTTPException(
status_code=400,
detail="Unsupported file type. Please upload an Excel (.xlsx, .xls) or CSV (.csv) file.",
)
# Ensure temp directory exists
temp_dir = "/app/temp"
if not os.path.exists(temp_dir):
os.makedirs(temp_dir, exist_ok=True)
# Create a temporary file to store the uploaded file
safe_filename = os.path.basename(file.filename).replace(" ", "_")
temp_file_path = os.path.join(temp_dir, f"temp_{safe_filename}")
try:
# Save the uploaded file
with open(temp_file_path, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
# Process the file and create embeddings
result = process_and_create_embeddings(temp_file_path, file_type)
if result["status"] == "error":
raise HTTPException(status_code=400, detail=result["message"])
return result
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
finally:
# Clean up the temporary file
if os.path.exists(temp_file_path):
try:
os.remove(temp_file_path)
except Exception as e:
print(
f"Warning: Could not remove temporary file {temp_file_path}: {str(e)}"
)
|