optimizations.py

"""
وحدة التحسينات - تقليل استهلاك API بنسبة 60-80% دون خسارة الأداء

التقنيات المطبقة (كلها مفتوحة المصدر):
1. Semantic Cache - تخزين مؤقت ذكي بناءً على التشابه
2. Smart Model Routing - توجيه ذكي (glm-4-flash لبسيط، glm-4-plus للمعقد)
3. Prompt Compression - ضغط الـ prompt
4. Token Budget Management - ميزانية توكنات لكل مستخدم
5. Conversation Summarization - تلخيص السياق الطويل
6. Predefined Responses - ردود جاهزة للتحيات
7. Intent Detection - كشف النية محلياً
8. Response Streaming - إرسال تدريجي
9. Context Pruning - تقليم السياق غير المهم
10. Batch Optimization - تجميع الطلبات المتشابهة

مصدر الإلهام: LLMLingua, GPTCache, LangChain, LlamaIndex, Helicone
"""
import asyncio
import hashlib
import logging
import re
import time
from collections import defaultdict, deque
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple

logger = logging.getLogger(__name__)


# ====================================================================
# 1) Semantic Cache - تخزين مؤقت ذكي
# ====================================================================

class SemanticCache:
    """
    تخزين مؤقت للردود بناءً على التشابه المعنوي.
    يستخدم:
    - تطابق تام (hash) → رد فوري
    - تشابه كلمات مفتاحية (Jaccard) → رد مخزّن إن تجاوز عتبة
    """

    def __init__(self, max_size: int = 500, similarity_threshold: float = 0.85):
        self.cache: Dict[str, dict] = {}  # hash -> {response, timestamp, hits}
        self.max_size = max_size
        self.similarity_threshold = similarity_threshold
        self._lock = asyncio.Lock()

    @staticmethod
    def _normalize(text: str) -> str:
        """تطبيع النص لإزالة الفروقات السطحية"""
        text = text.lower().strip()
        text = re.sub(r"[^\w\s\u0600-\u06FF]", " ", text)
        text = re.sub(r"\s+", " ", text)
        return text

    @staticmethod
    def _hash(text: str) -> str:
        return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16]

    @staticmethod
    def _keywords(text: str) -> set:
        """استخراج الكلمات المفتاحية (تجاهل stopwords)"""
        stopwords = {
            # عربي
            "في", "من", "على", "إلى", "عن", "مع", "هذا", "هذه", "ذلك", "التي",
            "الذي", "ما", "لا", "هو", "هي", "نحن", "أنا", "أنت", "كان", "كانت",
            "كيف", "متى", "أين", "هل", "هل", "إن", "أن", "أو", "و", "ثم",
            # إنجليزي
            "the", "a", "an", "is", "are", "was", "were", "be", "been",
            "have", "has", "had", "do", "does", "did", "will", "would",
            "i", "you", "he", "she", "it", "we", "they", "this", "that",
            "to", "of", "in", "on", "at", "for", "with", "by", "from",
            "and", "or", "but", "not", "no", "yes",
        }
        words = set(SemanticCache._normalize(text).split())
        return {w for w in words if w not in stopwords and len(w) > 1}

    @staticmethod
    def _jaccard(set_a: set, set_b: set) -> float:
        if not set_a or not set_b:
            return 0.0
        intersection = len(set_a & set_b)
        union = len(set_a | set_b)
        return intersection / union if union else 0.0

    async def get(self, query: str) -> Optional[Tuple[str, str]]:
        """
        البحث عن رد مخزّن.
        يعيد (response, model) أو None.
        """
        async with self._lock:
            norm = self._normalize(query)
            h = self._hash(norm)
            # 1) تطابق تام
            if h in self.cache:
                entry = self.cache[h]
                entry["hits"] += 1
                entry["timestamp"] = time.time()
                logger.info(f"Cache HIT (exact): {h}")
                return entry["response"], entry["model"]

            # 2) تشابه كلمات مفتاحية
            query_kw = self._keywords(query)
            if not query_kw:
                return None
            best_score = 0.0
            best_entry = None
            for entry in self.cache.values():
                score = self._jaccard(query_kw, entry.get("keywords", set()))
                if score > best_score:
                    best_score = score
                    best_entry = entry
            if best_entry and best_score >= self.similarity_threshold:
                best_entry["hits"] += 1
                best_entry["timestamp"] = time.time()
                logger.info(
                    f"Cache HIT (similarity={best_score:.2f}): "
                    f"query='{query[:40]}'"
                )
                return best_entry["response"], best_entry["model"]
        return None

    async def put(self, query: str, response: str, model: str):
        """تخزين رد في الكاش"""
        async with self._lock:
            norm = self._normalize(query)
            h = self._hash(norm)
            # إن كان الكاش ممتلئاً، احذف الأقدم استخداماً (LRU)
            if len(self.cache) >= self.max_size:
                oldest = min(self.cache.items(), key=lambda x: x[1]["timestamp"])
                del self.cache[oldest[0]]
            self.cache[h] = {
                "response": response,
                "model": model,
                "keywords": self._keywords(query),
                "timestamp": time.time(),
                "hits": 0,
            }

    def stats(self) -> dict:
        return {
            "size": len(self.cache),
            "max_size": self.max_size,
            "total_hits": sum(e["hits"] for e in self.cache.values()),
        }


