Update app.py

app.py

@@ -1,523 +1,352 @@
-import os
+from fastapi import FastAPI, Request, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import StreamingResponse, JSONResponse
+from pydantic import BaseModel
+from typing import List, Optional
+import asyncio
 import json
 import time
 import uuid
-import asyncio
 import logging
-import uvicorn
-import random
-from fastapi import FastAPI, HTTPException, Request, Response
-from fastapi.middleware.cors import CORSMiddleware
-from concurrent.futures import ThreadPoolExecutor
-from curl_cffi import requests as curl_requests
-from typing import Dict, List, Optional
-
-# =========================================================
-# 1. الإعدادات
-# =========================================================
-API_KEY          = os.environ.get("API_KEY", "sk-your-secret-key")
-PORT             = int(os.environ.get("PORT", 7860))
-MAX_WORKERS      = 50
-REFRESH_INTERVAL = 300          # ثواني بين دورات الاستكشاف
-CACHE_FILE       = "models_cache.json"
-
-# نماذج غير نصية — يُستبعد استيرادها
-MODEL_BLACKLIST = [
-    "embed", "embedding",
-    "image", "vision",
-    "audio", "tts", "whisper",
-    "moderation",
-    "dall-e", "stable-diffusion", "midjourney",
-]
-
-# ===================== الحالة العامة =====================
-WORKING_MODELS: Dict[str, dict] = {}       # { model_id: {providers, latency, health} }
-PROVIDER_MODEL_MAP: Dict[str, List[str]] = {}  # { chat_url: [model_id, ...] }
-
-STATE_LOCK      = asyncio.Lock()
-REQUEST_LIMITER = asyncio.Semaphore(30)
-EXECUTOR        = ThreadPoolExecutor(max_workers=MAX_WORKERS)
-
-app = FastAPI(title="Omega Orchestrator", version="3.0")
+
+import g4f
+from g4f.client import Client
+
+# =====================================================
+# LOGGING
+# =====================================================
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# =====================================================
+# CONFIG
+# =====================================================
+
+API_KEY = "sk-your-secret-key"
+
+# =====================================================
+# FASTAPI
+# =====================================================
+
+app = FastAPI(
+    title="Universal AI Gateway",
+    version="4.0.0"
+)
+
+# =====================================================
+# CORS
+# =====================================================
+
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],
+    allow_credentials=True,
     allow_methods=["*"],
     allow_headers=["*"],
 )
 
-logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
-logger = logging.getLogger("OMEGA")
+# =====================================================
+# MODELS
+# =====================================================
 
+class Message(BaseModel):
+    role: str
+    content: str
 
-# =========================================================
-# 2. مساعدات
-# =========================================================
-def stealth_headers() -> dict:
-    return {
-        "User-Agent": (
-            f"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
-            f"AppleWebKit/537.36 (KHTML, like Gecko) "
-            f"Chrome/{random.randint(120, 126)}.0.0.0 Safari/537.36"
-        ),
-        "Origin":           "https://g4f.space",
-        "Referer":          "https://g4f.space/",
-        "X-Requested-With": "XMLHttpRequest",
-        "Accept":           "application/json",
-        "Content-Type":     "application/json",
-    }
+class ChatRequest(BaseModel):
+    model: str
+    messages: List[Message]
+    stream: bool = False
+    temperature: Optional[float] = 0.7
+    max_tokens: Optional[int] = 4096
 
+# =====================================================
+# AUTH
+# =====================================================
 
-def is_blacklisted(model_id: str) -> bool:
-    low = model_id.lower()
-    return any(b in low for b in MODEL_BLACKLIST)
+def verify_api_key(req: Request):
 
+    auth = req.headers.get("Authorization")
 
-def verify_api_key(request: Request) -> bool:
-    """
-    يقبل المفتاح من ثلاث طرق:
-      - Authorization: Bearer <key>
-      - x-api-key: <key>
-      - api-key: <key>
-    وصول مفتوح إذا لم يُضبط API_KEY.
-    """
-    if not API_KEY or API_KEY == "sk-your-secret-key":
+    # السماح للاختبار
+    if not auth:
         return True
 
