Spaces:
Sleeping
Sleeping
| import os, time, html | |
| import requests | |
| import gradio as gr | |
| # ---------- Config (set in HF Space → Settings → Secrets) ---------- | |
| # If you don't set NGROK_URL in Secrets, we'll use your current tunnel as a fallback. | |
| NGROK_URL = os.environ.get("NGROK_URL", "https://bd8de44dca7e.ngrok-free.app/answer").strip() | |
| BACKEND_TOKEN = os.environ.get("BACKEND_TOKEN", "").strip() # optional: Bearer token | |
| REQUEST_TIMEOUT = float(os.environ.get("REQUEST_TIMEOUT", "30")) # seconds | |
| HEADERS = {"Content-Type": "application/json"} | |
| if BACKEND_TOKEN: | |
| HEADERS["Authorization"] = f"Bearer {BACKEND_TOKEN}" | |
| def _resolve_backend_url(raw: str) -> str: | |
| url = (raw or "").strip().rstrip("/") | |
| if not url: | |
| return "" | |
| if not url.endswith("/answer"): | |
| url = f"{url}/answer" | |
| return url | |
| def _endpoints(): | |
| """Return (answer_url, command_url) from NGROK_URL, tolerant if it already has /answer.""" | |
| answer_url = _resolve_backend_url(NGROK_URL) | |
| base = answer_url.rsplit("/", 1)[0] if answer_url.endswith("/answer") else answer_url | |
| command_url = f"{base}/command" | |
| return answer_url, command_url | |
| def _normalize_matches(matches): | |
| out = [] | |
| if isinstance(matches, list): | |
| for m in matches: | |
| if isinstance(m, (list, tuple)) and len(m) >= 2: | |
| q = str(m[0]); d = float(m[1]) | |
| elif isinstance(m, dict): | |
| q = str(m.get("question") or m.get("q") or m.get("text") or "") | |
| d = float(m.get("distance", m.get("score", 0.0))) | |
| else: | |
| q = str(m); d = 0.0 | |
| out.append((q, d)) | |
| return out | |
| def chat_handler(message, history, threshold): | |
| answer_url, command_url = _endpoints() | |
| if not answer_url: | |
| return "⚠️ يجب ضبط المتغير السري `NGROK_URL`." | |
| try: | |
| # Slash command → send to /command | |
| if message.strip().startswith("/"): | |
| r = requests.post(command_url, json={"cmd": message}, headers=HEADERS, timeout=REQUEST_TIMEOUT) | |
| r.raise_for_status() | |
| data = r.json() if r.content else {} | |
| return data.get("output", "—") | |
| # Normal question → send to /answer | |
| payload = {"question": message, "threshold": threshold} | |
| t0 = time.time() | |
| r = requests.post(answer_url, json=payload, headers=HEADERS, timeout=REQUEST_TIMEOUT) | |
| r.raise_for_status() | |
| data = r.json() if r.content else {} | |
| answer = data.get("answer") or data.get("message") or "لا توجد إجابة." | |
| # matches = _normalize_matches(data.get("matches", [])) | |
| latency = time.time() - t0 | |
| # lines = "\n".join([f"{i+1}. \"{q}\" (distance: {d:.4f})" for i, (q, d) in enumerate(matches[:5])]) | |
| # ... after we parse `answer` and compute `latency` | |
| md = ( | |
| f"**الإجابة**\n\n{answer}\n\n" | |
| f"<sub>زمن الاستجابة: {latency:.2f}s</sub>" | |
| ) | |
| return md | |
| except Exception as e: | |
| return f"تعذر الاتصال بالخادم: {e}" | |
| demo = gr.ChatInterface( | |
| fn=chat_handler, | |
| title="نظام الأسئلة والإجابات لجهاز الخدمة المدنية", | |
| description="اكتب سؤالك أو استخدم أوامر مثل: `/datasets`, `/use <name>`, `/train`, `/threshold 0.55`, `/probs : <سؤال>`", | |
| additional_inputs=[gr.Slider(minimum=0.0, maximum=2.0, step=0.1, value=0.6, label="عتبة المسافة")], | |
| ) | |
| demo.queue(default_concurrency_limit=8, max_size=32) | |
| if __name__ == "__main__": | |
| demo.launch(server_name="0.0.0.0", server_port=int(os.environ.get("PORT", "7860"))) | |