API: add FastAPI endpoints for Hair Swap flow; deps updated

README.md

@@ -11,4 +11,19 @@ license: mit
 hardware: cpu-basic
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+ 
+## API (FastAPI)
+- Base URL: `https://logicgoinfotechspaces-hair-stable-new.hf.space`
+- Auth: Bearer token `logicgo@123`
+
+Endpoints:
+- `GET /health` → {"status":"healthy"}
+- `POST /upload` (form-data: image=file) → {"id":"<id>","filename":"name.png"}
+- `POST /get-hairswap` (JSON body with source_id, reference_id, …) → {"result":"output_xxx.png"}
+- `GET /download/{filename}` → image file
+- `GET /logs` → recent uploads/results
+
+Local run:
+- Install deps: `python3 -m pip install -r requirements.txt`
+- Run API: `python3 -m uvicorn api.main:app --host 0.0.0.0 --port 7860`

api/main.py

@@ -0,0 +1,100 @@
+import os
+import uuid
+import shutil
+from datetime import datetime
+from typing import Dict, List, Optional
+
+from fastapi import FastAPI, UploadFile, File, HTTPException, Depends, Header
+from fastapi.responses import FileResponse, JSONResponse
+from pydantic import BaseModel
+
+# Simple settings
+UPLOAD_DIR = os.path.join(os.getcwd(), "uploads")
+OUTPUT_DIR = os.path.join(os.getcwd(), "outputs")
+AUTH_TOKEN = "logicgo@123"  # Bearer token
+
+os.makedirs(UPLOAD_DIR, exist_ok=True)
+os.makedirs(OUTPUT_DIR, exist_ok=True)
+
+app = FastAPI(title="Hair Swap API", version="1.0.0")
+
+# In-memory stores
+file_store: Dict[str, Dict[str, str]] = {}
+logs: List[Dict[str, str]] = []
+
+
+def bearer_auth(authorization: Optional[str] = Header(default=None)) -> None:
+    if authorization is None or not authorization.lower().startswith("bearer "):
+        raise HTTPException(status_code=401, detail="Unauthorized")
+    token = authorization.split(" ", 1)[1]
+    if token != AUTH_TOKEN:
+        raise HTTPException(status_code=403, detail="Forbidden")
+
+
+class HairSwapRequest(BaseModel):
+    source_id: str
+    reference_id: str
+    converter_scale: Optional[float] = 1.0
+    scale: Optional[float] = 1.0
+    guidance_scale: Optional[float] = 1.5
+    controlnet_conditioning_scale: Optional[float] = 1.0
+
+
+@app.get("/health")
+def health() -> Dict[str, str]:
+    return {"status": "healthy"}
+
+
+@app.post("/upload")
+def upload_image(image: UploadFile = File(...), _: None = Depends(bearer_auth)) -> Dict[str, str]:
+    ext = os.path.splitext(image.filename)[1] or ".png"
+    file_id = str(uuid.uuid4())
+    stored_name = f"{file_id}{ext}"
+    stored_path = os.path.join(UPLOAD_DIR, stored_name)
+
+    with open(stored_path, "wb") as f:
+        shutil.copyfileobj(image.file, f)
+
+    file_store[file_id] = {
+        "filename": image.filename,
+        "stored_name": stored_name,
+        "path": stored_path,
+        "timestamp": datetime.utcnow().isoformat(),
+    }
+    logs.append({"id": file_id, "filename": image.filename, "timestamp": datetime.utcnow().isoformat()})
+    return {"id": file_id, "filename": image.filename}
+
+
+@app.post("/get-hairswap")
+def get_hairswap(req: HairSwapRequest, _: None = Depends(bearer_auth)) -> Dict[str, str]:
+    if req.source_id not in file_store:
+        raise HTTPException(status_code=404, detail="source_id not found")
+    if req.reference_id not in file_store:
+        raise HTTPException(status_code=404, detail="reference_id not found")
+
+    # Placeholder: In lieu of a real hair-swap model, return the source image as the result
+    src_path = file_store[req.source_id]["path"]
+    result_name = f"output_{uuid.uuid4().hex}.png"
+    result_path = os.path.join(OUTPUT_DIR, result_name)
+    shutil.copyfile(src_path, result_path)
+
+    logs.append({"result": result_name, "timestamp": datetime.utcnow().isoformat()})
+    return {"result": result_name}
+
+
+@app.get("/download/{filename}")
+def download_file(filename: str, _: None = Depends(bearer_auth)):
+    # Serve from outputs first, then uploads as a fallback
+    path = os.path.join(OUTPUT_DIR, filename)
+    if not os.path.isfile(path):
+        alt = os.path.join(UPLOAD_DIR, filename)
+        if os.path.isfile(alt):
+            path = alt
+        else:
+            raise HTTPException(status_code=404, detail="file not found")
+    return FileResponse(path)
+
+
+@app.get("/logs")
+def get_logs(_: None = Depends(bearer_auth)) -> JSONResponse:
+    return JSONResponse(content=logs)

requirements.txt

@@ -22,4 +22,7 @@ kornia==0.5.0
 webdataset
 packaging
 wldhx.yadisk-direct
-altair<5
+altair<5
+fastapi
+uvicorn[standard]
+python-multipart