-    candidates: List[str] = []
-
-    auth = request.headers.get("Authorization", "")
-    if auth.startswith("Bearer "):
-        candidates.append(auth[7:].strip())
-
-    for h in ("x-api-key", "api-key"):
-        v = request.headers.get(h, "").strip()
-        if v:
-            candidates.append(v)
-
-    return any(c == API_KEY for c in candidates)
-
-
-# =========================================================
-# 3. استخراج المحتوى من أي رد JSON
-# =========================================================
-def extract_content(data) -> Optional[str]:
-    if not isinstance(data, dict):
-        return None
-
-    for f in ("response", "content", "text", "output", "result", "generated_text", "completion"):
-        v = data.get(f)
-        if isinstance(v, str) and v.strip():
-            return v.strip()
-
-    choices = data.get("choices")
-    if isinstance(choices, list) and choices:
-        c = choices[0]
-        if isinstance(c, dict):
-            msg = c.get("message", {})
-            if isinstance(msg, dict):
-                t = msg.get("content")
-                if isinstance(t, str) and t.strip():
-                    return t.strip()
-            txt = c.get("text", "")
-            if isinstance(txt, str) and txt.strip():
-                return txt.strip()
-            delta = c.get("delta", {})
-            if isinstance(delta, dict):
-                t = delta.get("content", "")
-                if isinstance(t, str) and t.strip():
-                    return t.strip()
-
-    msg = data.get("message")
-    if isinstance(msg, dict):
-        c = msg.get("content")
-        if isinstance(c, list):
-            parts = [b.get("text", "") for b in c if isinstance(b, dict) and b.get("type") == "text"]
-            joined = "".join(parts).strip()
-            if joined:
-                return joined
-        elif isinstance(c, str) and c.strip():
-            return c.strip()
-    elif isinstance(msg, str) and msg.strip():
-        return msg.strip()
-
-    inner = data.get("data")
-    if isinstance(inner, dict):
-        return extract_content(inner)
-
-    return None
-
-
-# =========================================================
-# 4. كلاس المزوّد الأساسي
-# =========================================================
-class BaseProvider:
-    def __init__(self, name: str, url_chat: str, url_models: str = ""):
-        self.name       = name
-        self.url_chat   = url_chat
-        self.url_models = url_models or url_chat
-        self.headers    = stealth_headers()
-        self.aliases: Dict[str, str] = {}
-        self.fails      = 0
-        self.success    = 0
-        self.cooldown   = 0.0
-        self.latency    = 0.0
-        self.health     = 100
-
-    def _record_success(self, latency: float):
-        self.success += 1
-        self.latency = (self.latency + latency) / 2 if self.latency else latency
-        self._calc_health()
-
-    def _record_fail(self):
-        self.fails += 1
-        self._calc_health()
-        if self.fails >= 3:
-            self.cooldown = time.time() + 60
-
-    def _calc_health(self):
-        total = self.success + self.fails
-        self.health = int(self.success / total * 100) if total else 100
-
-    # ── جلب النماذج — بدون تحقق، مباشرة من المزوّد ──────
-    async def fetch_models(self) -> List[str]:
-        loop = asyncio.get_event_loop()
-        try:
-            async with REQUEST_LIMITER:
-                raw = await loop.run_in_executor(EXECUTOR, self._fetch_models_sync)
-            return [m for m in set(raw) if m and not is_blacklisted(m)]
-        except Exception as e:
-            logger.warning(f"[{self.name}] fetch_models error: {e}")
-            return []
-
-    def _fetch_models_sync(self) -> List[str]:
-        # نقاط النهاية المحتملة بالترتيب
-        base = self.url_models.rstrip("/")
-        chat = self.url_chat.rstrip("/")
-        endpoints = list(dict.fromkeys([
-            base,
-            base + "/v1/models",
-            base + "/models",
-            chat + "/v1/models",
-            chat + "/models",
-        ]))
-
-        with curl_requests.Session() as session:
-            for ep in endpoints:
-                try:
-                    r = session.get(ep, headers=self.headers, impersonate="chrome124", timeout=12)
-                    if r.status_code != 200:
-                        continue
-                    ids = self._parse_models(r.json())
-                    if ids:
-                        logger.info(f"[{self.name}] {len(ids)} models ← {ep}")
-                        return ids
-                except Exception:
-                    continue
-        return []
-
-    @staticmethod
-    def _parse_models(data) -> List[str]:
-        """يستخرج معرّفات النماذج من أي هيكل JSON ممكن."""
-        ids: List[str] = []
-
-        if isinstance(data, list):
-            for item in data:
-                if isinstance(item, str) and item.strip():
-                    ids.append(item.strip())
-                elif isinstance(item, dict):
-                    for k in ("id", "name", "model", "model_id", "modelId"):
-                        v = item.get(k)
-                        if isinstance(v, str) and v.strip():
-                            ids.append(v.strip())
-                            break
-
-        elif isinstance(data, dict):
-            for key in ("data", "models", "result", "results", "items"):
-                sub = data.get(key)
-                if isinstance(sub, list):
-                    found = BaseProvider._parse_models(sub)
-                    if found:
-                        return found
-            for k in ("id", "name", "model"):
-                v = data.get(k)
-                if isinstance(v, str) and v.strip():
-                    ids.append(v.strip())
-
-        return ids
-
-    # ── إرسال الطلب ──────────────────────────────────────
-    async def attempt_request(self, payload: dict) -> Optional[str]:
-        payload = payload.copy()
-
-        if time.time() < self.cooldown:
-            return None
-
-        model_id = payload.get("model", "")
-        payload["model"] = self.aliases.get(model_id, model_id)
+    if not auth.startswith("Bearer "):
+        raise HTTPException(
+            status_code=401,
+            detail="Invalid Authorization Format"
+        )
 
