import json import os import logging import time import aiosqlite from selenium import webdriver from selenium.webdriver.chrome.options import Options from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC logger = logging.getLogger(__name__) class SteamLogin(): def __init__(self): if not os.path.exists('credentials_backup.json'): open('credentials_backup.json', 'w').write(json.dumps({'accounts': []})) self.options = Options() prefs = { "credentials_enable_service": False, "profile.password_manager_enabled": False, "profile.password_manager_leak_detection": False # Отключение проверки утечек паролей } self.options.add_experimental_option("prefs", prefs) self.driver = webdriver.Chrome( options=self.options ) def __wait_get(self, xpath: str, timeout: int = 15): try: return WebDriverWait(self.driver, timeout).until( EC.presence_of_element_located(('xpath', xpath)) ) except Exception as e: ... async def __write_to_database(self, account: dict): async with aiosqlite.connect('credentials.db') as db: # Создаем таблицу если её нет await db.execute(''' CREATE TABLE IF NOT EXISTS backup_accounts ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT NOT NULL UNIQUE, password TEXT NOT NULL, date TEXT NOT NULL, added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # Пробуем вставить новую запись (UNIQUE constraint предотвратит дубликаты) try: await db.execute( '''INSERT INTO backup_accounts (username, password, date) VALUES (?, ?, ?)''', (account['username'], account['password'], account['date']) ) await db.commit() except aiosqlite.IntegrityError: # Аккаунт уже существует в базе pass def __restart_driver(self): self.driver.quit() self.driver = webdriver.Chrome( options=self.options ) self.driver.get("https://steamcommunity.com/login/home/?goto=") # time.sleep(15) async def open_steam_login_page_and_type(self): # Открываем страницу входа в Steam self.driver.get("https://steamcommunity.com/login/home/?goto=") # Загружаем учетные данные из JSON файла async with aiosqlite.connect('credentials.db') as db: # Получаем свежие аккаунты из основной таблицы cursor = await db.execute('SELECT username, password, date FROM accounts') data = await cursor.fetchall() cursor = await db.execute('SELECT username, password, date FROM backup_accounts') archived_accounts = await cursor.fetchall() # Преобразуем в множества для быстрой проверки archived_set = {(acc[0], acc[1]) for acc in archived_accounts} # Фильтруем уникальные аккаунты unique_accounts = [ {'username': acc[0], 'password': acc[1], 'date': acc[2]} for acc in data if (acc[0], acc[1]) not in archived_set ] for account in unique_accounts: # time.sleep(30) username = account['username'] password = account['password'] # Задержка перед заходом в аккаунт WebDriverWait(self.driver, 15).until( EC.presence_of_element_located(("xpath", '//*[@id="responsive_page_template_content"]/div[1]/div[1]/div/div/div/div[2]/div/form/div[1]/input')) ) # Вводим имя пользователя self.driver.find_element("xpath", '//*[@id="responsive_page_template_content"]/div[1]/div[1]/div/div/div/div[2]/div/form/div[1]/input').send_keys(username) # Вводим пароль self.driver.find_element("xpath", '//*[@id="responsive_page_template_content"]/div[1]/div[1]/div/div/div/div[2]/div/form/div[2]/input').send_keys(password) # Нажимаем на кнопку входа self.driver.find_element("xpath", '//*[@id="responsive_page_template_content"]/div[1]/div[1]/div/div/div/div[2]/div/form/div[4]/button').click() time.sleep(5) is_code = self.__wait_get('//*[@id="responsive_page_template_content"]/div[1]/div[1]/div/div/div/div[2]/form/div/a', 5) if is_code: self.driver.get("https://steamcommunity.com/login/home/?goto=") await self.__write_to_database(account) logger.info(f"Вход для аккаунта {username} не удался. Ошибка: требуется код из почты") continue is_authed = self.__wait_get('//*[@id="account_pulldown"]', 5) if not is_authed: self.driver.get("https://steamcommunity.com/login/home/?goto=") await self.__write_to_database(account) logger.info(f"Вход для аккаунта {username} не удался. Ошибка: не авторизован") continue is_family = self.__wait_get('//*[@id="parental_main_content"]/div/div/form/div/div[4]/button', 5) if is_family: self.__wait_get('//*[@id="account_pulldown"]').click() self.__wait_get('//*[@id="account_dropdown"]/div/a[last()]').click() self.__restart_driver() await self.__write_to_database(account) logger.info(f"Вход для аккаунта {username} не удался. Ошибка: семейный доступ") continue # Проверяем, успешен ли вход, ищем нужный элемент try: self.driver.find_element('xpath', '//*[@id="account_pulldown"]').click() WebDriverWait(self.driver, 10).until( EC.presence_of_element_located(('xpath', '//*[@id="account_dropdown"]/div/a[2]')) ) self.driver.find_element('xpath', '//*[@id="account_dropdown"]/div/a[2]').click() self.driver.find_element('xpath', '//*[@id="main_content"]/div[1]/a[4]').click() WebDriverWait(self.driver, 10).until( EC.presence_of_element_located(('xpath', '//*[@id="main_content"]/div[2]/div/div[2]/div[3]/div[5]/div[2]/a')) ).click() # ключевой момент WebDriverWait(self.driver, 10).until( EC.presence_of_element_located(('xpath', '//*[@id="email_authenticator_form"]/div/div[2]/label')) ).click() is_red = self.__wait_get('/html/body/div[6]/div/div/div[1]/div/a/span') if is_red: self.__wait_get('//*[@id="account_pulldown"]').click() self.__wait_get('//*[@id="account_dropdown"]/div/a[last()]').click() self.__restart_driver() await self.__write_to_database(account) logger.info(f"Вход для аккаунта {username} не удался. Ошибка: красная табличка") continue WebDriverWait(self.driver, 10).until( EC.presence_of_element_located(('xpath', '//*[@id="deauthorize_devices_form"]/div/a/span')) ).click() WebDriverWait(self.driver, 10).until( EC.presence_of_element_located(('xpath', '/html/body/div[4]/div[3]/div/div[2]/div[1]')) ).click() self.__wait_get('//*[@id="account_pulldown"]').click() self.__wait_get('//*[@id="account_dropdown"]/div/a[last()]').click() logger.info(f"✅ Guard code установлен для аккаунта: {username}") time.sleep(5) except Exception as e: # Если элемент не найден, переход к следующему аккаунту logger.info(f"Вход для аккаунта {username} не удался. Ошибка: {str(e)}") # Возвращаемся на страницу входа для следующего аккаунта await self.__write_to_database(account) self.__restart_driver()