""" 🤖 LLM Sender Unified Module ماژول یکپارچه برای ارسال به ChatGPT و Grok """ import requests import os import logging from typing import Optional import time from abc import ABC, abstractmethod logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class LLMSender(ABC): """کلاس پایه برای ارسال به مدل‌های مختلف LLM""" def __init__(self, api_key: Optional[str] = None, model: str = None): self.api_key = api_key self.model = model self.base_url = "" @abstractmethod def get_default_model(self) -> str: """مدل پیش‌فرض""" pass @abstractmethod def get_base_url(self) -> str: """URL پایه API""" pass def set_api_key(self, api_key: str): """تنظیم کلید API""" self.api_key = api_key logger.info("✅ کلید API تنظیم شد") def set_model(self, model: str): """تغییر مدل""" self.model = model logger.info(f"✅ مدل تغییر یافت به: {model}") def send_simple(self, text: str, lang: str = 'fa') -> str: """ارسال ساده بدون system message سفارشی""" system_msg = ( "شما یک تحلیلگر متخصص هستید. متن حاوی کدهای ناشناس است. " "به درخواست‌ها با دقت و حرفه‌ای پاسخ دهید." if lang == 'fa' else "You are a professional analyst. The text contains anonymous codes. " "Answer requests accurately and professionally." ) return self.send(text, system_msg=system_msg, lang=lang) def send( self, text: str, system_msg: Optional[str] = None, max_tokens: int = 2000, temperature: float = 0.7, timeout: int = 60, lang: str = 'fa', retry_count: int = 3 ) -> str: """ارسال متن به LLM با کنترل کامل""" try: # بررسی اولیه if not text or not text.strip(): error_msg = "متن خالی است!" if lang == 'fa' else "Text is empty!" logger.error(f"❌ {error_msg}") return f"❌ {error_msg}" if not self.api_key: error_msg = "کلید API تنظیم نشده است!" if lang == 'fa' else "API Key not configured!" logger.error(f"❌ {error_msg}") return f"❌ {error_msg}" # تنظیم system message پیش‌فرض if system_msg is None: system_msg = ( "شما یک تحلیلگر مالی حرفه‌ای هستید. متن حاوی کدهای ناشناس است. " "به سوالات با دقت پاسخ دهید." if lang == 'fa' else "You are a professional financial analyst. The text contains anonymous codes. " "Answer questions accurately." ) # تهیه headers headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" } # ساخت request body data = { "model": self.model, "messages": [ {"role": "system", "content": system_msg}, {"role": "user", "content": text} ], "max_tokens": max_tokens, "temperature": temperature } # ارسال با retry mechanism for attempt in range(retry_count): try: logger.info(f"📤 ارسال درخواست به {self.__class__.__name__} (تلاش {attempt + 1}/{retry_count})...") response = requests.post( self.base_url, headers=headers, json=data, timeout=timeout ) # پردازش پاسخ موفق if response.status_code == 200: result = response.json() llm_response = result['choices'][0]['message']['content'] logger.info("✅ پاسخ دریافت شد") return llm_response # پردازش خطاهای مختلف elif response.status_code == 429: # Rate limiting wait_time = 5 * (attempt + 1) logger.warning(f"⚠️ Rate limit | صبر: {wait_time} ثانیه") if attempt < retry_count - 1: time.sleep(wait_time) continue else: return ( "❌ سهمیه API تمام شده است. لطفاً بعداً تلاش کنید." if lang == 'fa' else "❌ API quota exceeded. Please try later." ) elif response.status_code == 401: return ( "❌ کلید API نامعتبر است!" if lang == 'fa' else "❌ Invalid API key!" ) elif response.status_code in [502, 503, 504]: # Server errors wait_time = 2 * (attempt + 1) logger.warning(f"⚠️ Server error {response.status_code} | صبر: {wait_time} ثانیه") if attempt < retry_count - 1: time.sleep(wait_time) continue else: return ( f"❌ خطای سرور: {response.status_code}" if lang == 'fa' else f"❌ Server error: {response.status_code}" ) else: # خطای دیگر try: error_data = response.json() if response.content else {} # چک کنیم که error_data واقعاً dict باشه if isinstance(error_data, dict): error_msg = error_data.get('error', {}).get('message', response.text) else: # اگه string یا چیز دیگه‌ای بود error_msg = str(error_data) except: # اگه JSON parse نشد error_msg = response.text[:200] # فقط 200 کاراکتر اول logger.error(f"❌ API Error: {error_msg}") return f"❌ API Error: {error_msg}" except requests.exceptions.Timeout: logger.warning("⚠️ Timeout | صبر: 3 ثانیه و تلاش مجدد") if attempt < retry_count - 1: time.sleep(3) continue else: return ( "❌ خطای اتصال: timeout" if lang == 'fa' else "❌ Connection error: timeout" ) except requests.exceptions.ConnectionError as e: logger.warning("⚠️ Connection error | صبر: 2 ثانیه و تلاش مجدد") if attempt < retry_count - 1: time.sleep(2) continue else: return ( f"❌ خطای اتصال: {str(e)}" if lang == 'fa' else f"❌ Connection error: {str(e)}" ) except Exception as e: logger.error(f"❌ خطای غیرمنتظره: {str(e)}") return ( f"❌ خطا در ارتباط با LLM: {str(e)}" if lang == 'fa' else f"❌ Error connecting to LLM: {str(e)}" ) class ChatGPTSender(LLMSender): """کلاس برای ارسال به ChatGPT""" def __init__(self, api_key: Optional[str] = None, model: str = "gpt-4o-mini"): # ✅ پاکسازی API key از فضاهای خالی و newline raw_key = api_key or os.getenv("OPENAI_API_KEY", "") cleaned_key = raw_key.strip() if raw_key else "" super().__init__(cleaned_key, model) self.base_url = self.get_base_url() if not self.api_key: logger.warning("⚠️ کلید OpenAI API تنظیم نشده است!") def get_default_model(self) -> str: return "gpt-4o-mini" def get_base_url(self) -> str: return "https://api.openai.com/v1/chat/completions" class GrokSender(LLMSender): """کلاس برای ارسال به Grok (xAI)""" def __init__(self, api_key: Optional[str] = None, model: str = "grok-beta"): # ✅ پاکسازی API key از فضاهای خالی و newline raw_key = api_key or os.getenv("XAI_API_KEY", "") cleaned_key = raw_key.strip() if raw_key else "" super().__init__(cleaned_key, model) self.base_url = self.get_base_url() if not self.api_key: logger.warning("⚠️ کلید xAI API تنظیم نشده است!") def get_default_model(self) -> str: return "grok-beta" def get_base_url(self) -> str: return "https://api.x.ai/v1/chat/completions" # Factory function برای ایجاد sender مناسب def create_llm_sender( provider: str = "chatgpt", api_key: Optional[str] = None, model: Optional[str] = None ) -> LLMSender: """ ایجاد LLM sender بر اساس provider Args: provider: "chatgpt" یا "grok" api_key: کلید API (اختیاری) model: مدل (اختیاری، از پیش‌فرض استفاده می‌شود) Returns: instance از ChatGPTSender یا GrokSender """ provider = provider.lower() if provider == "chatgpt": if model is None: model = "gpt-4o-mini" return ChatGPTSender(api_key=api_key, model=model) elif provider == "grok": if model is None: model = "grok-beta" return GrokSender(api_key=api_key, model=model) else: raise ValueError(f"Provider نامعتبر: {provider}. باید 'chatgpt' یا 'grok' باشد") # مدل‌های موجود برای هر provider AVAILABLE_MODELS = { "chatgpt": [ "gpt-4o-mini", "gpt-4o", "gpt-4-turbo", "gpt-3.5-turbo" ], "grok": [ "grok-beta", # رایگان در بتا "grok-3-mini", # ارزان‌تر "grok-3", # flagship "grok-2-1212" # نسخه قدیمی‌تر ] } if __name__ == "__main__": print("=" * 60) print("🤖 LLM Sender Unified - مثال استفاده") print("=" * 60) # مثال 1: ChatGPT print("\n1️⃣ ChatGPT:") chatgpt = create_llm_sender("chatgpt") if chatgpt.api_key: response = chatgpt.send_simple("سلام") print(f"✅ پاسخ: {response[:100]}...") else: print("⚠️ کلید OpenAI API تنظیم نشده است!") # مثال 2: Grok print("\n2️⃣ Grok:") grok = create_llm_sender("grok", model="grok-beta") if grok.api_key: response = grok.send_simple("سلام") print(f"✅ پاسخ: {response[:100]}...") else: print("⚠️ کلید xAI API تنظیم نشده است!") # مثال 3: لیست مدل‌ها print("\n3️⃣ مدل‌های موجود:") for provider, models in AVAILABLE_MODELS.items(): print(f"\n{provider.upper()}:") for model in models: print(f" - {model}")