-        try:
-            async with REQUEST_LIMITER:
-                loop = asyncio.get_event_loop()
-                t0   = time.time()
-                text = await loop.run_in_executor(EXECUTOR, self._post_sync, payload)
-                lat  = time.time() - t0
-
-            if text:
-                self._record_success(lat)
-                return text
-
-            self._record_fail()
-            return None
-        except Exception:
-            self._record_fail()
-            return None
-
-    def _post_sync(self, payload: dict) -> Optional[str]:
-        with curl_requests.Session() as session:
-            try:
-                r = session.post(
-                    self.url_chat,
-                    headers=self.headers,
-                    json=payload,
-                    impersonate="chrome124",
-                    timeout=30,
-                )
-                if r.status_code == 200:
-                    content = extract_content(r.json())
-                    if content:
-                        return content
-            except Exception:
-                pass
-        return None
-
-
-# =========================================================
-# 5. تعريف المزوّدين
-#    أضف أي مزوّد جديد هنا بنفس الطريقة
-# =========================================================
-PROVIDER_INSTANCES: List[BaseProvider] = [
-    BaseProvider("G4F-Groq",        "https://g4f.space/api/groq"),
-    BaseProvider("G4F-Gemini",      "https://g4f.space/api/gemini"),
-    BaseProvider("G4F-Pollinations","https://g4f.space/api/pollinations"),
-    BaseProvider("G4F-Ollama",      "https://g4f.space/api/ollama"),
-    BaseProvider("G4F-Perplexity",  "https://g4f.space/api/perplexity"),
-    BaseProvider("G4F-OpenAI",      "https://g4f.space/api/openai"),
-    BaseProvider("G4F-DeepSeek",    "https://g4f.space/api/deepseek"),
-    BaseProvider("G4F-Mistral",     "https://g4f.space/api/mistral"),
-    BaseProvider("G4F-Claude",      "https://g4f.space/api/claude"),
-    BaseProvider("G4F-Meta",        "https://g4f.space/api/meta"),
-]
-
-
-# =========================================================
-# 6. التخزين المؤقت
-# =========================================================
-async def load_cache():
-    global WORKING_MODELS, PROVIDER_MODEL_MAP
-    try:
-        if os.path.exists(CACHE_FILE):
-            with open(CACHE_FILE, "r", encoding="utf-8") as f:
-                d = json.load(f)
-            async with STATE_LOCK:
-                WORKING_MODELS     = d.get("WORKING_MODELS", {})
-                PROVIDER_MODEL_MAP = d.get("PROVIDER_MODEL_MAP", {})
-            logger.info(f"✅ Cache loaded — {len(WORKING_MODELS)} models.")
-    except Exception as e:
-        logger.error(f"Cache load error: {e}")
+    token = auth.replace("Bearer ", "").strip()
 
