kbones 26 мая в 13:25

Разработка Telegram-бота для мониторинга цен на Авито: пошаговое руководство

Средний

4 мин

3.2K

Привет, Хабр! Сегодня я расскажу о том, как я разработал Telegram-бота для мониторинга цен на Авито. Бот умеет отслеживать изменения цен в объявлениях и уведомлять пользователей об изменениях. В статье я поделюсь всеми этапами разработки, от проектирования до финальной реализации.

Идея создания бота появилась, когда я хотел сделать агента под свои нужды, не буду говорить какие. И дело дошло до автоматизации процесса пользованием авито.

Что умеет бот?

Поиск объявлений по различным параметрам (название, категория, город, ценовой диапазон)
Отслеживание цен в реальном времени
Уведомления при изменении цены (настраиваемый порог изменения)
Управление списком отслеживаемых объявлений
Поддержка нескольких объявлений для каждого пользователя

Технический стек

Python 3.9+
python-telegram-bot 20.7
aiohttp 3.9.1
pydantic 2.5.2
python-dotenv 1.0.0

Структура проекта

avito-monitor/
├── bot.py           # Основной файл бота
├── avito_api.py     # Класс для работы с API Авито
├── config.py        # Конфигурация
├── requirements.txt # Зависимости
└── README.md        # Документация

Шаг 1: Настройка окружения

python -m venv venv
source venv/bin/activate  # для Linux/macOS
pip install python-telegram-bot requests python-dotenv aiohttp beautifulsoup4 pydantic aiofiles

Шаг 2: Конфигурация

Создаем файл config.py для хранения настроек:

import os
from dotenv import load_dotenv

load_dotenv()

BOT_TOKEN = os.getenv('BOT_TOKEN')
AVITO_CLIENT_ID = os.getenv('AVITO_CLIENT_ID')
AVITO_CLIENT_SECRET = os.getenv('AVITO_CLIENT_SECRET')
AVITO_ACCESS_TOKEN = os.getenv('AVITO_ACCESS_TOKEN')

# Настройки мониторинга
CHECK_INTERVAL = 300  # 5 минут
MAX_ITEMS_PER_USER = 10
PRICE_CHANGE_THRESHOLD = 5  # процент изменения цены

Шаг 3: Реализация API клиента

class AvitoAPI:
    def __init__(self):
        self.base_url = AVITO_API_BASE_URL
        self.access_token = AVITO_ACCESS_TOKEN
        
    async def search_items(self, **params):
        """Поиск объявлений"""
        return await self._make_request('GET', '/items', params=params)
        
    async def get_item_details(self, item_id: str):
        """Получение деталей объявления"""
        return await self._make_request('GET', f'/items/{item_id}')

Важные особенности реализации:

Асинхронные запросы через aiohttp
Система повторных попыток при ошибках
Автоматическое обновление токена
Обработка различных HTTP-статусов

Шаг 4: Разработка Telegram бота

class AvitoBot:
    def __init__(self):
        self.api = AvitoAPI()

    async def start(self, update, context):
        """Обработчик команды /start"""
        await update.message.reply_text(
            "Привет! Я бот для мониторинга объявлений на Авито..."
        )

    async def check_prices(self, context):
        """Периодическая проверка цен"""
        for user_id, items in user_items.items():
            for item_id, last_price in items.copy().items():
                try:
                    item_details = await self.api.get_item_details(item_id)
                    current_price = float(item_details.get('price', 0))
                    if self._price_changed_significantly(last_price, current_price):
                        await self._notify_user(user_id, item_id, last_price, current_price)
                except Exception as e:
                    logger.error(f"Error checking price: {e}")

Шаг 5: Реализация поиска

Поиск реализован в двух форматах:

Простой: пользователь отправляет только поисковый запрос
Расширенный: запрос в формате "Запрос | Категория | Город | Цена от | Цена до"

async def search_items(self, update, query, category=None, location=None, 
                      price_from=None, price_to=None):
    try:
        # Получаем ID категории и локации если указаны
        category_id = await self._get_category_id(category)
        location_id = await self._get_location_id(location)

        # Выполняем поиск
        results = await self.api.search_items(
            category_id=category_id,
            location_id=location_id,
            search_query=query,
            price_from=price_from,
            price_to=price_to
        )

        await self._send_search_results(update, results)
    except Exception as e:
        logger.error(f"Search error: {e}")

Шаг 6: Система мониторинга цен

Мониторинг реализован через job_queue библиотеки python-telegram-bot:

def main():
    application = Application.builder().token(BOT_TOKEN).build()
    job_queue = application.job_queue
    job_queue.run_repeating(avito_bot.check_prices, 
                           interval=CHECK_INTERVAL, first=10)

Для простоты я использовал словари для хранения данных (In-memory хранилище):

user_items: Dict[int, Dict[str, float]] = {}  # user_id -> {item_id: last_price}

В реальном проекте лучше использовать базу данных (например, PostgreSQL или Redis).

Обработка ошибок

Важный момент - правильная обработка ошибок API:

async def _make_request(self, method, endpoint, params=None, data=None, 
                       retry_count=0):
    if retry_count >= MAX_RETRIES:
        raise Exception(f"Превышено количество попыток")

    try:
        async with session.request(method, url, params=params, 
                                 json=data) as response:
            if response.status == 401:
                await self.refresh_token()
                return await self._make_request(method, endpoint, 
                                              params, data, retry_count + 1)
            # ... остальная обработка
    except aiohttp.ClientError:
        await asyncio.sleep(RETRY_DELAY)
        return await self._make_request(method, endpoint, 
                                      params, data, retry_count + 1)

Возможные улучшения

Добавить поддержку базы данных
Реализовать систему фильтров для уведомлений
Добавить статистику изменения цен
Реализовать поддержку нескольких площадок
Добавить систему подписок на поисковые запросы

Заключение

Разработка бота была интересным опытом. Основные сложности были связаны с:

Реализацией асинхронных запросов
Правильной организацией системы уведомлений

Исходный код проекта будет доступен на GitHub

P.S. Если у вас есть вопросы или предложения по улучшению бота - пишите в комментариях!

Хабы:

Python