Update whale_monitor/rpc_manager.py

whale_monitor/rpc_manager.py

@@ -1,5 +1,5 @@
 # whale_monitor/rpc_manager.py
-# (V3.2 - GEM-Architect: Full Enterprise Edition)
+# (V3.3 - GEM-Architect: Full Enterprise Edition + V7 Crash Fix)
 # يتضمن: Web3 Engine + Raw RPC Fallback + CSV Injection + Strict API Handling
 
 import asyncio
@@ -13,16 +13,22 @@ import traceback
 from collections import defaultdict, deque
 from typing import Dict, Any, Optional, List, Union
 
-# Web3 Imports
+# Web3 Imports with Crash-Safety Logic
 from web3 import AsyncWeb3
 from web3.providers import AsyncHTTPProvider
-from web3.middleware import async_geth_poa_middleware
+
+# [CRITICAL FIX] معالجة التغييرات في إصدارات Web3 المختلفة
+try:
+    from web3.middleware import async_geth_poa_middleware
+except ImportError:
+    # في حالة web3 v7+ أو إصدارات مختلفة، نحدد المتغير كـ None لتجنب الكسر
+    async_geth_poa_middleware = None
+    print("⚠️ [RPCManager] Web3 Middleware 'async_geth_poa_middleware' not found (running v7?). PoA injection disabled.")
 
 # Local Imports
 from .config import DEFAULT_NETWORK_CONFIGS, COINGECKO_BASE_URL
 
 # --- Constants & Limits ---
-# التأكد من عدم الحظر من قبل API العامة
 RPC_HEALTH_CHECK_WINDOW = 15
 RPC_ERROR_THRESHOLD = 5
 RPC_CIRCUIT_BREAKER_DURATION = 300  # 5 Minutes
@@ -45,7 +51,7 @@ class AdaptiveRpcManager:
     """
 
     def __init__(self, http_client: httpx.AsyncClient):
-        print("🔄 [RPCManager V3.2] Initializing Enterprise Connection Engine...")
+        print("🔄 [RPCManager V3.3] Initializing Enterprise Connection Engine...")
         self.http_client = http_client
         
         # 1. Load Secure Keys
@@ -86,7 +92,7 @@ class AdaptiveRpcManager:
         # 6. Web3 Engine Cache
         self.web3_instances = {}
 
-        print(f"✅ [RPCManager V3.2] System Ready. Tracking {len(self.network_configs)} networks.")
+        print(f"✅ [RPCManager V3.3] System Ready. Tracking {len(self.network_configs)} networks.")
 
     # ==============================================================================
     # 🛠️ Initialization & Config Loading (Full Logic)
@@ -104,7 +110,6 @@ class AdaptiveRpcManager:
                 reader = csv.DictReader(f)
                 count = 0
                 for row in reader:
-                    # التنظيف والتحقق
                     net = row.get('network', '').strip().lower()
                     url = row.get('url', '').strip()
                     
@@ -127,7 +132,6 @@ class AdaptiveRpcManager:
         final_configs = {}
 
         for network, config in configs.items():
-            # Deep Copy لتجنب تعديل الافتراضيات
             new_config = config.copy()
             current_endpoints = new_config.get('rpc_endpoints', [])
             
@@ -137,17 +141,16 @@ class AdaptiveRpcManager:
                 if "<INFURA_KEY>" in ep:
                     if infura_key:
                         processed_endpoints.append(ep.replace("<INFURA_KEY>", infura_key))
-                    # else: skip infura endpoint if no key
                 else:
                     processed_endpoints.append(ep)
             
-            # 2. إضافة نقاط CSV (في المقدمة للأولوية)
+            # 2. إضافة نقاط CSV
             if network in custom_rpcs:
                 csv_points = custom_rpcs[network]
                 print(f"   ➕ [Config] Injecting {len(csv_points)} custom RPCs for {network}")
                 processed_endpoints = csv_points + processed_endpoints
             
-            # 3. إزالة التكرار والحفاظ على الترتيب
+            # 3. إزالة التكرار
             unique_endpoints = []
             seen = set()
             for ep in processed_endpoints:
@@ -155,8 +158,7 @@ class AdaptiveRpcManager:
                     unique_endpoints.append(ep)
                     seen.add(ep)
             
-            # 4. خلط عشوائي جزئي (للنقاط العامة فقط) لضمان توزيع الحمل
-            # ملاحظة: نحافظ على أول نقطة إذا كانت من CSV غالباً
+            # 4. خلط عشوائي (مع الحفاظ على الأولوية)
             if len(unique_endpoints) > 1:
                 first = unique_endpoints[0]
                 rest = unique_endpoints[1:]
@@ -165,7 +167,6 @@ class AdaptiveRpcManager:
 
             new_config['rpc_endpoints'] = unique_endpoints
             
-            # 5. حقن مفاتيح المستكشفات (Explorers)
             if config.get('explorer'):
                 key_name = config['explorer'].get('api_key_name')
                 if key_name and key_name in self.api_keys:
@@ -176,21 +177,18 @@ class AdaptiveRpcManager:
         return final_configs
 
     # ==============================================================================
-    # ⚡ Web3 Engine (High Performance)
+    # ⚡ Web3 Engine (High Performance & V7 Safe)
     # ==============================================================================
 
     def get_web3(self, network: str) -> Optional[AsyncWeb3]:
         """
