|
from fastapi import FastAPI, File, UploadFile, HTTPException, Form |
|
from fastapi.responses import FileResponse |
|
from fastapi.staticfiles import StaticFiles |
|
import requests |
|
import os |
|
import uuid |
|
import shutil |
|
import logging |
|
import time |
|
|
|
|
|
logging.basicConfig(level=logging.INFO) |
|
logger = logging.getLogger(__name__) |
|
|
|
app = FastAPI() |
|
|
|
|
|
STATIC_DIR = "/app/static" |
|
os.makedirs(STATIC_DIR, exist_ok=True) |
|
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") |
|
|
|
|
|
CHANGE_CLOTHES_API = os.environ.get('CHANGE_CLOTHES_API', 'https://changeclothesai.online/api/try-on/edge') |
|
|
|
@app.post("/try-on/") |
|
async def try_on( |
|
human_img: UploadFile = File(...), |
|
garment_img: UploadFile = File(...), |
|
garment_desc: str = Form("dresses"), |
|
category: str = Form("dresses") |
|
): |
|
try: |
|
|
|
human_filename = f"{uuid.uuid4()}_{human_img.filename}" |
|
garment_filename = f"{uuid.uuid4()}_{garment_img.filename}" |
|
|
|
|
|
human_path = os.path.join(STATIC_DIR, human_filename) |
|
garment_path = os.path.join(STATIC_DIR, garment_filename) |
|
|
|
logger.info(f"Saving human image to: {human_path}") |
|
with open(human_path, "wb") as f: |
|
shutil.copyfileobj(human_img.file, f) |
|
logger.info(f"Human image saved: {os.path.exists(human_path)}") |
|
|
|
logger.info(f"Saving garment image to: {garment_path}") |
|
with open(garment_path, "wb") as f: |
|
shutil.copyfileobj(garment_img.file, f) |
|
logger.info(f"Garment image saved: {os.path.exists(garment_path)}") |
|
|
|
|
|
base_url = "https://tejani-tryapi.hf.space" |
|
human_url = f"{base_url}/static/{human_filename}" |
|
garment_url = f"{base_url}/static/{garment_filename}" |
|
logger.info(f"Human URL: {human_url}") |
|
logger.info(f"Garment URL: {garment_url}") |
|
|
|
|
|
time.sleep(1) |
|
try: |
|
url_check = requests.head(human_url, timeout=5) |
|
logger.info(f"Human URL status: {url_check.status_code}") |
|
if url_check.status_code != 200: |
|
raise HTTPException(status_code=500, detail=f"Human image URL inaccessible: {human_url}") |
|
url_check = requests.head(garment_url, timeout=5) |
|
logger.info(f"Garment URL status: {url_check.status_code}") |
|
if url_check.status_code != 200: |
|
raise HTTPException(status_code=500, detail=f"Garment image URL inaccessible: {garment_url}") |
|
except requests.RequestException as e: |
|
raise HTTPException(status_code=500, detail=f"Error accessing URLs: {str(e)}") |
|
|
|
|
|
headers = { |
|
"accept": "*/*", |
|
"f": str(uuid.uuid4()).replace("-", ""), |
|
} |
|
cookies = {} |
|
data = { |
|
"humanImg": human_url, |
|
"garment": garment_url, |
|
"garmentDesc": garment_desc, |
|
"category": category, |
|
} |
|
|
|
|
|
logger.info(f"Calling API: {CHANGE_CLOTHES_API}") |
|
response = requests.post(CHANGE_CLOTHES_API, headers=headers, cookies=cookies, data=data, timeout=60) |
|
logger.info(f"API Response: {response.status_code}, {response.text}") |
|
|
|
|
|
if response.status_code != 200: |
|
raise HTTPException(status_code=response.status_code, detail=f"Error calling ChangeClothesAI API: {response.text}") |
|
|
|
|
|
os.remove(human_path) |
|
os.remove(garment_path) |
|
logger.info("Temporary files deleted") |
|
|
|
return response.json() |
|
|
|
except Exception as e: |
|
|
|
if os.path.exists(human_path): |
|
os.remove(human_path) |
|
if os.path.exists(garment_path): |
|
os.remove(garment_path) |
|
logger.error(f"Error processing request: {str(e)}") |
|
raise HTTPException(status_code=500, detail=f"Error processing files: {str(e)}") |
|
|
|
finally: |
|
human_img.file.close() |
|
garment_img.file.close() |
|
|
|
if __name__ == "__main__": |
|
import uvicorn |
|
uvicorn.run(app, host="0.0.0.0", port=7860) |