reverse-dictionary/backend/app/search.py

# ============================================================
# 📄 الملف: app/search.py
# 🎯 الغرض: منطق البحث الدلالي (Semantic Search) الأساسي.
#
# هذا قلب المشروع: يأخذ تعريفًا (استعلام) ويحوّله إلى متجه،
# ثم يقارنه بكل متجهات البيانات عبر تشابه الجيب التمام (cosine)،
# ويرجّع أقرب الكلمات المطابقة فوق حدّ معيّن (threshold).
#
# يحتوي على دالتين:
#   search_one_model  → بحث في نموذج واحد.
#   search_all_models → بحث في النماذج الثلاثة معًا (المستخدم في الـ API).
# ============================================================

import numpy as np
import pandas as pd
from sklearn.metrics.pairwise import cosine_similarity

import config
from app.models_loader import resources
from app.text_utils import strip_diacritics


def _safe(value) -> str:
    """يحوّل القيمة إلى نص نظيف ويتعامل مع القيم الفارغة (NaN)."""
    if value is None or (isinstance(value, float) and pd.isna(value)):
        return ""
    text = str(value)
    return "" if text.lower() == "nan" else text


def _build_item(row, score: float) -> dict:
    """يبني عنصر نتيجة واحد من صف البيانات ودرجة التشابه."""
    return {
        "lemma": _safe(row.get("lemma")),
        "definition": _safe(row.get("definition")),
        "example": _safe(row.get("example")),
        "pos": _safe(row.get("pos_label")),
        "score": round(float(score), 3),
    }


def search_one_model(query: str, model_cfg: dict, threshold: float,
                     top_k: int | None = None) -> list[dict]:
    """
    يبحث في نموذج واحد ويرجّع قائمة نتائج مرتّبة تنازليًا حسب التشابه.

    المعاملات:
        query     : نص التعريف المُدخل.
        model_cfg : إعدادات النموذج (من config.MODELS).
        threshold : أقل قيمة تشابه مقبولة.
        top_k     : حدّ أقصى للنتائج يتجاوز قيمة config (اختياري). إن لم
                    يُمرَّر، يُستخدم model_cfg["top_k"].
    """
    df = resources.get_dataframe()
    doc_embeds = resources.get_embeddings(model_cfg["key"])
    model = resources.get_model(model_cfg["key"], model_cfg["hf_id"])

    # حذف التشكيل من الاستعلام ليطابق معالجة المستندات (نفس خطوة البناء)
    query = strip_diacritics(query)

    # نماذج e5 تتطلّب بادئة "query:" للاستعلام
    prepared_query = f"query: {query}" if model_cfg["use_prefix"] else query

    # ترميز الاستعلام إلى متجه مُطبّع
    q_emb = model.encode([prepared_query], normalize_embeddings=True, convert_to_numpy=True)

    # حساب التشابه مع كل المتجهات ثم الترتيب تنازليًا
    sims = cosine_similarity(q_emb, doc_embeds)[0]
    idx_sorted = np.argsort(sims)[::-1]

    # top_k الممرّر يتجاوز قيمة config (وإلا نستخدم قيمة النموذج)
    top_k = top_k if top_k is not None else model_cfg["top_k"]

    # كوساين + threshold + top_k
    results = []
    for i in idx_sorted:
        score = float(sims[i])
        if score < threshold:
            break  # القائمة مرتّبة، فلا داعي للاستمرار
        results.append(_build_item(df.iloc[i], score))
        if top_k is not None and len(results) >= top_k:
            break
    return results


def search_all_models(query: str, threshold: float) -> list[dict]:
    """يبحث في النماذج الثلاثة ويرجّع نتائج كل واحد منفصلة."""
    all_results = []
    for index, model_cfg in enumerate(config.MODELS):
        items = search_one_model(query, model_cfg, threshold)
        all_results.append({
            "model_index": index,
            "model_key": model_cfg["key"],
            "display_name": model_cfg["display_name"],
            "items": items,
            "top_lemma": items[0]["lemma"] if items else None,
            "top_score": items[0]["score"] if items else None,
        })
    return all_results