llm_client.py

"""Mode B: ask a language model a question, then verify its answer.

Supported providers (the user supplies their own API key; nothing is stored by this module):

* ``gemini``       Google Gemini ``generateContent`` REST endpoint
* ``openai``       OpenAI chat completions
* ``huggingface``  Hugging Face Inference Providers (OpenAI-compatible router)

Request building and response parsing are pure functions so they can be tested without network access. The same
request shapes are implemented in JavaScript in the static page (``build_static_space.py``).
"""
from __future__ import annotations

import json
import urllib.error
import urllib.request
from dataclasses import dataclass
from typing import Dict, Optional, Tuple

SYSTEM_PROMPT = (
    "أنت مساعد معرفي في العلوم الإسلامية. أجب بالعربية بإيجاز ودقة. عند الاستشهاد بآية قرآنية أو حديث نبوي اكتب نصه كاملًا "
    "بين علامتي تنصيص مزدوجتين \"...\" بعد عبارة تمهيدية مثل: قال الله تعالى: أو قال رسول الله ﷺ:. "
    "لا تضع بين علامات التنصيص إلا نص الآية أو الحديث، واذكر السورة ورقم الآية أو مصدر الحديث بعد الاقتباس."
)

PROVIDERS: Dict[str, Dict[str, str]] = {
    "gemini": {"label": "Google Gemini", "model": "gemini-2.5-flash",
               "base_url": "https://generativelanguage.googleapis.com/v1beta"},
    "openai": {"label": "OpenAI", "model": "gpt-4o-mini", "base_url": "https://api.openai.com/v1"},
    "huggingface": {"label": "Hugging Face", "model": "Qwen/Qwen2.5-72B-Instruct", "base_url": "https://router.huggingface.co/v1"},
}


class LLMError(RuntimeError):
    """Raised with a user-presentable Arabic message."""


@dataclass
class LLMSettings:
    provider: str
    api_key: str
    model: str = ""
    base_url: str = ""          # override for tests or compatible gateways
    timeout: float = 60.0


def build_request(settings: LLMSettings, prompt: str) -> Tuple[str, Dict[str, str], dict]:
    """``(url, headers, json_body)`` for the chosen provider."""
    if settings.provider not in PROVIDERS:
        raise LLMError("مزوّد النموذج غير مدعوم.")
    info = PROVIDERS[settings.provider]
    base = (settings.base_url or info["base_url"]).rstrip("/")
    model = settings.model.strip() or info["model"]
    if settings.provider == "gemini":
        url = f"{base}/models/{model}:generateContent"
        headers = {"Content-Type": "application/json", "x-goog-api-key": settings.api_key}
        body = {"systemInstruction": {"parts": [{"text": SYSTEM_PROMPT}]}, "contents": [{"role": "user", "parts": [{"text": prompt}]}]}
        return url, headers, body
    url = f"{base}/chat/completions"
    headers = {"Content-Type": "application/json", "Authorization": f"Bearer {settings.api_key}"}
    body = {"model": model, "messages": [{"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": prompt}]}
    return url, headers, body


def parse_response(provider: str, payload: dict) -> str:
    try:
        if provider == "gemini":
            text = "".join(part.get("text", "") for part in payload["candidates"][0]["content"]["parts"])
        else:
            text = payload["choices"][0]["message"]["content"]
    except (KeyError, IndexError, TypeError) as exc:
        raise LLMError("وصلت استجابة غير متوقعة من النموذج.") from exc
    if not text or not text.strip():
        raise LLMError("لم يُرجع النموذج أي نص.")
    return text.strip()


_HTTP_MESSAGES = {
    400: "رفض المزوّد الطلب؛ تحقق من اسم النموذج.",
    401: "مفتاح الواجهة البرمجية غير صالح.",
    403: "المفتاح غير مصرّح له باستخدام هذا النموذج.",
    404: "النموذج غير موجود عند المزوّد؛ تحقق من اسمه.",
    429: "تجاوزت حد الاستخدام المسموح؛ حاول لاحقًا.",
}


def generate(settings: LLMSettings, prompt: str) -> str:
    """Return the model's answer. Raises ``LLMError`` with an Arabic message on any failure."""
    if not prompt or not prompt.strip():
        raise LLMError("اكتب سؤالًا أولًا.")
    if not settings.api_key or not settings.api_key.strip():
        raise LLMError("أدخل مفتاح الواجهة البرمجية للمزوّد.")
    url, headers, body = build_request(settings, prompt.strip())
    request = urllib.request.Request(url, data=json.dumps(body).encode("utf-8"), headers=headers, method="POST")
    try:
        with urllib.request.urlopen(request, timeout=settings.timeout) as response:
            payload = json.loads(response.read().decode("utf-8"))
    except urllib.error.HTTPError as exc:
        raise LLMError(_HTTP_MESSAGES.get(exc.code, f"فشل الطلب (الرمز {exc.code}).")) from exc
    except (urllib.error.URLError, TimeoutError) as exc:
        raise LLMError("تعذّر الاتصال بالمزوّد؛ تحقق من الإنترنت.") from exc
    except json.JSONDecodeError as exc:
        raise LLMError("وصلت استجابة غير مقروءة من المزوّد.") from exc
    return parse_response(settings.provider, payload)