feat: improved stability and job application logic

- Limit is now calculated based on successful applications
- Updated modal window selectors for HH.ru
- Added "skipped" status for problematic vacancies
- Fixed search logic without auto-adding keywords
- Improved logging and CLI settings
This commit is contained in:
itqop 2025-06-28 18:58:36 +03:00
parent 6ed9453b8e
commit fd8da44b84
13 changed files with 501 additions and 259 deletions

View File

@ -1,7 +1,3 @@
"""
🚀 HH.ru Автоматизация - Главный пакет
"""
__version__ = "2.0.0"
__author__ = "HH Bot Team"

View File

@ -1,12 +1,7 @@
"""
🚀 HH.ru Автоматизация - Entry point для python -m hh_bot
"""
from .cli import CLIInterface
def main():
"""Главная функция"""
CLIInterface.run_application()

View File

@ -1,17 +1,11 @@
"""
🖥 Интерфейс командной строки для HH.ru автоматизации
"""
from ..core.job_application_manager import JobApplicationManager
from ..config.settings import settings, ResumeFileManager, UIFormatter
class CLIInterface:
"""Интерфейс командной строки"""
@staticmethod
def print_welcome():
"""Приветственное сообщение"""
print("🚀 HH.ru АВТОМАТИЗАЦИЯ v2.0")
print(UIFormatter.create_separator())
print("🏗️ Архитектурно правильная версия")
@ -21,19 +15,16 @@ class CLIInterface:
@staticmethod
def print_settings_info():
"""Информация о настройках"""
print("\n⚙️ ТЕКУЩИЕ НАСТРОЙКИ:")
print(f"🔍 Ключевые слова: {settings.hh_search.keywords}")
print(f"📊 Максимум заявок: {settings.application.max_applications}")
print(
f"🤖 Gemini AI: "
f"{'✅ Доступен' if settings.enable_ai_matching() else '❌ Недоступен'}"
)
print(f"🌐 Режим браузера: " f"{'Фоновый' if settings.browser.headless else 'Видимый'}")
ai_status = "✅ Доступен" if settings.enable_ai_matching() else "❌ Недоступен"
print(f"🤖 Gemini AI: {ai_status}")
browser_mode = "Фоновый" if settings.browser.headless else "Видимый"
print(f"🌐 Режим браузера: {browser_mode}")
@staticmethod
def get_user_preferences():
"""Получение предпочтений пользователя"""
print("\n🎯 НАСТРОЙКА ПОИСКА:")
keywords = input(f"Ключевые слова [{settings.hh_search.keywords}]: ").strip()
@ -48,21 +39,44 @@ class CLIInterface:
print("⚠️ AI фильтрация недоступна (нет GEMINI_API_KEY)")
use_ai = False
max_apps_input = input(
f"Максимум заявок [{settings.application.max_applications}]: "
).strip()
excludes = ", ".join(settings.get_exclude_keywords()[:5])
print(f"\n🚫 Текущие исключения: {excludes}...")
exclude_choice = input("Изменить список исключений? [y/n]: ").lower()
if exclude_choice == "y":
CLIInterface._configure_exclude_keywords()
prompt = f"Максимум заявок [{settings.application.max_applications}]: "
max_apps_input = input(prompt).strip()
try:
max_apps = (
int(max_apps_input) if max_apps_input else settings.application.max_applications
int(max_apps_input)
if max_apps_input
else settings.application.max_applications
)
except ValueError:
max_apps = settings.application.max_applications
return keywords, use_ai, max_apps
@staticmethod
def _configure_exclude_keywords():
print("\n⚙️ НАСТРОЙКА ИСКЛЮЧЕНИЙ:")
print("Введите слова через запятую (или Enter для значений по умолчанию):")
current_excludes = ", ".join(settings.get_exclude_keywords())
print(f"Текущие: {current_excludes}")
new_excludes = input("Новые исключения: ").strip()
if new_excludes:
exclude_list = [
word.strip() for word in new_excludes.split(",") if word.strip()
]
settings.get_exclude_keywords = lambda: exclude_list
print(f"✅ Обновлены исключения: {exclude_list}")
else:
print("✅ Оставлены значения по умолчанию")
@staticmethod
def print_final_stats(stats):
"""Вывод итоговой статистики"""
UIFormatter.print_section_header("📊 ИТОГОВАЯ СТАТИСТИКА:", long=True)
if "error" in stats:
@ -71,6 +85,9 @@ class CLIInterface:
print(f"📝 Всего заявок: {stats['total_applications']}")
print(f"✅ Успешных: {stats['successful']}")
print(f"❌ Неудачных: {stats['failed']}")
if "skipped" in stats and stats["skipped"] > 0:
print(f"⏭️ Пропущено: {stats['skipped']}")
if stats["successful"] > 0:
print(f"\n🎉 Отлично! Отправлено {stats['successful']} заявок!")
@ -81,7 +98,6 @@ class CLIInterface:
@staticmethod
def run_application():
"""Главная функция запуска приложения"""
try:
cli = CLIInterface()

View File

