Create app.py

app.py

@@ -0,0 +1,97 @@
+"""
+KG Embedding Server — FastAPI on HuggingFace Spaces
+يشتغل كـ REST API، يحمل الموديل مرة واحدة بس
+"""
+
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel
+from typing import Optional
+import math
+import re
+import contextlib
+from io import StringIO
+
+app = FastAPI(title="KG Embedding Server")
+
+# ══════════════════════════════════════════════════════
+#  GLOBALS — الموديل بيتحمل مرة واحدة عند startup
+# ══════════════════════════════════════════════════════
+_model = None
+_use_st = False
+
+@app.on_event("startup")
+def load_model():
+    global _model, _use_st
+    try:
+        from sentence_transformers import SentenceTransformer
+        with contextlib.redirect_stdout(StringIO()), contextlib.redirect_stderr(StringIO()):
+            _model = SentenceTransformer("all-MiniLM-L6-v2")
+        _use_st = True
+        print("[Server] sentence-transformers loaded ✓")
+    except Exception as e:
+        print(f"[Server] ST unavailable: {e}")
+        _use_st = False
+
+# ══════════════════════════════════════════════════════
+#  REQUEST / RESPONSE MODELS
+# ══════════════════════════════════════════════════════
+class EmbedRequest(BaseModel):
+    texts: list[str]
+
+class EmbedResponse(BaseModel):
+    embeddings: list[list[float]]
+    model: str = "all-MiniLM-L6-v2"
+
+class SimilarityRequest(BaseModel):
+    query: str
+    candidates: list[str]   # النصوص اللي هنقيس عليها
+
+class SimilarityResponse(BaseModel):
+    scores: list[float]     # cosine similarity لكل candidate
+
+class HealthResponse(BaseModel):
+    status: str
+    model_loaded: bool
+
+# ══════════════════════════════════════════════════════
+#  ENDPOINTS
+# ══════════════════════════════════════════════════════
+
+@app.get("/health", response_model=HealthResponse)
+def health():
+    return {"status": "ok", "model_loaded": _use_st}
+
+@app.post("/embed", response_model=EmbedResponse)
+def embed(req: EmbedRequest):
+    if not _use_st or _model is None:
+        raise HTTPException(503, "Model not loaded")
+    if not req.texts:
+        return EmbedResponse(embeddings=[])
+    with contextlib.redirect_stdout(StringIO()), contextlib.redirect_stderr(StringIO()):
+        vecs = _model.encode(req.texts, show_progress_bar=False)
+    return EmbedResponse(embeddings=[v.tolist() for v in vecs])
+
+@app.post("/similarity", response_model=SimilarityResponse)
+def similarity(req: SimilarityRequest):
+    """Query + candidates → cosine scores (الأسرع لو عايز ترتب نتايج)"""
+    if not _use_st or _model is None:
+        raise HTTPException(503, "Model not loaded")
+    texts = [req.query] + req.candidates
+    with contextlib.redirect_stdout(StringIO()), contextlib.redirect_stderr(StringIO()):
+        vecs = _model.encode(texts, show_progress_bar=False)
+    qvec = vecs[0]
+    scores = [_cosine(qvec, v) for v in vecs[1:]]
+    return SimilarityResponse(scores=scores)
+
+def _cosine(a, b) -> float:
+    dot = sum(x * y for x, y in zip(a, b))
+    na = math.sqrt(sum(x * x for x in a)) or 1e-9
+    nb = math.sqrt(sum(x * x for x in b)) or 1e-9
+    return dot / (na * nb)
+
+# ══════════════════════════════════════════════════════
+#  PING — لمنع الـ Space من النوم (اعمل cron job يبعته كل 4 دقايق)
+# ══════════════════════════════════════════════════════
+@app.get("/ping")
+def ping():
+    return {"pong": True}