Update whale_monitor/rpc_manager.py

whale_monitor/rpc_manager.py

@@ -1,5 +1,7 @@
 # whale_monitor/rpc_manager.py
-# هذا هو "الوكيل الذكي" لإدارة اتصالات RPC والمستكشفات
+# (V2 - الموجه الذكي)
+# هذا هو "الوكيل الذكي" لإدارة اتصالات RPC والمستكشفات (APIs)
+# يدير منظمات الطلبات (Rate Limiters) والإحصائيات
 
 import asyncio
 import httpx
@@ -7,8 +9,10 @@ import time
 import ssl
 import json
 import os
+import csv
 from collections import deque, defaultdict
 import random
+from typing import Dict, Any, Optional, List
 
 # استيراد الإعدادات الثابتة
 from .config import DEFAULT_NETWORK_CONFIGS, COINGECKO_BASE_URL
@@ -21,51 +25,122 @@ RPC_CIRCUIT_BREAKER_DURATION = 300 # 5 دقائق إيقاف مؤقت
 # (إضافة ثابت لفرض تأخير بين طلبات CoinGecko)
 COINGECKO_REQUEST_DELAY = 2.0  # 2.0 ثانية (يساوي 30 طلب/دقيقة كحد أقصى)
 
+# (تحديد المسار الحالي لملف rpc_manager.py)
+_CURRENT_DIR = os.path.dirname(os.path.abspath(__file__))
+# (تحديد مسار ملف CSV داخل نفس المجلد)
+CUSTOM_RPC_CSV_FILE = os.path.join(_CURRENT_DIR, 'rpc_endpoints (1).csv')
+
 class AdaptiveRpcManager:
     """
-    مدير RPC ذكي يتتبع صحة النقاط، ويدير حدود الطلبات،
-    ويختار أفضل نقطة متاحة تلقائياً.
+    (محدث V2)
+    مدير RPC و API ذكي:
+    1. يدير صحة نقاط RPC العامة (Public).
+    2. يدير منظمات الطلبات (Rate Limiters) للمفاتيح الخاصة (Infura, Moralis, Scanners).
+    3. يتتبع إحصائيات الجلسة (Session Stats) لجميع الطلبات.
     """
-    def __init__(self, http_client: httpx.AsyncClient, api_keys: dict):
+    def __init__(self, http_client: httpx.AsyncClient):
         self.http_client = http_client
-        self.api_keys = api_keys
         
-        # منظمات الطلبات (Semaphores)
-        self.rpc_semaphore = asyncio.Semaphore(5)  # حد عام لـ RPC
-        self.coingecko_semaphore = asyncio.Semaphore(1) # حد خاص لـ CoinGecko
+        # 1. تحميل المفاتيح الخاصة من متغيرات البيئة
+        self.api_keys = {
+            "infura": os.getenv("INFURA_KEY"),
+            "moralis": os.getenv("MORALIS_KEY"),
+            "etherscan": os.getenv("ETHERSCAN_KEY"),
+            "bscscan": os.getenv("BSCSCAN_KEY"),
+            "polygonscan": os.getenv("POLYGONSCAN_KEY"),
+            # (يمكن إضافة المزيد من مفاتيح Scanners هنا، مثل ARBISCAN_KEY)
+        }
+        print("✅ [RPCManager V2] تم تحميل المفاتيح الخاصة من البيئة.")
+
+        # 2. تهيئة منظمات الطلبات (Semaphores)
+        # (بناءً على الحدود التي قدمتها)
+        
+        # لـ Infura (500 credits/sec) - سنستخدم 450 كحد أمان
+        self.infura_semaphore = asyncio.Semaphore(450)
+        
+        # لـ Moralis (40k/day ~ 0.46/sec) - سنستخدم 1 لضمان طلب واحد في كل مرة
+        self.moralis_semaphore = asyncio.Semaphore(1)
+        
+        # لـ Etherscan, BscScan, PolygonScan (الحد المشترك 5/sec)
+        self.scanner_semaphore = asyncio.Semaphore(5)
         
