File size: 4,778 Bytes
f750267 6be194e f332e2d 6be194e f750267 d2116ed a842a46 6be194e f332e2d 6be194e f332e2d 6be194e f332e2d d2116ed f750267 d2116ed f750267 6be194e f750267 d2116ed f750267 d2116ed f750267 6be194e f750267 6be194e f750267 6be194e d2116ed 6be194e f750267 d2116ed f332e2d 6be194e d2116ed |
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 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 |
from fastapi import FastAPI, HTTPException, UploadFile, File
from pydantic import BaseModel
import requests
from fastapi.middleware.cors import CORSMiddleware
import os
import uuid
from pathlib import Path
import logging
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI()
# Add CORS middleware to allow requests from Hugging Face Spaces frontend
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Adjust for production to specific origins
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Define the request model
class TryOnRequest(BaseModel):
garmentDesc: str
category: str
# Base directory for file storage (use /tmp/gradio for Hugging Face Spaces)
UPLOAD_DIR = Path("/tmp/gradio")
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
# Helper function to save file and generate public URL
async def save_file_and_get_url(file: UploadFile) -> str:
try:
# Generate unique filename
file_extension = file.filename.split(".")[-1]
unique_filename = f"{uuid.uuid4()}.{file_extension}"
file_path = UPLOAD_DIR / unique_filename
# Save file
logger.info(f"Saving file to {file_path}")
with file_path.open("wb") as buffer:
content = await file.read()
buffer.write(content)
# Verify file exists
if not file_path.exists():
logger.error(f"File {file_path} was not saved correctly")
raise HTTPException(status_code=500, detail="Failed to save file")
# Generate public URL
# Use SPACE_ID environment variable or fallback to placeholder
public_url = f"https://tejani-tryapi.hf.space/file={str(file_path)}"
logger.info(f"Generated public URL: {public_url}")
# Test URL accessibility
try:
response = requests.head(public_url, timeout=5)
if response.status_code != 200:
logger.warning(f"Public URL {public_url} returned status {response.status_code}")
except requests.exceptions.RequestException as e:
logger.error(f"Failed to access public URL {public_url}: {str(e)}")
return public_url
except Exception as e:
logger.error(f"Error in save_file_and_get_url: {str(e)}")
raise HTTPException(status_code=500, detail=f"Error processing file: {str(e)}")
# Updated endpoint to handle file uploads and proxy the request
@app.post("/try-on")
async def try_on(
human_img: UploadFile = File(...),
garment: UploadFile = File(...),
garment_desc: str = "",
category: str = "upper_body"
):
try:
# Save files and get public URLs
human_img_url = await save_file_and_get_url(human_img)
garment_url = await save_file_and_get_url(garment)
# Original API endpoint
url = "https://changeclothesai.online/api/try-on/edge"
headers = {
"accept": "*/*",
"f": "sdfdsfsKaVgUoxa5j1jzcFtziPx",
}
data = {
"humanImg": human_img_url,
"garment": garment_url,
"garmentDesc": garment_desc,
"category": category
}
logger.info(f"Forwarding request to {url} with data: {data}")
# Forward request to the original API
response = requests.post(url, headers=headers, cookies={}, data=data)
response.raise_for_status()
return {
"status_code": response.status_code,
"response": response.json() if response.headers.get('content-type') == 'application/json' else response.text,
"human_img_url": human_img_url,
"garment_url": garment_url
}
except requests.exceptions.RequestException as e:
logger.error(f"Error forwarding request: {str(e)}")
raise HTTPException(status_code=500, detail=f"Error forwarding request: {str(e)}")
except Exception as e:
logger.error(f"Error in try_on endpoint: {str(e)}")
raise HTTPException(status_code=500, detail=f"Error processing request: {str(e)}")
# Health check endpoint for Hugging Face Spaces
@app.get("/")
async def root():
return {"message": "FastAPI proxy for try-on API with file upload is running"}
# Debug endpoint to list stored files
@app.get("/list-files")
async def list_files():
try:
files = [str(f) for f in UPLOAD_DIR.glob("*") if f.is_file()]
logger.info(f"Files in {UPLOAD_DIR}: {files}")
return {"files": files}
except Exception as e:
logger.error(f"Error listing files: {str(e)}")
raise HTTPException(status_code=500, detail=f"Error listing files: {str(e)}") |