self.user_agents = [ "Mozilla/5.0 (Wind

def init(self): self.DP_PATTERN = re.compile(r'/dp/([A-Z0-9]{10})') self.cookies_file = "amazon_cookies.json" self.USE_TOR = True self.TOR_PORT = 9150 self.TOR_TIMES = 1 self.TOR_MAX_TIMES = 30 self.COOKIES_TIMES = 1 self.COOKIES_MAX_TIMES = 100 self.DAY = 86400 self.WEEK = self.DAY * 7 self.MONTH = self.DAY * 30 self.HEADLESS = True self.button_submit_text = "تابع التسوق" self.session_lock = threading.Lock() self._sessions = {} self.COUNTRY = "sa" self.proxy_playwright = {'server': f'socks5://127.0.0.1:{self.TOR_PORT}'} if self.USE_TOR else None self.proxy_requests = {'socks5h': f'127.0.0.1:{self.TOR_PORT}'} if self.USE_TOR else None self.proxy_aiohttp = f'socks5://127.0.0.1:{self.TOR_PORT}' if self.USE_TOR else None self.user_agents = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.0.3 Safari/605.1.15", "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.114 Safari/537.36", ] self.create_products_availability_db() self.create_search_db() self.create_products_db() def _get_session(self): """الحصول على session خاص بالـ thread الحالي (آمن للخيوط)""" thread_id = threading.get_ident() if thread_id not in self._sessions: session = requests.Session() if self.USE_TOR: proxy = { 'http': f'socks5h://127.0.0.1:{self.TOR_PORT}', 'https': f'socks5h://127.0.0.1:{self.TOR_PORT}' } session.proxies.update(self.proxy_requests) adapter = requests.adapters.HTTPAdapter( pool_connections=10, pool_maxsize=10, max_retries=3 ) session.mount('http://', adapter) session.mount('https://', adapter) self._sessions[thread_id] = session return self._sessions[thread_id]