""" 🌐 БСрвис для Ρ€Π°Π±ΠΎΡ‚Ρ‹ с Π±Ρ€Π°ΡƒΠ·Π΅Ρ€ΠΎΠΌ """ 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