Create app.py

app.py

@@ -0,0 +1,225 @@
+from __future__ import annotations
+
+import os
+import re
+import time
+from typing import Any, Dict, List, Optional
+
+import numpy as np
+import pandas as pd
+import faiss
+from flask import Flask, request, jsonify
+from flask_cors import CORS
+from sentence_transformers import SentenceTransformer
+
+
+# =========================
+# Config
+# =========================
+INDEX_PATH = os.getenv("HADITH_INDEX_PATH", "hadith_ar.faiss")
+META_PATH  = os.getenv("HADITH_META_PATH",  "hadith_meta.parquet")
+MODEL_NAME = os.getenv("HADITH_MODEL_NAME", "intfloat/multilingual-e5-base")
+
+DEFAULT_TOP_K = 10
+MAX_TOP_K = 50
+
+# If you want a smaller response payload
+DEFAULT_INCLUDE_TEXT = True
+
+
+# =========================
+# Arabic normalization
+# =========================
+_AR_DIACRITICS = re.compile(r"""
+    [\u0610-\u061A]
+  | [\u064B-\u065F]
+  | [\u0670]
+  | [\u06D6-\u06ED]
+""", re.VERBOSE)
+
+def normalize_ar(text: str) -> str:
+    """Remove tashkeel + normalize common Arabic letter variants."""
+    if text is None:
+        return ""
+    text = str(text)
+    text = _AR_DIACRITICS.sub("", text)
+    text = text.replace("ـ", "")
+    text = re.sub(r"[إأآٱ]", "ا", text)
+    text = text.replace("ى", "ي")
+    text = text.replace("ؤ", "و")
+    text = text.replace("ئ", "ي")
+    text = re.sub(r"\s+", " ", text).strip()
+    return text
+
+
+# =========================
+# Load model + index + meta (once)
+# =========================
+if not os.path.exists(INDEX_PATH):
+    raise FileNotFoundError(f"FAISS index not found: {INDEX_PATH}")
+
+if not os.path.exists(META_PATH):
+    raise FileNotFoundError(f"Meta parquet not found: {META_PATH}")
+
+model = SentenceTransformer(MODEL_NAME)
+index = faiss.read_index(INDEX_PATH)
+meta  = pd.read_parquet(META_PATH)
+
+required_cols = {"hadithID", "collection", "hadith_number", "arabic", "english"}
+missing = required_cols - set(meta.columns)
+if missing:
+    raise ValueError(f"Meta is missing required columns: {missing}")
+
+if "arabic_clean" not in meta.columns:
+    meta["arabic_clean"] = ""
+
+# Normalize column types to avoid NaN surprises
+for col in ["arabic", "english", "arabic_clean", "collection"]:
+    if col in meta.columns:
+        meta[col] = meta[col].fillna("").astype(str)
+
+
+def semantic_search(query: str, top_k: int = DEFAULT_TOP_K) -> pd.DataFrame:
+    q = str(query or "").strip()
+    if not q:
+        return meta.iloc[0:0].copy()
+
+    top_k = max(1, min(int(top_k), MAX_TOP_K))
+
+    q_norm = normalize_ar(q)
+    q_emb = model.encode(["query: " + q_norm], normalize_embeddings=True).astype("float32")
+    scores, idx = index.search(q_emb, top_k)
+
+    res = meta.iloc[idx[0]].copy()
+    res["score"] = scores[0].astype(float)
+    res = res.sort_values("score", ascending=False)
+
+    # Ensure no empty Arabic (avoid useless results)
+    res["arabic"] = res["arabic"].fillna("").astype(str)
+    res = res[res["arabic"].str.strip() != ""]
+
+    return res
+
+
+def row_to_json(row: pd.Series, include_text: bool = True) -> Dict[str, Any]:
+    arabic = str(row.get("arabic", "") or "")
+    arabic_clean = str(row.get("arabic_clean", "") or "").strip()
+    if not arabic_clean:
+        arabic_clean = normalize_ar(arabic)
+
+    base = {
+        "score": float(row.get("score", 0.0)),
+        "hadithID": int(row.get("hadithID")),
+        "collection": str(row.get("collection", "")),
+        "hadith_number": int(row.get("hadith_number")),
+    }
+
+    if include_text:
+        base.update({
+            "arabic": arabic,
+            "arabic_clean": arabic_clean,
+            "english": str(row.get("english", "") or ""),
+        })
+
+    return base
+
+
+# =========================
+# Flask API app
+# =========================
+app = Flask(__name__)
+CORS(app, resources={r"/*": {"origins": "*"}})  # allow calls from other hosts
+
+
+@app.get("/health")
+def health():
+    return jsonify({
+        "ok": True,
+        "rows": int(len(meta)),
+        "index_ntotal": int(getattr(index, "ntotal", -1)),
+        "model": MODEL_NAME
+    })
+
+
+@app.post("/search")
+def search():
+    """
+    JSON body:
+    {
+      "q": "الزرق و سبيل الرزق",
+      "k": 10,
+      "include_text": true
+    }
+    """
+    payload = request.get_json(silent=True) or {}
+    q = (payload.get("q") or "").strip()
+    k = payload.get("k", DEFAULT_TOP_K)
+    include_text = payload.get("include_text", DEFAULT_INCLUDE_TEXT)
+
+    # Validate
+    if not q:
+        return jsonify({"ok": False, "error": "Missing 'q'"}), 400
+    try:
+        k = int(k)
+    except Exception:
+        k = DEFAULT_TOP_K
+    k = max(1, min(k, MAX_TOP_K))
+
+    t0 = time.time()
+    res_df = semantic_search(q, top_k=k)
+    took_ms = int((time.time() - t0) * 1000)
+
+    results = [row_to_json(r, include_text=bool(include_text)) for _, r in res_df.iterrows()]
+
+    return jsonify({
+        "ok": True,
+        "query": q,
+        "query_norm": normalize_ar(q),
+        "k": k,
+        "took_ms": took_ms,
+        "results_count": len(results),
+        "results": results
+    })
+
+
+@app.get("/search")
+def search_get():
+    """
+    GET /search?q=...&k=10&include_text=1
+    Useful for quick testing in browser.
+    """
+    q = (request.args.get("q") or "").strip()
+    k = request.args.get("k", str(DEFAULT_TOP_K))
+    include_text = request.args.get("include_text", "1")
+
+    if not q:
+        return jsonify({"ok": False, "error": "Missing 'q'"}), 400
+
+    try:
+        k_int = int(k)
+    except Exception:
+        k_int = DEFAULT_TOP_K
+    k_int = max(1, min(k_int, MAX_TOP_K))
+
+    include_text_bool = include_text not in ("0", "false", "False", "")
+
+    t0 = time.time()
+    res_df = semantic_search(q, top_k=k_int)
+    took_ms = int((time.time() - t0) * 1000)
+
+    results = [row_to_json(r, include_text=include_text_bool) for _, r in res_df.iterrows()]
+
+    return jsonify({
+        "ok": True,
+        "query": q,
+        "query_norm": normalize_ar(q),
+        "k": k_int,
+        "took_ms": took_ms,
+        "results_count": len(results),
+        "results": results
+    })
+
+
+if __name__ == "__main__":
+    # For local debug only. On HF Spaces, gunicorn/uvicorn handles it.
+    app.run(host="0.0.0.0", port=7860, debug=False)