+    if token != API_KEY:
+        raise HTTPException(
+            status_code=403,
+            detail="Invalid API Key"
+        )
 
-async def save_cache():
-    try:
-        async with STATE_LOCK:
-            snap = {
-                "WORKING_MODELS":    dict(WORKING_MODELS),
-                "PROVIDER_MODEL_MAP": dict(PROVIDER_MODEL_MAP),
-            }
-        with open(CACHE_FILE, "w", encoding="utf-8") as f:
-            json.dump(snap, f, indent=2, ensure_ascii=False)
-    except Exception as e:
-        logger.error(f"Cache save error: {e}")
+    return True
 
+# =====================================================
+# ROOT
+# =====================================================
 
-# =========================================================
-# 7. محرك الاستكشاف — استيراد مباشر بلا تحقق مسبق
-# =========================================================
-async def discovery_engine():
-    """
-    يجلب النماذج مباشرةً من كل مزوّد في نفس الوقت
-    ويضيفها للسجل فوراً — بدون خطوة تحقق تُعيق الاستيراد.
-    الفلترة الوحيدة: استبعاد القائمة السوداء (صور/صوت/تضمين).
-    """
-    await load_cache()
+@app.get("/")
+async def root():
 
-    while True:
-        logger.info("🔍 Discovery cycle …")
+    return {
+        "status": "online",
+        "service": "Universal AI Gateway",
+        "version": "4.0.0"
+    }
 
-        fresh_models: Dict[str, dict]      = {}
-        fresh_map:    Dict[str, List[str]] = {}
+# =====================================================
+# MODELS
+# =====================================================
 
-        # جلب متوازٍ من جميع المزوّدين
-        tasks = {p: asyncio.create_task(p.fetch_models()) for p in PROVIDER_INSTANCES}
+@app.get("/v1/models")
+async def get_models():
 
-        for provider, task in tasks.items():
-            try:
-                models = await task
-            except Exception as e:
-                logger.warning(f"[{provider.name}] task error: {e}")
-                models = []
+    models_data = []
 
-            fresh_map[provider.url_chat] = models
+    fallback_models = [
+        "gpt-4o-mini",
+        "gpt-4o",
+        "gpt-4",
+        "gpt-3.5-turbo",
+        "claude-3-haiku",
+        "llama-3.1-70b",
+        "mixtral-8x7b",
+        "deepseek-chat",
+        "gemini-pro"
+    ]
 
-            for m in models:
-                if m not in fresh_models:
-                    fresh_models[m] = {
-                        "providers": [],
-                        "latency":   0.0,
-                        "health":    provider.health,
-                    }
-                if provider.url_chat not in fresh_models[m]["providers"]:
-                    fresh_models[m]["providers"].append(provider.url_chat)
+    try:
 
-            logger.info(f"[{provider.name}] → {len(models)} models imported.")
+        if hasattr(g4f.models, "_all_models"):
 
-        # نشر ذري
-        async with STATE_LOCK:
-            global WORKING_MODELS, PROVIDER_MODEL_MAP
-            WORKING_MODELS     = fresh_models
-            PROVIDER_MODEL_MAP = fresh_map
+            all_models = list(g4f.models._all_models)
 
-        await save_cache()
-        logger.info(f"🚀 Ready — {len(WORKING_MODELS)} total models.")
-        await asyncio.sleep(REFRESH_INTERVAL)
+            for model in all_models[:50]:
 
+                models_data.append({
+                    "id": str(model),
+                    "object": "model",
+                    "created": int(time.time()),
+                    "owned_by": "g4f"
+                })
 
-# =========================================================
-# 8. المعالج المركزي
-# =========================================================
-@app.on_event("startup")
-async def on_startup():
-    asyncio.create_task(discovery_engine())
+    except Exception as e:
 
