Create app.py

app.py

@@ -0,0 +1,186 @@
+import os
+import asyncio
+from typing import Optional, List, Dict, Any
+
+import httpx
+from fastapi import FastAPI, UploadFile, File, Form, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
+
+# =========================
+# إعدادات عامة
+# =========================
+
+HF_API_TOKEN = os.getenv("HF_API_TOKEN")
+if not HF_API_TOKEN:
+    raise RuntimeError("Please set HF_API_TOKEN as a Secret in your Space.")
+
+# موديلات Hugging Face المستخدمة
+DETECTOR_MODEL = os.getenv("DETECTOR_MODEL", "Tinny-Robot/acne")
+SEVERITY_MODEL = os.getenv("SEVERITY_MODEL", "imfarzanansari/skintelligent-acne")
+CONDITION_MODEL = os.getenv("CONDITION_MODEL", "Tanishq77/skin-condition-classifier")
+
+HF_API_BASE = "https://api-inference.huggingface.co/models"
+
+HEADERS = {"Authorization": f"Bearer {HF_API_TOKEN}"}
+TIMEOUT = 60.0  # ثواني
+
+
+# =========================
+# Schemas
+# =========================
+
+class SeverityOut(BaseModel):
+    raw: str
+    label_ar: str
+    score: float
+
+
+class ConditionOut(BaseModel):
+    label: str
+    score: float
+
+
+class AnalysisResponse(BaseModel):
+    num_lesions: int
+    severity: SeverityOut
+    condition: ConditionOut
+    meta: dict
+    # ممكن تضيف حقول أخرى لاحقًا (مثلاً توزيع الحبوب)
+
+
+# =========================
+# Utilities
+# =========================
+
+def map_severity_label_ar(label: str) -> str:
+    m = {
+        "clear": "صافية",
+        "mild": "خفيفة",
+        "moderate": "متوسطة",
+        "severe": "شديدة",
+    }
+    return m.get(label.lower(), label)
+
+
+async def hf_object_detection(client: httpx.AsyncClient, image_bytes: bytes) -> List[Dict[str, Any]]:
+    url = f"{HF_API_BASE}/{DETECTOR_MODEL}"
+    resp = await client.post(url, headers=HEADERS, content=image_bytes)
+    if resp.status_code != 200:
+        raise HTTPException(status_code=500, detail=f"Detector error: {resp.text}")
+    data = resp.json()
+    # نتوقع list
+    if not isinstance(data, list):
+        # بعض الموديلات ترجع dict مع "error"
+        raise HTTPException(status_code=500, detail=f"Unexpected detector response: {data}")
+    return data
+
+
+async def hf_image_classification(client: httpx.AsyncClient, image_bytes: bytes, model_name: str) -> List[Dict[str, Any]]:
+    url = f"{HF_API_BASE}/{model_name}"
+    resp = await client.post(url, headers=HEADERS, content=image_bytes)
+    if resp.status_code != 200:
+        raise HTTPException(status_code=500, detail=f"{model_name} error: {resp.text}")
+    data = resp.json()
+    if not isinstance(data, list):
+        raise HTTPException(status_code=500, detail=f"Unexpected response from {model_name}: {data}")
+    return data
+
+
+# =========================
+# FastAPI app
+# =========================
+
+app = FastAPI(title="Acne Orchestrator", version="0.1.0")
+
+# CORS عشان تربطه مع Lovable / فرونت ثاني
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],  # عدلها لو تبي Origins محددة
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+@app.get("/")
+async def root():
+    return {
+        "message": "Acne Orchestrator is running.",
+        "models": {
+            "detector": DETECTOR_MODEL,
+            "severity": SEVERITY_MODEL,
+            "condition": CONDITION_MODEL,
+        },
+    }
+
+
+@app.post("/analyze", response_model=AnalysisResponse)
+async def analyze(
+    file: UploadFile = File(...),
+    age: Optional[int] = Form(None),
+    skin_type: Optional[str] = Form(None),
+    notes: Optional[str] = Form(None),
+):
+    """
+    Endpoint رئيسي:
+    يستقبل صورة + معلومات بسيطة ويرجع:
+    - عدد الحبوب (من object detection)
+    - شدة الحالة (severity model)
+    - نوع الحالة الجلدية (condition model)
+    """
+
+    # قراءة الصورة كـ bytes
+    image_bytes = await file.read()
+    if not image_bytes:
+        raise HTTPException(status_code=400, detail="Empty image file.")
+
+    async with httpx.AsyncClient(timeout=TIMEOUT) as client:
+        # نطلق النداءات الثلاثة بالتوازي
+        det_task = hf_object_detection(client, image_bytes)
+        sev_task = hf_image_classification(client, image_bytes, SEVERITY_MODEL)
+        cond_task = hf_image_classification(client, image_bytes, CONDITION_MODEL)
+
+        det_json, sev_json, cond_json = await asyncio.gather(det_task, sev_task, cond_task)
+
+    # ========== عدد الحبوب ==========
+    # HF object-detection عادة ترجع list of dicts
+    num_lesions = len(det_json)
+
+    # ========== severity ==========
+    # HF image-classification: list of {"label": ..., "score": ...}
+    sev_top = max(sev_json, key=lambda x: x.get("score", 0))
+    severity_raw = str(sev_top.get("label", "unknown"))
+    severity_score = float(sev_top.get("score", 0.0))
+    severity_ar = map_severity_label_ar(severity_raw)
+
+    severity_obj = SeverityOut(
+        raw=severity_raw,
+        label_ar=severity_ar,
+        score=severity_score,
+    )
+
+    # ========== condition ==========
+    cond_top = max(cond_json, key=lambda x: x.get("score", 0))
+    condition_label = str(cond_top.get("label", "unknown"))
+    condition_score = float(cond_top.get("score", 0.0))
+
+    condition_obj = ConditionOut(
+        label=condition_label,
+        score=condition_score,
+    )
+
+    meta = {
+        "age": age,
+        "skin_type": skin_type,
+        "notes": notes,
+        "filename": file.filename,
+        "content_type": file.content_type,
+    }
+
+    return AnalysisResponse(
+        num_lesions=num_lesions,
+        severity=severity_obj,
+        condition=condition_obj,
+        meta=meta,
+    )