Всем привет! Существует такая проблема, связанная с тем, что хорошие объявления на Avito исчезают буквально за минуты. Хотите купить студию по выгодной цене? Или найти iPhone дешевле рынка? Пока вы открываете сайт — кто-то уже договорился. А стандартное APi ограничивает число запросов.

Сегодня мы соберём максимально простого помощника, который будет сам следить за новыми объявлениями на Avito и моментально отправлять вам уведомление в Телеграм. Он будет работать в фоне, не требуя лишних окон или настроек. И главное, мы сделаем так, что Авито не будет блокировать его по числу запросов, выдавая 429 ответ.

Суть работы

Наш парсер регулярно обращается к Avito API по списку ссылок, указанному в переменной AVITO_URLS. Парсер получает список объявлений, сравнивает их с ранее сохраненными и отправляет в Телеграм только те, которые ещё не были показаны. Бот присылает только новые и уникальные предложения, без дублей и повторов.

Авито может ограничивать частые запросы или временно блокировать IP адреса. Чтобы избежать подобных ограничений, я запущу скрипт в облачном сервисе Amvera, который предоставляет бесплатные ротируемые (изменяющиеся) IP адреса.

Работа с Amvera даст нам следующую возможность:

• Безопасно обращаться к API Avito без блокировок. Каждый запрос будет идти через новый IP.

• Не покупать собственные прокси.

• И, что полезно, быстрый и простой деплой через три команды в терминале.

Что такое Amvera

Amvera — облачный сервис для развёртывания IT-приложений, направленный в первую очередь на простоту и скорость деплоя. Amvera после регистрации совершенно бесплатно предоставляет 111 рублей для тестирования. Этого уже будет достаточно для некоторого времени работы нашего парсера.

Теоретическая часть

Как мы уже говорили, наш парсер будет отправлять уведомления о новых объявлениях из выбранной нами категории. Но нужно разобраться в этом подробнее: что именно и как мы парсим?

Avito API

Если зайти в любую категорию на авито и открыть вкладку Network в DevTools вашего браузера, среди множества различных запросов вы увидите тот самый API, с которым мы будем работать — это https://www.avito.ru/web/1/main/items.

Его параметры:

• locationId=637640 — регион (например, Москва);

• categoryId=24 — категория (например, квартиры);

• sort=date — сортировка по дате (самые новые в начале);

• limit=50 — количество объявлений за раз;

• page=1 — страница (можно использовать пагинацию);

• lastStamp=0 — специальный параметр, обычно оставляем 0.

Этот запрос возвращает JSON с ключом items, где каждое объявление содержит id, заголовок, цену, изображения и т.д. Именно этот API мы и используем в нашем парсере.

Хранение уже обработанных объявлений

Хранить само объявление после отправки в Телеграм мы будем с помощью сохранения его уникального ID в файл seen.json. Это обеспечит нам актуальность всех уведомлений и экономичное хранение данных.

Практическая часть. Создаём парсер на Python

Парсер у нас достаточно простой, при этом его легко доработать — теории оказалось немного. Теперь можно переходить к практике: напишем полноценный скрипт, разберём его логику и запустим в фоновом режиме.

Подробный разбор кода

Теперь рассмотрим наш скрипт по частям:

import asyncio
import aiohttp
import os
import json
from aiogram import Bot, Dispatcher
from aiogram.types import InputMediaPhoto
from aiogram.exceptions import TelegramBadRequest, TelegramNetworkError
from dotenv import load_dotenv

Импортируем все необходимые библиотеки. asyncio и aiohttp - для асинхронной загрузки, aiogram - для работы с Telegram Bot API, dotenv - для чтения переменных окружения.

load_dotenv()
TOKEN = os.getenv("TG_TOKEN")
CHAT_ID = int(os.getenv("TG_CHAT_ID"))
AVITO_URLS = os.getenv("AVITO_URLS", "").split(",")
SEEN_FILE = os.getenv("SEEN_FILE", "/data/seen.json")

Загружаем переменные окружения: токен бота, ID чата, список ссылок на API Avito и путь до файла, где мы будем хранить id уже отправленных объявлений. Здесь довольно важно — путь к seen.json должен начинаться со /data — это директория постоянного хранилища в Amvera. Если сохранять в другое место, то после пересборки все сохранённые данные пропадут.

