274 lines
9.9 KiB
Python
274 lines
9.9 KiB
Python
"""
|
||
🌐 Сервис для работы с браузером
|
||
"""
|
||
|
||
import time
|
||
import random
|
||
import logging
|
||
from typing import Optional
|
||
from selenium import webdriver
|
||
from selenium.webdriver.common.by import By
|
||
from selenium.webdriver.support.ui import WebDriverWait
|
||
from selenium.webdriver.chrome.options import Options
|
||
from selenium.webdriver.chrome.service import Service
|
||
from selenium.common.exceptions import NoSuchElementException
|
||
from webdriver_manager.chrome import ChromeDriverManager
|
||
|
||
from ..config.settings import settings, AppConstants, UIFormatter
|
||
from ..models.vacancy import ApplicationResult
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class BrowserInitializer:
|
||
"""Отвечает за инициализацию браузера"""
|
||
|
||
@staticmethod
|
||
def create_chrome_options(headless: bool) -> Options:
|
||
"""Создание опций Chrome"""
|
||
options = Options()
|
||
|
||
if headless:
|
||
options.add_argument("--headless")
|
||
|
||
options.add_argument("--no-sandbox")
|
||
options.add_argument("--disable-dev-shm-usage")
|
||
options.add_argument("--disable-blink-features=AutomationControlled")
|
||
|
||
return options
|
||
|
||
@staticmethod
|
||
def hide_automation(driver: webdriver.Chrome) -> None:
|
||
"""Скрытие признаков автоматизации"""
|
||
try:
|
||
driver.execute_script(
|
||
"""
|
||
Object.defineProperty(navigator, 'webdriver', {
|
||
get: () => undefined
|
||
});
|
||
"""
|
||
)
|
||
except Exception as e:
|
||
logger.warning(f"Не удалось скрыть признаки автоматизации: {e}")
|
||
|
||
|
||
class AuthenticationHandler:
|
||
"""Отвечает за процесс авторизации"""
|
||
|
||
LOGIN_URL = f"{AppConstants.HH_SITE_URL}/account/login"
|
||
|
||
def __init__(self, driver: webdriver.Chrome):
|
||
self.driver = driver
|
||
|
||
def authenticate_interactive(self) -> bool:
|
||
"""Интерактивная авторизация на HH.ru"""
|
||
try:
|
||
logger.info("Переход на страницу авторизации...")
|
||
self.driver.get(self.LOGIN_URL)
|
||
|
||
print("\n🔐 РЕЖИМ РУЧНОЙ АВТОРИЗАЦИИ")
|
||
print("1. Авторизуйтесь в браузере")
|
||
print("2. Нажмите Enter для продолжения")
|
||
|
||
input("⏳ Авторизуйтесь и нажмите Enter...")
|
||
|
||
if self._check_authentication():
|
||
logger.info("Авторизация успешна!")
|
||
return True
|
||
else:
|
||
logger.error("Авторизация не завершена")
|
||
return False
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при авторизации: {e}")
|
||
return False
|
||
|
||
def _check_authentication(self) -> bool:
|
||
"""Проверка успешности авторизации"""
|
||
try:
|
||
current_url = self.driver.current_url
|
||
page_text = self.driver.page_source.lower()
|
||
|
||
success_indicators = [
|
||
"applicant" in current_url,
|
||
"account" in current_url and "login" not in current_url,
|
||
"мои резюме" in page_text,
|
||
]
|
||
|
||
return any(success_indicators)
|
||
except Exception as e:
|
||
logger.error(f"Ошибка проверки авторизации: {e}")
|
||
return False
|
||
|
||
|
||
class VacancyApplicator:
|
||
"""Отвечает за подачу заявок на вакансии"""
|
||
|
||
APPLY_SELECTORS = [
|
||
'[data-qa="vacancy-response-link-top"]',
|
||
'[data-qa="vacancy-response-button"]',
|
||
'.bloko-button[data-qa*="response"]',
|
||
'button[data-qa*="response"]',
|
||
".vacancy-response-link",
|
||
'a[href*="response"]',
|
||
]
|
||
|
||
ALREADY_APPLIED_INDICATORS = [
|
||
"откликнулись",
|
||
"отклик отправлен",
|
||
"заявка отправлена",
|
||
"response sent",
|
||
"уже откликнулись",
|
||
"чат",
|
||
]
|
||
|
||
def __init__(self, driver: webdriver.Chrome):
|
||
self.driver = driver
|
||
|
||
def apply_to_vacancy(self, vacancy_url: str, vacancy_name: str) -> ApplicationResult:
|
||
"""Подача заявки на вакансию"""
|
||
try:
|
||
truncated_name = UIFormatter.truncate_text(vacancy_name)
|
||
logger.info(f"Переход к вакансии: {truncated_name}...")
|
||
self.driver.get(vacancy_url)
|
||
time.sleep(3)
|
||
|
||
apply_button = self._find_apply_button()
|
||
if not apply_button:
|
||
return ApplicationResult(
|
||
vacancy_id="",
|
||
vacancy_name=vacancy_name,
|
||
success=False,
|
||
error_message="Кнопка отклика не найдена",
|
||
)
|
||
|
||
button_text = apply_button.text.lower()
|
||
if self._is_already_applied(button_text):
|
||
return ApplicationResult(
|
||
vacancy_id="",
|
||
vacancy_name=vacancy_name,
|
||
success=False,
|
||
already_applied=True,
|
||
error_message="Уже откликались на эту вакансию",
|
||
)
|
||
|
||
self.driver.execute_script("arguments[0].click();", apply_button)
|
||
time.sleep(2)
|
||
|
||
logger.info("Кнопка отклика нажата")
|
||
return ApplicationResult(vacancy_id="", vacancy_name=vacancy_name, success=True)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при подаче заявки: {e}")
|
||
return ApplicationResult(
|
||
vacancy_id="",
|
||
vacancy_name=vacancy_name,
|
||
success=False,
|
||
error_message=str(e),
|
||
)
|
||
|
||
def _find_apply_button(self):
|
||
"""Поиск кнопки отклика"""
|
||
for selector in self.APPLY_SELECTORS:
|
||
try:
|
||
button = self.driver.find_element(By.CSS_SELECTOR, selector)
|
||
return button
|
||
except NoSuchElementException:
|
||
continue
|
||
return None
|
||
|
||
def _is_already_applied(self, button_text: str) -> bool:
|
||
"""Проверка, не откликались ли уже"""
|
||
return any(indicator in button_text for indicator in self.ALREADY_APPLIED_INDICATORS)
|
||
|
||
|
||
class BrowserService:
|
||
"""Главный сервис для управления браузером и автоматизации"""
|
||
|
||
def __init__(self):
|
||
self.driver: Optional[webdriver.Chrome] = None
|
||
self.wait: Optional[WebDriverWait] = None
|
||
self._is_authenticated = False
|
||
self.auth_handler: Optional[AuthenticationHandler] = None
|
||
self.applicator: Optional[VacancyApplicator] = None
|
||
|
||
def initialize(self, headless: bool = None) -> bool:
|
||
"""Инициализация браузера"""
|
||
if headless is None:
|
||
headless = settings.browser.headless
|
||
|
||
try:
|
||
logger.info("Инициализация браузера...")
|
||
|
||
options = BrowserInitializer.create_chrome_options(headless)
|
||
service = Service(ChromeDriverManager().install())
|
||
|
||
self.driver = webdriver.Chrome(service=service, options=options)
|
||
self.wait = WebDriverWait(self.driver, settings.browser.wait_timeout)
|
||
|
||
self.auth_handler = AuthenticationHandler(self.driver)
|
||
self.applicator = VacancyApplicator(self.driver)
|
||
|
||
BrowserInitializer.hide_automation(self.driver)
|
||
|
||
logger.info("Браузер успешно инициализирован")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка инициализации браузера: {e}")
|
||
return False
|
||
|
||
def authenticate_interactive(self) -> bool:
|
||
"""Интерактивная авторизация на HH.ru"""
|
||
if not self.driver or not self.auth_handler:
|
||
logger.error("Браузер не инициализирован")
|
||
return False
|
||
|
||
success = self.auth_handler.authenticate_interactive()
|
||
if success:
|
||
self._is_authenticated = True
|
||
return success
|
||
|
||
def apply_to_vacancy(self, vacancy_url: str, vacancy_name: str) -> ApplicationResult:
|
||
"""Подача заявки на вакансию"""
|
||
if not self.is_ready():
|
||
return ApplicationResult(
|
||
vacancy_id="",
|
||
vacancy_name=vacancy_name,
|
||
success=False,
|
||
error_message="Браузер не готов или нет авторизации",
|
||
)
|
||
|
||
return self.applicator.apply_to_vacancy(vacancy_url, vacancy_name)
|
||
|
||
def add_random_pause(self) -> None:
|
||
"""Случайная пауза между действиями"""
|
||
try:
|
||
pause_time = random.uniform(
|
||
settings.application.pause_min, settings.application.pause_max
|
||
)
|
||
logger.info(f"Пауза {pause_time:.1f} сек...")
|
||
time.sleep(pause_time)
|
||
except Exception as e:
|
||
logger.warning(f"Ошибка паузы: {e}")
|
||
time.sleep(3)
|
||
|
||
def close(self) -> None:
|
||
"""Закрытие браузера"""
|
||
try:
|
||
if self.driver:
|
||
self.driver.quit()
|
||
logger.info("Браузер закрыт")
|
||
except Exception as e:
|
||
logger.warning(f"Ошибка при закрытии: {e}")
|
||
finally:
|
||
self.driver = None
|
||
self.wait = None
|
||
self._is_authenticated = False
|
||
self.auth_handler = None
|
||
self.applicator = None
|
||
|
||
def is_ready(self) -> bool:
|
||
"""Проверка готовности к работе"""
|
||
return self.driver is not None and self._is_authenticated and self.applicator is not None
|