Привет, Хабр! Сегодня я расскажу о том, как я разработал 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. Если у вас есть вопросы или предложения по улучшению бота - пишите в комментариях!
