обучает
#!/usr/bin/env python3
"""
Φ-HR-Coach v2.0 — Обучает любого под вакансию за пол дня.
Теперь с YandexGPT, адаптивным обучением и исправленными ошибками.
"""
import os
import re
import json
import hashlib
import time
import asyncio
from typing import Dict, List, Optional
from dataclasses import dataclass, field
from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command
from aiogram.types import Message
# Настоящий клиент YandexGPT (если ключа нет — работает в автономном режиме)
try:
from yandex_gpt import YandexGPT
HAS_YANDEX = True
except ImportError:
HAS_YANDEX = False
print("⚠️ YandexGPT не установлен. Работаем без AI-генерации.")
# ================================================================
# НАСТРОЙКИ
# ================================================================
BOT_TOKEN = os.getenv("BOT_TOKEN")
YANDEX_API_KEY = os.getenv("YANDEX_API_KEY")
DATA_DIR = "./hr_coach_data"
os.makedirs(DATA_DIR, exist_ok=True)
if HAS_YANDEX and YANDEX_API_KEY:
yandex = YandexGPT(api_key=YANDEX_API_KEY)
else:
yandex = None
bot = Bot(token=BOT_TOKEN)
dp = Dispatcher()
# ================================================================
# ЭТАЛОН ВАКАНСИИ (ROM-образ)
# ================================================================
VACANCY = {
"title": "Системный администратор розничной сети",
"skills": [
{"name": "Сети TCP/IP", "keywords": ["tcp/ip", "ip-адрес", "шлюз", "dns", "dhcp", "vlan", "маршрут", "switch", "коммутатор"], "weight": 5},
{"name": "MikroTik", "keywords": ["mikrotik", "routeros", "winbox", "hotspot", "firewall", "queue"], "weight": 5},
{"name": "Кассы и POS", "keywords": ["касса", "ккт", "онлайн-касса", "frontol", "atol", "эквайринг", "pos-терминал"], "weight": 5},
{"name": "1С и Frontol", "keywords": ["1с", "1c", "управление торговлей", "frontol", "кассир", "номенклатура"], "weight": 4},
{"name": "Windows Admin", "keywords": ["windows", "active directory", "ad", "пользователи", "групповые политики", "домен"], "weight": 4},
{"name": "Мониторинг", "keywords": ["zabbix", "мониторинг", "nagios", "графики", "алерты", "диагностика"], "weight": 3},
{"name": "Видеонаблюдение", "keywords": ["видеонаблюдение", "ip-камера", "cctv", "регистратор", "arch"], "weight": 3},
{"name": "Soft Skills", "keywords": ["общение", "помощь", "коллеги", "команда", "объяснить"], "weight": 2},
],
"tasks": [
"Настройка и обслуживание касс, 1С УТ, Frontol 6",
"Настройка видеонаблюдения и сети",
"Устранение неполадок оборудования",
"Инструктаж и поддержка персонала",
"Настройка доступа (логины, пароли, AD)",
]
}
# ================================================================
# ХРАНИЛИЩЕ ДАННЫХ
# ================================================================
@dataclass
class UserSession:
user_id: int
plan: List[dict] = field(default_factory=list)
current_topic_index: int = 0
answers: List[dict] = field(default_factory=list)
last_question: Optional[str] = None
last_topic: Optional[str] = None
sessions: Dict[int, UserSession] = {}
def save_qa(question: str, answer: str, score: int, topic: str) -> str:
"""Сохраняет Q&A пару в датасет."""
entry_id = hashlib.md5(f"{question}{answer}".encode()).hexdigest()[:8]
entry = {
"id": entry_id,
"vacancy": VACANCY["title"],
"topic": topic,
"question": question,
"answer": answer,
"user_level": score,
"timestamp": time.time(),
}
with open(f"{DATA_DIR}/qa_dataset.jsonl", "a", encoding="utf-8") as f:
f.write(json.dumps(entry, ensure_ascii=False) + "\n")
return entry_id
# ================================================================
# АНАЛИЗ ОПЫТА (ROM)
# ================================================================
def analyze_experience(text: str) -> dict:
text_lower = text.lower()
found = []
missing = []
for skill in VACANCY["skills"]:
matches = sum(1 for kw in skill["keywords"] if kw.lower() in text_lower)
if matches > 0:
found.append({"name": skill["name"], "matches": matches, "weight": skill["weight"]})
else:
missing.append(skill["name"])
total_weight = sum(s["weight"] for s in found)
max_weight = sum(s["weight"] for s in VACANCY["skills"])
score = int(total_weight / max_weight * 100) if max_weight else 0
exp_match = re.search(r"(\d+)\s*(?:лет|год|года|years?)", text_lower)
years = int(exp_match.group(1)) if exp_match else 0
verdict = "hire" if score >= 70 else "maybe" if score >= 40 else "learn"
return {
"score": score,
"years": years,
"found": found,
"missing": missing,
"verdict": verdict,
}
def generate_plan(missing_skills: List[str]) -> List[dict]:
plan = []
for skill_name in missing_skills[:5]:
skill = next((s for s in VACANCY["skills"] if s["name"] == skill_name), None)
if skill:
plan.append({
"topic": skill_name,
"priority": skill["weight"],
"questions": [
f"Что вы знаете про {skill_name}?",
f"Опишите ваш опыт работы с {skill['keywords'][0] if skill['keywords'] else skill_name}.",
f"Как быстро вы найдёте и устраните проблему с {skill['keywords'][0] if skill['keywords'] else skill_name}?",
]
})
return plan
def check_answer(answer: str, topic: str) -> int:
"""Оценивает ответ по ключевым словам темы (0-100)."""
skill = next((s for s in VACANCY["skills"] if s["name"] == topic), None)
if not skill:
return 0
answer_lower = answer.lower()
matches = sum(1 for kw in skill["keywords"] if kw.lower() in answer_lower)
max_matches = len(skill["keywords"])
score = int(matches / max_matches * 100) if max_matches else 0
return min(score, 100)
async def ai_generate_question(topic: str) -> str:
"""Генерирует вопрос с помощью YandexGPT, если доступен."""
if yandex:
try:
prompt = f"Ты — HR-эксперт. Придумай один конкретный вопрос для проверки знаний по теме '{topic}'. Вопрос должен требовать развёрнутого ответа."
response = await yandex.generate(prompt)
if response:
return response.strip()
except Exception:
pass
return f"Опишите ваш опыт работы с {topic} (чем подробнее, тем лучше)."
# ================================================================
# КОМАНДЫ БОТА
# ================================================================
@dp.message(Command("start"))
async def cmd_start(message: Message):
user_id = message.from_user.id
sessions[user_id] = UserSession(user_id=user_id)
await message.answer(
"⚡ **Φ-HR-Coach v2.0**\n\n"
"Обучаю любого под вакансию за пол дня.\n\n"
"/vacancy — показать эталон вакансии\n"
"/experience <текст> — оценить ваш опыт\n"
"/plan — показать план обучения\n"
"/train — начать тренировку по плану\n"
"/next — следующий вопрос (если отвечен текущий)\n"
"/status — статистика датасета\n"
"/ai <запрос> — задать вопрос напрямую YandexGPT (если настроен)",
parse_mode="Markdown"
)
@dp.message(Command("vacancy"))
async def cmd_vacancy(message: Message):
skills = "\n".join(f"{i+1}. {s['name']} (вес {s['weight']})" for i, s in enumerate(VACANCY["skills"]))
text = f"📋 **Эталон вакансии:** {VACANCY['title']}\n\n**Навыки:**\n{skills}"
await message.answer(text, parse_mode="Markdown")
@dp.message(Command("experience"))
async def cmd_experience(message: Message):
text = message.text.replace("/experience", "", 1).strip()
if not text:
await message.answer("❌ Опишите ваш опыт после команды.\nПример: `/experience Работал 2 года в магазине, настраивал кассы и сети`")
return
analysis = analyze_experience(text)
user_id = message.from_user.id
if user_id not in sessions:
sessions[user_id] = UserSession(user_id=user_id)
# Сохраняем план для пользователя на основе missing навыков
plan = generate_plan(analysis["missing"])
sessions[user_id].plan = plan
sessions[user_id].current_topic_index = 0
sessions[user_id].answers = []
result = f"📊 **Анализ опыта:**\n"
result += f"Оценка: {analysis['score']}%\n"
result += f"Опыт (лет): {analysis['years']}\n"
result += f"Вердикт: *{analysis['verdict'].upper()}*\n\n"
if analysis['found']:
result += f"✅ Найдено ({len(analysis['found'])}):\n"
for s in analysis['found']:
result += f" • {s['name']} ({s['matches']} совпадений)\n"
if analysis['missing']:
result += f"\n❌ Требуется изучить ({len(analysis['missing'])}):\n"
for m in analysis['missing'][:5]:
result += f" • {m}\n"
result += "\n/plan — показать план обучения."
# Сохраняем для датасета
save_qa("experience_analysis", json.dumps(analysis, ensure_ascii=False), analysis["score"], "analysis")
await message.answer(result, parse_mode="Markdown")
@dp.message(Command("plan"))
async def cmd_plan(message: Message):
user_id = message.from_user.id
session = sessions.get(user_id)
if not session or not session.plan:
await message.answer("Сначала отправьте /experience, чтобы я оценил ваш опыт.")
return
plan = session.plan
if not plan:
await message.answer("План пуст — вы уже знаете все необходимые навыки!")
return
text = "📚 **Ваш план обучения:**\n\n"
for i, p in enumerate(plan, 1):
text += f"{i}. *{p['topic']}* (приоритет: {p['priority']})\n"
text += "\n/train — начать тренировку"
await message.answer(text, parse_mode="Markdown")
@dp.message(Command("train"))
async def cmd_train(message: Message):
user_id = message.from_user.id
session = sessions.get(user_id)
if not session or not session.plan:
await message.answer("Сначала /experience, затем /plan, а потом уже /train.")
return
if session.current_topic_index >= len(session.plan):
await message.answer("🎉 Вы прошли все темы! Начните заново с /experience (можно добавить новые навыки).")
return
# Показываем первый вопрос первой темы
topic = session.plan[session.current_topic_index]["topic"]
session.last_topic = topic
# Генерируем вопрос (если есть AI – используем)
if yandex:
question = await ai_generate_question(topic)
else:
question = f"Что вы знаете о {topic}? Опишите подробно."
session.last_question = question
await message.answer(f"🧠 **Тема:** {topic}\n\n❓ {question}\n\n_Ответьте текстом на этот вопрос._", parse_mode="Markdown")
@dp.message(Command("next"))
async def cmd_next(message: Message):
user_id = message.from_user.id
session = sessions.get(user_id)
if not session or not session.plan:
await message.answer("Сначала /experience, /train.")
return
if session.current_topic_index >= len(session.plan):
await message.answer("Все темы пройдены. Поздравляю! 🎉")
return
# Если есть последний вопрос без ответа, напомним
if session.last_question is not None and session.last_topic is not None:
# Проверяем, не отвечали ли уже на этот вопрос (можно пропустить)
await message.answer(f"Сначала ответьте на текущий вопрос: {session.last_question}")
return
session.current_topic_index += 1
if session.current_topic_index >= len(session.plan):
await message.answer("✅ Вы завершили весь план обучения!")
return
topic = session.plan[session.current_topic_index]["topic"]
session.last_topic = topic
if yandex:
question = await ai_generate_question(topic)
else:
question = f"Что вы знаете о {topic}? Опишите подробно."
session.last_question = question
await message.answer(f"🧠 **Тема:** {topic}\n\n❓ {question}")
@dp.message(Command("status"))
async def cmd_status(message: Message):
dataset_path = f"{DATA_DIR}/qa_dataset.jsonl"
if not os.path.exists(dataset_path):
await message.answer("📊 Пока нет сохранённых данных. Пройдите /train и отвечайте на вопросы.")
return
with open(dataset_path, "r", encoding="utf-8") as f:
lines = f.readlines()
total = len(lines)
topics = {}
for line in lines:
try:
entry = json.loads(line)
topic = entry.get("topic", "unknown")
topics[topic] = topics.get(topic, 0) + 1
except:
pass
text = f"📊 **Статистика датасета**\nВсего Q&A пар: {total}\n\n"
for topic, cnt in sorted(topics.items(), key=lambda x: -x[1])[:10]:
text += f"• {topic}: {cnt}\n"
text += f"\nФайл: `{dataset_path}`"
await message.answer(text, parse_mode="Markdown")
@dp.message(Command("ai"))
async def cmd_ai(message: Message):
if not yandex:
await message.answer("❌ YandexGPT не настроен. Установите `yandex_gpt` и задайте YANDEX_API_KEY.")
return
prompt = message.text.replace("/ai", "", 1).strip()
if not prompt:
await message.answer("Напишите запрос после /ai, например: `/ai Чем отличается VLAN от подсети?`")
return
await message.answer("🤖 Запрос к YandexGPT... ⏳")
try:
response = await yandex.generate(prompt)
if response:
await message.answer(response[:4000])
else:
await message.answer("❌ Не удалось получить ответ от AI.")
except Exception as e:
await message.answer(f"❌ Ошибка: {e}")
# ================================================================
# ОБРАБОТКА ТЕКСТОВЫХ ОТВЕТОВ (проверка знаний)
# ================================================================
@dp.message()
async def handle_answer(message: Message):
user_id = message.from_user.id
text = message.text.strip()
if text.startswith("/"):
return # команда
session = sessions.get(user_id)
if not session or session.last_question is None or session.last_topic is None:
# Нет активного диалога
await message.answer("Не понял. Используйте команды: /start, /experience, /train")
return
# Оцениваем ответ
score = check_answer(text, session.last_topic)
# Сохраняем в датасет
qa_id = save_qa(session.last_question, text, score, session.last_topic)
# Запоминаем ответ в сессии
session.answers.append({
"topic": session.last_topic,
"question": session.last_question,
"answer": text,
"score": score,
"qa_id": qa_id
})
# Формируем обратную связь
if score >= 70:
feedback = "✅ Отлично! Вы хорошо знаете эту тему."
elif score >= 40:
feedback = "⚡ Средний уровень. Повторите ключевые слова темы."
else:
feedback = "❌ Слабый ответ. Рекомендую изучить тему по документации."
# Показываем результат
await message.answer(
f"📝 **Оценка:** {score}%\n{feedback}\n\n"
f"Тема: {session.last_topic}\n"
f"Нажмите /next для следующего вопроса или /train для повторного прохождения."
)
# Сбрасываем последний вопрос, чтобы можно было вызвать /next
session.last_question = None
session.last_topic = None
# ================================================================
# ЗАПУСК
# ================================================================
async def main():
await dp.start_polling(bot)
if __name__ == "__main__":
asyncio.run(main())
Запускай:
bash
export BOT_TOKEN=...
export YANDEX_API_KEY=...
python hr_coach.py
Без токена YandexGPT бот работает в автономном режиме (без AI-генерации).
С YandexGPT – генерирует умные вопросы и может общаться свободно.
4
В чате с Алисой замените данные в скобках [...] на свои
Опубликовано: 28 апреля 2026