Compare commits

...

10 Commits

Author SHA1 Message Date
itqop c0a5d97e6c Fix Rate Limiter 2025-06-28 20:35:35 +03:00
itqop 4585d181f8 Add RateLimiter 2025-06-28 20:06:40 +03:00
itqop acaa9cae29 Fix logging output 2025-06-28 19:59:02 +03:00
itqop 092f553f84 Fix typo 2025-06-28 19:57:29 +03:00
itqop 62a3dfa445 Fix typo and logs 2025-06-28 19:51:41 +03:00
itqop 1b5193b842 Fix promt 2025-06-28 19:46:30 +03:00
itqop c2971645ad feat: add AI cover letter CLI option and update README
- Add CLI toggle for AI-generated cover letters
- Update README with cover letter documentation and examples
- Display cover letter status in launch summary
- Independent control from AI vacancy filtering
2025-06-28 19:39:23 +03:00
itqop 5973d0f70e feat: add AI-powered cover letter generation and improve modal handling
- Add automatic cover letter generation using Gemini AI
- Implement smart detection of "Add cover letter" button in modals
- Generate personalized cover letters based on real resume data and vacancy text
- Improve modal window detection with updated selectors for HH.ru
- Fix code formatting to comply with black and flake8 standards
- Add .flake8 configuration with modern standards (88 char limit)
- Handle cover letter field detection and auto-filling
- Graceful fallback to default letter if AI generation fails

Features:
- Smart cover letter button detection
- Personalized content generation via Gemini AI
- Robust error handling for cover letter functionality
- Maintains existing functionality without cover letters
2025-06-28 19:30:56 +03:00
itqop fd8da44b84 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
2025-06-28 18:58:36 +03:00
itqop 6ed9453b8e Edit gitignore file 2025-06-28 18:26:34 +03:00
16 changed files with 925 additions and 456 deletions

11
.flake8 Normal file
View File

@ -0,0 +1,11 @@
[flake8]
max-line-length = 88
extend-ignore = E203, W503
exclude =
.git,
__pycache__,
.pytest_cache,
venv,
env,
.venv,
.env

214
.gitignore vendored
View File

