189 lines
8.9 KiB
Python
189 lines
8.9 KiB
Python
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()
|
|
|