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 # Set up logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI() # Mount a static directory to serve files STATIC_DIR = "/app/static" os.makedirs(STATIC_DIR, exist_ok=True) app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") # API endpoint for ChangeClothesAI 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(""), # Match Trinket category: str = Form("upper_body") # Match Trinket ): try: # Generate unique filenames human_filename = f"{uuid.uuid4()}_{human_img.filename}" garment_filename = f"{uuid.uuid4()}_{garment_img.filename}" # Save files temporarily in the static directory 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)}") # Generate public URLs base_url = "https://tejani-tryapi.hf.space" # Verify this is correct 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}") # Skip URL verification for debugging # try: # url_check = requests.head(human_url, timeout=10) # 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=10) # 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)}") # Prepare headers, cookies, and data for the API headers = { "accept": "*/*", "f": "sdfdsfsKaVgUoxa5j1jzcFtziPx", # Match Trinket } cookies = {} data = { "humanImg": human_url, "garment": garment_url, "garmentDesc": garment_desc, "category": category, } # Test with Trinket URLs (uncomment to test) # data = { # "humanImg": "https://images.pexels.com/photos/1391498/pexels-photo-1391498.jpeg", # "garment": "https://r2.changeclothesai.online/1752560324456537514.jpeg", # "garmentDesc": "", # "category": "upper_body" # } # Call the ChangeClothesAI API logger.info(f"Sending API request: URL={CHANGE_CLOTHES_API}, Headers={headers}, Data={data}") response = requests.post(CHANGE_CLOTHES_API, headers=headers, cookies=cookies, data=data, timeout=60) logger.info(f"API Response: Status={response.status_code}, Body={response.text}, Headers={response.headers}") # Check if the API call was successful if response.status_code != 200: raise HTTPException(status_code=response.status_code, detail=f"Error calling ChangeClothesAI API: {response.text}") # Delete temporary files (commented out for debugging) # os.remove(human_path) # os.remove(garment_path) # logger.info("Temporary files deleted") return response.json() except Exception as e: # Clean up in case of errors (commented out for debugging) # 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)