-        # (متغير لتتبع وقت آخر طلب لـ CoinGecko)
+        # لـ CoinGecko (عام، سنكون حذرين)
+        self.coingecko_semaphore = asyncio.Semaphore(1)
+        
+        # لـ مجمع RPC العام (Public Pool) (لحمايتهم من الضغط)
+        self.public_rpc_semaphore = asyncio.Semaphore(10)
+
         self.last_coingecko_call = 0.0
+        self.last_moralis_call = 0.0
+        
+        # 3. تهيئة إحصائيات الجلسة
+        self.session_stats = defaultdict(int)
         
-        # تهيئة إعدادات الشبكة ونقاط RPC
+        # 4. تهيئة إعدادات الشبكة ونقاط RPC
         self.network_configs = self._initialize_network_configs(DEFAULT_NETWORK_CONFIGS)
         
-        # نظام تتبع الصحة (التعلم)
+        # 5. نظام تتبع الصحة (فقط لنقاط RPC العامة)
         self.endpoint_health = defaultdict(lambda: defaultdict(lambda: {
             "latency": deque(maxlen=RPC_HEALTH_CHECK_WINDOW),
-            "consecutive_errors": 0,
-            "total_errors": 0,
-            "last_error_time": None,
+            "consecutive_errors": 0, "total_errors": 0, "last_error_time": None,
             "circuit_open": False,
         }))
         
-        print("✅ مدير RPC الذكي (الوكيل) مهيأ.")
+        print("✅ [RPCManager V2] مدير RPC/API الذكي (V2) مهيأ.")
+        
+    def _load_rpc_from_csv(self, csv_file_path: str) -> Dict[str, List[str]]:
+        """
+        (جديد V2)
+        قراءة ملف rpc_endpoints (1).csv ودمج النقاط.
+        يتوقع الملف أن يحتوي على عمودين: 'network' و 'url'
+        """
+        custom_rpcs = defaultdict(list)
+        if not os.path.exists(csv_file_path):
+            print(f"⚠️ [RPCManager V2] ملف CSV المخصص '{csv_file_path}' غير موجود. سيتم تخطيه.")
+            return custom_rpcs
+            
+        try:
+            with open(csv_file_path, mode='r', encoding='utf-8') as f:
+                reader = csv.DictReader(f)
+                for row in reader:
+                    network = row.get('network')
+                    url = row.get('url')
+                    if network and url and url.startswith('http'):
+                        custom_rpcs[network].append(url)
+            print(f"✅ [RPCManager V2] تم تحميل {sum(len(v) for v in custom_rpcs.values())} نقطة RPC مخصصة من {csv_file_path}")
+            return custom_rpcs
+        except Exception as e:
+            print(f"❌ [RPCManager V2] فشل في قراءة ملف CSV '{csv_file_path}': {e}")
+            return defaultdict(list)
 
     def _initialize_network_configs(self, configs):
         """
-        يقوم بحقن مفاتيح API في إعدادات الشبكة عند بدء التشغيل.
+        (محدث V2)
+        يقوم بدمج CSV، وحقن مفاتيح API، وإعداد الشبكات.
         """
         initialized_configs = {}
+        
+        # 1. تحميل نقاط CSV المخصصة
+        custom_rpcs = self._load_rpc_from_csv(CUSTOM_RPC_CSV_FILE)
+        
         for network, config in configs.items():
             new_config = config.copy()
             
-            # حقن مفاتيح RPC
+            # 2. حقن مفاتيح Infura
             new_config['rpc_endpoints'] = self._inject_api_keys(
                 config['rpc_endpoints'], 
                 self.api_keys.get('infura')
             )
             
