Spaces:
Running
Running
File size: 2,596 Bytes
4a672f4 7d4226e 4a672f4 f984e57 4a672f4 f984e57 4a672f4 f984e57 4a672f4 797f638 f984e57 |
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 |
from fastapi import FastAPI, File, UploadFile, HTTPException
from fastapi.responses import JSONResponse
from pydantic import BaseModel
import shutil
import os
import json
app = FastAPI(docs_url=None, redoc_url=None)
# Create folders to store uploaded images
UPLOAD_FOLDER = "uploads"
IMAGE_FOLDER = os.path.join(UPLOAD_FOLDER, "res")
VIDEO_FOLDER = os.path.join(UPLOAD_FOLDER, "videos")
for folder in [UPLOAD_FOLDER, IMAGE_FOLDER, VIDEO_FOLDER]:
os.makedirs(folder, exist_ok=True)
# Create a dictionary to store available images
available_images = {}
class ImageInfo(BaseModel):
url: str
name: str
@app.post("/upload")
async def upload_image(url: str, file: UploadFile = File(...)):
allowed_formats = [".png", ".jpg"]
file_format = os.path.splitext(file.filename)[1]
if file_format not in allowed_formats:
raise HTTPException(status_code=400, detail="Unsupported file format")
# Save the uploaded file
with open(os.path.join(UPLOAD_FOLDER, file.filename), "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
# Determine the folder based on the file format
folder = "images" if file_format in [".png", ".jpg"] else "videos"
# Move the file to the appropriate folder
os.rename(
os.path.join(UPLOAD_FOLDER, file.filename),
os.path.join(UPLOAD_FOLDER, folder, file.filename)
)
# Update available images dictionary
available_images[file.filename] = ImageInfo(url=url, name=file.filename)
return JSONResponse(content={"message": "File uploaded successfully"})
@app.get("/get-all-images")
async def get_all_images(password: str, from_folder: str):
if password != "Kastg@123":
raise HTTPException(status_code=401, detail="Unauthorized")
folder_path = os.path.join(UPLOAD_FOLDER, from_folder)
if not os.path.exists(folder_path):
raise HTTPException(status_code=404, detail="Folder not found")
images = os.listdir(folder_path)
return JSONResponse(content={"images": images})
@app.delete("/delete-image")
async def delete_image(image_name: str, password: str):
if password != "Kastg@123":
raise HTTPException(status_code=401, detail="Unauthorized")
image_path = os.path.join(UPLOAD_FOLDER, image_name)
if not os.path.exists(image_path):
raise HTTPException(status_code=404, detail="Image not found")
os.remove(image_path)
available_images.pop(image_name, None)
return JSONResponse(content={"message": "Image deleted successfully"})
if __name__ == "__main__":
app.run(host="0.0.0.0", port=7860, debug=True) |