333 lines
13 KiB
Python
333 lines
13 KiB
Python
"""
|
||
🤖 Сервис для работы с Gemini AI
|
||
"""
|
||
|
||
import json
|
||
import requests
|
||
import logging
|
||
from typing import Dict, Optional, Tuple, List
|
||
import traceback
|
||
from pathlib import Path
|
||
|
||
from ..config.settings import settings, AppConstants
|
||
from ..models.vacancy import Vacancy
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class GeminiApiClient:
|
||
"""Клиент для работы с Gemini API"""
|
||
|
||
def __init__(self, api_key: str):
|
||
self.api_key = api_key
|
||
self.base_url = AppConstants.GEMINI_BASE_URL
|
||
self.model = AppConstants.GEMINI_MODEL
|
||
|
||
def generate_content(self, prompt: str) -> Optional[Dict]:
|
||
"""Генерация контента через Gemini API"""
|
||
url = f"{self.base_url}/models/{self.model}:generateContent"
|
||
|
||
headers = {"Content-Type": "application/json"}
|
||
params = {"key": self.api_key}
|
||
|
||
payload = {
|
||
"contents": [{"parts": [{"text": prompt}]}],
|
||
"generationConfig": {
|
||
"temperature": AppConstants.GEMINI_TEMPERATURE,
|
||
"maxOutputTokens": AppConstants.GEMINI_MAX_OUTPUT_TOKENS,
|
||
},
|
||
}
|
||
|
||
try:
|
||
logger.info("Отправка запроса к Gemini API")
|
||
response = requests.post(
|
||
url,
|
||
headers=headers,
|
||
params=params,
|
||
json=payload,
|
||
timeout=AppConstants.DEFAULT_TIMEOUT,
|
||
)
|
||
|
||
if response.status_code != 200:
|
||
logger.error(f"Ошибка API Gemini: {response.status_code}, {response.text}")
|
||
return None
|
||
|
||
result = response.json()
|
||
return self._parse_response(result)
|
||
|
||
except requests.RequestException as e:
|
||
logger.error(f"Ошибка сети при запросе к Gemini: {e}")
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"Неожиданная ошибка Gemini API: {e}")
|
||
return None
|
||
|
||
def _parse_response(self, result: Dict) -> Optional[Dict]:
|
||
"""Парсинг ответа от Gemini API"""
|
||
try:
|
||
if "candidates" not in result or not result["candidates"]:
|
||
logger.warning("Пустой ответ от Gemini")
|
||
return None
|
||
|
||
content = result["candidates"][0]["content"]["parts"][0]["text"]
|
||
return self._extract_json_from_text(content)
|
||
|
||
except (KeyError, IndexError) as e:
|
||
logger.error(f"Ошибка структуры ответа Gemini: {e}")
|
||
return None
|
||
|
||
def _extract_json_from_text(self, content: str) -> Optional[Dict]:
|
||
"""Извлечение JSON из текстового ответа"""
|
||
try:
|
||
json_start = content.find("{")
|
||
json_end = content.rfind("}") + 1
|
||
|
||
if json_start >= 0 and json_end > json_start:
|
||
json_str = content[json_start:json_end]
|
||
parsed_response = json.loads(json_str)
|
||
|
||
score = parsed_response.get("match_score", 0)
|
||
logger.info(f"Gemini анализ завершен: {score}")
|
||
return parsed_response
|
||
else:
|
||
logger.error("JSON не найден в ответе Gemini")
|
||
return None
|
||
|
||
except json.JSONDecodeError as e:
|
||
logger.error(f"Ошибка парсинга JSON от Gemini: {e}")
|
||
logger.debug(f"Контент: {content}")
|
||
return None
|
||
|
||
|
||
class ResumeDataLoader:
|
||
"""Загрузчик данных резюме из файлов"""
|
||
|
||
def __init__(self):
|
||
self._cache: Optional[Dict[str, str]] = None
|
||
|
||
def load(self) -> Dict[str, str]:
|
||
"""Загрузка данных резюме с кэшированием"""
|
||
if self._cache is not None:
|
||
return self._cache
|
||
|
||
try:
|
||
resume_data = self._load_from_files()
|
||
self._cache = resume_data
|
||
return resume_data
|
||
except Exception as e:
|
||
logger.error(f"Ошибка загрузки файлов резюме: {e}")
|
||
return self._get_default_resume_data()
|
||
|
||
def _load_from_files(self) -> Dict[str, str]:
|
||
"""Загрузка из файлов"""
|
||
resume_data = {}
|
||
|
||
file_mappings = {
|
||
"experience": settings.resume.experience_file,
|
||
"about_me": settings.resume.about_me_file,
|
||
"skills": settings.resume.skills_file,
|
||
}
|
||
|
||
for key, relative_path in file_mappings.items():
|
||
file_path = self._get_file_path(relative_path)
|
||
|
||
if file_path.exists():
|
||
resume_data[key] = file_path.read_text(encoding="utf-8")
|
||
logger.info(f"Загружен {key} из {file_path}")
|
||
else:
|
||
resume_data[key] = self._get_default_value(key)
|
||
logger.warning(f"Файл {file_path} не найден, используем заглушку")
|
||
|
||
return resume_data
|
||
|
||
def _get_file_path(self, relative_path: str) -> Path:
|
||
"""Получение абсолютного пути к файлу"""
|
||
if Path(relative_path).is_absolute():
|
||
return Path(relative_path)
|
||
return Path.cwd() / relative_path
|
||
|
||
def _get_default_value(self, key: str) -> str:
|
||
"""Получение значений по умолчанию"""
|
||
defaults = {
|
||
"experience": "Без опыта работы. Начинающий разработчик.",
|
||
"about_me": "Начинающий Python разработчик, изучающий программирование.",
|
||
"skills": "Python, SQL, Git, основы веб-разработки",
|
||
}
|
||
return defaults.get(key, "Не указано")
|
||
|
||
def _get_default_resume_data(self) -> Dict[str, str]:
|
||
"""Полный набор данных по умолчанию"""
|
||
return {
|
||
"experience": self._get_default_value("experience"),
|
||
"about_me": self._get_default_value("about_me"),
|
||
"skills": self._get_default_value("skills"),
|
||
}
|
||
|
||
|
||
class VacancyAnalyzer:
|
||
"""Анализатор соответствия вакансий"""
|
||
|
||
def __init__(self, api_client: GeminiApiClient, resume_loader: ResumeDataLoader):
|
||
self.api_client = api_client
|
||
self.resume_loader = resume_loader
|
||
self.match_threshold = settings.gemini.match_threshold
|
||
|
||
def analyze(self, vacancy: Vacancy) -> Tuple[float, List[str]]:
|
||
"""Анализ соответствия вакансии резюме"""
|
||
try:
|
||
resume_data = self.resume_loader.load()
|
||
prompt = self._create_prompt(vacancy, resume_data)
|
||
|
||
response = self.api_client.generate_content(prompt)
|
||
|
||
if response and "match_score" in response:
|
||
score = float(response["match_score"])
|
||
reasons = response.get("match_reasons", ["AI анализ выполнен"])
|
||
return self._validate_score(score), reasons
|
||
else:
|
||
logger.warning("Ошибка анализа Gemini, используем базовую фильтрацию")
|
||
return self._basic_analysis(vacancy)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка в анализе Gemini: {e}")
|
||
logger.debug(f"Traceback: {traceback.format_exc()}")
|
||
return self._basic_analysis(vacancy)
|
||
|
||
def _create_prompt(self, vacancy: Vacancy, resume_data: Dict[str, str]) -> str:
|
||
"""Создание промпта для анализа соответствия"""
|
||
prompt = f"""
|
||
Проанализируй соответствие между резюме кандидата и вакансией.
|
||
Верни ТОЛЬКО JSON с такой структурой:
|
||
{{
|
||
"match_score": 0.85,
|
||
"match_reasons": ["причина1", "причина2"],
|
||
"recommendation": "стоит откликаться"
|
||
}}
|
||
|
||
РЕЗЮМЕ КАНДИДАТА:
|
||
Опыт работы: {resume_data.get('experience', 'Не указан')}
|
||
О себе: {resume_data.get('about_me', 'Не указано')}
|
||
Навыки: {resume_data.get('skills', 'Не указаны')}
|
||
|
||
ВАКАНСИЯ:
|
||
Название: {vacancy.name}
|
||
Компания: {vacancy.employer.name}
|
||
Требования: {vacancy.snippet.requirement or 'Не указаны'}
|
||
Обязанности: {vacancy.snippet.responsibility or 'Не указаны'}
|
||
Опыт: {vacancy.experience.name}
|
||
|
||
Оцени соответствие от {AppConstants.MIN_AI_SCORE} до {AppConstants.MAX_AI_SCORE}, где:
|
||
- 0.0-0.3: Не подходит
|
||
- 0.4-0.6: Частично подходит
|
||
- 0.7-0.9: Хорошо подходит
|
||
- 0.9-1.0: Отлично подходит
|
||
|
||
В match_reasons укажи конкретные причины оценки.
|
||
В recommendation: "стоит откликаться", "стоит подумать" или "не стоит откликаться".
|
||
"""
|
||
return prompt.strip()
|
||
|
||
def _validate_score(self, score: float) -> float:
|
||
"""Валидация оценки AI"""
|
||
return max(AppConstants.MIN_AI_SCORE, min(AppConstants.MAX_AI_SCORE, score))
|
||
|
||
def _basic_analysis(self, vacancy: Vacancy) -> Tuple[float, List[str]]:
|
||
"""Базовый анализ без AI (фолбэк)"""
|
||
score = 0.0
|
||
reasons = []
|
||
|
||
try:
|
||
|
||
if vacancy.has_python():
|
||
score += 0.4
|
||
reasons.append("Содержит Python в требованиях")
|
||
else:
|
||
reasons.append("Не содержит Python в требованиях")
|
||
return 0.1, reasons
|
||
|
||
if vacancy.is_junior_level():
|
||
score += 0.3
|
||
reasons.append("Подходящий уровень (junior)")
|
||
elif vacancy.experience.id in ["noExperience", "between1And3"]:
|
||
score += 0.2
|
||
reasons.append("Приемлемый опыт работы")
|
||
|
||
if not vacancy.has_test:
|
||
score += 0.1
|
||
reasons.append("Без обязательного тестирования")
|
||
else:
|
||
reasons.append("Есть обязательное тестирование")
|
||
|
||
if not vacancy.archived:
|
||
score += 0.1
|
||
reasons.append("Актуальная вакансия")
|
||
|
||
return min(score, 1.0), reasons
|
||
except Exception as e:
|
||
logger.error(f"Ошибка базового анализа: {e}")
|
||
return 0.0, ["Ошибка анализа"]
|
||
|
||
def should_apply(self, vacancy: Vacancy) -> bool:
|
||
"""Принятие решения о подаче заявки"""
|
||
try:
|
||
score, reasons = self.analyze(vacancy)
|
||
|
||
vacancy.ai_match_score = score
|
||
vacancy.ai_match_reasons = reasons
|
||
|
||
should_apply = score >= self.match_threshold
|
||
|
||
if should_apply:
|
||
logger.info(f"Рекомендуется откликаться (score: {score:.2f})")
|
||
else:
|
||
logger.info(f"Не рекомендуется откликаться (score: {score:.2f})")
|
||
|
||
return should_apply
|
||
except Exception as e:
|
||
logger.error(f"Ошибка в should_apply: {e}")
|
||
return False
|
||
|
||
|
||
class GeminiAIService:
|
||
"""Главный сервис для анализа вакансий с помощью Gemini AI"""
|
||
|
||
def __init__(self):
|
||
self.api_key = settings.gemini.api_key
|
||
|
||
if self.api_key:
|
||
self.api_client = GeminiApiClient(self.api_key)
|
||
else:
|
||
self.api_client = None
|
||
|
||
self.resume_loader = ResumeDataLoader()
|
||
|
||
if self.api_client:
|
||
self.analyzer = VacancyAnalyzer(self.api_client, self.resume_loader)
|
||
else:
|
||
self.analyzer = None
|
||
|
||
def is_available(self) -> bool:
|
||
"""Проверка доступности сервиса"""
|
||
return bool(self.api_key)
|
||
|
||
def load_resume_data(self) -> Dict[str, str]:
|
||
"""Загрузка данных резюме из файлов"""
|
||
return self.resume_loader.load()
|
||
|
||
def analyze_vacancy_match(self, vacancy: Vacancy) -> Tuple[float, List[str]]:
|
||
"""Анализ соответствия вакансии резюме"""
|
||
if not self.is_available() or not self.analyzer:
|
||
logger.warning("Gemini API недоступен, используем базовую фильтрацию")
|
||
return VacancyAnalyzer(None, self.resume_loader)._basic_analysis(vacancy)
|
||
|
||
return self.analyzer.analyze(vacancy)
|
||
|
||
def should_apply(self, vacancy: Vacancy) -> bool:
|
||
"""Принятие решения о подаче заявки"""
|
||
if not self.is_available() or not self.analyzer:
|
||
|
||
score, _ = VacancyAnalyzer(None, self.resume_loader)._basic_analysis(vacancy)
|
||
return score >= settings.gemini.match_threshold
|
||
|
||
return self.analyzer.should_apply(vacancy)
|