-            # حقن مفاتيح المستكشف (Explorer)
+            # 3. دمج نقاط CSV
+            if network in custom_rpcs:
+                new_config['rpc_endpoints'].extend(custom_rpcs[network])
+                print(f"   ... دمج {len(custom_rpcs[network])} نقاط مخصصة لشبكة {network}")
+
+            # 4. خلط القائمة النهائية (لضمان التوزيع)
+            random.shuffle(new_config['rpc_endpoints'])
+            
+            # 5. حقن مفاتيح المستكشف (Explorer)
             if config.get('explorer'):
                 explorer_key_name = config['explorer'].get('api_key_name')
                 if explorer_key_name and explorer_key_name in self.api_keys:
@@ -86,175 +161,222 @@ class AdaptiveRpcManager:
         
         return [ep.replace("<INFURA_KEY>", infura_key) for ep in endpoints]
 
+    # --- (دوال مساعدة للحصول على الإعدادات) ---
+    
     def get_network_configs(self):
-        """
-        إرجاع إعدادات الشبكة المهيأة (للاستخدام في core.py)
-        """
         return self.network_configs
 
     def get_explorer_config(self, network: str):
-        """
-        إرجاع إعدادات المستكشف (URL والمفتاح) لشبكة معينة.
-        """
         config = self.network_configs.get(network, {})
         return config.get('explorer')
 
-    def _get_healthy_endpoints(self, network: str):
+    def get_api_key(self, key_name: str) -> Optional[str]:
+        """(جديد V2) جلب مفتاح API بأمان"""
+        return self.api_keys.get(key_name)
+
+    # --- (دوال إدارة الإحصائيات V2) ---
+    
+    def reset_session_stats(self):
+        """(جديد V2) تصفير عدادات الإحصائيات للجلسة الجديدة"""
+        self.session_stats = defaultdict(int)
+        print("📊 [RPCManager V2] تم تصفير عدادات إحصائيات الجلسة.")
+
+    def get_session_stats(self) -> Dict[str, int]:
+        """(جديد V2) إرجاع نسخة من الإحصائيات الحالية"""
+        return self.session_stats.copy()
+
+    # --- (دوال الصحة لـ Public RPCs - لا تغيير) ---
+
+    def _get_healthy_public_endpoints(self, network: str):
         """
-        "الذكاء" - يرتب نقاط RPC بناءً على الصحة.
-        يستبعد النقاط التي في وضع "قاطع الدائرة" (Circuit Breaker).
+        (معدل V2)
+        يرتب نقاط RPC العامة (فقط) بناءً على الصحة.
         """
-        if network not in self.network_configs:
-            return []
+        if network not in self.network_configs: return []
             
         endpoints = self.network_configs[network]['rpc_endpoints']
         healthy_endpoints = []
-        
         current_time = time.time()
         
         for ep in endpoints:
+            # (تخطي Infura، هذا المجمع للعام فقط)
+            if "infura.io" in ep:
+                continue
+                
             health = self.endpoint_health[network][ep]
-            
-            # التحقق من قاطع الدائرة
             if health['circuit_open']:
                 if current_time - health['last_error_time'] > RPC_CIRCUIT_BREAKER_DURATION:
-                    # إعادة فتح الدائرة للمحاولة
-                    health['circuit_open'] = False
-                    health['consecutive_errors'] = 0
-                    print(f"ℹ️ [RPC Manager] إعادة تفعيل نقطة RPC: {ep.split('//')[-1]}")
+                    health['circuit_open'] = False; health['consecutive_errors'] = 0
                 else:
                     continue # النقطة لا تزال معطلة مؤقتاً
             
-            # حساب متوسط زمن الاستجابة
             avg_latency = sum(health['latency']) / len(health['latency']) if health['latency'] else float('inf')
-            
-            # (يمكن إضافة منطق ترتيب أكثر تعقيداً هنا مستقبلاً)
             healthy_endpoints.append((ep, avg_latency, health['consecutive_errors']))
 
-        # الترتيب: الأقل أخطاء متتالية، ثم الأقل زمن استجابة
         healthy_endpoints.sort(key=lambda x: (x[2], x[1]))
         