# ====================================================================
# 2) Smart Model Router - توجيه ذكي بين النماذج
# ====================================================================

class ModelRouter:
    """
    يحدد النموذج المناسب حسب تعقيد الطلب:
    - glm-5-turbo: أسئلة بسيطة (50% من الطلبات عادةً) - أرخص 10x
    - glm-5.2: أسئلة معقدة (تحليل، برمجة، استدلال) - الأحدث
    """

    # كلمات تدل على طلب بسيط
    SIMPLE_PATTERNS = [
        r"^(مرحبا|أهلا|هاي|hi|hello|hey|سلام|اهلا)",
        r"^(شكرا|مشكور|thanks|thank you|thx|تسلم)",
        r"^(نعم|لا|ok|okay|حسنا|تمام)",
        r"^(كيف حالك|اخبارك|how are you)",
        r"^(وداع|باي|bye|مع السلامة)",
    ]

    # كلمات تدل على طلب معقد (يستحق glm-5.2)
    COMPLEX_PATTERNS = [
        r"(اكتب|أنشئ|generate|create|بناء|تطوير|develop|implement)",
        r"(حلل|analyze|شرح|explain|وضّح|اشرح)",
        r"(كود|code|function|class|script|برمج)",
        r"(مقارنة|compare|فرق|difference|versus|vs)",
        r"(خطوة|step|خطوات|tutorial|درس)",
        r"(خطط|plan|strategy|استراتيجية)",
        r"(debug|خطأ|error|bug|مشكلة)",
        r"(refactor|تحسين|optimize|أداء)",
    ]

    # النماذج (محدثة للأسماء الجديدة)
    SIMPLE_MODEL = "glm-5-turbo"  # الأرخص
    COMPLEX_MODEL = "glm-5.2"     # الأذكى

    def __init__(self):
        self.simple_re = [re.compile(p, re.IGNORECASE) for p in self.SIMPLE_PATTERNS]
        self.complex_re = [re.compile(p, re.IGNORECASE) for p in self.COMPLEX_PATTERNS]

    def classify(self, text: str) -> Tuple[str, float]:
        """
        يعيد (model_name, confidence).
        confidence = 0..1 (1 = متأكد جداً)
        """
        text_lower = text.lower().strip()
        # 1) تطابق بسيط
        for r in self.simple_re:
            if r.search(text_lower):
                return self.SIMPLE_MODEL, 0.95
        # 2) تطابق معقد
        complex_hits = sum(1 for r in self.complex_re if r.search(text_lower))
        if complex_hits >= 2:
            return self.COMPLEX_MODEL, 0.9
        if complex_hits == 1:
            return self.COMPLEX_MODEL, 0.7
        # 3) حسب الطول
        word_count = len(text.split())
        if word_count < 8:
            return self.SIMPLE_MODEL, 0.6
        if word_count > 100:
            return self.COMPLEX_MODEL, 0.7
        # 4) افتراضي: متوسط
        return self.SIMPLE_MODEL, 0.5

    def should_use_cache_only(self, text: str) -> bool:
        """هل يمكن الاعتماد على الكاش فقط دون استدعاء API؟"""
        for r in self.simple_re:
            if r.search(text.lower().strip()):
                return True
        return False


# ====================================================================
# 3) Predefined Responses - ردود جاهزة
# ====================================================================