bot = Bot(token=TOKEN)
dp = Dispatcher()

Создаём экземпляр бота и диспетчер (хотя диспетчер в этом случае мы не используем).

def load_seen_ids():
    if os.path.exists(SEEN_FILE):
        with open(SEEN_FILE, "r") as f:
            try:
                return set(json.load(f))
            except Exception:
                return set()
    return set()

Функция загрузки всех ранее сохранённых id из seen.json. Если файла нет — возвращаем пустое множество.

def save_seen_ids(ids):
    with open(SEEN_FILE, "w") as f:
        json.dump(list(ids), f)

Функция сохраняет множество id в файл.

async def fetch_avito(session, url):
    try:
        async with session.get(url.strip()) as response:
            if response.status == 200:
                data = await response.json()
                return data.get("items", [])
            else:
                print(f"[!] Ошибка {response.status} при запросе {url}")
    except Exception as e:
        print(f"[!] Ошибка при запросе {url}: {e}")
    return []

Отправляет запрос к API Avito, возвращает список новых объявлений. Если что-то пошло не так — выводит ошибку и возвращает пустой список.

async def send_ad(item):
    try:
        title = item.get("title", "Без названия")
        price = item.get("priceDetailed", {}).get("value", "Неизвестно")
        url = f"https://www.avito.ru{item.get('urlPath', '')}"
        text = f"? <b>{title}</b>\n? Цена: {price}\n? <a href="\&quot;{url}\&quot;">Открыть объявление</a>"

        images = item.get("images", [])
        if images:
            photo_url = images[0].get("864x864") or images[0].get("416x416")
            await bot.send_photo(CHAT_ID, photo_url, caption=text, parse_mode="HTML")
        else:
            await bot.send_message(CHAT_ID, text, parse_mode="HTML")

    except (TelegramBadRequest, TelegramNetworkError) as e:
        print(f"[!] Ошибка телеграм: {e}")

Формирует текстовое сообщение об объявлении и отправляет в Телеграм. Если есть фото — прикладывает его.

async def main():
    seen_ids = load_seen_ids()

    async with aiohttp.ClientSession() as session:
        all_items = []
        for url in AVITO_URLS:
            items = await fetch_avito(session, url)
            all_items.extend(items)

        new_items = [item for item in all_items if item["id"] not in seen_ids]

        for item in sorted(new_items, key=lambda x: x.get("sortDate", 0), reverse=True):
            await send_ad(item)
            seen_ids.add(item["id"])

        save_seen_ids(seen_ids)

    await bot.session.close()

Главная логика: загружаем старые id, получаем объявления по всем ссылкам, сравниваем с уже отправленными, сортируем по дате, отправляем новые объявления и сохраняем их id.

async def run_forever():
    while True:
        try:
            await main()
        except Exception as e:
            print(f"[!] Ошибка в main: {e}")
        await asyncio.sleep(60)

Функция, которая запускает main() раз в минуту - бесконечный цикл.

if __name__ == "__main__":
    asyncio.run(run_forever())

Полный код:

import asyncio
import aiohttp
import os
import json
from aiogram import Bot, Dispatcher
from aiogram.types import InputMediaPhoto
from aiogram.exceptions import TelegramBadRequest, TelegramNetworkError
from dotenv import load_dotenv

load_dotenv()

TOKEN = os.getenv("TG_TOKEN")
CHAT_ID = int(os.getenv("TG_CHAT_ID"))
AVITO_URLS = os.getenv("AVITO_URLS", "").split(",")
SEEN_FILE = os.getenv("SEEN_FILE", "/data/seen.json")

bot = Bot(token=TOKEN)
dp = Dispatcher()

def load_seen_ids():
    if os.path.exists(SEEN_FILE):
        with open(SEEN_FILE, "r") as f:
            try:
                return set(json.load(f))
            except Exception:
                return set()
    return set()

def save_seen_ids(ids):
    with open(SEEN_FILE, "w") as f:
        json.dump(list(ids), f)

async def fetch_avito(session, url):
    try:
        async with session.get(url.strip()) as response:
            if response.status == 200:
                data = await response.json()
                return data.get("items", [])
            else:
                print(f"[!] Ошибка {response.status} при запросе {url}")
    except Exception as e:
        print(f"[!] Ошибка при запросе {url}: {e}")
    return []