-        # خلط بسيط للنقاط السليمة لمنع التحميل الزائد على نقطة واحدة
+        # (فصل السليم تماماً عن غير السليم)
         healthy_list = [ep[0] for ep in healthy_endpoints if ep[2] == 0]
-        random.shuffle(healthy_list)
-        
+        random.shuffle(healthy_list) # خلط السليم للتوزيع
         unhealthy_list = [ep[0] for ep in healthy_endpoints if ep[2] > 0]
         
         return healthy_list + unhealthy_list
 
     def _update_health(self, network: str, endpoint: str, success: bool, latency: float):
-        """
-        "التعلم" - تحديث إحصائيات صحة نقطة RPC.
-        """
+        """(لا تغيير) تحديث إحصائيات صحة نقطة RPC العامة."""
         health = self.endpoint_health[network][endpoint]
-        
         if success:
             health['latency'].append(latency)
             health['consecutive_errors'] = 0
             health['circuit_open'] = False
         else:
-            health['consecutive_errors'] += 1
-            health['total_errors'] += 1
+            health['consecutive_errors'] += 1; health['total_errors'] += 1
             health['last_error_time'] = time.time()
-            
-            # 🔴 --- START OF CHANGE --- 🔴
-            # (تصحيح الخطأ الطباعي: استخدام RPC_ERROR_THRESHOLD)
             if health['consecutive_errors'] >= RPC_ERROR_THRESHOLD:
-            # 🔴 --- END OF CHANGE --- 🔴
                 health['circuit_open'] = True
-                print(f"🚨 [RPC Manager] قاطع الدائرة مفعل! إيقاف مؤقت لـ: {endpoint.split('//')[-1]}")
+                print(f"🚨 [RPCManager V2] قاطع الدائرة مفعل (عام)! إيقاف مؤقت لـ: {endpoint.split('//')[-1]}")
+
+    # --- (دوال الاتصال الأساسية V2 - محدثة بالكامل) ---
 
-    async def post(self, network: str, payload: dict, timeout: float = 20.0):
+    async def post_rpc(self, network: str, payload: dict, timeout: float = 20.0):
         """
-        إرسال طلب POST إلى أفضل نقطة RPC متاحة للشبكة.
-        يعيد المحاولة تلقائياً عند الفشل.
+        (محدث V2)
+        إرسال طلب POST (JSON-RPC)
+        سيحاول مع Infura أولاً (إذا كان متاحاً)، ثم يلجأ إلى المجمع العام.
         """
-        endpoints = self._get_healthy_endpoints(network)
+        
+        # 1. محاولة Infura أولاً (الأولوية)
+        infura_key = self.api_keys.get('infura')
+        infura_endpoint = next((ep for ep in self.network_configs.get(network, {}).get('rpc_endpoints', []) if "infura.io" in ep), None)
+        
+        if infura_key and infura_endpoint:
+            start_time = time.time()
+            try:
+                async with self.infura_semaphore:
+                    response = await self.http_client.post(infura_endpoint, json=payload, timeout=timeout)
+                    response.raise_for_status()
+                
+                self.session_stats['infura_success'] += 1
+                latency = time.time() - start_time
+                print(f"✅ [RPC Infura] {network} - {latency:.2f}s")
+                return response.json()
+            
+            except Exception as e:
+                self.session_stats['infura_fail'] += 1
+                print(f"⚠️ [RPC Infura] فشل {network}: {type(e).__name__}. اللجوء إلى المجمع العام...")
+
+        # 2. اللجوء إلى المجمع العام (Public Pool)
+        endpoints = self._get_healthy_public_endpoints(network)
         if not endpoints:
-            print(f"❌ [RPC Manager] لا توجد نقاط RPC متاحة أو سليمة لشبكة {network}")
+            print(f"❌ [RPCManager V2] لا توجد نقاط RPC ع��مة متاحة لشبكة {network}")
             return None
             
