Update app.py

app.py

@@ -1,498 +1,210 @@
 import time
 import random
 import traceback
-import asyncio
-import hashlib
-from typing import Optional, Dict, List, Any
-from dataclasses import dataclass, field
-from datetime import datetime
-from contextlib import asynccontextmanager
-
-from fastapi import FastAPI, BackgroundTasks, Request, HTTPException
-from fastapi.responses import JSONResponse
-from fastapi.middleware.cors import CORSMiddleware
+
+from fastapi import FastAPI
 from pydantic import BaseModel
 import uvicorn
 
 from g4f.client import Client
 from g4f.Provider import __providers__
 
-# إعدادات النظام العامة
 MODEL = "gpt-4o-mini"
-MAX_CONCURRENT_REQUESTS = 3      # الحد الأقصى للطلبات المتزامنة من الخادم
-RESPONSE_TIMEOUT = 35            # ثواني
-CACHE_MAX_SIZE = 100             # حجم الكاش
-CACHE_TTL = 300                  # الكاش يعيش 5 دقائق
-
-# بلاك ليست للمزودين المعروفين بأنهم يحتاجون auth أو يعطون مشاكل دائمة
-BLOCKED_KEYWORDS = [
-    "openai", "qwen", "copilot", "gemini", "claude",
-    "google", "meta", "tecnai", "liaobots", "deepinfra",
-    "sambanova", "moonshot", "theb", "huggingchat"
-]
-
-
-# ==================== نظام الإحصائيات والتتبع ====================
-@dataclass
-class ProviderStats:
-    """إحصائيات كل مزود لاتخاذ قرارات ذكية"""
-    name: str
-    success_count: int = 0
-    fail_count: int = 0
-    avg_response_time: float = 0.0
-    last_used: float = 0.0
-    consecutive_failures: int = 0
-    circuit_status: str = "closed"  # closed (عادي), open (مفصول), half-open (اختبار)
-    cooldown_until: float = 0.0
-    
-    @property
-    def total_requests(self) -> int:
-        return self.success_count + self.fail_count
-    
-    @property
-    def success_rate(self) -> float:
-        if self.total_requests == 0:
-            return 1.0
-        return self.success_count / self.total_requests
-    
-    @property
-    def weight(self) -> float:
-        """وزن ديناميكي: يفضل المزود الناجح السريع"""
-        if self.circuit_status == "open":
-            return 0.0
-        if self.circuit_status == "half-open":
-            return 0.5
-        
-        # وزن يعتمد على معدل النجاح والسرعة
-        speed_factor = max(0.1, 1.0 / (self.avg_response_time + 1))
-        return self.success_rate * speed_factor
-    
-    def record_success(self, response_time: float):
-        self.success_count += 1
-        self.consecutive_failures = 0
-        # حساب المتوسط المتحرك
-        if self.avg_response_time == 0:
-            self.avg_response_time = response_time
-        else:
-            self.avg_response_time = (self.avg_response_time * 0.7) + (response_time * 0.3)
-        self.last_used = time.time()
-        
-        # إعادة Circuit Breaker للعمل
-        if self.circuit_status == "half-open":
-            self.circuit_status = "closed"
-    
-    def record_failure(self):
-        self.fail_count += 1
-        self.consecutive_failures += 1
-        self.last_used = time.time()
-        
-        # Circuit Breaker Logic
-        if self.consecutive_failures >= 3:
-            self.circuit_status = "open"
-            self.cooldown_until = time.time() + 120  # 2 دقيقة حظر
-        elif self.consecutive_failures >= 1:
-            # حظر خفيف عند أول فشل
-            self.cooldown_until = time.time() + 30
-
-
-# ==================== الذاكرة المؤقتة (Cache) ====================
-class LRUCache:
-    """كاش محلي ذكي لتجنب طلب نفس السؤال مراراً"""
-    def __init__(self, maxsize: int = 100, ttl: int = 300):
-        self.maxsize = maxsize
-        self.ttl = ttl
-        self.cache: Dict[str, Dict[str, Any]] = {}
-        self.access_order: List[str] = []
-        self.lock = asyncio.Lock()
-    
-    def _generate_key(self, prompt: str, model: str) -> str:
-        return hashlib.md5(f"{model}:{prompt}".encode()).hexdigest()
-    
-    async def get(self, prompt: str, model: str) -> Optional[Dict]:
-        key = self._generate_key(prompt, model)
-        async with self.lock:
-            if key not in self.cache:
-                return None
-            
-            entry = self.cache[key]
-            if time.time() - entry["timestamp"] > self.ttl:
-                del self.cache[key]
-                self.access_order.remove(key)
-                return None
-            
-            # تحديث أمر الوصول (LRU)
-            self.access_order.remove(key)
-            self.access_order.append(key)
-            return entry["data"]
-    
-    async def set(self, prompt: str, model: str, data: Dict):
-        key = self._generate_key(prompt, model)
-        async with self.lock:
-            if key in self.cache:
-                self.access_order.remove(key)
-            
-            while len(self.access_order) >= self.maxsize:
-                oldest = self.access_order.pop(0)
-                del self.cache[oldest]
-            
-            self.cache[key] = {
-                "timestamp": time.time(),
-                "data": data
-            }
-            self.access_order.append(key)
-
-
-# ==================== مدير المزودين الذكي ====================
-class ProviderManager:
-    def __init__(self, providers: list):
-        self.providers = providers
-        self.stats: Dict[str, ProviderStats] = {
-            p.__name__: ProviderStats(name=p.__name__) 
-            for p in providers
-        }
-        self.lock = asyncio.Lock()
-        self.global_semaphore = asyncio.Semaphore(MAX_CONCURRENT_REQUESTS)
-        self.rate_limiters: Dict[str, asyncio.Semaphore] = {
-            p.__name__: asyncio.Semaphore(1) for p in providers
-        }
-    
-    async def get_ready_providers(self) -> List[Any]:
-        """جلب المزودين الجاهزين فقط"""
-        ready = []
-        now = time.time()
-        async with self.lock:
-            for provider in self.providers:
-                stat = self.stats[provider.__name__]
-                
-                # فحص التبريد
-                if stat.cooldown_until > now:
-                    if stat.circuit_status == "open" and now > stat.cooldown_until:
-                        stat.circuit_status = "half-open"
-                        stat.cooldown_until = 0
-                    else:
-                        continue
-                
-                if stat.weight > 0:
-                    ready.append((provider, stat.weight))
-            
-            # ترتيب حسب الوزن (الأفضل أولاً)
-            ready.sort(key=lambda x: x[1], reverse=True)
-            return [p for p, _ in ready]
-    
-    async def mark_success(self, provider, elapsed: float):
-        async with self.lock:
-            self.stats[provider.__name__].record_success(elapsed)
-    
-    async def mark_failure(self, provider, error: str):
-        async with self.lock:
-            self.stats[provider.__name__].record_failure()
-    
-    async def get_stats(self) -> Dict:
-        async with self.lock:
-            return {
-                name: {
-                    "success_rate": round(s.success_rate * 100, 2),
-                    "avg_time": round(s.avg_response_time, 2),
-                    "circuit": s.circuit_status,
-                    "total": s.total_requests
-                }
-                for name, s in self.stats.items()
-            }
+
+MAX_RETRIES = 10
+TIMEOUT = 60
+
+SAFE_PROVIDERS = []
+
+for provider in __providers__:
+
+    try:
+
+        name = provider.__name__.lower()
+
+        if not getattr(provider, "working", False):
+            continue
+
+        if getattr(provider, "needs_auth", False):
+            continue
+
+        if getattr(provider, "use_nodriver", False):
+            continue
+
+        blocked = [
+            "openai",
+            "qwen",
+            "copilot",
+            "gemini",
+            "claude",
+        ]
+
+        if any(x in name for x in blocked):
+            continue
+
+        SAFE_PROVIDERS.append(provider)
+
+    except:
+        pass
+
+random.shuffle(SAFE_PROVIDERS)
+
+print(f"[+] SAFE PROVIDERS: {len(SAFE_PROVIDERS)}")
+
+client = Client()
 
 
-# ==================== المحرك الرئيسي (Engine) ====================
 class SmartG4F:
+
     def __init__(self):
-        # جمع المزودين
-        self.all_providers = self._collect_providers()
-        random.shuffle(self.all_providers)
-        
-        self.manager = ProviderManager(self.all_providers)
-        self.cache = LRUCache(maxsize=CACHE_MAX_SIZE, ttl=CACHE_TTL)
-        self.client = Client()
-        self.request_queue: asyncio.Queue = asyncio.Queue(maxsize=50)
-        self.processing = set()  # تتبع الطلبات النشطة
-    
-    def _collect_providers(self) -> list:
+
+        self.good = []
+        self.bad = {}
+
+    def mark_bad(self, provider, cooldown=300):
+
+        self.bad[provider.__name__] = time.time() + cooldown
+
+    def is_bad(self, provider):
+
+        expire = self.bad.get(provider.__name__)
+
+        if not expire:
+            return False
+
+        if time.time() > expire:
+            del self.bad[provider.__name__]
+            return False
+
+        return True
+
+    def provider_pool(self):
+
         providers = []
-        for provider in __providers__:
-            try:
-                if not getattr(provider, "working", False):
-                    continue
-                if getattr(provider, "needs_auth", False):
-                    continue
-                if getattr(provider, "use_nodriver", False):
-                    continue
-                
-                name = provider.__name__.lower()
-                if any(x in name for x in BLOCKED_KEYWORDS):
-                    continue
-                
-                providers.append(provider)
-            except:
-                continue
+
+        providers.extend(self.good)
+
+        for p in SAFE_PROVIDERS:
+
+            if p not in providers and not self.is_bad(p):
+                providers.append(p)
+
+        random.shuffle(providers)
+
         return providers