async def send_ad(item):
    try:
        title = item.get("title", "Без названия")
        price = item.get("priceDetailed", {}).get("value", "Неизвестно")
        url = f"https://www.avito.ru{item.get('urlPath', '')}"
        text = f"? <b>{title}</b>\n? Цена: {price}\n? <a href="\&quot;{url}\&quot;">Открыть объявление</a>"

        images = item.get("images", [])
        if images:
            photo_url = images[0].get("864x864") or images[0].get("416x416")
            await bot.send_photo(CHAT_ID, photo_url, caption=text, parse_mode="HTML")
        else:
            await bot.send_message(CHAT_ID, text, parse_mode="HTML")

    except (TelegramBadRequest, TelegramNetworkError) as e:
        print(f"[!] Ошибка телеграм: {e}")

async def main():
    seen_ids = load_seen_ids()

    async with aiohttp.ClientSession() as session:
        all_items = []
        for url in AVITO_URLS:
            items = await fetch_avito(session, url)
            all_items.extend(items)

        new_items = [item for item in all_items if item["id"] not in seen_ids]

        for item in sorted(new_items, key=lambda x: x.get("sortDate", 0), reverse=True):
            await send_ad(item)
            seen_ids.add(item["id"])

        save_seen_ids(seen_ids)

    await bot.session.close()

async def run_forever():
    while True:
        try:
            await main()
        except Exception as e:
            print(f"[!] Ошибка в main: {e}")
        await asyncio.sleep(60) 

if __name__ == "__main__":
    asyncio.run(run_forever())

Теперь скрипт будет запускаться каждые 60 секунд и проверять объявления по указанным ссылкам.

Деплой

Теперь, когда все готово, мы можем перейти к деплою приложения в Amvera:

1. Регистрируемся по ссылке и подтверждаем почту. Получаем 111 рублей на баланс для тестов.

   Открываем страницу проектов и создаем новый проект со следующими параметрами:

   1. Тип сервиса: Приложение. Жмём далее.

   2. Название проекта: по вашему желанию.

   3. Тариф: тарифа "Начальный плюс" будет достаточно для стабильной работы парсера.

   4. Все остальные шаги можно пропустить — все файлы и конфигурацию мы загрузим уже после создания проекта.

После создания проекта открываем его страницу и сразу переходим во вкладку Репозиторий. Тут нам нужно загрузить 2 файла: parser.py (код выше) и requirements.txt со следующим содержимым:

requests==2.32.4
beautifulsoup4==4.13.4
python-dotenv==1.1.1
aiogram==3.21.0
aiohttp
python-dateutil

Далее, переходим во вкладку Конфигурация и выставляем следующие параметры как на скриншоте:

После заполнения конфигурации остаётся последний штрих — создание всех необходимых переменных окружения. Это делается во вкладке Переменные.

Там мы создаем:

1. Переменную PYTHONUNBUFFERED в значении 1.

2. Секрет TG_TOKEN. В значение вставьте ваш токен.

3. Переменную TG_CHAT_ID. В значении - чат, куда бот будет присылать уведомления.

4. Переменную AVITO_URLS - тут будут все API URL, с которыми будет работать парсер в рассмотренном ранее формате.

5. Переменную SEEN_FILE с путем к seen.json. По умолчанию — /data/seen.json

Всё — теперь мы можем собирать проект во вкладке Конфигурация и ждать запуска!

Дополнительно мы можем воспользоваться встроенным горизонтальным масштабированием и запустить сервис (для бота может не подходить) в нескольких инстансах, что будет хоть и дороже, но увеличит количество используемых IP-адресов.

Готово, наш парсинг Авито работает, отсылая результаты в чат.

Итог разработки парсера Авито

В этой статье мы показали, как создать собственный парсер Avito, который отслеживает новые объявления на Avito и отправляет их прямо в чат.

Это решение хорошо подойдёт тем, кто хочет получать актуальные предложения первым. И легко расширяется под любые задачи: от фильтрации по ключевым словам до комментирования прямо в Телеграм.

А что бы добавили вы? Или, может, есть идея для следующей статьи? Напишите в комментариях, и я постараюсь подробнее разобрать аспекты парсинга в следующих статьях.