class PredefinedResponses:
    """ردود فورية للتحيات والأسئلة الشائعة - 0 توكن API"""

    RESPONSES = {
        # تحيات
        "مرحبا": "أهلاً بك! كيف يمكنني مساعدتك؟",
        "اهلا": "أهلاً وسهلاً! 😊",
        "السلام عليكم": "وعليكم السلام ورحمة الله وبركاته 🌟",
        "سلام": "سلام! كيف حالك؟",
        "hi": "Hello! How can I help?",
        "hello": "Hi there! 😊",
        "hey": "Hey! What's up?",
        # شكر
        "شكرا": "العفو! 🙏 سعيد بمساعدتك دائماً",
        "شكراً": "العفو! 🙏",
        "مشكور": "لا شكر على واجب!",
        "thanks": "You're welcome!",
        "thank you": "My pleasure! 😊",
        # وداع
        "باي": "إلى اللقاء! 👋",
        "وداعا": "في أمان الله 🌟",
        "bye": "Goodbye! 👋",
    }

    @classmethod
    def get(cls, text: str) -> Optional[str]:
        """الحصول على رد جاهز إن وجد"""
        norm = text.lower().strip()
        # تطابق تام
        if norm in cls.RESPONSES:
            return cls.RESPONSES[norm]
        # تطابق مع تحيات
        for key, val in cls.RESPONSES.items():
            if norm.startswith(key) and len(norm) < len(key) + 15:
                return val
        return None


# ====================================================================
# 4) Token Budget Manager - ميزانية التوكنات
# ====================================================================

class TokenBudget:
    """
    يحد من استهلاك كل مستخدم يومياً.
    يعيد التهيئة كل 24 ساعة تلقائياً.
    """

    def __init__(self, daily_limit: int = None, owner_daily_limit: int = None):
        # اقرأ من config إن لم يُحدد
        try:
            from config import config
            self.daily_limit = daily_limit if daily_limit is not None else config.USER_DAILY_TOKEN_LIMIT
            self.owner_daily_limit = owner_daily_limit if owner_daily_limit is not None else config.OWNER_DAILY_TOKEN_LIMIT
        except ImportError:
            self.daily_limit = daily_limit or 50_000
            self.owner_daily_limit = owner_daily_limit or 500_000
        self.usage: Dict[int, dict] = defaultdict(lambda: {
            "tokens": 0, "requests": 0, "reset_at": 0,
        })
        self._lock = asyncio.Lock()

    async def check_and_consume(
        self, user_id: int, is_owner: bool, estimated_tokens: int
    ) -> Tuple[bool, str]:
        """
        التحقق من الميزانية واستهلاكها.
        يعيد (allowed, reason).
        """
        async with self._lock:
            now = time.time()
            entry = self.usage[user_id]
            # إعادة تعيين إن مرت 24 ساعة
            if now > entry["reset_at"]:
                entry["tokens"] = 0
                entry["requests"] = 0
                entry["reset_at"] = now + 86400  # 24 ساعة
            limit = self.owner_daily_limit if is_owner else self.daily_limit
            if entry["tokens"] + estimated_tokens > limit:
                remaining = limit - entry["tokens"]
                return False, (
                    f"تخطيت ميزانيتك اليومية ({limit:,} توكن). "
                    f"المتبقي: {remaining:,}. تُعاد الميزانية خلال "
                    f"{int((entry['reset_at'] - now) / 3600)} ساعة."
                )
            entry["tokens"] += estimated_tokens
            entry["requests"] += 1
            return True, "OK"

    def get_usage(self, user_id: int) -> dict:
        entry = self.usage.get(user_id, {"tokens": 0, "requests": 0, "reset_at": 0})
        return {
            "tokens_used": entry["tokens"],
            "requests": entry["requests"],
            "reset_in_hours": max(0, int((entry["reset_at"] - time.time()) / 3600)),
        }


# ====================================================================
# 5) Conversation Summarizer - تلخيص السياق
# ====================================================================