@ -1,196 +1,42 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
.python-version
# pipenv
Pipfile.lock
# poetry
poetry.lock
# pdm
.pdm.toml
# PEP 582
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
.venv/
venv/
ENV/
env.bak/
venv.bak/
env/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
.idea/
# VS Code
.vscode/
# Specific to this project
# =======================
# API Keys and sensitive data
.env
*.key
secrets/
# Logs
logs/
*.log
# Resume data (may contain personal info)
data/experience.txt
data/about_me.txt
data/skills.txt
# Browser data
*.profile/
downloads/
# Config files with personal data
config.local
local_config.py
# Selenium WebDriver logs
geckodriver.log
chromedriver.log
# Temporary files
tmp/
temp/
*.tmp
# Screenshots (may contain sensitive info)
screenshots/
*.png
*.jpg
*.jpeg
# Database files
*.db
*.sqlite
# OS specific
.DS_Store
Thumbs.db
desktop.ini
# IDE specific
.idea/
*.swp
*.swo
*~
.env
config.local
data/*.txt
logs/
*.log
hh_bot.log
geckodriver.log
chromedriver.log
selenium.log
*.profile/
browser_data/
user_data/
chrome_data/
session_data/
cookies/
*.session
hh_cookies.json
test_*.py
*_test.py
!test_basic.py

View File

@ -11,6 +11,7 @@
- **🔍 Умный поиск** - Находит релевантные вакансии по ключевым словам
- **🤖 ИИ-анализ** - Gemini AI оценивает соответствие резюме требованиям
- **📝 Автоматические отклики** - Отправляет заявки на подходящие вакансии
- **✉️ ИИ-сопроводительные письма** - Автоматически генерирует персональные письма для каждой вакансии
- **⚙️ Гибкая настройка** - Настраиваемые критерии поиска и фильтрации
- **📊 Детальная статистика** - Отчёты о проделанной работе
- **🛡️ Безопасность** - Имитация человеческого поведения, паузы между действиями
@ -81,6 +82,7 @@ python main.py
🎯 НАСТРОЙКА ПОИСКА:
Ключевые слова [python junior]: python разработчик
Использовать AI фильтрацию? [y/n]: y
Использовать ИИ-сопроводительные письма? [y/n]: y
Максимум заявок [40]: 25
Начать автоматизацию? [y/n]: y
@ -132,6 +134,9 @@ class AppConstants:
# ИИ анализ
DEFAULT_AI_THRESHOLD = 0.7
# Сопроводительные письма
USE_AI_COVER_LETTERS = True
# Таймауты
DEFAULT_TIMEOUT = 30
API_PAUSE_SECONDS = 0.5
@ -159,6 +164,40 @@ class AppConstants:
📈 % одобрения: 53.3%
```
## ✉️ ИИ-сопроводительные письма
### Как работает автоматическая генерация писем
1. **Обнаружение возможности** - Ищет кнопку "Добавить сопроводительное письмо"
2. **Анализ вакансии** - Извлекает требования, обязанности, название и компанию
3. **Персонализация** - Использует ваши реальные данные из файлов резюме
4. **Генерация письма** - Gemini AI создает уникальное письмо для каждой вакансии
5. **Автозаполнение** - Вставляет письмо в поле и отправляет заявку
### Особенности ИИ-писем
- **Персональные** - Каждое письмо уникально для конкретной вакансии
- **Честные** - Не придумывает несуществующий опыт
- **Человечные** - Дружелюбный тон без официоза
- **Контактные** - Всегда включает контакт "Telegram — @itqen"
- **Адаптивные** - Работает с любыми типами вакансий
### Пример сгенерированного письма
```
Добрый день!
Заинтересовался вашей вакансией Python-разработчика.
У меня есть опыт работы с Python, Django и базами данных,
что соответствует вашим требованиям.
Готов обсудить возможности сотрудничества и рассказать
подробнее о своем опыте.
С уважением,
Telegram — @itqen
```
## 📊 Статистика работы
После завершения работы бот предоставляет подробную статистику:

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,25 @@ 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}")
if settings.enable_ai_matching():
try:
from ..services.gemini_service import GeminiAIService
gemini_service = GeminiAIService()
print(f" {gemini_service.get_api_status()}")
except Exception:
pass
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 +48,52 @@ class CLIInterface:
print("⚠️ AI фильтрация недоступна (нет GEMINI_API_KEY)")
use_ai = False
max_apps_input = input(
f"Максимум заявок [{settings.application.max_applications}]: "
).strip()
use_ai_cover_letters = True
if settings.enable_ai_matching():
cover_letter_choice = input("Использовать ИИ-сопроводительные письма? [y/n]: ").lower()
use_ai_cover_letters = cover_letter_choice != "n"
else:
print("⚠️ ИИ-сопроводительные письма недоступны (нет GEMINI_API_KEY)")
use_ai_cover_letters = False
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
return keywords, use_ai, use_ai_cover_letters, 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:
@ -72,30 +103,42 @@ class CLIInterface:
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']} заявок!")
else:
print("\n😕 Заявки не были отправлены")
if settings.enable_ai_matching():
try:
from ..services.gemini_service import GeminiAIService
gemini_service = GeminiAIService()
print(f"\n{gemini_service.get_api_status()}")
except Exception:
pass
print(UIFormatter.create_separator(long=True))
@staticmethod
def run_application():
"""Главная функция запуска приложения"""
try:
cli = CLIInterface()
cli.print_welcome()
ResumeFileManager.create_sample_files()
cli.print_settings_info()
keywords, use_ai, max_apps = cli.get_user_preferences()
keywords, use_ai, use_ai_cover_letters, max_apps = cli.get_user_preferences()
settings.update_search_keywords(keywords)
settings.application.max_applications = max_apps
settings.application.use_ai_cover_letters = use_ai_cover_letters
print("\n🎯 ЗАПУСК С ПАРАМЕТРАМИ:")
print(f"🔍 Поиск: {keywords}")
print(f"🤖 AI: {'Включен' if use_ai else 'Отключен'}")
print(f"📝 ИИ-письма: {'Включены' if use_ai_cover_letters else 'Отключены'}")
print(f"📊 Максимум заявок: {max_apps}")
confirm = input("\nНачать автоматизацию? [y/n]: ").lower()

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 = False,
) -> 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,27 +60,28 @@ class BrowserConfig:
@dataclass
class ApplicationConfig:
"""Настройки подачи заявок"""
max_applications: int = AppConstants.DEFAULT_MAX_APPLICATIONS
pause_min: float = 3.0
pause_max: float = 6.0
manual_login: bool = True
use_ai_cover_letters: bool = True
@dataclass
class GeminiConfig:
"""Настройки Gemini AI"""
api_key: str = ""
model: str = AppConstants.GEMINI_MODEL
base_url: str = AppConstants.GEMINI_BASE_URL
match_threshold: float = AppConstants.DEFAULT_AI_THRESHOLD
max_requests_per_minute: int = 15
rate_limit_window_seconds: int = 61
@dataclass
class ResumeConfig:
"""Настройки резюме"""
experience_file: str = AppConstants.DEFAULT_EXPERIENCE_FILE
about_me_file: str = AppConstants.DEFAULT_ABOUT_FILE
@ -95,11 +89,9 @@ class ResumeConfig:
class ResumeFileManager:
"""Менеджер для работы с файлами резюме"""
@staticmethod
def create_sample_files() -> None:
"""Создание примеров файлов резюме"""
data_dir = Path("data")
data_dir.mkdir(exist_ok=True)
@ -108,12 +100,7 @@ class ResumeFileManager:
experience_file.write_text(
"""
Опыт работы:
- Изучаю Python уже 6 месяцев
- Прошел курсы по основам программирования
- Делал учебные проекты: калькулятор, игра в крестики-нолики
- Изучаю Django и Flask для веб-разработки
- Базовые знания SQL и работы с базами данных
- Знаком с Git для контроля версий
- ноль
""".strip(),
encoding="utf-8",
)
@ -124,11 +111,7 @@ class ResumeFileManager:
about_file.write_text(
"""
О себе:
Начинающий Python разработчик с большим желанием учиться и развиваться.
Интересуюсь веб-разработкой и анализом данных.
Быстро обучаюсь, ответственно подхожу к работе.
Готов к стажировке или junior позиции для получения практического опыта.
Хочу работать в команде опытных разработчиков и вносить вклад в интересные проекты.
Котенок.
""".strip(),
encoding="utf-8",
)
@ -139,15 +122,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 +130,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 +156,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 +163,8 @@ class UIFormatter:
class Settings:
"""Главный класс настроек"""
def __init__(self):
self._load_env()
self.hh_search = HHSearchConfig()
@ -202,7 +176,6 @@ class Settings:
self._validate_config()
def _load_env(self) -> None:
"""Загрузка переменных окружения"""
try:
from dotenv import load_dotenv
@ -211,7 +184,6 @@ class Settings:
print("💡 Установите python-dotenv для работы с .env файлами")
def _validate_config(self) -> None:
"""Валидация настроек"""
if not self.gemini.api_key:
print("⚠️ GEMINI_API_KEY не установлен в переменных окружения")
@ -222,13 +194,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():
if use_ai and self._get_ai_service().is_available():
vacancies = self._ai_filter_vacancies(vacancies)
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,48 @@ 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} "
f"(успешных: {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)
application_results.append(result)
self._log_application_result(result)
if i < len(vacancies_to_process):
if result.success:
successful_count += 1
logger.info(
f" 🎉 Успешных заявок: "
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 +183,23 @@ class AutomationOrchestrator:
)
application_results.append(error_result)
logger.info(
f"🏁 Обработка завершена. Обработано вакансий: {processed_count}, "
f"успешных заявок: {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 +217,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 +227,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 +249,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 +266,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 +285,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,31 +1,116 @@
"""
🌐 Сервис для работы с браузером
"""
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
from webdriver_manager.chrome import ChromeDriverManager
from ..config.settings import settings, AppConstants, UIFormatter
from ..models.vacancy import ApplicationResult
from ..models.vacancy import ApplicationResult, Vacancy
class SubmissionResult(Enum):
SUCCESS = "success"
FAILED = "failed"
SKIPPED = "skipped"
ALREADY_APPLIED = "already_applied"
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 +119,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 +163,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("Авторизация не завершена")
@ -119,25 +247,26 @@ class VacancyApplicator:
"заявка отправлена",
"response sent",
"уже откликнулись",
"повторно",
"чат",
]
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: Vacancy) -> ApplicationResult:
"""Подача заявки на вакансию"""
try:
truncated_name = UIFormatter.truncate_text(vacancy_name)
truncated_name = UIFormatter.truncate_text(vacancy.name)
logger.info(f"Переход к вакансии: {truncated_name}...")
self.driver.get(vacancy_url)
self.driver.get(vacancy.alternate_url)
time.sleep(3)
apply_button = self._find_apply_button()
if not apply_button:
return ApplicationResult(
vacancy_id="",
vacancy_name=vacancy_name,
vacancy_name=vacancy.name,
success=False,
error_message="Кнопка отклика не найдена",
)
@ -146,7 +275,7 @@ class VacancyApplicator:
if self._is_already_applied(button_text):
return ApplicationResult(
vacancy_id="",
vacancy_name=vacancy_name,
vacancy_name=vacancy.name,
success=False,
already_applied=True,
error_message="Уже откликались на эту вакансию",
@ -155,14 +284,49 @@ 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(vacancy)
if submit_result == SubmissionResult.SUCCESS:
logger.info("✅ Заявка успешно отправлена")
return ApplicationResult(
vacancy_id="", vacancy_name=vacancy.name, success=True
)
elif submit_result == SubmissionResult.ALREADY_APPLIED:
logger.warning("⚠️ Уже откликались на эту вакансию")
return ApplicationResult(
vacancy_id="",
vacancy_name=vacancy.name,
success=False,
already_applied=True,
error_message="Уже откликались на эту вакансию",
)
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}")
return ApplicationResult(
vacancy_id="",
vacancy_name=vacancy_name,
vacancy_name=vacancy.name,
success=False,
error_message=str(e),
)
@ -179,7 +343,276 @@ 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, vacancy: Vacancy) -> 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
submit_button = None
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:
break
except Exception:
continue
if not submit_button:
logger.warning("Кнопка отправки в модальном окне не найдена")
return SubmissionResult.FAILED
button_text = submit_button.text.strip().lower()
if self._is_already_applied(button_text):
logger.warning(
f"⚠️ Кнопка указывает что уже откликались: "
f"{submit_button.text.strip()}"
)
return SubmissionResult.ALREADY_APPLIED
self._add_cover_letter_if_possible(vacancy)
time.sleep(1)
logger.info(
f"Нажимаем кнопку отправки: "
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 as e:
logger.error(f"Ошибка в модальном окне: {e}")
return SubmissionResult.FAILED
def _add_cover_letter_if_possible(self, vacancy: Vacancy) -> None:
"""Добавление сопроводительного письма если возможно"""
try:
if not settings.application.use_ai_cover_letters:
logger.info("ИИ-сопроводительные письма отключены в настройках")
return
logger.info("Ищем кнопку сопроводительного письма...")
cover_letter_button_selectors = [
'[data-qa="add-cover-letter"]',
'button[data-qa*="cover-letter"]',
'button:contains("Добавить сопроводительное")',
'button:contains("сопроводительное")',
]
cover_letter_button = None
for selector in cover_letter_button_selectors:
try:
if selector.startswith("button:contains"):
buttons = self.driver.find_elements(By.TAG_NAME, "button")
text_to_find = selector.split('"')[1].lower()
for button in buttons:
if text_to_find in button.text.lower():
cover_letter_button = button
break
else:
cover_letter_button = self.driver.find_element(
By.CSS_SELECTOR, selector
)
if cover_letter_button:
break
except Exception:
continue
if not cover_letter_button:
logger.info("📝 Кнопка сопроводительного письма не найдена")
return
logger.info("📝 Найдена кнопка сопроводительного письма, нажимаем...")
self.driver.execute_script("arguments[0].click();", cover_letter_button)
time.sleep(2)
logger.info("Ищем поле для ввода письма...")
cover_letter_field_selectors = [
'textarea[data-qa*="cover-letter"]',
'textarea[name*="letter"]',
'textarea[placeholder*="письм"]',
'textarea[id*="letter"]',
".modal textarea",
"form textarea",
]
cover_letter_field = None
for selector in cover_letter_field_selectors:
try:
cover_letter_field = WebDriverWait(self.driver, 3).until(
EC.presence_of_element_located((By.CSS_SELECTOR, selector))
)
if cover_letter_field:
logger.info(f"Поле найдено: {selector}")
break
except Exception:
continue
if not cover_letter_field:
logger.warning("📝 Поле для сопроводительного письма не найдено")
return
logger.info("📝 Генерация сопроводительного письма...")
from ..services.gemini_service import GeminiAIService
gemini_service = GeminiAIService()
cover_letter_text = gemini_service.generate_cover_letter(vacancy)
if cover_letter_text:
logger.info("📝 Заполняем сопроводительное письмо...")
cover_letter_field.clear()
cover_letter_field.send_keys(cover_letter_text)
logger.info("✅ Сопроводительное письмо добавлено")
else:
logger.warning("📝 Не удалось сгенерировать сопроводительное письмо")
except Exception as e:
logger.warning(f"📝 Ошибка при добавлении сопроводительного письма: {e}")
logger.info("Продолжаем без сопроводительного письма")
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} - "
f"{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,17 +662,17 @@ 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: Vacancy) -> ApplicationResult:
"""Подача заявки на вакансию"""
if not self.is_ready():
return ApplicationResult(
vacancy_id="",
vacancy_name=vacancy_name,
vacancy_name=vacancy.name,
success=False,
error_message="Браузер не готов или нет авторизации",
)
return self.applicator.apply_to_vacancy(vacancy_url, vacancy_name)
return self.applicator.apply_to_vacancy(vacancy)
def add_random_pause(self) -> None:
"""Случайная пауза между действиями"""
@ -270,4 +703,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,11 +1,9 @@
"""
🤖 Сервис для работы с Gemini AI
"""
import json
import requests
import logging
import time
from typing import Dict, Optional, Tuple, List
from collections import deque
import traceback
from pathlib import Path
@ -15,6 +13,55 @@ from ..models.vacancy import Vacancy
logger = logging.getLogger(__name__)
class RateLimiter:
"""Ограничитель скорости запросов к API"""
def __init__(self, max_requests: int, window_seconds: int):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.request_times = deque()
def wait_if_needed(self) -> None:
"""Ожидание если превышен лимит запросов"""
while True:
current_time = time.time()
self._cleanup_old_requests(current_time)
if len(self.request_times) < self.max_requests:
break
oldest_request_time = self.request_times[0]
wait_time = self.window_seconds - (current_time - oldest_request_time) + 0.1
logger.info(f"⏳ Достигнут лимит {self.max_requests} запросов. "
f"Ожидание {wait_time:.1f} секунд...")
time.sleep(wait_time)
def record_request(self) -> None:
"""Записать новый запрос"""
current_time = time.time()
self._cleanup_old_requests(current_time)
self.request_times.append(current_time)
def _cleanup_old_requests(self, current_time: float) -> None:
"""Удаление старых запросов из окна"""
while (self.request_times and
current_time - self.request_times[0] >= self.window_seconds):
self.request_times.popleft()
def get_remaining_requests(self) -> int:
"""Количество оставшихся запросов в текущем окне"""
current_time = time.time()
self._cleanup_old_requests(current_time)
return max(0, self.max_requests - len(self.request_times))
def get_status(self) -> str:
"""Статус rate limiter для логирования"""
remaining = self.get_remaining_requests()
return (f"📊 API лимит: {remaining}/{self.max_requests} запросов осталось "
f"(окно {self.window_seconds}с)")
class GeminiApiClient:
"""Клиент для работы с Gemini API"""
@ -23,8 +70,17 @@ class GeminiApiClient:
self.base_url = AppConstants.GEMINI_BASE_URL
self.model = AppConstants.GEMINI_MODEL
# Инициализируем rate limiter
self.rate_limiter = RateLimiter(
max_requests=settings.gemini.max_requests_per_minute,
window_seconds=settings.gemini.rate_limit_window_seconds
)
def generate_content(self, prompt: str) -> Optional[Dict]:
"""Генерация контента через Gemini API"""
self.rate_limiter.wait_if_needed()
self.rate_limiter.record_request()
url = f"{self.base_url}/models/{self.model}:generateContent"
headers = {"Content-Type": "application/json"}
@ -39,7 +95,8 @@ class GeminiApiClient:
}
try:
logger.info("Отправка запроса к Gemini API")
status_after = self.rate_limiter.get_status()
logger.info(f"Отправка запроса к Gemini API. {status_after}")
response = requests.post(
url,
headers=headers,
@ -49,7 +106,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()
@ -86,8 +145,14 @@ class GeminiApiClient:
json_str = content[json_start:json_end]
parsed_response = json.loads(json_str)
if "match_score" in parsed_response:
score = parsed_response.get("match_score", 0)
logger.info(f"Gemini анализ завершен: {score}")
elif "cover_letter" in parsed_response:
logger.info("Gemini сгенерировал сопроводительное письмо")
else:
logger.info("Получен ответ от Gemini")
return parsed_response
else:
logger.error("JSON не найден в ответе Gemini")
@ -185,7 +250,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 +391,101 @@ 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)
def generate_cover_letter(self, vacancy: Vacancy) -> Optional[str]:
"""Генерация сопроводительного письма для вакансии"""
if not self.is_available():
logger.warning("Gemini API недоступен, используем базовое письмо")
return self._get_default_cover_letter()
try:
resume_data = self.resume_loader.load()
vacancy_text = self._get_vacancy_full_text(vacancy)
experience_text = resume_data.get("experience", "")
about_me_text = resume_data.get("about_me", "")
skills_text = resume_data.get("skills", "")
my_profile = f"""
Опыт работы:
{experience_text}
О себе:
{about_me_text}
Навыки и технологии:
{skills_text}
"""
prompt_text = (
"Напиши короткое, человечное и честное сопроводительное письмо "
"для отклика на вакансию на русском языке. Не придумывай опыт, "
"которого нет. Используй только мой реальный опыт и навыки ниже. "
"Но если какого-то опыта нет, то не пиши про это."
"Пиши по делу, дружелюбно и без официоза. Не делай письмо слишком "
"длинным. Всегда заканчивай строкой «Telegram — @itqen»."
)
prompt = f"""{prompt_text}
**Верни только JSON с ключом "cover_letter", без других пояснений.**
Пример формата вывода:
{{"cover_letter": "текст письма здесь"}}
**Вот мой опыт:**
{my_profile}
**Вот текст вакансии:**
{vacancy_text}"""
logger.info("Генерация сопроводительного письма через Gemini")
response = self.api_client.generate_content(prompt)
if response and "cover_letter" in response:
cover_letter = response["cover_letter"]
logger.info("Сопроводительное письмо сгенерировано")
return cover_letter
else:
logger.error("Не удалось получить сопроводительное письмо от Gemini")
return self._get_default_cover_letter()
except Exception as e:
logger.error(f"Ошибка генерации сопроводительного письма: {e}")
return self._get_default_cover_letter()
def _get_vacancy_full_text(self, vacancy: Vacancy) -> str:
"""Получение полного текста вакансии"""
parts = [
f"Название: {vacancy.name}",
f"Компания: {vacancy.employer.name}",
]
if vacancy.snippet.requirement:
parts.append(f"Требования: {vacancy.snippet.requirement}")
if vacancy.snippet.responsibility:
parts.append(f"Обязанности: {vacancy.snippet.responsibility}")
return "\n\n".join(parts)
def _get_default_cover_letter(self) -> str:
"""Базовое сопроводительное письмо на случай ошибки"""
return """Добрый день!
Заинтересован в данной вакансии. Готов обсудить детали и возможности сотрудничества.
С уважением,
Telegram @itqen"""
def get_api_status(self) -> str:
"""Получение статуса API лимитов"""
if not self.api_client:
return "❌ Gemini API недоступен"
return self.api_client.rate_limiter.get_status()

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():