-        # محاولة أفضل 3 نقاط متاحة
-        for endpoint in endpoints[:3]:
+        for endpoint in endpoints[:3]: # محاولة أفضل 3
             start_time = time.time()
-            ep_name = endpoint.split('//')[-1].split('/')[0] # للاختصار
-            
+            ep_name = endpoint.split('//')[-1].split('/')[0]
             try:
-                async with self.rpc_semaphore:
+                async with self.public_rpc_semaphore:
                     response = await self.http_client.post(endpoint, json=payload, timeout=timeout)
-                    response.raise_for_status() # يطلق استثناء لأخطاء 4xx/5xx
+                    response.raise_for_status()
                 
                 latency = time.time() - start_time
                 self._update_health(network, endpoint, success=True, latency=latency)
-                
-                print(f"✅ [RPC] {network} ({ep_name}) - {latency:.2f}s")
+                self.session_stats['public_rpc_success'] += 1
+                print(f"✅ [RPC Public] {network} ({ep_name}) - {latency:.2f}s")
                 return response.json()
                 
-            except (httpx.HTTPStatusError, httpx.RequestError, ssl.SSLError, asyncio.TimeoutError, json.JSONDecodeError) as e:
-                latency = time.time() - start_time
-                self._update_health(network, endpoint, success=False, latency=latency)
-                print(f"⚠️ [RPC] فشل {network} ({ep_name}): {type(e).__name__}")
-                continue # انتقل إلى النقطة التالية
             except Exception as e:
                 latency = time.time() - start_time
                 self._update_health(network, endpoint, success=False, latency=latency)
-                print(f"❌ [RPC] فشل فادح {network} ({ep_name}): {e}")
-                continue # انتقل إلى النقطة التالية
+                self.session_stats['public_rpc_fail'] += 1
+                print(f"⚠️ [RPC Public] فشل {network} ({ep_name}): {type(e).__name__}")
+                continue
 
-        print(f"❌ [RPC Manager] فشلت جميع المحاولات لشبكة {network} للطلب: {payload.get('method', 'N/A')}")
+        print(f"❌ [RPCManager V2] فشلت جميع محاولات RPC لشبكة {network}")
         return None
 
-    async def get(self, base_url: str, params: dict, headers: dict = None, timeout: float = 15.0, use_coingecko_semaphore: bool = False):
+    async def get_scanner_api(self, base_url: str, params: dict, timeout: float = 15.0):
+        """
+        (جديد V2)
+        إجراء طلب GET لواجهات Scanners (Etherscan, BscScan, etc.)
+        يستخدم المنظم المشترك (5/ثانية).
+        """
+        self.session_stats['scanner_total'] += 1
+        try:
+            async with self.scanner_semaphore:
+                response = await self.http_client.get(base_url, params=params, headers=None, timeout=timeout)
+                response.raise_for_status()
+            self.session_stats['scanner_success'] += 1
+            return response.json()
+        except Exception as e:
+            self.session_stats['scanner_fail'] += 1
+            print(f"❌ [Scanner API] فشل الطلب من {base_url.split('//')[-1]}: {e}")
+            return None
+
+    async def get_moralis_api(self, base_url: str, params: dict, timeout: float = 20.0):
         """
-        (محدث) إرسال طلب GET (للمستكشفات أو CoinGecko).
-        يدير حدود الطلبات بذكاء (Throttling).
+        (جديد V2)
+        إجراء طلب GET لـ Moralis API.
+        يستخدم المنظم الخاص به (1/ثانية) ومفتاح API.
         """
-        semaphore = self.coingecko_semaphore if use_coingecko_semaphore else self.rpc_semaphore
+        moralis_key = self.api_keys.get('moralis')
+        if not moralis_key:
+            print("❌ [Moralis API] لا يوجد مفتاح MORALIS_KEY.")
+            return None
+            
+        headers = {"accept": "application/json", "X-API-Key": moralis_key}
+        self.session_stats['moralis_total'] += 1
         
         try:
