reverse-dictionary/backend/app/database.py

# ============================================================
# 📄 الملف: app/database.py
# 🎯 الغرض: طبقة قاعدة البيانات الوحيدة — MongoDB.
#
# يُقرأ رابط الاتصال MONGODB_URI من البيئة/الأسرار (لا يُكتب في الكود):
#   • محليًا   : من ملف .env  (مثال: mongodb://localhost:27017)
#   • على HF   : من أسرار الـ Space (رابط Atlas: mongodb+srv://...)
#
# مجموعتان (collections):
#   evaluators : المقيّمون (email فريد).
#   ratings    : وثيقة لكل كلمة قُيّمت (صح/خطأ) + اسم وإيميل المقيّم.
#
# أمان: عمليات الحذف في الواجهة مقصورة على إيميل المقيّم الحالي،
#       فلا يقدر مقيّم على حذف تقييمات غيره.
# ============================================================

import os
from datetime import datetime

import certifi
import pandas as pd
import streamlit as st

# اسم قاعدة البيانات الافتراضي إن لم يُحدَّد داخل الرابط
DEFAULT_DB_NAME = "reverse_dictionary"

# خرائط الأعمدة → أسماء عربية للعرض في الواجهة
_AR_COLUMNS = {
    "created_at": "التاريخ والوقت",
    "query": "الاستعلام",
    "word_type": "التصنيف",
    "model": "النموذج",
    "word": "الكلمة",
    "similarity": "نسبة التشابه",
    "verdict": "التقييم",
    "evaluator_name": "المقيّم",
    "evaluator_email": "الإيميل",
}


# ------------------------------------------------------------
# 🔌 قراءة الإعدادات والاتصال
# ------------------------------------------------------------
def _setting(key: str) -> str | None:
    """يقرأ إعدادًا من متغيّرات البيئة أولًا (HF Spaces / .env)، ثم من
    st.secrets فقط إن وُجد ملف secrets.toml — لتفادي تحذير «No secrets found»."""
    val = os.getenv(key)
    if val:
        return val
    secret_paths = (
        "/root/.streamlit/secrets.toml",
        "/app/.streamlit/secrets.toml",
        os.path.expanduser("~/.streamlit/secrets.toml"),
        ".streamlit/secrets.toml",
    )
    if any(os.path.exists(p) for p in secret_paths):
        try:
            return st.secrets[key]
        except Exception:
            return None
    return None


def is_configured() -> bool:
    """هل MONGODB_URI مضبوط؟ (التطبيق يتطلّبه للتخزين)."""
    return _setting("MONGODB_URI") is not None


@st.cache_resource(show_spinner=False)
def _get_db():
    """يفتح الاتصال مرة واحدة، وينشئ فهرس الإيميل الفريد.
    tlsCAFile يُستخدم تلقائيًا لروابط Atlas (mongodb+srv) ويُتجاهل محليًا."""
    from pymongo import MongoClient

    uri = _setting("MONGODB_URI") or ""
    # شهادات TLS لازمة لروابط Atlas (mongodb+srv/ssl)، ومضرّة محليًا (localhost)
    low = uri.lower()
    use_tls = ("mongodb+srv" in low) or ("tls=true" in low) or ("ssl=true" in low)
    kwargs = {"tlsCAFile": certifi.where()} if use_tls else {}
    client = MongoClient(uri, **kwargs)
    try:
        db = client.get_default_database()   # اسم القاعدة من الرابط إن وُجد
    except Exception:
        db = None
    if db is None:                           # الرابط بلا اسم قاعدة → الافتراضي
        db = client[DEFAULT_DB_NAME]
    db["evaluators"].create_index("email", unique=True)
    return db


def backend_label() -> str:
    """وصف مكان التخزين (يُعرض في الشريط الجانبي)."""
    return "MongoDB"


# ------------------------------------------------------------
# 👤 المقيّمون
# ------------------------------------------------------------
def upsert_evaluator(name: str, email: str) -> None:
    """يُدرج المقيّم أو يحدّث اسمه (المفتاح: الإيميل)."""
    _get_db()["evaluators"].update_one(
        {"email": email},
        {"$set": {"name": name},
         "$setOnInsert": {"email": email, "created_at": datetime.now()}},
        upsert=True,
    )


# ------------------------------------------------------------
# 💾 حفظ تقييم استعلام كامل (كل النماذج وكلماتها دفعة واحدة)
# ------------------------------------------------------------
def save_evaluation(query: str, per_model: list[dict],
                    evaluator_name: str, evaluator_email: str,
                    word_type: str = "") -> int:
    """per_model = [{"model": str, "items": [{"lemma","score","chosen"}, ...]}, ...].
    word_type = تصنيف الكلمة المختار في الواجهة (فعل / معنى معجمي متقدم / معنى عام).
    يرجّع عدد الوثائق المحفوظة."""
    ts = datetime.now()   # وقت واحد لكل وثائق هذا الحفظ (لتماسك التجميع)
    docs = []
    for block in per_model:
        for item in block["items"]:
            docs.append({
                "created_at": ts,
                "query": query,
                "word_type": word_type,
                "model": block["model"],
                "word": item["lemma"],
                "similarity": round(float(item["score"]), 3),
                "verdict": "صح" if item["chosen"] else "خطأ",
                "evaluator_name": evaluator_name,
                "evaluator_email": evaluator_email,
            })
    if not docs:
        return 0
    _get_db()["ratings"].insert_many(docs)
    return len(docs)


# ------------------------------------------------------------
# 📖 قراءة التقييمات (يمكن قصرها على مقيّم) — ترجع DataFrame بأعمدة عربية
# ------------------------------------------------------------
def load_ratings(evaluator_email: str | None = None) -> pd.DataFrame:
    flt = {"evaluator_email": evaluator_email} if evaluator_email else {}
    rows = list(_get_db()["ratings"].find(flt).sort("created_at", 1))
    if not rows:
        return pd.DataFrame()
    df = pd.DataFrame(rows).drop(columns=["_id"], errors="ignore")
    return df.rename(columns=_AR_COLUMNS)


# ------------------------------------------------------------
# 📊 ملخّص أداء النماذج (نسبة الصح) — يمكن قصره على مقيّم
# ------------------------------------------------------------
def get_summary(evaluator_email: str | None = None) -> pd.DataFrame:
    df = load_ratings(evaluator_email)
    if df.empty:
        return pd.DataFrame()
    summary = df.groupby("النموذج")["التقييم"].value_counts().unstack(fill_value=0)
    for col in ["صح", "خطأ"]:
        if col not in summary.columns:
            summary[col] = 0
    summary["الإجمالي"] = summary["صح"] + summary["خطأ"]
    summary["نسبة الصح"] = (summary["صح"] / summary["الإجمالي"] * 100).round(1).astype(str) + "%"
    return summary[["صح", "خطأ", "الإجمالي", "نسبة الصح"]]


# ------------------------------------------------------------
# 🗑️ حذف (مقصور على إيميل المقيّم الحالي للأمان)
# ------------------------------------------------------------
def delete_query(created_at, query: str, evaluator_email: str) -> int:
    """يحذف تقييم استعلام واحد لهذا المقيّم."""
    res = _get_db()["ratings"].delete_many({
        "created_at": created_at,
        "query": query,
        "evaluator_email": evaluator_email,
    })
    return res.deleted_count


def delete_all_for(evaluator_email: str) -> int:
    """يحذف كل تقييمات المقيّم الحالي فقط."""
    res = _get_db()["ratings"].delete_many({"evaluator_email": evaluator_email})
    return res.deleted_count