Upload app.py

app.py

@@ -0,0 +1,132 @@
+from fastapi import FastAPI, File, UploadFile, Request
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel
+import httpx
+import io
+import uuid
+import os
+from sse_starlette.sse import EventSourceResponse
+import asyncio
+
+app = FastAPI()
+
+@app.get("/")
+async def root():
+    return {"message": "FastAPI to forward image to external API"}
+
+
+@app.post("/forward-upload")
+async def forward_upload(file: UploadFile = File(...)):
+    if file.content_type not in ["image/png", "image/jpeg"]:
+        return JSONResponse(status_code=400, content={"error": "Invalid file type. Only PNG or JPEG allowed."})
+
+    target_url = "https://kwai-kolors-kolors-virtual-try-on.hf.space/upload?upload_id=r9ywzifblvk"
+
+    try:
+        file_content = await file.read()
+        files = {"files": (file.filename, file_content, file.content_type)}
+
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                target_url,
+                files=files,
+                headers={
+                    "accept": "*/*",
+                    "accept-language": "en-GB,en-US;q=0.9,en;q=0.8,gu;q=0.7",
+                    "origin": "https://kwai-kolors-kolors-virtual-try-on.hf.space",
+                    "referer": "https://kwai-kolors-kolors-virtual-try-on.hf.space/?__theme=system",
+                    "user-agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 16_6 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.6 Mobile/15E148 Safari/604.1",
+                },
+            )
+
+        if response.status_code == 200:
+            return JSONResponse(content={"status": "Success", "response": response.json()})
+        else:
+            return JSONResponse(status_code=response.status_code, content={"error": response.text})
+
+    except httpx.HTTPError as e:
+        return JSONResponse(status_code=500, content={"error": f"HTTP error occurred: {str(e)}"})
+    except Exception as e:
+        return JSONResponse(status_code=500, content={"error": f"An error occurred: {str(e)}"})
+
+
+class TryOnRequest(BaseModel):
+    person_image_path: str
+    garment_image_path: str
+    session_hash: str 
+
+@app.post("/virtual-try-on")
+async def join(request: TryOnRequest):
+    url = "https://kwai-kolors-kolors-virtual-try-on.hf.space/queue/join?__theme=system"
+
+    headers = {
+        "accept": "*/*",
+        "content-type": "application/json",
+    }
+
+    # Extract orig_name from file path
+    person_orig_name = os.path.basename(request.person_image_path)
+    garment_orig_name = os.path.basename(request.garment_image_path)
+
+    #session_hash = str(uuid.uuid4())[:11]
+
+    payload = {
+        "data": [
+            {
+                "path": request.person_image_path,
+                "url": f"https://kwai-kolors-kolors-virtual-try-on.hf.space/file={request.person_image_path}",
+                "orig_name": person_orig_name,
+                "size": None,
+                "mime_type": None,
+                "is_stream": False,
+                "meta": {"_type": "gradio.FileData"}
+            },
+            {
+                "path": request.garment_image_path,
+                "url": f"https://kwai-kolors-kolors-virtual-try-on.hf.space/file={request.garment_image_path}",
+                "orig_name": garment_orig_name,
+                "size": None,
+                "mime_type": None,
+                "is_stream": False,
+                "meta": {"_type": "gradio.FileData"}
+            },
+            0,
+            True
+        ],
+        "event_data": None,
+        "fn_index": 2,
+        "trigger_id": 26,
+        "session_hash": request.session_hash
+    }
+
+    try:
+        async with httpx.AsyncClient() as client:
+            response = await client.post(url, headers=headers, json=payload)
+
+        if response.status_code == 200:
+            return response.json()
+        else:
+            raise HTTPException(status_code=response.status_code, detail=response.text)
+
+    except httpx.RequestError as e:
+        raise HTTPException(status_code=500, detail=f"Request failed: {str(e)}")
+
+
+async def event_generator(session_hash: str, request: Request):
+    SSE_URL = f"https://kwai-kolors-kolors-virtual-try-on.hf.space/queue/data?session_hash={session_hash}"
+
+    headers = {
+        "accept": "text/event-stream",
+        "content-type": "application/json",
+    }
+
+    async with httpx.AsyncClient(timeout=None) as client:
+        async with client.stream("GET", SSE_URL, headers=headers) as response:
+            async for line in response.aiter_lines():
+                if line.startswith("data:"):
+                    yield line.removeprefix("data:").strip()
+                await asyncio.sleep(0.01)
+
+@app.get("/sse-proxy")
+async def sse_proxy(session_hash: str, request: Request):
+    return EventSourceResponse(event_generator(session_hash, request))