app.py

"""مُدقِّق المحتوى الإسلامي: Gradio interface.

    python app.py                      # http://127.0.0.1:7860

Mode A verifies pasted text. Mode B asks a language model first and verifies its answer. API keys are taken from the
form or, preferably, from environment variables (GEMINI_API_KEY, OPENAI_API_KEY, HF_TOKEN) so that a deployment can
keep them as secrets.
"""
from __future__ import annotations

import json
import logging
import os
import threading
from pathlib import Path
from typing import List, Optional

import ui
from llm_client import PROVIDERS, LLMError, LLMSettings, generate
from verifier import MAX_INPUT_CHARS, IslamicContentVerifier

logger = logging.getLogger(__name__)

EXAMPLES_PATH = Path(__file__).resolve().parent / "demo" / "examples.json"
PROVIDER_LABELS = {"gemini": "جوجل جيميناي", "openai": "أوبن إيه آي", "huggingface": "هاغينغ فيس"}
KEY_ENV = {"gemini": "GEMINI_API_KEY", "openai": "OPENAI_API_KEY", "huggingface": "HF_TOKEN"}

_pipeline: Optional[IslamicContentVerifier] = None
_lock = threading.Lock()


def get_pipeline() -> IslamicContentVerifier:
    """Created once. The Quran index loads immediately; the Hadith index loads lazily (see ``warm_in_background``)."""
    global _pipeline
    with _lock:
        if _pipeline is None:
            _pipeline = IslamicContentVerifier()
        return _pipeline


def warm_in_background() -> None:
    threading.Thread(target=lambda: get_pipeline().retriever.warm(), daemon=True).start()


def load_examples(path: Path = EXAMPLES_PATH) -> List[dict]:
    try:
        with open(path, encoding="utf-8") as handle:
            return json.load(handle)
    except (OSError, json.JSONDecodeError):
        logger.exception("Could not load demo examples from %s", path)
        return []


def verify_text(text: str) -> str:
    """Mode A. Never raises: problems become Arabic notices."""
    if not text or not text.strip():
        return ui.render_message("الرجاء إدخال نص للتحقق منه.", "warn")
    try:
        return ui.render_results(get_pipeline().analyze(text))
    except ValueError:
        return ui.render_message(f"النص طويل جدًا (الحد الأقصى {MAX_INPUT_CHARS} حرف).", "warn")
    except Exception:
        logger.exception("Verification failed")
        return ui.render_message("حدث خطأ غير متوقع أثناء التحقق.", "bad")


def verify_generated_answer(answer: str) -> str:
    """Verify a model answer and show it above the report (also used by the in-browser page)."""
    try:
        return ui.render_results(get_pipeline().analyze(answer), generated_answer=answer)
    except Exception:
        logger.exception("Verification of the generated answer failed")
        return ui.render_message("تعذّر التحقق من إجابة النموذج.", "bad")


def ask_then_verify(prompt: str, provider: str, model: str, api_key: str) -> str:
    """Mode B: generate an answer with the chosen model, then verify every quotation in it."""
    key = (api_key or "").strip() or os.environ.get(KEY_ENV.get(provider, ""), "")
    try:
        answer = generate(LLMSettings(provider=provider, api_key=key, model=model or ""), prompt)
    except LLMError as exc:
        return ui.render_message(str(exc), "warn")
    return verify_generated_answer(answer)


def build_interface():
    import gradio as gr

    examples = load_examples()

    def next_example(index: int):
        if not examples:
            return "", 0
        return examples[index % len(examples)]["text"], (index + 1) % len(examples)

    def show_default_model(provider: str):
        return PROVIDERS[provider]["model"]

    with gr.Blocks(title="مُدقِّق المحتوى الإسلامي", css=ui.CSS, theme=gr.themes.Base(primary_hue="emerald", neutral_hue="stone")) as demo:
        gr.HTML(ui.HERO)
        with gr.Tabs():
            with gr.Tab("تحقق مباشر"):
                example_index = gr.State(0)
                text_input = gr.Textbox(label="النص المراد التحقق منه", lines=9, max_lines=24, placeholder=ui.PLACEHOLDER,
                                        rtl=True, elem_classes="input-area")
                with gr.Row():
                    verify_button = gr.Button("تحقّق من النص", variant="primary", scale=3)
                    example_button = gr.Button("جرّب مثالًا", variant="secondary", scale=2)
                results = gr.HTML(elem_classes="results")
                verify_button.click(verify_text, inputs=text_input, outputs=results)
                example_button.click(next_example, inputs=example_index, outputs=[text_input, example_index]).then(
                    verify_text, inputs=text_input, outputs=results)
            with gr.Tab("اسأل ثم تحقّق"):
                gr.HTML('<div class="icv"><div class="notice">اكتب سؤالًا، وسيجيب النموذج اللغوي الذي تختاره، ثم يفحص النظام كل آية '
                        'وحديث ورد في إجابته ويعرض الأخطاء والتصحيحات. مفتاحك يُستخدم لهذا الطلب فقط ولا يُحفظ.</div></div>')
                with gr.Row(elem_classes="llm-row"):
                    provider = gr.Dropdown(choices=[(PROVIDER_LABELS[k], k) for k in PROVIDERS], value="gemini", label="مزوّد النموذج", scale=2)
                    model = gr.Textbox(value=PROVIDERS["gemini"]["model"], label="اسم النموذج (قابل للتعديل)", scale=3)
                    api_key = gr.Textbox(label="مفتاح الواجهة البرمجية", type="password", scale=3,
                                         placeholder="اتركه فارغًا إن كان محفوظًا كمتغير بيئة")
                prompt = gr.Textbox(label="سؤالك", lines=3, placeholder=ui.PROMPT_PLACEHOLDER, rtl=True, elem_classes="input-area")
                ask_button = gr.Button("اسأل ثم تحقّق", variant="primary")
                answer_results = gr.HTML(elem_classes="results")
                provider.change(show_default_model, inputs=provider, outputs=model)
                ask_button.click(ask_then_verify, inputs=[prompt, provider, model, api_key], outputs=answer_results)
        gr.HTML(ui.DISCLAIMER)
    return demo


def main() -> None:
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
    get_pipeline()
    warm_in_background()
    build_interface().queue().launch(share=os.environ.get("ICV_SHARE") == "1")


if __name__ == "__main__":
    main()