+        logger.error(f"Models error: {e}")
 
-@app.api_route("/{path:path}", methods=["GET", "HEAD", "POST", "OPTIONS"])
-async def omega_handler(request: Request, path: str):
+    # fallback
+    if not models_data:
 
-    # CORS preflight
-    if request.method == "OPTIONS":
-        return Response(
-            status_code=204,
-            headers={
-                "Access-Control-Allow-Origin":  "*",
-                "Access-Control-Allow-Methods": "GET, POST, OPTIONS",
-                "Access-Control-Allow-Headers": "*",
-            },
-        )
+        for model in fallback_models:
 
-    p = path.lower().strip("/")
-
-    # ── قائمة النماذج ─────────────────────────────────────
-    if request.method in ("GET", "HEAD"):
-        if "models" in p or p in ("", "v1", "v1/"):
-            if "models" in p:
-                async with STATE_LOCK:
-                    ids = sorted(WORKING_MODELS.keys())
-                return {
-                    "object": "list",
-                    "data": [
-                        {"id": m, "object": "model", "created": int(time.time()), "owned_by": "omega"}
-                        for m in ids
-                    ],
-                }
-            return Response(status_code=200)
+            models_data.append({
+                "id": model,
+                "object": "model",
+                "created": int(time.time()),
+                "owned_by": "g4f"
+            })
 
-    # ── إرسال رسائل ───────────────────────────────────────
-    if request.method == "POST" and any(x in p for x in ("messages", "completions", "chat")):
+    return {
+        "object": "list",
+        "data": models_data
+    }
 
-        if not verify_api_key(request):
-            raise HTTPException(status_code=401, detail="Unauthorized.")
+# =====================================================
+# CHAT COMPLETIONS
+# =====================================================
 
-        try:
-            body = await request.json()
-        except Exception:
-            raise HTTPException(status_code=400, detail="Invalid JSON.")
+@app.post("/v1/chat/completions")
+async def chat_completions(
+    req: Request,
+    body: ChatRequest
+):
 
-        model    = body.get("model", "")
-        messages = body.get("messages", [])
+    verify_api_key(req)
 
-        if not messages:
-            raise HTTPException(status_code=400, detail="messages is required.")
+    messages = [
+        {
+            "role": m.role,
+            "content": m.content
+        }
+        for m in body.messages
+    ]
 
-        # اختيار المزوّدين القادرين على تقديم هذا النموذج
-        async with STATE_LOCK:
-            info = WORKING_MODELS.get(model)
-            target_urls = list(info["providers"]) if info and info["providers"] else [p.url_chat for p in PROVIDER_INSTANCES]
+    logger.info(
+        f"Request model={body.model} stream={body.stream}"
+    )
 
-        providers = [p for p in PROVIDER_INSTANCES if p.url_chat in target_urls]
-        providers.sort(key=lambda p: (p.fails, -p.health, p.latency))
+    # =================================================
+    # STREAMING
+    # =================================================
 
-        reply: Optional[str] = None
+    if body.stream:
 
-        for provider in providers:
-            if time.time() < provider.cooldown:
-                continue
-            reply = await provider.attempt_request(body)
-            if reply:
-                logger.info(f"✅ [{provider.name}] served model={model}")
-                break
+        async def generate_stream():
 
-        # Fallback عبر g4f
-        if not reply:
             try:
-                from g4f.client import Client
-                loop = asyncio.get_event_loop()
-                reply = await loop.run_in_executor(
-                    EXECUTOR,
-                    lambda: Client()
-                        .chat.completions.create(model=model, messages=messages)
-                        .choices[0].message.content,
+
+                client = Client()
+
+                response = client.chat.completions.create(
+                    model=body.model,
+                    messages=messages,
+                    stream=True
                 )
-                logger.info("🔄 g4f fallback used.")
+
+                chunk_id = f"chatcmpl-{uuid.uuid4().hex}"
+
+                for chunk in response:
+
+                    try:
+
+                        content = ""
+
+                        if (
+                            chunk.choices
+                            and chunk.choices[0].delta
+                            and chunk.choices[0].delta.content
+                        ):
+                            content = chunk.choices[0].delta.content
+
+                        if content:
+
+                            payload = {
+                                "id": chunk_id,
+                                "object": "chat.completion.chunk",
+                                "created": int(time.time()),
+                                "model": body.model,
+                                "choices": [
+                                    {
+                                        "index": 0,
+                                        "delta": {
+                                            "content": content
+                                        },
+                                        "finish_reason": None
+                                    }
+                                ]
+                            }
+
+                            yield f"data: {json.dumps(payload)}\n\n"
+
+                            await asyncio.sleep(0)
+
+                    except Exception as chunk_error:
+
+                        logger.error(
+                            f"Chunk error: {chunk_error}"
+                        )
+
+                final_payload = {
+                    "id": chunk_id,
+                    "object": "chat.completion.chunk",
+                    "created": int(time.time()),
+                    "model": body.model,
+                    "choices": [
+                        {
+                            "index": 0,
+                            "delta": {},
+                            "finish_reason": "stop"
+                        }
+                    ]
+                }
+
+                yield f"data: {json.dumps(final_payload)}\n\n"
+
+                yield "data: [DONE]\n\n"
+
             except Exception as e:
-                logger.warning(f"g4f fallback failed: {e}")
-
-        if not reply:
-            raise HTTPException(status_code=502, detail="All routes exhausted.")
-
-        # ─ تنسيق الرد ─
-        if "messages" in p and "chat" not in p:
-            # Anthropic format
-            return {
-                "id":          f"msg_{uuid.uuid4().hex}",
-                "type":        "message",
-                "role":        "assistant",
-                "model":       model,
-                "content":     [{"type": "text", "text": reply}],
-                "stop_reason": "end_turn",
-                "usage":       {"input_tokens": 0, "output_tokens": 0},
+
+                logger.error(f"Streaming error: {e}")
+
+                error_payload = {
+                    "error": {
+                        "message": str(e),
+                        "type": "server_error"
+                    }
+                }
+
+                yield f"data: {json.dumps(error_payload)}\n\n"
+
+        return StreamingResponse(
+            generate_stream(),
+            media_type="text/event-stream",
+            headers={
+                "Cache-Control": "no-cache",
+                "Connection": "keep-alive",
+                "X-Accel-Buffering": "no"
             }
+        )
+
+    # =================================================
+    # NORMAL RESPONSE
+    # =================================================
+
+    try:
+
+        client = Client()
+
+        response = await asyncio.to_thread(
+            client.chat.completions.create,
+            model=body.model,
+            messages=messages
+        )
+
+        assistant_message = ""
+
+        try:
+            assistant_message = response.choices[0].message.content
+        except:
+            assistant_message = str(response)
 
-        # OpenAI format
-        return {
-            "id":      f"chatcmpl-{uuid.uuid4().hex}",
-            "object":  "chat.completion",
+        return JSONResponse({
+            "id": f"chatcmpl-{uuid.uuid4().hex}",
+            "object": "chat.completion",
             "created": int(time.time()),
-            "model":   model,
+            "model": body.model,
             "choices": [
                 {
-                    "index":         0,
-                    "message":       {"role": "assistant", "content": reply},
-                    "finish_reason": "stop",
+                    "index": 0,
+                    "message": {
+                        "role": "assistant",
+                        "content": assistant_message
+                    },
+                    "finish_reason": "stop"
                 }
             ],
-            "usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0},
-        }
+            "usage": {
+                "prompt_tokens": 0,
+                "completion_tokens": 0,
+                "total_tokens": 0
+            }
+        })
 
-    return Response(status_code=404)
+    except Exception as e:
 
+        logger.error(f"Chat error: {e}")
+
+        raise HTTPException(
+            status_code=500,
+            detail=str(e)
+        )
+
+# =====================================================
+# RUN
+# =====================================================
 
-# =========================================================
-# 9. نقطة الدخول
-# =========================================================
 if __name__ == "__main__":
-    uvicorn.run(app, host="0.0.0.0", port=PORT, log_level="info")
+
+    import uvicorn
+
+    uvicorn.run(
+        app,
+        host="0.0.0.0",
+        port=7860
+    )