-        إرجاع مثيل Web3 حيوي للشبكة.
-        يستخدم Cache، وإذا فشل يقوم بتدوير النقاط.
+        إرجاع مثيل Web3 حيوي.
         """
         config = self.network_configs.get(network)
         if not config or config.get('type') != 'evm':
             return None
             
-        # التحقق من الكاش
         if network in self.web3_instances:
-            # هنا يمكن إضافة فحص صحة الاتصال (isConnected) إذا لزم الأمر
             return self.web3_instances[network]
             
         endpoints = config['rpc_endpoints']
@@ -198,17 +196,21 @@ class AdaptiveRpcManager:
             print(f"❌ [Web3] No endpoints available for {network}")
             return None
 
-        # محاولة الاتصال (نختار أفضل نقطة متاحة)
-        # في بيئة الإنتاج، نفضل النقطة الأولى (التي قد تكون Private Node من CSV)
         endpoint = endpoints[0]
         
         try:
             provider = AsyncHTTPProvider(endpoint, request_kwargs={'timeout': 20.0})
             w3 = AsyncWeb3(provider)
             
-            # إضافة Middleware للشبكات التي تحتاج POA (مثل BSC, Polygon)
+            # [CRITICAL FIX] حقن Middleware فقط إذا كان مدعوماً ومتاحاً
+            # هذا يمنع الانهيار في web3 v7
             if network in ['bsc', 'polygon', 'avalanche', 'fantom']:
-                w3.middleware_onion.inject(async_geth_poa_middleware, layer=0)
+                if async_geth_poa_middleware and hasattr(w3, 'middleware_onion'):
+                    w3.middleware_onion.inject(async_geth_poa_middleware, layer=0)
+                else:
+                    # في الإصدارات الجديدة، قد لا نحتاج للحقن أو يتم بطريقة أخرى
+                    # نتخطى بصمت للحفاظ على عمل النظام
+                    pass
             
             self.web3_instances[network] = w3
             print(f"   🔌 [Web3] Connected to {network} via {endpoint.split('//')[-1][:20]}...")
@@ -222,11 +224,6 @@ class AdaptiveRpcManager:
     # ==============================================================================
 
     async def post_rpc(self, network: str, payload: dict, timeout: float = 20.0) -> Optional[Dict]:
-        """
-        إرسال طلب JSON-RPC خام عبر HTTP.
-        ضروري للوظائف التي لا تغطيها Web3 أو عند فشل Web3.
-        يدعم نظام Circuit Breaker والصحة.
-        """
         config = self.network_configs.get(network)
         if not config: return None
         
@@ -234,14 +231,10 @@ class AdaptiveRpcManager:
         valid_endpoints = self._get_healthy_endpoints(network, endpoints)
         
         if not valid_endpoints:
-            print(f"❌ [Raw RPC] No healthy endpoints for {network}")
+            # print(f"❌ [Raw RPC] No healthy endpoints for {network}")
             return None
             
-        # المحاولة مع أفضل 3 نقاط
         for endpoint in valid_endpoints[:3]:
-            ep_name = endpoint.split('//')[-1]
-            
-            # تحديد Semaphore المناسب
             semaphore = self.semaphores['infura'] if 'infura.io' in endpoint else self.semaphores['public_rpc']
             
             async with semaphore:
@@ -251,42 +244,30 @@ class AdaptiveRpcManager:
                     response.raise_for_status()
                     data = response.json()
                     
-                    # تحديث الصحة
-                    latency = time.time() - start_time
-                    self._update_health(network, endpoint, True, latency)
+                    self._update_health(network, endpoint, True, time.time() - start_time)
                     self.session_stats['rpc_raw_success'] += 1
-                    
                     return data
                     
-                except Exception as e:
+                except Exception:
                     self._update_health(network, endpoint, False, 0.0)
                     self.session_stats['rpc_raw_fail'] += 1
-                    # print(f"⚠️ [Raw RPC] Fail {ep_name}: {str(e)}") # سجل صامت لتفادي الإزعاج
                     continue
         
         return None
 
     def _get_healthy_endpoints(self, network: str, endpoints: List[str]) -> List[str]:
-        """تصفية وترتيب النقاط بناءً على الصحة والأخطاء السابقة"""
         now = time.time()
         healthy = []
-        
         for ep in endpoints:
             stats = self.endpoint_health[network][ep]
-            
-            # التحقق من قاطع الدائرة (Circuit Breaker)
             if stats['circuit_open']:
                 if now - stats['last_error'] > RPC_CIRCUIT_BREAKER_DURATION:
                     stats['circuit_open'] = False
                     stats['errors'] = 0
                 else:
-                    continue # تجاوز هذه النقطة
-            
-            # حساب متوسط التأخير (Latency Score)
+                    continue
             avg_lat = sum(stats['latency']) / len(stats['latency']) if stats['latency'] else 1.0
             healthy.append((ep, avg_lat))
-            
-        # الترتيب حسب الأسرع
         healthy.sort(key=lambda x: x[1])
         return [h[0] for h in healthy]
 
@@ -308,18 +289,13 @@ class AdaptiveRpcManager:
     # ==============================================================================
 
     async def get_coingecko_api(self, path: str, params: dict, retries: int = 1) -> Optional[Dict]:
-        """
-        التعامل الدقيق مع CoinGecko لتفادي 404 و 429.
-        """
         full_url = f"{COINGECKO_BASE_URL}{path}"
         
         async with self.semaphores['coingecko']:
-            # Throttling
             now = time.time()
             diff = now - self.last_calls['coingecko']
             if diff < COINGECKO_REQUEST_DELAY:
-                wait_time = COINGECKO_REQUEST_DELAY - diff
-                await asyncio.sleep(wait_time)
+                await asyncio.sleep(COINGECKO_REQUEST_DELAY - diff)
             
             for attempt in range(retries + 1):
                 try:
@@ -330,122 +306,84 @@ class AdaptiveRpcManager:
                     if response.status_code == 200:
                         self.session_stats['coingecko_success'] += 1
                         return response.json()
-                    
-                    elif response.status_code == 429: # Rate Limit
-                        wait = 10.0 * (attempt + 1)
-                        print(f"⚠️ [CoinGecko] Rate Limit (429). Waiting {wait}s...")
-                        await asyncio.sleep(wait)
+                    elif response.status_code == 429:
+                        await asyncio.sleep(10.0 * (attempt + 1))
                         continue
-                        
                     elif response.status_code == 404:
-                        # 404 تعني العملة غير موجودة أو المسار خطأ
-                        # لا نعتبرها خطأ اتصال، بل استجابة فارغة
                         return None
-                        
                     else:
                         response.raise_for_status()
-                        
-                except Exception as e:
+                except Exception:
                     self.session_stats['coingecko_fail'] += 1
-                    print(f"❌ [CoinGecko] Error: {e}")
                     return None
             return None
 
     async def get_solscan_api(self, path: str, params: dict) -> Optional[Dict]:
-        """جلب بيانات Solscan مع التحقق من المفتاح"""
         key = self.api_keys.get('solscan')
-        if not key:
-            # print("⚠️ [Solscan] Key missing. Request skipped.")
-            return None
-            
+        if not key: return None
+        
         base_url = "https://pro-api.solscan.io"
         full_url = f"{base_url}{path}"
         headers = {"token": key, "accept": "application/json"}
         
         async with self.semaphores['solscan']:
-            # Solscan Throttling
             now = time.time()
             if now - self.last_calls['solscan'] < SOLSCAN_RATE_LIMIT:
                 await asyncio.sleep(SOLSCAN_RATE_LIMIT)
-                
             try:
                 self.session_stats['solscan_total'] += 1
                 response = await self.http_client.get(full_url, params=params, headers=headers)
                 self.last_calls['solscan'] = time.time()
-                
                 if response.status_code == 200:
                     self.session_stats['solscan_success'] += 1
                     return response.json()
-                else:
-                    print(f"⚠️ [Solscan] Error {response.status_code}: {response.text}")
-                    return None
-            except Exception as e:
+                return None
+            except Exception:
                 self.session_stats['solscan_fail'] += 1
-                print(f"❌ [Solscan] Exception: {e}")
                 return None
 
     async def get_moralis_api(self, params: dict) -> Optional[Dict]:
-        """جلب بيانات Moralis"""
         key = self.api_keys.get('moralis')
         if not key: return None
-        
         url = "https://deep-index.moralis.io/api/v2.2/erc20/transfers"
         headers = {"X-API-Key": key, "accept": "application/json"}
-        
         async with self.semaphores['moralis']:
             now = time.time()
             if now - self.last_calls['moralis'] < MORALIS_RATE_LIMIT:
                 await asyncio.sleep(MORALIS_RATE_LIMIT)
-                
             try:
                 self.session_stats['moralis_total'] += 1
                 response = await self.http_client.get(url, params=params, headers=headers)
                 self.last_calls['moralis'] = time.time()
-                
                 if response.status_code == 200:
                     self.session_stats['moralis_success'] += 1
                     return response.json()
                 return None
-            except Exception as e:
+            except Exception:
                 self.session_stats['moralis_fail'] += 1
                 return None
 
     async def get_scanner_api(self, base_url: str, params: dict) -> Optional[Dict]:
-        """Etherscan/BscScan compatible API calls"""
         async with self.semaphores['scanners']:
             try:
                 self.session_stats['scanner_total'] += 1
                 response = await self.http_client.get(base_url, params=params, timeout=15.0)
-                
                 if response.status_code == 200:
                     self.session_stats['scanner_success'] += 1
                     return response.json()
                 return None
-            except Exception as e:
+            except Exception:
                 self.session_stats['scanner_fail'] += 1
                 return None
 
     # ==============================================================================
     # 📊 Utilities
     # ==============================================================================
-
-    def reset_session_stats(self):
-        self.session_stats = defaultdict(int)
-
-    def get_session_stats(self):
-        return self.session_stats.copy()
-
-    def get_api_key(self, name: str) -> Optional[str]:
-        return self.api_keys.get(name)
-    
-    def get_explorer_config(self, network: str) -> Optional[Dict]:
-        return self.network_configs.get(network, {}).get('explorer')
-
-    def get_network_configs(self) -> Dict:
-        return self.network_configs
-
+    def reset_session_stats(self): self.session_stats = defaultdict(int)
+    def get_session_stats(self): return self.session_stats.copy()
+    def get_api_key(self, name: str): return self.api_keys.get(name)
+    def get_explorer_config(self, network: str): return self.network_configs.get(network, {}).get('explorer')
+    def get_network_configs(self): return self.network_configs
     async def close(self):
-        """Cleanup resources if needed"""
-        # Usually http_client is managed externally, but we can clear caches
         self.web3_instances.clear()
         print("🛑 [RPCManager] Shutting down connection engine.")