Шаг 1: Легальность и получение доступа. Прежде всего, изучите «Соглашение об использовании API HeadHunter». Автоматизация не должна нарушать правила платформы. Для доступа необходимо зарегистрировать приложение в разделе для разработчиков (https://dev.hh.ru). Вы получите Client ID и Client Secret. Важно: определитесь с типами токенов. User Token (от имени пользователя) подходит для автоматизации действий конкретного аккаунта (например, откликов на вакансии). Client Token (беспользовательский) — для доступа к публичным данным (поиск вакансий, работодателей). Для продакшена чаще требуется именно Client Token. Храните секреты не в коде, а в защищенных хранилищах (HashiCorp Vault, AWS Secrets Manager, или, как минимум, в переменных окружения сервера).
Шаг 2: Проектирование архитектуры. Прямые вызовы API с продакшен-сервера — плохая идея. Необходимо создать слой абстракции и очереди задач. Рекомендуемая архитектура:
- Менеджер задач (например, Celery для Python или Bull для Node.js), который ставит в очередь запросы: «найти резюме по параметрам», «отправить приглашение», «собрать статистику».
- Воркер-сервис, который берет задачи из очереди, выполняет запросы к HH API и обрабатывает ответы.
- Кэширующий слой (Redis/Memcached) для временного хранения результатов частых запросов (список регионов, профессиональные роли) и соблюдения rate limit.
- База данных (PostgreSQL) для сохранения полученных структурированных данных (вакансии, резюме, отклики) и логов.
- Мониторинг (Prometheus/Grafana) для отслеживания количества запросов, ошибок API и времени ответа.
```
import requests
import time
from collections import deque
import logging
class HHAPIClient:
def __init__(self, token, requests_per_second=5):
self.token = token
self.rate_limit = requests_per_second
self.request_times = deque(maxlen=requests_per_second)
self.session = requests.Session()
self.session.headers.update({'Authorization': f'Bearer {self.token}', 'User-Agent': 'MyProdApp/1.0'})
def _respect_limit(self):
now = time.time()
if len(self.request_times) == self.rate_limit:
elapsed = now - self.request_times[0]
if elapsed < 1.0:
sleep_time = 1.0 - elapsed
time.sleep(sleep_time)
self.request_times.append(time.time())
def safe_request(self, method, url, **kwargs):
self._respect_limit()
try:
response = self.session.request(method, url, **kwargs)
response.raise_for_status()
# Проверяем заголовки на наличие оставшихся запросов (X-RateLimit-Remaining)
return response.json()
except requests.exceptions.HTTPError as e:
if e.response.status_code == 429: # Too Many Requests
retry_after = int(e.response.headers.get('Retry-After', 60))
logging.warning(f"Rate limit hit. Sleeping for {retry_after} seconds.")
time.sleep(retry_after)
return self.safe_request(method, url, **kwargs) # Рекурсивный повтор
else:
logging.error(f"HTTP error: {e}")
raise
```
Этот код реализует простой rate limiter и обработку ошибки 429.
Шаг 4: Обработка ошибок и повторные попытки. Сеть ненадежна, API может временно не работать. Используйте библиотеки с экспоненциальной backoff-задержкой (tenacity для Python, retry для JS). Всегда проверяйте статус ответа и структуру JSON. Логируйте все неудачи для последующего анализа.
Шаг 5: Работа с данными. Полученные данные (вакансии, резюме) часто имеют сложную вложенную структуру. Создайте модели данных (Pydantic модели в Python, классы в TypeScript) для валидации и нормализации. Решите, какие данные вам нужны в сыром виде (JSON), а какие в структурированном (разложенные по таблицам БД). Для поисковых запросов используйте пагинацию: обходите все страницы результатов, но с обязательными паузами между запросами.
Шаг 6: Планирование задач. Используйте планировщик (cron, Celery Beat, AWS EventBridge) для регулярных задач. Например:
- Ежедневно в 03:00: сбор новых вакансий по ключевым навыкам.
- Каждый час: проверка статуса отправленных приглашений.
- Раз в неделю: генерация отчета по рынку.
Шаг 8: Деплой и мониторинг. Разверните вашу систему на надежной инфраструктуре (облачные VM, Kubernetes). Настройте алерты в мониторинге на ключевые метрики: рост количества ошибок 4xx/5xx, приближение к rate limit, отсутствие новых данных в БД. Ведите детальный аудит-лог всех действий, особенно тех, что меняют состояние (отправка приглашения, отклика).
Пример production-задачи: Ежедневный сбор вакансий для аналитики.
```
# Задача в очереди Celery
@app.task(bind=True, max_retries=3)
def fetch_daily_vacancies(self, search_text):
client = HHAPIClient(token=os.getenv('HH_CLIENT_TOKEN'))
all_vacancies = []
page = 0
while True:
try:
data = client.safe_request('GET', 'https://api.hh.ru/vacancies',
params={'text': search_text, 'page': page, 'per_page': 100})
except Exception as exc:
self.retry(exc=exc, countdown=60) # Повтор через 60 сек
items = data.get('items', [])
if not items:
break
all_vacancies.extend(items)
# Проверяем, последняя ли это страница (API часто возвращает признак)
if page >= data.get('pages', 0) - 1:
break
page += 1
time.sleep(0.5) # Дополнительная пауза между страницами
# Сохранение в БД
save_vacancies_to_db(all_vacancies)
return len(all_vacancies)
```
Следуя этой инструкции, вы создадите не просто скрипт, а масштабируемую, надежную и законную систему автоматизации, которая станет ценным активом для вашего бизнеса, а не источником проблем.
Комментарии (5)