-            async with semaphore:
+            async with self.moralis_semaphore:
+                # (ضمان وجود ثانية واحدة على الأقل بين الطلبات لتوزيع 40k على اليوم)
+                current_time = time.time()
+                if current_time - self.last_moralis_call < 1.0:
+                    await asyncio.sleep(1.0 - (current_time - self.last_moralis_call))
+                self.last_moralis_call = time.time()
+                
+                response = await self.http_client.get(base_url, params=params, headers=headers, timeout=timeout)
+                response.raise_for_status()
                 
+            self.session_stats['moralis_success'] += 1
+            return response.json()
+        except Exception as e:
+            self.session_stats['moralis_fail'] += 1
+            print(f"❌ [Moralis API] فشل الطلب: {e}")
+            return None
+
+    async def get_coingecko_api(self, params: dict, headers: dict = None, timeout: float = 15.0):
+        """
+        (معدل V2)
+        إجراء طلب GET لـ CoinGecko (يستخدم الآن إحصائيات ومنظم خاص).
+        """
+        base_url = COINGECKO_BASE_URL
+        self.session_stats['coingecko_total'] += 1
+        try:
+            async with self.coingecko_semaphore:
                 # (تطبيق "الخنق" لـ CoinGecko)
-                if use_coingecko_semaphore:
-                    current_time = time.time()
-                    time_since_last = current_time - self.last_coingecko_call
-                    
-                    if time_since_last < COINGECKO_REQUEST_DELAY:
-                        wait_time = COINGECKO_REQUEST_DELAY - time_since_last
-                        # print(f"   [CoinGecko Throttler] الانتظار {wait_time:.2f} ثانية...")
-                        await asyncio.sleep(wait_time)
-                    
-                    # (تحديث وقت آخر استدعاء *قبل* الطلب)
-                    self.last_coingecko_call = time.time()
+                current_time = time.time()
+                time_since_last = current_time - self.last_coingecko_call
+                if time_since_last < COINGECKO_REQUEST_DELAY:
+                    wait_time = COINGECKO_REQUEST_DELAY - time_since_last
+                    await asyncio.sleep(wait_time)
+                self.last_coingecko_call = time.time()
 
                 response = await self.http_client.get(base_url, params=params, headers=headers, timeout=timeout)
                 
                 if response.status_code == 429: # Too Many Requests
-                    # (زيادة وقت الانتظار)
                     wait_duration = 15.0
-                    print(f"⚠️ [GET] خطأ 429 (Rate Limit) من {base_url}. الانتظار {wait_duration} ثوان...")
+                    print(f"⚠️ [CoinGecko] خطأ 429 (Rate Limit). الانتظار {wait_duration} ثوان...")
                     await asyncio.sleep(wait_duration)
-                    
-                    # (تحديث وقت آخر استدعاء مرة أخرى بعد الانتظار)
-                    if use_coingecko_semaphore:
-                        self.last_coingecko_call = time.time()
-
-                    # إعادة المحاولة مرة واحدة
+                    self.last_coingecko_call = time.time()
                     response = await self.http_client.get(base_url, params=params, headers=headers, timeout=timeout)
                 
                 response.raise_for_status()
             
+            self.session_stats['coingecko_success'] += 1
             return response.json()
             
-        except httpx.HTTPStatusError as e:
-            print(f"❌ [GET] خطأ HTTP من {base_url}: {e.response.status_code}")
-            return None
-        except (httpx.RequestError, asyncio.TimeoutError) as e:
-            print(f"❌ [GET] خطأ اتصال/مهلة من {base_url}: {e}")
-            return None
-        except json.JSONDecodeError as e:
-            print(f"❌ [GET] خطأ JSON من {base_url}: {e}")
-            return None
         except Exception as e:
-            print(f"❌ [GET] خطأ غير متوقع من {base_url}: {e}")
+            self.session_stats['coingecko_fail'] += 1
+            print(f"❌ [CoinGecko] فشل الطلب: {e}")
             return None