@ -1,7 +1,3 @@
"""
📝 Конфигурация логирования для HH.ru автоматизации
"""
import logging
import logging.handlers
from pathlib import Path
@ -14,7 +10,9 @@ class LoggingConfigurator:
@staticmethod
def setup_logging(
log_level: int = logging.INFO, log_file: Optional[str] = None, console_output: bool = True
log_level: int = logging.INFO,
log_file: Optional[str] = None,
console_output: bool = True,
) -> None:
"""
Настройка системы логирования
@ -31,7 +29,8 @@ class LoggingConfigurator:
root_logger.handlers.clear()
formatter = logging.Formatter(
fmt="%(asctime)s - %(name)s - %(levelname)s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S"
fmt="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
if console_output:

View File

@ -1,21 +1,16 @@
"""
Конфигурация для HH.ru автоматизации
"""
import os
from dataclasses import dataclass
from pathlib import Path
class AppConstants:
"""Константы приложения"""
HH_BASE_URL = "https://api.hh.ru"
HH_SITE_URL = "https://hh.ru"
GEMINI_BASE_URL = "https://generativelanguage.googleapis.com/v1beta"
GEMINI_MODEL = "gemini-2.0-flash"
DEFAULT_TIMEOUT = 30
DEFAULT_TIMEOUT = 20
API_PAUSE_SECONDS = 0.5
AI_REQUEST_PAUSE = 1
@ -45,9 +40,8 @@ class AppConstants:
@dataclass
class HHSearchConfig:
"""Настройки поиска вакансий"""
keywords: str = "python junior"
keywords: str = "python"
area: str = "1"
experience: str = "noExperience"
per_page: int = AppConstants.MAX_VACANCIES_PER_PAGE
@ -57,7 +51,6 @@ class HHSearchConfig:
@dataclass
class BrowserConfig:
"""Настройки браузера"""
headless: bool = False
wait_timeout: int = 15
@ -67,7 +60,6 @@ class BrowserConfig:
@dataclass
class ApplicationConfig:
"""Настройки подачи заявок"""
max_applications: int = AppConstants.DEFAULT_MAX_APPLICATIONS
pause_min: float = 3.0
@ -77,7 +69,6 @@ class ApplicationConfig:
@dataclass
class GeminiConfig:
"""Настройки Gemini AI"""
api_key: str = ""
model: str = AppConstants.GEMINI_MODEL
@ -87,7 +78,6 @@ class GeminiConfig:
@dataclass
class ResumeConfig:
"""Настройки резюме"""
experience_file: str = AppConstants.DEFAULT_EXPERIENCE_FILE
about_me_file: str = AppConstants.DEFAULT_ABOUT_FILE
@ -95,11 +85,9 @@ class ResumeConfig:
class ResumeFileManager:
"""Менеджер для работы с файлами резюме"""
@staticmethod
def create_sample_files() -> None:
"""Создание примеров файлов резюме"""
data_dir = Path("data")
data_dir.mkdir(exist_ok=True)
@ -108,12 +96,7 @@ class ResumeFileManager:
experience_file.write_text(
"""
Опыт работы:
- Изучаю Python уже 6 месяцев
- Прошел курсы по основам программирования
- Делал учебные проекты: калькулятор, игра в крестики-нолики
- Изучаю Django и Flask для веб-разработки
- Базовые знания SQL и работы с базами данных
- Знаком с Git для контроля версий
- ноль
""".strip(),
encoding="utf-8",
)
@ -124,11 +107,7 @@ class ResumeFileManager:
about_file.write_text(
"""
О себе:
Начинающий Python разработчик с большим желанием учиться и развиваться.
Интересуюсь веб-разработкой и анализом данных.
Быстро обучаюсь, ответственно подхожу к работе.
Готов к стажировке или junior позиции для получения практического опыта.
Хочу работать в команде опытных разработчиков и вносить вклад в интересные проекты.
Котенок.
""".strip(),
encoding="utf-8",
)
@ -139,15 +118,7 @@ class ResumeFileManager:
skills_file.write_text(
"""
Технические навыки:
- Python (основы, ООП, модули)
- SQL (SELECT, JOIN, базовые запросы)
- Git (commit, push, pull, merge)
- HTML/CSS (базовые знания)
- Django (учебные проекты)
- Flask (микрофреймворк)
- PostgreSQL, SQLite
- Linux (базовые команды)
- VS Code, PyCharm
- Мяу
""".strip(),
encoding="utf-8",
)
@ -155,23 +126,25 @@ class ResumeFileManager:
class UIFormatter:
"""Утилиты для форматирования пользовательского интерфейса"""
@staticmethod
def create_separator(long: bool = False) -> str:
"""Создание разделительной линии"""
length = AppConstants.LONG_SEPARATOR_LENGTH if long else AppConstants.SHORT_SEPARATOR_LENGTH
length = (
AppConstants.LONG_SEPARATOR_LENGTH
if long
else AppConstants.SHORT_SEPARATOR_LENGTH
)
return "=" * length
@staticmethod
def truncate_text(text: str, medium: bool = False) -> str:
"""Обрезание текста до заданного лимита"""
limit = AppConstants.MEDIUM_TEXT_LIMIT if medium else AppConstants.SHORT_TEXT_LIMIT
limit = (
AppConstants.MEDIUM_TEXT_LIMIT if medium else AppConstants.SHORT_TEXT_LIMIT
)
return text[:limit]
@staticmethod
def format_percentage(value: float, total: float) -> str:
"""Форматирование процентного соотношения"""
if total <= 0:
return "0.0%"
percentage = (value / total) * AppConstants.PERCENT_MULTIPLIER
@ -179,7 +152,6 @@ class UIFormatter:
@staticmethod
def print_section_header(title: str, long: bool = False) -> None:
"""Печать заголовка секции с разделителями"""
separator = UIFormatter.create_separator(long)
print(f"\n{separator}")
print(title)
@ -187,10 +159,8 @@ class UIFormatter:
class Settings:
"""Главный класс настроек"""
def __init__(self):
self._load_env()
self.hh_search = HHSearchConfig()
@ -202,7 +172,6 @@ class Settings:
self._validate_config()
def _load_env(self) -> None:
"""Загрузка переменных окружения"""
try:
from dotenv import load_dotenv
@ -211,7 +180,6 @@ class Settings:
print("💡 Установите python-dotenv для работы с .env файлами")
def _validate_config(self) -> None:
"""Валидация настроек"""
if not self.gemini.api_key:
print("⚠️ GEMINI_API_KEY не установлен в переменных окружения")
@ -222,13 +190,14 @@ class Settings:
logs_dir.mkdir(exist_ok=True)
def update_search_keywords(self, keywords: str) -> None:
"""Обновление ключевых слов поиска"""
self.hh_search.keywords = keywords
print(f"🔄 Обновлены ключевые слова: {keywords}")
def enable_ai_matching(self) -> bool:
"""Проверяем можно ли использовать AI сравнение"""
return bool(self.gemini.api_key)
def get_exclude_keywords(self) -> list:
return ['стажер', 'cv']
settings = Settings()

View File

@ -1,7 +1,3 @@
"""
🎯 Главный менеджер для автоматизации откликов на вакансии
"""
import logging
from typing import List, Dict, Optional
import time
@ -21,8 +17,13 @@ class AutomationOrchestrator:
def __init__(self):
self.api_service = HHApiService()
self.ai_service = GeminiAIService()
self.browser_service = BrowserService()
self.ai_service = None
def _get_ai_service(self):
if self.ai_service is None:
self.ai_service = GeminiAIService()
return self.ai_service
def execute_automation_pipeline(
self, keywords: Optional[str] = None, use_ai: bool = True
@ -34,12 +35,13 @@ class AutomationOrchestrator:
if not vacancies:
return {"error": "Подходящие вакансии не найдены"}
if use_ai and self.ai_service.is_available():
vacancies = self._ai_filter_vacancies(vacancies)
if use_ai and self._get_ai_service().is_available():
vacancies = self._ai_filter_vacancies(vacancies[:3])
if not vacancies:
return {"error": "После AI фильтрации не осталось подходящих вакансий"}
return {"error": "После AI фильтрации подходящих вакансий нет"}
if not self._initialize_browser_and_auth():
init_ok = self._initialize_browser_and_auth()
if not init_ok:
return {"error": "Ошибка инициализации браузера или авторизации"}
application_results = self._apply_to_vacancies(vacancies)
@ -55,7 +57,9 @@ class AutomationOrchestrator:
finally:
self._cleanup()
def _search_and_filter_vacancies(self, keywords: Optional[str] = None) -> List[Vacancy]:
def _search_and_filter_vacancies(
self, keywords: Optional[str] = None
) -> List[Vacancy]:
"""Поиск и базовая фильтрация вакансий"""
logger.info("🔍 ЭТАП 1: Поиск вакансий")
@ -65,7 +69,9 @@ class AutomationOrchestrator:
logger.warning("Вакансии не найдены через API")
return []
suitable_vacancies = self.api_service.filter_suitable_vacancies(all_vacancies)
suitable_vacancies = self.api_service.filter_suitable_vacancies(
all_vacancies, search_keywords=keywords or ""
)
self._log_search_results(all_vacancies, suitable_vacancies)
return suitable_vacancies
@ -86,7 +92,7 @@ class AutomationOrchestrator:
logger.info(f"Анализ {i}/{total_count}: {truncated_name}...")
try:
if self.ai_service.should_apply(vacancy):
if self._get_ai_service().should_apply(vacancy):
ai_suitable.append(vacancy)
logger.info("✅ Добавлено в список для отклика")
else:
@ -123,25 +129,39 @@ class AutomationOrchestrator:
return False
def _apply_to_vacancies(self, vacancies: List[Vacancy]) -> List[ApplicationResult]:
"""Подача заявок на вакансии"""
max_apps = settings.application.max_applications
vacancies_to_process = vacancies[:max_apps]
max_successful_apps = settings.application.max_applications
logger.info(f"📨 ЭТАП 4: Подача заявок (максимум {max_apps})")
logger.info("💡 Между заявками добавляются паузы для безопасности")
logger.info(f"📨 ЭТАП 4: Подача заявок (максимум {max_successful_apps} успешных)")
logger.info("💡 Между заявками добавляются паузы")
logger.info("💡 Лимит считается только по успешным заявкам")
application_results = []
successful_count = 0
processed_count = 0
for i, vacancy in enumerate(vacancies_to_process, 1):
for vacancy in vacancies:
if successful_count >= max_successful_apps:
logger.info(f"🎯 Достигнут лимит успешных заявок: {max_successful_apps}")
break
processed_count += 1
truncated_name = UIFormatter.truncate_text(vacancy.name, medium=True)
logger.info(f"Обработка {i}/{len(vacancies_to_process)}: {truncated_name}...")
logger.info(
f"Обработка {processed_count}: {truncated_name} (успешных: {successful_count}/{max_successful_apps})"
)
try:
result = self.browser_service.apply_to_vacancy(vacancy.alternate_url, vacancy.name)
result = self.browser_service.apply_to_vacancy(
vacancy.alternate_url, vacancy.name
)
application_results.append(result)
self._log_application_result(result)
if i < len(vacancies_to_process):
if result.success:
successful_count += 1
logger.info(f" 🎉 Успешных заявок: {successful_count}/{max_successful_apps}")
if processed_count < len(vacancies) and successful_count < max_successful_apps:
self.browser_service.add_random_pause()
except Exception as e:
@ -154,15 +174,20 @@ class AutomationOrchestrator:
)
application_results.append(error_result)
logger.info(f"🏁 Обработка завершена. Обработано вакансий: {processed_count}, успешных заявок: {successful_count}")
return application_results
def _log_search_results(self, all_vacancies: List[Vacancy], suitable: List[Vacancy]):
def _log_search_results(
self, all_vacancies: List[Vacancy], suitable: List[Vacancy]
):
"""Логирование результатов поиска"""
logger.info("📊 Результат базовой фильтрации:")
logger.info(f" 🔍 Всего: {len(all_vacancies)}")
logger.info(f" ✅ Подходящих: {len(suitable)}")
if len(all_vacancies) > 0:
percentage = UIFormatter.format_percentage(len(suitable), len(all_vacancies))
percentage = UIFormatter.format_percentage(
len(suitable), len(all_vacancies)
)
logger.info(f" 📈 % соответствия: {percentage}")
def _log_ai_results(self, total_analyzed: int, ai_suitable: List[Vacancy]):
@ -180,6 +205,8 @@ class AutomationOrchestrator:
logger.info(" ✅ Заявка отправлена успешно")
elif result.already_applied:
logger.info(" ⚠️ Уже откликались ранее")
elif result.skipped:
logger.warning(f" ⏭️ Пропущено: {result.error_message}")
else:
logger.warning(f" ❌ Ошибка: {result.error_message}")
@ -188,13 +215,15 @@ class AutomationOrchestrator:
total_applications = len(application_results)
successful = sum(1 for r in application_results if r.success)
already_applied = sum(1 for r in application_results if r.already_applied)
failed = total_applications - successful - already_applied
skipped = sum(1 for r in application_results if r.skipped)
failed = total_applications - successful - already_applied - skipped
return {
"total_applications": total_applications,
"successful": successful,
"failed": failed,
"already_applied": already_applied,
"skipped": skipped,
}
def _cleanup(self):
@ -208,12 +237,16 @@ class JobApplicationManager:
def __init__(self):
LoggingConfigurator.setup_logging(log_file="logs/hh_bot.log", console_output=False)
LoggingConfigurator.setup_logging(
log_file="logs/hh_bot.log", console_output=True
)
self.orchestrator = AutomationOrchestrator()
self.application_results: List[ApplicationResult] = []
def run_automation(self, keywords: Optional[str] = None, use_ai: bool = True) -> Dict:
def run_automation(
self, keywords: Optional[str] = None, use_ai: bool = True
) -> Dict:
"""Запуск полного цикла автоматизации"""
print("🚀 Запуск автоматизации HH.ru")
print(UIFormatter.create_separator())
@ -221,8 +254,7 @@ class JobApplicationManager:
stats = self.orchestrator.execute_automation_pipeline(keywords, use_ai)
if "error" not in stats:
pass
self.application_results = []
return stats
@ -241,6 +273,7 @@ class JobApplicationManager:
print(f"📝 Всего попыток подачи заявок: {stats['total_applications']}")
print(f"✅ Успешно отправлено: {stats['successful']}")
print(f"⚠️ Уже откликались ранее: {stats['already_applied']}")
print(f"⏭️ Пропущено (тестовые/ошибки): {stats['skipped']}")
print(f"❌ Неудачных попыток: {stats['failed']}")
if stats["total_applications"] > 0:

View File

@ -1,7 +1,3 @@
"""
📋 Модели данных для работы с вакансиями HH.ru
"""
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Any
import re
@ -9,7 +5,6 @@ import re
@dataclass
class Employer:
"""Информация о работодателе"""
id: str
name: str
@ -22,7 +17,6 @@ class Employer:
@dataclass
class Experience:
"""Информация об опыте работы"""
id: str
name: str
@ -30,7 +24,6 @@ class Experience:
@dataclass
class Snippet:
"""Краткая информация о вакансии"""
requirement: Optional[str] = None
responsibility: Optional[str] = None
@ -38,7 +31,6 @@ class Snippet:
@dataclass
class Salary:
"""Информация о зарплате"""
from_value: Optional[int] = None
to_value: Optional[int] = None
@ -48,7 +40,6 @@ class Salary:
@dataclass
class Vacancy:
"""Модель вакансии HH.ru"""
id: str
name: str
@ -69,7 +60,6 @@ class Vacancy:
@classmethod
def from_api_response(cls, data: Dict[str, Any]) -> "Vacancy":
"""Создание экземпляра из ответа API HH.ru"""
try:
employer_data = data.get("employer", {})
@ -132,9 +122,9 @@ class Vacancy:
)
def has_python(self) -> bool:
"""Проверка упоминания Python в вакансии"""
text_to_check = (
f"{self.name} {self.snippet.requirement or ''} " f"{self.snippet.responsibility or ''}"
f"{self.name} {self.snippet.requirement or ''} "
f"{self.snippet.responsibility or ''}"
)
python_patterns = [
r"\bpython\b",
@ -151,8 +141,20 @@ class Vacancy:
return True
return False
def matches_keywords(self, keywords: str) -> bool:
text_to_check = (
f"{self.name} {self.snippet.requirement or ''} "
f"{self.snippet.responsibility or ''}"
).lower()
search_terms = [term.strip().lower() for term in keywords.split()]
for term in search_terms:
if term in text_to_check:
return True
return False
def is_junior_level(self) -> bool:
"""Проверка на junior уровень"""
junior_keywords = [
"junior",
"джуниор",
@ -173,7 +175,6 @@ class Vacancy:
return False
def get_salary_info(self) -> str:
"""Получение информации о зарплате в читаемом виде"""
if not self.salary:
return "Зарплата не указана"
@ -192,7 +193,6 @@ class Vacancy:
return "Зарплата не указана"
def get_full_text(self) -> str:
"""Получение полного текста вакансии для анализа"""
text_parts = [
self.name,
self.employer.name,
@ -205,17 +205,16 @@ class Vacancy:
@dataclass
class ApplicationResult:
"""Результат подачи заявки на вакансию"""
vacancy_id: str
vacancy_name: str
success: bool
already_applied: bool = False
skipped: bool = False
error_message: Optional[str] = None
timestamp: Optional[str] = None
def __post_init__(self):
"""Устанавливаем timestamp если не указан"""
if self.timestamp is None:
from datetime import datetime
@ -224,7 +223,6 @@ class ApplicationResult:
@dataclass
class SearchStats:
"""Статистика поиска вакансий"""
total_found: int = 0
pages_processed: int = 0
@ -235,13 +233,13 @@ class SearchStats:
without_test: int = 0
def __str__(self) -> str:
return f"""
📊 Статистика поиска:
📋 Всего найдено: {self.total_found}
📄 Страниц обработано: {self.pages_processed}
Прошло фильтрацию: {self.filtered_count}
🐍 Python вакансий: {self.python_vacancies}
👶 Junior уровня: {self.junior_vacancies}
💰 С указанной ЗП: {self.with_salary}
📝 Без тестов: {self.without_test}
"""
return (
f"📊 Статистика поиска:\n"
f" 📋 Всего найдено: {self.total_found}\n"
f" 📄 Страниц обработано: {self.pages_processed}\n"
f" ✅ Прошло фильтрацию: {self.filtered_count}\n"
f" 🐍 Python вакансий: {self.python_vacancies}\n"
f" 👶 Junior уровня: {self.junior_vacancies}\n"
f" 💰 С указанной ЗП: {self.with_salary}\n"
f" 📝 Без тестов: {self.without_test}"
)

View File

@ -1,3 +0,0 @@
"""
🔧 Пакет сервисов
"""

View File

@ -1,14 +1,14 @@
"""
🌐 Сервис для работы с браузером
"""
import time
import random
import logging
import json
from typing import Optional
from pathlib import Path
from enum import Enum
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from selenium.common.exceptions import NoSuchElementException
@ -17,15 +17,97 @@ from webdriver_manager.chrome import ChromeDriverManager
from ..config.settings import settings, AppConstants, UIFormatter
from ..models.vacancy import ApplicationResult
class SubmissionResult(Enum):
SUCCESS = "success"
FAILED = "failed"
SKIPPED = "skipped"
logger = logging.getLogger(__name__)
class SessionManager:
"""Управление сессиями авторизации HH.ru"""
def __init__(self, driver: webdriver.Chrome):
self.driver = driver
self.session_dir = Path("session_data")
self.cookies_file = self.session_dir / "hh_cookies.json"
self.session_dir.mkdir(exist_ok=True)
def save_session(self) -> bool:
"""Сохранение текущей сессии в файл"""
try:
cookies = self.driver.get_cookies()
session_data = {
"cookies": cookies,
"user_agent": self.driver.execute_script("return navigator.userAgent;"),
"timestamp": time.time(),
}
with open(self.cookies_file, 'w', encoding='utf-8') as f:
json.dump(session_data, f, indent=2, ensure_ascii=False)
logger.info(f"Сессия сохранена в {self.cookies_file}")
return True
except Exception as e:
logger.error(f"Ошибка сохранения сессии: {e}")
return False
def load_session(self) -> bool:
"""Загрузка сохраненной сессии"""
try:
if not self.cookies_file.exists():
logger.info("Файл сессии не найден")
return False
with open(self.cookies_file, 'r', encoding='utf-8') as f:
session_data = json.load(f)
if not self._is_session_valid(session_data):
logger.info("Сессия устарела или невалидна")
return False
self.driver.get(AppConstants.HH_SITE_URL)
time.sleep(2)
for cookie in session_data["cookies"]:
try:
self.driver.add_cookie(cookie)
except Exception as e:
logger.warning(f"Не удалось добавить cookie: {e}")
logger.info("Сессия загружена из файла")
return True
except Exception as e:
logger.error(f"Ошибка загрузки сессии: {e}")
return False
def clear_session(self) -> None:
"""Удаление сохраненной сессии"""
try:
if self.cookies_file.exists():
self.cookies_file.unlink()
logger.info("Сессия удалена")
except Exception as e:
logger.warning(f"Ошибка удаления сессии: {e}")
def _is_session_valid(self, session_data: dict) -> bool:
"""Проверка валидности сессии по времени"""
try:
session_age = time.time() - session_data.get("timestamp", 0)
max_age = 7 * 24 * 3600 # 7 дней
return session_age < max_age
except Exception:
return False
class BrowserInitializer:
"""Отвечает за инициализацию браузера"""
@staticmethod
def create_chrome_options(headless: bool) -> Options:
"""Создание опций Chrome"""
options = Options()
if headless:
@ -34,6 +116,25 @@ class BrowserInitializer:
options.add_argument("--no-sandbox")
options.add_argument("--disable-dev-shm-usage")
options.add_argument("--disable-blink-features=AutomationControlled")
options.add_argument("--disable-gpu")
options.add_argument("--disable-features=VizDisplayCompositor")
options.add_argument("--disable-web-security")
options.add_argument("--disable-features=VoiceInteraction")
options.add_argument("--disable-speech-api")
options.add_argument("--disable-background-networking")
options.add_argument("--disable-background-timer-throttling")
options.add_argument("--disable-renderer-backgrounding")
options.add_argument("--disable-backgrounding-occluded-windows")
options.add_argument("--disable-client-side-phishing-detection")
options.add_argument("--disable-sync")
options.add_argument("--disable-translate")
options.add_argument("--disable-ipc-flooding-protection")
options.add_argument("--log-level=3")
options.add_experimental_option("excludeSwitches", ["enable-automation"])
options.add_experimental_option("useAutomationExtension", False)
options.add_experimental_option("excludeSwitches", ["enable-logging"])
options.add_experimental_option("excludeSwitches", ["disable-logging"])
return options
@ -59,21 +160,45 @@ class AuthenticationHandler:
def __init__(self, driver: webdriver.Chrome):
self.driver = driver
self.session_manager = SessionManager(driver)
def authenticate_interactive(self) -> bool:
"""Интерактивная авторизация на HH.ru"""
"""Интерактивная авторизация на HH.ru с поддержкой сохраненной сессии"""
try:
logger.info("Переход на страницу авторизации...")
self.driver.get(self.LOGIN_URL)
print("\n🔐 ПРОВЕРКА СОХРАНЕННОЙ СЕССИИ")
logger.info("Проверяем сохраненную сессию...")
if self.session_manager.load_session():
self.driver.refresh()
time.sleep(3)
if self._check_authentication():
print("✅ Использована сохраненная сессия")
logger.info("Авторизация через сохраненную сессию успешна!")
return True
else:
print("❌ Сохраненная сессия недействительна")
logger.warning("Сохраненная сессия недействительна")
self.session_manager.clear_session()
print("\n🔐 РЕЖИМ РУЧНОЙ АВТОРИЗАЦИИ")
print("1. Авторизуйтесь в браузере")
print("2. Нажмите Enter для продолжения")
print("3. Сессия будет сохранена для повторного использования")
logger.info("Переход на страницу авторизации...")
self.driver.get(self.LOGIN_URL)
input("⏳ Авторизуйтесь и нажмите Enter...")
if self._check_authentication():
logger.info("Авторизация успешна!")
if self.session_manager.save_session():
print("✅ Сессия сохранена для следующих запусков")
else:
print("⚠️ Не удалось сохранить сессию")
return True
else:
logger.error("Авторизация не завершена")
@ -125,7 +250,9 @@ class VacancyApplicator:
def __init__(self, driver: webdriver.Chrome):
self.driver = driver
def apply_to_vacancy(self, vacancy_url: str, vacancy_name: str) -> ApplicationResult:
def apply_to_vacancy(
self, vacancy_url: str, vacancy_name: str
) -> ApplicationResult:
"""Подача заявки на вакансию"""
try:
truncated_name = UIFormatter.truncate_text(vacancy_name)
@ -155,8 +282,32 @@ class VacancyApplicator:
self.driver.execute_script("arguments[0].click();", apply_button)
time.sleep(2)
logger.info("Кнопка отклика нажата")
return ApplicationResult(vacancy_id="", vacancy_name=vacancy_name, success=True)
logger.info("Кнопка отклика нажата, ищем форму заявки...")
submit_result = self._submit_application_form()
if submit_result == SubmissionResult.SUCCESS:
logger.info("✅ Заявка успешно отправлена")
return ApplicationResult(
vacancy_id="", vacancy_name=vacancy_name, success=True
)
elif submit_result == SubmissionResult.SKIPPED:
logger.warning("⚠️ Вакансия пропущена (нет модального окна)")
return ApplicationResult(
vacancy_id="",
vacancy_name=vacancy_name,
success=False,
skipped=True,
error_message="Модальное окно не найдено (возможно тестовая вакансия)",
)
else: # FAILED
logger.warning("Не удалось отправить заявку в модальном окне")
return ApplicationResult(
vacancy_id="",
vacancy_name=vacancy_name,
success=False,
error_message="Ошибка отправки в модальном окне",
)
except Exception as e:
logger.error(f"Ошибка при подаче заявки: {e}")
@ -179,7 +330,158 @@ class VacancyApplicator:
def _is_already_applied(self, button_text: str) -> bool:
"""Проверка, не откликались ли уже"""
return any(indicator in button_text for indicator in self.ALREADY_APPLIED_INDICATORS)
return any(
indicator in button_text for indicator in self.ALREADY_APPLIED_INDICATORS
)
def _submit_application_form(self) -> SubmissionResult:
"""Отправка заявки в модальном окне"""
try:
modal_selectors = [
'[data-qa="modal-overlay"]',
'.magritte-modal-overlay',
'[data-qa="modal"]',
'[data-qa="vacancy-response-popup"]',
'.vacancy-response-popup',
'.modal',
'.bloko-modal',
]
submit_selectors = [
'[data-qa="vacancy-response-submit-popup"]',
'button[form="RESPONSE_MODAL_FORM_ID"]',
'button[type="submit"][form="RESPONSE_MODAL_FORM_ID"]',
'[data-qa="vacancy-response-letter-submit"]',
'button[data-qa*="submit"]',
'button[type="submit"]',
'input[type="submit"]',
'.bloko-button[data-qa*="submit"]',
]
modal_found = False
for selector in modal_selectors:
try:
modal = WebDriverWait(self.driver, 5).until(
EC.presence_of_element_located((By.CSS_SELECTOR, selector))
)
if modal:
modal_found = True
logger.info(f"Модальное окно найдено: {selector}")
break
except Exception:
continue
if not modal_found:
logger.warning("⚠️ Модальное окно не найдено - пропускаем вакансию (возможно тестовая или ошибка)")
return SubmissionResult.SKIPPED
form_selectors = [
'form[name="vacancy_response"]',
'form[id="RESPONSE_MODAL_FORM_ID"]',
'form[data-qa*="response"]',
]
form_found = False
for form_selector in form_selectors:
try:
form = self.driver.find_element(By.CSS_SELECTOR, form_selector)
if form:
form_found = True
logger.info(f"Форма отклика найдена: {form_selector}")
break
except Exception:
continue
if not form_found:
logger.warning("Форма отклика не найдена в модальном окне - пропускаем")
return SubmissionResult.SKIPPED
time.sleep(1)
for selector in submit_selectors:
try:
submit_button = WebDriverWait(self.driver, 3).until(
EC.element_to_be_clickable((By.CSS_SELECTOR, selector))
)
if submit_button:
logger.info(f"Нажимаем кнопку отправки: {submit_button.text.strip()}")
self.driver.execute_script("arguments[0].click();", submit_button)
time.sleep(3)
if self._check_success_message():
return SubmissionResult.SUCCESS
else:
return SubmissionResult.FAILED
except Exception:
continue
logger.warning("Кнопка отправки в модальном окне не найдена")
return SubmissionResult.FAILED
except Exception as e:
logger.error(f"Ошибка в модальном окне: {e}")
return SubmissionResult.FAILED
def _check_success_message(self) -> bool:
"""Проверка успешной отправки заявки"""
try:
success_indicators = [
"отклик отправлен",
"заявка отправлена",
"успешно отправлено",
"спасибо за отклик",
"ваш отклик получен",
"response sent",
"отклик на вакансию отправлен",
"резюме отправлено",
"откликнулись на вакансию",
]
success_selectors = [
'[data-qa*="success"]',
'[data-qa*="sent"]',
'.success-message',
'.response-sent',
'[class*="success"]',
]
for selector in success_selectors:
try:
success_element = self.driver.find_element(By.CSS_SELECTOR, selector)
if success_element and success_element.is_displayed():
logger.info(f"Найден элемент успеха: {selector} - {success_element.text}")
return True
except Exception:
continue
page_text = self.driver.page_source.lower()
for indicator in success_indicators:
if indicator in page_text:
logger.info(f"Найдено подтверждение: '{indicator}'")
return True
current_url = self.driver.current_url
if "sent" in current_url or "success" in current_url or "response" in current_url:
logger.info(f"URL указывает на успешную отправку: {current_url}")
return True
modal_disappeared = True
try:
self.driver.find_element(By.CSS_SELECTOR, '[data-qa="modal-overlay"]')
modal_disappeared = False
except NoSuchElementException:
pass
if modal_disappeared:
logger.info("Модальное окно исчезло - возможно отклик отправлен")
return True
logger.info("Подтверждение отправки не найдено")
return False
except Exception as e:
logger.error(f"Ошибка проверки успеха: {e}")
return False
class BrowserService:
@ -229,7 +531,9 @@ class BrowserService:
self._is_authenticated = True
return success
def apply_to_vacancy(self, vacancy_url: str, vacancy_name: str) -> ApplicationResult:
def apply_to_vacancy(
self, vacancy_url: str, vacancy_name: str
) -> ApplicationResult:
"""Подача заявки на вакансию"""
if not self.is_ready():
return ApplicationResult(
@ -270,4 +574,8 @@ class BrowserService:
def is_ready(self) -> bool:
"""Проверка готовности к работе"""
return self.driver is not None and self._is_authenticated and self.applicator is not None
return (
self.driver is not None
and self._is_authenticated
and self.applicator is not None
)

View File

@ -1,7 +1,3 @@
"""
🤖 Сервис для работы с Gemini AI
"""
import json
import requests
import logging
@ -49,7 +45,9 @@ class GeminiApiClient:
)
if response.status_code != 200:
logger.error(f"Ошибка API Gemini: {response.status_code}, {response.text}")
logger.error(
f"Ошибка API Gemini: {response.status_code}, {response.text}"
)
return None
result = response.json()
@ -185,7 +183,7 @@ class VacancyAnalyzer:
reasons = response.get("match_reasons", ["AI анализ выполнен"])
return self._validate_score(score), reasons
else:
logger.warning("Ошибка анализа Gemini, используем базовую фильтрацию")
logger.error("Ошибка анализа Gemini, используем базовую фильтрацию")
return self._basic_analysis(vacancy)
except Exception as e:
@ -326,7 +324,9 @@ class GeminiAIService:
"""Принятие решения о подаче заявки"""
if not self.is_available() or not self.analyzer:
score, _ = VacancyAnalyzer(None, self.resume_loader)._basic_analysis(vacancy)
score, _ = VacancyAnalyzer(None, self.resume_loader)._basic_analysis(
vacancy
)
return score >= settings.gemini.match_threshold
return self.analyzer.should_apply(vacancy)

View File

@ -1,7 +1,3 @@
"""
🔍 Сервис для работы с API HH.ru
"""
import requests
import time
from typing import List, Dict, Any, Optional
@ -12,14 +8,12 @@ from ..models.vacancy import Vacancy, SearchStats
class VacancySearcher:
"""Отвечает только за поиск вакансий"""
def __init__(self):
self.base_url = AppConstants.HH_BASE_URL
self.headers = {"User-Agent": "HH-Search-Bot/2.0 (job-search-automation)"}
def search(self, keywords: Optional[str] = None) -> List[Vacancy]:
"""Поиск вакансий через API"""
if keywords:
settings.update_search_keywords(keywords)
@ -36,7 +30,9 @@ class VacancySearcher:
break
all_vacancies.extend(page_vacancies)
print(f"📋 Найдено {len(page_vacancies)} вакансий на странице {page + 1}")
print(
f"📋 Найдено {len(page_vacancies)} вакансий на странице {page + 1}"
)
time.sleep(AppConstants.API_PAUSE_SECONDS)
@ -49,7 +45,6 @@ class VacancySearcher:
return []
def _fetch_page(self, page: int) -> List[Vacancy]:
"""Получение одной страницы результатов"""
params = self._build_search_params(page)
try:
@ -83,97 +78,54 @@ class VacancySearcher:
return []
def _build_search_params(self, page: int) -> Dict[str, str]:
"""Построение параметров поиска"""
config = settings.hh_search
search_query = QueryBuilder.build_search_query(config.keywords)
params = {
"text": search_query,
"area": config.area,
"experience": config.experience,
"per_page": str(config.per_page),
"per_page": str(min(config.per_page, 20)),
"page": str(page),
"order_by": config.order_by,
"employment": "full,part",
"schedule": "fullDay,remote,flexible",
"only_with_salary": "false",
"order_by": "publication_time",
}
return params
class QueryBuilder:
"""Отвечает за построение поисковых запросов"""
@staticmethod
def build_search_query(keywords: str) -> str:
"""Построение умного поискового запроса"""
base_queries = [
keywords,
f"{keywords} junior",
f"{keywords} стажер",
f"{keywords} начинающий",
f"{keywords} без опыта",
]
return " OR ".join(f"({query})" for query in base_queries)
@staticmethod
def suggest_keywords(base_keyword: str = "python") -> List[str]:
"""Предложения ключевых слов для поиска"""
return [
f"{base_keyword} junior",
f"{base_keyword} стажер",
f"{base_keyword} django",
f"{base_keyword} flask",
f"{base_keyword} fastapi",
f"{base_keyword} web",
f"{base_keyword} backend",
f"{base_keyword} разработчик",
f"{base_keyword} developer",
f"{base_keyword} программист",
]
return keywords
class VacancyFilter:
"""Отвечает за фильтрацию вакансий"""
EXCLUDE_KEYWORDS = [
"senior",
"lead",
"старший",
"ведущий",
"главный",
"team lead",
"tech lead",
"архитектор",
"head",
"руководитель",
"manager",
"director",
]
@staticmethod
def filter_suitable(vacancies: List[Vacancy]) -> List[Vacancy]:
"""Фильтрация подходящих вакансий"""
def filter_suitable(
vacancies: List[Vacancy], search_keywords: str = ""
) -> List[Vacancy]:
suitable = []
for vacancy in vacancies:
if VacancyFilter._is_suitable_basic(vacancy):
if VacancyFilter._is_suitable_basic(vacancy, search_keywords):
suitable.append(vacancy)
print(f"✅ После базовой фильтрации: {len(suitable)} подходящих вакансий")
return suitable
@staticmethod
def _is_suitable_basic(vacancy: Vacancy) -> bool:
"""Базовая проверка подходящести вакансии"""
def _is_suitable_basic(vacancy: Vacancy, search_keywords: str = "") -> bool:
if search_keywords and not vacancy.matches_keywords(search_keywords):
print(f"❌ Пропускаем '{vacancy.name}' - не соответствует ключевым словам")
return False
if not vacancy.has_python():
if not search_keywords and not vacancy.has_python():
print(f"❌ Пропускаем '{vacancy.name}' - нет Python")
return False
text = vacancy.get_full_text().lower()
for exclude in VacancyFilter.EXCLUDE_KEYWORDS:
for exclude in settings.get_exclude_keywords():
if exclude in text:
print(f"❌ Пропускаем '{vacancy.name}' - содержит '{exclude}'")
return False
@ -187,14 +139,12 @@ class VacancyFilter:
class VacancyDetailsFetcher:
"""Отвечает за получение детальной информации о вакансиях"""
def __init__(self):
self.base_url = AppConstants.HH_BASE_URL
self.headers = {"User-Agent": "HH-Search-Bot/2.0 (job-search-automation)"}
def get_details(self, vacancy_id: str) -> Optional[Dict[str, Any]]:
"""Получение детальной информации о вакансии"""
try:
response = requests.get(
f"{self.base_url}/vacancies/{vacancy_id}",
@ -210,7 +160,6 @@ class VacancyDetailsFetcher:
class HHApiService:
"""Главный сервис для работы с API HH.ru"""
def __init__(self):
self.searcher = VacancySearcher()
@ -219,34 +168,28 @@ class HHApiService:
self.stats = SearchStats()
def search_vacancies(self, keywords: Optional[str] = None) -> List[Vacancy]:
"""Поиск вакансий с фильтрацией"""
vacancies = self.searcher.search(keywords)
self.stats.total_found = len(vacancies)
return vacancies
def filter_suitable_vacancies(
self, vacancies: List[Vacancy], use_basic_filter: bool = True
self,
vacancies: List[Vacancy],
use_basic_filter: bool = True,
search_keywords: str = "",
) -> List[Vacancy]:
"""Фильтрация подходящих вакансий"""
if not use_basic_filter:
return vacancies
suitable = self.filter.filter_suitable(vacancies)
suitable = self.filter.filter_suitable(vacancies, search_keywords)
self.stats.filtered_count = len(suitable)
return suitable
def get_vacancy_details(self, vacancy_id: str) -> Optional[Dict[str, Any]]:
"""Получение детальной информации о вакансии"""
return self.details_fetcher.get_details(vacancy_id)
def get_search_stats(self) -> SearchStats:
"""Получение статистики поиска"""
return self.stats
def reset_stats(self) -> None:
"""Сброс статистики"""
self.stats = SearchStats()
def suggest_keywords(self, base_keyword: str = "python") -> List[str]:
"""Предложения ключевых слов для поиска"""
return QueryBuilder.suggest_keywords(base_keyword)

View File

@ -1,12 +1,7 @@
"""
🚀 HH.ru Автоматизация - Точка входа для прямого запуска
"""
from hh_bot.cli import CLIInterface
def main():
"""Главная функция"""
CLIInterface.run_application()

View File

@ -73,7 +73,7 @@ class TestSettings:
"""Тест создания настроек"""
settings = Settings()
assert settings.hh_search.keywords == "python junior"
assert settings.hh_search.keywords == "python"
assert settings.application.max_applications == 40
assert settings.browser.headless is False
@ -92,38 +92,31 @@ class TestServices:
"""Тест создания Gemini сервиса"""
service = GeminiAIService()
assert service.model == "gemini-2.0-flash"
assert service.match_threshold == 0.7
assert service is not None
def test_hh_api_service_creation(self):
"""Тест создания HH API сервиса"""
service = HHApiService()
assert service.base_url == "https://api.hh.ru"
assert service is not None
def test_gemini_basic_analysis(self):
"""Тест базового анализа Gemini"""
service = GeminiAIService()
def test_vacancy_matches_keywords(self):
"""Тест проверки соответствия ключевым словам"""
employer = Employer(id="123", name="Test Company")
experience = Experience(id="noExperience", name="Без опыта")
snippet = Snippet(requirement="Python", responsibility="Программирование")
snippet = Snippet(requirement="Python ML", responsibility="Машинное обучение")
vacancy = Vacancy(
id="test_id",
name="Python Developer",
name="ML Engineer",
alternate_url="https://test.url",
employer=employer,
experience=experience,
snippet=snippet,
)
score, reasons = service._basic_analysis(vacancy)
assert isinstance(score, float)
assert 0.0 <= score <= 1.0
assert isinstance(reasons, list)
assert len(reasons) > 0
assert vacancy.matches_keywords("python ml") is True
assert vacancy.matches_keywords("java") is False
def test_imports():