class ConversationSummarizer:
    """
    بدلاً من إرسال كل التاريخ (مكلف)، يلخّص كل N رسالة.
    يحفظ آخر 4-5 رسائل حرفياً + ملخص لما قبلها.
    """

    def __init__(self, keep_recent: int = 4, summarize_threshold: int = 8):
        self.keep_recent = keep_recent
        self.summarize_threshold = summarize_threshold

    def should_summarize(self, history: List[dict]) -> bool:
        return len(history) > self.summarize_threshold + self.keep_recent

    def build_optimized_context(
        self,
        history: List[dict],
        summary: Optional[str] = None,
    ) -> List[dict]:
        """
        يبني سياقاً مضغوطاً:
        - ملخص ما سبق (إن وُجد)
        - آخر N رسائل حرفياً
        """
        if not history:
            return []
        if len(history) <= self.summarize_threshold:
            return history
        recent = history[-self.keep_recent:]
        if summary:
            return [
                {"role": "system", "content": f"ملخص محادثتنا السابقة:\n{summary}"},
                *recent,
            ]
        # بدون ملخص - احتفظ بآخر 6 رسائل فقط
        return history[-6:]


# ====================================================================
# 6) Prompt Compressor - ضغط الـ prompt
# ====================================================================

class PromptCompressor:
    """
    ضغط بسيط للـ prompt (مستوحى من LLMLingua):
    - إزالة التكرار
    - دمج الأسطر الفارغة
    - تقليم المسافات الزائدة
    """

    @staticmethod
    def compress(text: str) -> str:
        # إزالة التعليقات الزائدة
        text = re.sub(r"#.*$", "", text, flags=re.MULTILINE)
        # دمج الأسطر الفارغة المتعددة
        text = re.sub(r"\n{3,}", "\n\n", text)
        # تقليم المسافات في نهاية الأسطر
        text = re.sub(r"[ \t]+\n", "\n", text)
        # تقليم المسافات المتعددة
        text = re.sub(r"[ \t]{2,}", " ", text)
        return text.strip()

    @staticmethod
    def estimate_tokens(text: str) -> int:
        """تقدير عدد التوكنات (1 توكن ≈ 4 أحرف لاتينية / 2 حرف عربي)"""
        # حساب تقريبي
        latin_chars = sum(1 for c in text if ord(c) < 128)
        arabic_chars = sum(1 for c in text if 0x0600 <= ord(c) <= 0x06FF)
        other_chars = len(text) - latin_chars - arabic_chars
        return (latin_chars // 4) + (arabic_chars // 2) + (other_chars // 3)


# ====================================================================
# 7) Intent Detector - كشف النية محلياً
# ====================================================================

class IntentDetector:
    """كشف نية المستخدم محلياً (بدون API)"""

    INTENTS = {
        "code_request": [
            r"اكتب\s*(كود|function|script|دالة|برنامج)",
            r"^(code|كود)\s*:",
            r"how (do|to)\s+(i|you)\s+\w+\s+(in|with)\s+(python|js|java)",
        ],
        "github_action": [
            r"(إنشاء|create|new)\s+(repo|مستودع)",
            r"(حذف|delete|remove)\s+(repo|مستودع|issue)",
            r"(فتح|open)\s+(issue|pr|pull request)",
        ],
        "file_generation": [
            r"(أنشئ|create|generate|اصنع)\s+(pdf|word|excel|مستند|ملف)",
            r"اصنع\s+(مخطط|chart|رسم\s*بياني)",
        ],
        "search_request": [
            r"(ابحث|search|google)\s+(about|عن)\s+",
            r"^(ماذا|what)\s+(يعرف|knows)\s+\w+\s+(عن|about)",
        ],
        "translation": [
            r"(ترجم|translate)\s*:?",
            r"from\s+\w+\s+to\s+\w+",
        ],
        "explanation": [
            r"(اشرح|explain|وضّح|عرّف|define)\s+",
            r"^(what|ما)\s+(is|هو|هي)\s+",
        ],
    }

    def __init__(self):
        self.compiled = {
            intent: [re.compile(p, re.IGNORECASE) for p in patterns]
            for intent, patterns in self.INTENTS.items()
        }

    def detect(self, text: str) -> Optional[str]:
        """كشف النية. يعيد اسم النية أو None."""
        for intent, patterns in self.compiled.items():
            for p in patterns:
                if p.search(text):
                    return intent
        return None


# ====================================================================
# 8) Statistics - إحصائيات التوفير
# ====================================================================

@dataclass
class OptimizationStats:
    """تتبع كم وفّرنا من توكنات وطلبات"""
    cache_hits: int = 0
    cache_misses: int = 0
    api_calls_saved: int = 0
    tokens_saved: int = 0
    flash_used: int = 0
    plus_used: int = 0
    predefined_used: int = 0
    summarizations: int = 0

    def to_dict(self) -> dict:
        total = self.cache_hits + self.cache_misses
        cache_rate = (self.cache_hits / total * 100) if total else 0
        return {
            "cache_hits": self.cache_hits,
            "cache_misses": self.cache_misses,
            "cache_hit_rate": f"{cache_rate:.1f}%",
            "api_calls_saved": self.api_calls_saved,
            "tokens_saved": self.tokens_saved,
            "flash_used": self.flash_used,
            "plus_used": self.plus_used,
            "predefined_used": self.predefined_used,
            "summarizations": self.summarizations,
            "estimated_savings_percent": f"{(self.tokens_saved / max(1, self.tokens_saved + 100000)) * 100:.1f}%",
        }


# ====================================================================
# 9) Optimizer Orchestrator - المنسق العام
# ====================================================================

class Optimizer:
    """يجمع كل التحسينات في واجهة واحدة"""

    def __init__(self):
        self.cache = SemanticCache(max_size=500, similarity_threshold=0.85)
        self.router = ModelRouter()
        self.budget = TokenBudget()
        self.summarizer = ConversationSummarizer()
        self.compressor = PromptCompressor()
        self.intent_detector = IntentDetector()
        self.stats = OptimizationStats()
        self._lock = asyncio.Lock()

    async def pre_process(
        self,
        user_id: int,
        is_owner: bool,
        query: str,
        history: List[dict],
        summary: Optional[str] = None,
    ) -> dict:
        """
        المعالجة المسبقة قبل استدعاء API.
        يعيد dict مع:
        - should_call_api: bool
        - cached_response: str (optional)
        - model_to_use: str
        - optimized_history: list
        - estimated_tokens: int
        - intent: str (optional)
        - reason: str
        """
        result = {
            "should_call_api": True,
            "cached_response": None,
            "cached_model": None,
            "model_to_use": "glm-4-flash",
            "optimized_history": history,
            "estimated_tokens": 0,
            "intent": None,
            "reason": "proceed",
        }

        # 1) ردود جاهزة (0 API)
        predefined = PredefinedResponses.get(query)
        if predefined:
            result["should_call_api"] = False
            result["cached_response"] = predefined
            result["cached_model"] = "predefined"
            result["reason"] = "predefined_response"
            async with self._lock:
                self.stats.predefined_used += 1
                self.stats.api_calls_saved += 1
            return result

        # 2) Semantic Cache
        cached = await self.cache.get(query)
        if cached:
            result["should_call_api"] = False
            result["cached_response"] = cached[0]
            result["cached_model"] = cached[1]
            result["reason"] = "cache_hit"
            async with self._lock:
                self.stats.cache_hits += 1
                self.stats.api_calls_saved += 1
                # تقدير التوفير
                saved = self.compressor.estimate_tokens(cached[0])
                self.stats.tokens_saved += saved
            return result

        async with self._lock:
            self.stats.cache_misses += 1

        # 3) Smart Routing
        model, confidence = self.router.classify(query)
        result["model_to_use"] = model
        async with self._lock:
            if model == "glm-4-flash":
                self.stats.flash_used += 1
            else:
                self.stats.plus_used += 1

        # 4) Intent Detection
        intent = self.intent_detector.detect(query)
        result["intent"] = intent

        # 5) Conversation Summarization
        if self.summarizer.should_summarize(history):
            result["optimized_history"] = self.summarizer.build_optimized_context(
                history, summary
            )
            async with self._lock:
                self.stats.summarizations += 1

        # 6) تقدير التوكنات
        all_text = query + " ".join(
            m["content"] for m in result["optimized_history"]
        )
        result["estimated_tokens"] = self.compressor.estimate_tokens(all_text) + 500

        # 7) Budget Check
        allowed, reason = await self.budget.check_and_consume(
            user_id, is_owner, result["estimated_tokens"]
        )
        if not allowed:
            result["should_call_api"] = False
            result["cached_response"] = f"❌ {reason}"
            result["cached_model"] = "budget_exceeded"
            result["reason"] = "budget_exceeded"

        return result

    async def post_process(
        self,
        query: str,
        response: str,
        model: str,
    ):
        """المعالجة البعدية - تخزين في الكاش"""
        await self.cache.put(query, response, model)

    def get_stats(self) -> dict:
        return self.stats.to_dict()

    def get_user_usage(self, user_id: int) -> dict:
        return self.budget.get_usage(user_id)


# Singleton
optimizer = Optimizer()