-    
-    def _extract_error_code(self, error_str: str) -> str:
-        """استخراج نوع الخطأ بطريقة ذكية"""
-        err = error_str.lower()
-        if "429" in err or "too many requests" in err or "rate limit" in err:
-            return "RATE_LIMIT"
-        elif "timeout" in err or "timed out" in err:
-            return "TIMEOUT"
-        elif "cloudflare" in err or "cf-ray" in err:
-            return "CLOUDFLARE"
-        elif "403" in err or "forbidden" in err:
-            return "FORBIDDEN"
-        elif "502" in err or "503" in err or "bad gateway" in err:
-            return "SERVER_ERROR"
-        elif "empty response" in err:
-            return "EMPTY"
-        else:
-            return "UNKNOWN"
-    
-    async def ask(self, prompt: str, max_retries: int = 3) -> Dict:
-        # 1. التحقق من الكاش أولاً
-        cached = await self.cache.get(prompt, MODEL)
-        if cached:
-            return {**cached, "cached": True}
-        
-        # 2. إذا كان الخادم تحت الضغط، ارفض الطلبات الجديدة بأسلوب ناعم
-        if len(self.processing) >= MAX_CONCURRENT_REQUESTS * 2:
-            return {
-                "success": False,
-                "error": "Server is currently at high load. Please retry in 10 seconds.",
-                "retry_after": 10
-            }
-        
-        request_id = id(prompt) + time.time()
-        self.processing.add(request_id)
-        
-        try:
-            errors = []
-            last_exception = None
-            
-            for attempt in range(max_retries):
-                providers = await self.manager.get_ready_providers()
-                
-                if not providers:
-                    # انتظار قصير قبل إعادة المحاولة لعل بعض المزودين يتاحون
-                    await asyncio.sleep(min(2 ** attempt, 10))
-                    continue
-                
-                # اختيار المزود الأفضل للمحاولة الحالية (وليس عشوائياً)
-                provider = providers[0]  # الأعلى وزناً
-                
-                # Rate Limiting لكل مزود (لا نضرب نفس المزود بطلبين في نفس الثانية)
-                async with self.manager.rate_limiters[provider.__name__]:
-                    print(f"[TRY {attempt+1}] Using {provider.__name__} (weight: {self.manager.stats[provider.__name__].weight:.2f})")
-                    
-                    start_time = time.time()
-                    
-                    try:
-                        # تشغيل الطلب في Thread منفصل
-                        response = await asyncio.wait_for(
-                            asyncio.to_thread(
-                                self.client.chat.completions.create,
-                                model=MODEL,
-                                provider=provider,
-                                messages=[{"role": "user", "content": prompt}],
-                                timeout=RESPONSE_TIMEOUT,
-                            ),
-                            timeout=RESPONSE_TIMEOUT + 5
-                        )
-                        
-                        text = response.choices[0].message.content
-                        if not text:
-                            raise Exception("Empty response")
-                        
-                        elapsed = time.time() - start_time
-                        
-                        # تسجيل النجاح
-                        await self.manager.mark_success(provider, elapsed)
-                        
-                        result = {
-                            "success": True,
-                            "provider": provider.__name__,
-                            "response": text,
-                            "response_time": round(elapsed, 2),
-                            "attempt": attempt + 1,
-                            "cached": False
+
+    def ask(self, prompt):
+
+        errors = []
+
+        for attempt in range(MAX_RETRIES):
+
+            pool = self.provider_pool()
+
+            if not pool:
+
+                return {
+                    "success": False,
+                    "error": "No providers available"
+                }
+
+            provider = random.choice(pool)
+
+            print(f"[TRY {attempt+1}] {provider.__name__}")
+
+            try:
+
+                response = client.chat.completions.create(
+                    model=MODEL,
+                    provider=provider,
+                    messages=[
+                        {
+                            "role": "user",
+                            "content": prompt
                         }
-                        
-                        # تخزين النتيجة في الكاش
-                        await self.cache.set(prompt, MODEL, result)
-                        
-                        return result
-                        
-                    except asyncio.TimeoutError:
-                        await self.manager.mark_failure(provider, "timeout")
-                        errors.append({
-                            "provider": provider.__name__,
-                            "error": "TIMEOUT",
-                            "attempt": attempt + 1
-                        })
-                        # لا ننتظر كثيراً مع Timeout، ننتقل سريعاً
-                        continue
-                        
-                    except Exception as e:
-                        err_str = str(e)
-                        err_type = self._extract_error_code(err_str)
-                        await self.manager.mark_failure(provider, err_str)
-                        
-                        errors.append({
-                            "provider": provider.__name__,
-                            "error": err_type,
-                            "detail": err_str[:200],
-                            "attempt": attempt + 1
-                        })
-                        
-                        # Backoff ذكي: لا ننام إذا عندنا مزودين آخرين جاهزين
-                        remaining = await self.manager.get_ready_providers()
-                        if len(remaining) <= 1 and attempt < max_retries - 1:
-                            await asyncio.sleep(random.uniform(1.0, 3.0))
-            
-            return {
-                "success": False,
-                "error": "All providers failed after retries",
-                "errors": errors
-            }
-            
-        finally:
-            self.processing.discard(request_id)
-    
-    async def health_check(self) -> Dict:
-        """فحص صحة النظام"""
-        providers = await self.manager.get_ready_providers()
-        stats = await self.manager.get_stats()
-        
-        best_providers = sorted(
-            stats.items(), 
-            key=lambda x: x[1].get("success_rate", 0), 
-            reverse=True
-        )[:5]
-        
+                    ],
+                    timeout=TIMEOUT,
+                )
+
+                text = response.choices[0].message.content
+
+                if not text:
+                    raise Exception("Empty response")
+
+                if provider not in self.good:
+                    self.good.append(provider)
+
+                return {
+                    "success": True,
+                    "provider": provider.__name__,
+                    "response": text
+                }
+
+            except Exception as e:
+
+                err = str(e).lower()
+
+                traceback.print_exc()
+
+                errors.append({
+                    "provider": provider.__name__,
+                    "error": str(e)
+                })
+
+                if "429" in err:
+
+                    self.mark_bad(provider, 600)
+                    time.sleep(random.randint(10, 20))
+
+                elif "cloudflare" in err:
+
+                    self.mark_bad(provider, 1200)
+                    time.sleep(20)
+
+                elif "timeout" in err:
+
+                    self.mark_bad(provider, 300)
+
+                else:
+
+                    self.mark_bad(provider, 180)
+
+                continue
+
         return {
-            "status": "healthy" if len(providers) > 0 else "degraded",
-            "available_providers": len(providers),
-            "total_providers": len(self.all_providers),
-            "active_requests": len(self.processing),
-            "top_performers": [name for name, _ in best_providers],
-            "provider_stats": stats
+            "success": False,
+            "errors": errors
         }
 
 
-# ==================== إعداد FastAPI ====================
-@asynccontextmanager
-async def lifespan(app: FastAPI):
-    # بدء التشغيل
-    print(f"[+] Total providers collected: {len(ai.all_providers)}")
-    print("[+] System ready and optimized for minimal rate-limiting")
-    
-    # بدء مهمة Health Check دورية في الخلفية (اختياري)
-    # asyncio.create_task(periodic_health_reporter())
-    
-    yield
-    
-    # إيقاف التشغيل بأمان
-    print("[+] Shutting down gracefully...")
-
-app = FastAPI(
-    title="Smart G4F Gateway",
-    description="AI Gateway optimized against Rate Limits",
-    version="4.0.0",
-    lifespan=lifespan
-)
-
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
+app = FastAPI()
 
 ai = SmartG4F()
 
 
-# ==================== نماذج البيانات ====================
 class Query(BaseModel):
     prompt: str
-    max_retries: Optional[int] = 3
-
 
-class BatchQuery(BaseModel):
-    prompts: List[str]
 
-
-# ==================== Endpoints ====================
 @app.get("/")
 async def home():
-    health = await ai.health_check()
-    return health
-
-
-@app.get("/stats")
-async def stats():
-    return await ai.health_check()
 
-
-@app.post("/ask")
-async def ask(query: Query, request: Request):
-    # التحقق من حجم الطلب
-    if len(query.prompt) > 8000:
-        raise HTTPException(status_code=413, detail="Prompt too large")
-    
-    result = await ai.ask(query.prompt, max_retries=min(query.max_retries or 3, 5))
-    
-    if result.get("success"):
-        return result
-    else:
-        return JSONResponse(
-            status_code=503 if "retry_after" in result else 424,
-            content=result,
-            headers={"Retry-After": str(result.get("retry_after", 5))} if "retry_after" in result else {}
-        )
-
-
-@app.post("/ask/batch")
-async def ask_batch(batch: BatchQuery):
-    """معالجة دفعة من الأسئلة بأمان (لكن ليس في نفس اللحظة بالضبط)"""
-    if len(batch.prompts) > 10:
-        raise HTTPException(status_code=400, detail="Max 10 prompts per batch")
-    
-    results = []
-    for prompt in batch.prompts:
-        result = await ai.ask(prompt, max_retries=2)
-        results.append(result)
-        # تأخير 1 ثانية بين كل دفعة والثانية لتجنب Rate Limit الشامل
-        await asyncio.sleep(1)
-    
-    successful = sum(1 for r in results if r.get("success"))
     return {
-        "batch_size": len(batch.prompts),
-        "successful": successful,
-        "failed": len(batch.prompts) - successful,
-        "results": results
+        "status": "running",
+        "providers": len(SAFE_PROVIDERS)
     }
 
 
-@app.get("/providers")
-async def list_providers():
-    """عرض حالة المزودين"""
-    stats = await ai.manager.get_stats()
-    ready = [p.__name__ for p in await ai.manager.get_ready_providers()]
-    return {
-        "total": len(ai.all_providers),
-        "ready": ready,
-        "stats": stats
-    }
+@app.post("/ask")
+async def ask(query: Query):
+
+    return ai.ask(query.prompt)
 
 
-# ==================== التشغيل ====================
 if __name__ == "__main__":
+
     uvicorn.run(
         app,
         host="0.0.0.0",
-        port=7860,
-        workers=1  # لا تزيد عن 1 إلا إذا فصلت الذاكرة (state)
-    )
+        port=7860
+    )