Введение
Хотите получать актуальные вакансии с HH.ru (HeadHunter) напрямую в свой Telegram-канал? В этой статье мы расскажем, как создать автоматизированный скрипт на Python, который будет искать вакансии по нужным вам параметрам и отправлять их в Telegram. Этот процесс можно настроить для любых вакансий и регионов, что упростит поиск работы или мониторинг вакансий для HR-менеджеров.
Шаг 1: Подготовка к работе с API HH.ru
Первым шагом будет получение доступа к API HH.ru. Для этого вам нужно зарегистрировать своё приложение и получить авторизационные данные.
- Регистрация на HH.ru
- Зайдите на сайт HeadHunter API и зарегистрируйте своё приложение.
- Получите уникальные
client_id
иclient_secret
. Эти данные необходимы для авторизации в системе.
- Настройка OAuth авторизации
Мы будем использовать OAuth для доступа к данным вакансий. Это популярный метод аутентификации, который обеспечивает безопасное получение данных без необходимости сохранять пароли.
Код для авторизации:
В скрипте вам нужно будет указать свой client_id
, client_secret
и redirect_uri
. После этого вы сможете получить авторизационный код для дальнейшего использования:
client_id = 'YOUR_CLIENT_ID'
client_secret = 'YOUR_CLIENT_SECRET'
redirect_uri = 'YOUR_REDIRECT_URI'
Получение кода авторизации:
params_auth = {
'response_type': 'code',
'client_id': client_id,
'redirect_uri': redirect_uri,
}
authorization_url = f"{auth_url}?{urlencode(params_auth)}"
webbrowser.open(authorization_url)
При открытии ссылки в браузере вам нужно будет авторизоваться и получить код, который следует использовать для получения токена доступа.
Шаг 2: Получение access_token
Для работы с API HH.ru необходим токен доступа. После получения авторизационного кода, вы можете запросить access_token
.
- Запрос токена:
Отправляем POST-запрос на сервер HH.ru для обмена кода наaccess_token
:
data_token = {
'grant_type': 'authorization_code',
'client_id': client_id,
'client_secret': client_secret,
'redirect_uri': redirect_uri,
'code': authorization_code
}
response = requests.post(token_url, data=data_token)
- Обработка ответа:
Если запрос выполнен успешно, вы получите access_token
, который будет использоваться для запросов к API.
if response.status_code == 200:
token_info = response.json()
access_token = token_info['access_token']
Шаг 3: Получение данных о работодателе
Каждая вакансия связана с работодателем, и важно иметь информацию о компании, публикующей вакансию. Мы можем использовать API для получения более детализированной информации о работодателе.
- Запрос информации о работодателе:
Используяemployer_id
, вы можете запросить данные о компании:
employer_url = 'https://api.hh.ru/employers/'
response = requests.get(employer_url + str(employer_id), headers=headers)
- Парсинг данных: Получив ответ от API, вы сможете извлечь имя работодателя, описание и рейтинг компании.
employer_name = employer_details.get('name', 'Имя недоступно')
employer_description_html = employer_details.get('description', '')
employer_description = BeautifulSoup(employer_description_html, 'html.parser').get_text().strip()
Шаг 4: Получение контактных данных
Контактные данные — это важная часть вакансии, без которых сложно связаться с работодателем. Мы можем запросить контактные данные через API по каждой вакансии.
- Запрос контактных данных:
Используя ID вакансии, запросим контактные данные:
url = f'https://api.hh.ru/vacancies/{vacancy_id}'
response = requests.get(url, headers=headers)
contacts = vacancy_details.get('contacts')
- Извлечение контактной информации: Мы можем извлечь имя контактного лица, email и номер телефона (если они указаны):
contact_name = contacts.get('name', 'Имя не указано')
contact_email = contacts.get('email', 'Email не указан')
Шаг 5: Получение и форматирование данных вакансии
Теперь давайте соберём полную информацию о вакансии, включая описание, зарплату, опыт работы, и отправим её в Telegram.
- Запрос данных вакансии:
Используем API для получения полной информации по каждой вакансии:
response = requests.get(url, headers=headers)
vacancy_details = response.json()
description = BeautifulSoup(vacancy_details.get('description', ''), 'html.parser').get_text().strip()
- Форматирование данных: Приводим данные в удобный для пользователя вид. Например, форматируем зарплату и описание:
salary_info = f"от {salary['from']} до {salary['to']} {salary['currency']}" if salary else 'Зарплата не указана'
Шаг 6: Отправка вакансии в Telegram
Теперь, когда у нас есть полная информация о вакансии, мы можем отправить её в Telegram-канал.
- Настройка Telegram-бота:
Создайте своего Telegram-бота через BotFather и получите токен:
telegram_bot_token = 'YOUR_BOT_TOKEN'
- Отправка сообщения в Telegram: Используя метод Telegram API
sendMessage
, отправим данные о вакансии в канал:
url = f"https://api.telegram.org/bot{telegram_bot_token}/sendMessage?chat_id={telegram_channel_id}&text={message}"
response = requests.get(url)
- Форматирование сообщения: Пример сообщения для отправки:
message = (
f"📢 Новая вакансия: {job_details['title']}\n"
f"💰 Зарплата: {job_details['salary']}\n"
f"📧 Контактное лицо: {job_details['contact_name']} \n 📬 Email: {job_details['contact_email']}\n"
)
Шаг 7: Автоматизация процесса
Чтобы скрипт постоянно отслеживал новые вакансии, мы добавим цикличность и временные интервалы для выполнения запросов.
- Цикличный поиск вакансий:
Добавим главный цикл, который будет отправлять запросы к API каждые несколько минут, проверяя новые вакансии:
while True:
response = requests.get(vacancies_url, headers=headers, params=params)
params['page'] += 1
time.sleep(5)
- Отслеживание отправленных вакансий: Используем файл для хранения отправленных вакансий, чтобы не отправлять одну и ту же вакансию несколько раз:
sent_vacancies_file = 'sent_vacancies.txt'
Заключение
Скрипт позволяет автоматизировать процесс поиска и публикации вакансий с HH.ru в ваш Telegram-канал. Благодаря использованию API, вы можете настроить параметры поиска под свои нужды и получать вакансии в реальном времени. Это может быть полезно как для HR-специалистов, так и для тех, кто хочет оперативно следить за новыми предложениями работы.
В итоге, мы настроили полностью автоматизированный процесс, который публикует вакансии на вахту в Москве в Telegram-канал. Благодаря интеграции с API HH.ru и Telegram-ботом, вы теперь можете получать актуальные вакансии с полным описанием, контактными данными работодателей и ссылками на оригинальные объявления, что упрощает процесс поиска работы. Бот регулярно проверяет новые вакансии, избегает повторных публикаций и гарантирует, что все объявления поступают в канал оперативно и без вашего участия. Это мощный инструмент для тех, кто ищет работу или занимается подбором персонала на вахтовые позиции.
Полный код на Python
import requests
import openpyxl
from bs4 import BeautifulSoup
import webbrowser
from urllib.parse import urlencode
import time
from telegram import Bot
import os
from datetime import datetime
from dateutil import parser
# Данные приложения (замените на свои значения)
client_id = '********************************' # Ваш Client ID
client_secret = '****************************' # Ваш Client Secret
redirect_uri = 'https://example.com/page' # Ваш Redirect URI
auth_url = 'https://hh.ru/oauth/authorize' # URL для авторизации
token_url = 'https://hh.ru/oauth/token' # URL для получения токена
vacancies_url = 'https://api.hh.ru/vacancies' # URL для получения списка вакансий
employer_url = 'https://api.hh.ru/employers/' # URL для получения данных работодателя
# Telegram настройки
telegram_bot_token = '******:************************' # Токен вашего бота
telegram_channel_id = '-1************' # ID вашего канала (можно указать username канала с @)
bot = Bot(token=telegram_bot_token)
# Параметры для поиска вакансий
params = {
'text': 'вахта', # Поисковый запрос
'area': 1, # Регион (1 - Москва)
'page': 0, # Номер страницы
'per_page': 1 # Количество вакансий на странице
}
# Файл для хранения отправленных идентификаторов вакансий
sent_vacancies_file = 'sent_vacancies.txt'
# Функция загрузки идентификаторов вакансий
def load_sent_vacancies():
if os.path.exists(sent_vacancies_file):
with open(sent_vacancies_file, 'r') as file:
return set(line.strip() for line in file)
return set()
# Функция сохранения идентификаторов вакансий
def save_sent_vacancy(vacancy_id):
with open(sent_vacancies_file, 'a') as file:
file.write(vacancy_id + '\n')
# Функция отправки сообщений в Telegram
def split_message(message, max_length=4096):
"""Разделяет сообщение на части, если оно слишком длинное."""
return [message[i:i + max_length] for i in range(0, len(message), max_length)]
def send_message_to_telegram(message):
messages = split_message(message) # Разделяем сообщение на части
for msg in messages:
url = f"https://api.telegram.org/bot{telegram_bot_token}/sendMessage?chat_id={telegram_channel_id}&text={msg}"
response = requests.get(url)
if response.status_code == 200:
print("Сообщение успешно отправлено в Telegram.")
else:
print(f"Ошибка при отправке сообщения в Telegram: {response.text}")
# Шаг 1: Получение авторизационного кода
def get_authorization_code():
params_auth = {
'response_type': 'code',
'client_id': client_id,
'redirect_uri': redirect_uri,
}
authorization_url = f"{auth_url}?{urlencode(params_auth)}"
webbrowser.open(authorization_url)
authorization_code = input("Введите код авторизации: ")
return authorization_code
# Шаг 2: Получение access_token
def get_access_token(authorization_code):
data_token = {
'grant_type': 'authorization_code',
'client_id': client_id,
'client_secret': client_secret,
'redirect_uri': redirect_uri,
'code': authorization_code
}
response = requests.post(token_url, data=data_token)
if response.status_code == 200:
token_info = response.json()
access_token = token_info['access_token']
return access_token
else:
print(f"Ошибка {response.status_code}: {response.text}")
return None
# Шаг 3: Функция для получения информации о компании
def get_employer_details(employer_id, headers):
response = requests.get(employer_url + str(employer_id), headers=headers)
if response.status_code == 200:
employer_details = response.json()
employer_name = employer_details.get('name', 'Имя недоступно')
employer_description_html = employer_details.get('description', '')
if employer_description_html is None:
employer_description = "Описание недоступно."
else:
employer_description = BeautifulSoup(employer_description_html, 'html.parser').get_text().strip()
badges = employer_details.get('badges', [])
employer_rating = ''
for badge in badges:
if badge['type'] == 'employer-hh-rating':
employer_rating = f"{badge.get('position', '')} место в рейтинге работодателей ({badge.get('year', '')})"
break
if not employer_rating:
employer_rating = 'Рейтинг не указан'
return employer_name, employer_description, employer_rating
else:
return 'Ошибка получения данных', 'Ошибка получения данных', 'Ошибка получения данных'
# Шаг 4: Функция для получения контактных данных по вакансии
def get_contacts(vacancy_id, headers):
url = f'https://api.hh.ru/vacancies/{vacancy_id}'
response = requests.get(url, headers=headers)
if response.status_code == 200:
vacancy_details = response.json()
contacts = vacancy_details.get('contacts')
if contacts:
contact_name = contacts.get('name', 'Имя не указано')
contact_email = contacts.get('email', 'Email не указан')
phones = contacts.get('phones', [])
phone_numbers = ', '.join([f"+{phone['country']} ({phone['city']}) {phone['number']}" for phone in phones]) if phones else 'Телефон не указан'
else:
contact_name = 'Имя не указано'
contact_email = 'Email не указан'
phone_numbers = 'Телефон не указан'
return contact_name, contact_email, phone_numbers
else:
print(f"Ошибка получения контактных данных: {response.status_code}")
return 'Ошибка', 'Ошибка', 'Ошибка'
# Шаг 5: Функция для получения описания вакансии, её данных и адреса
def get_job_details(vacancy_id, headers):
url = f'https://api.hh.ru/vacancies/{vacancy_id}'
response = requests.get(url, headers=headers)
if response.status_code == 200:
vacancy_details = response.json()
description_html = vacancy_details.get('description', '')
description = BeautifulSoup(description_html, 'html.parser').get_text().strip()
salary = vacancy_details.get('salary')
salary_info = f"от {salary['from']} до {salary['to']} {salary['currency']}" if salary else 'Зарплата не указана'
experience = vacancy_details.get('experience', {}).get('name', 'Опыт не указан')
employer_id = vacancy_details['employer']['id']
published_at = vacancy_details.get('published_at')
if published_at:
published_date = parser.isoparse(published_at).strftime('%Y-%m-%d %H:%M:%S')
else:
published_date = 'Дата публикации не указана'
employer_name, employer_description, employer_rating = get_employer_details(employer_id, headers)
contact_name, contact_email, phone_numbers = get_contacts(vacancy_id, headers)
return {
'title': vacancy_details['name'],
'description': description,
'salary': salary_info,
'experience': experience,
'employer_name': employer_name,
'employer_description': employer_description,
'employer_rating': employer_rating,
'contact_name': contact_name,
'contact_email': contact_email,
'phone_numbers': phone_numbers,
'alternate_url': vacancy_details['alternate_url'],
'published_date': published_date
}
else:
print(f"Ошибка получения данных вакансии: {response.status_code}")
return None
# Главный цикл
authorization_code = get_authorization_code()
access_token = get_access_token(authorization_code)
if access_token:
headers = {'Authorization': f'Bearer {access_token}'}
sent_vacancies = load_sent_vacancies() # Загружаем отправленные вакансии один раз перед циклом
while True:
response = requests.get(vacancies_url, headers=headers, params=params)
if response.status_code == 200:
vacancies = response.json().get('items', [])
for vacancy in vacancies:
vacancy_id = vacancy['id']
# Пропускаем, если вакансия уже отправлена
if vacancy_id in sent_vacancies:
continue
job_details = get_job_details(vacancy_id, headers)
if job_details:
if job_details['contact_email'] == 'Email не указан' and job_details['phone_numbers'] == 'Телефон не указан':
print(f"Пропуск вакансии: {job_details['title']} - отсутствуют контактные данные.")
continue
message = (
f"📢 Новая вакансия: {job_details['title']}\n"
f"📝 Описание: {job_details['description']}\n"
f"💰 Зарплата: {job_details['salary']}\n"
f"🔍 Опыт: {job_details['experience']}\n"
f"🏢 Работодатель: {job_details['employer_name']} \n (Рейтинг: {job_details['employer_rating']})\n"
f"📅 Дата публикации: {job_details['published_date']}\n"
f"📧 Контактное лицо: {job_details['contact_name']} \n 📬 Email: {job_details['contact_email']} \n ☎️ Телефон: {job_details['phone_numbers']}\n"
)
send_message_to_telegram(message)
save_sent_vacancy(vacancy_id) # Сохраняем ID вакансии
sent_vacancies.add(vacancy_id) # Добавляем в память
else:
print(f"Не удалось получить детали для вакансии с ID: {vacancy_id}")
params['page'] += 1
time.sleep(5) # Задержка между запросами
else:
print(f"Ошибка получения вакансий: {response.status_code}")
break
time.sleep(3600) # Задержка перед следующим циклом
else:
print("Не удалось получить access_token.")