https://pastein.ru/t/lQg
Загрузка данных
import os
import re
import json
import time
import random
import requests
from bs4 import BeautifulSoup
from playwright.sync_api import sync_playwright
# ==================== НАСТРОЙКИ TELEGRAM ====================
TELEGRAM_BOT_TOKEN = "ВАШ_ТОКЕН_ОТ_BOTFATHER"
TELEGRAM_CHAT_ID = "ВАШ_CHAT_ID"
# Если Telegram требует прокси:
TG_PROXY = None
# ==================== НАСТРОЙКИ ПОИСКА И ЦЕН ====================
MAX_PRICE = 3000
# Твоя ссылка на выдачу (сортировка "По дате")
AVITO_URL = "https://www.avito.ru/rossiya/odezhda_obuv_aksessuary/aksessuary/golovnye_ubory-ASgBAgICAUT8AtgM?q=gatta&s=104"
# Варианты бренда (ищем строго в названии)
BRAND_VARIANTS = ["gutta", "гутта", "gatta", "гатта"]
# Ключевые слова кепок
CAP_KEYWORDS = ["кепк", "бейсболк", "снепбек", "snapback", "тракер", "trucker", "cap"]
# Маркеры договорной цены / цены в ЛС
NEGOTIABLE_MARKERS = ["в лс", "в личк", "договорн", "предлож", "обмен"]
# Минус-слова
STOP_WORDS = ["type", "тайп", "колготк", "чулк", "носк", "лосин", "белье", "трусы"]
# Требовать ли Авито Доставку?
REQUIRE_DELIVERY = True
# Задержка между регулярными проверками (в секундах)
MIN_DELAY = 40
MAX_DELAY = 75
TELEGRAM_SEND_DELAY = 3 # КД между отправками фото
# Пути к файлам строго рядом со скриптом
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
SEEN_FILE = os.path.join(BASE_DIR, "seen_items.json")
USER_DATA_DIR = os.path.join(BASE_DIR, "avito_profile")
def send_telegram_photo(caption: str, photo_bytes: bytes, item_url: str) -> bool:
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendPhoto"
reply_markup = {
"inline_keyboard": [
[{"text": "🔥 Забрать на Авито", "url": item_url}]
]
}
data = {
"chat_id": TELEGRAM_CHAT_ID,
"caption": caption,
"parse_mode": "HTML",
"reply_markup": json.dumps(reply_markup)
}
files = {"photo": ("cap.png", photo_bytes, "image/png")}
proxies = {"http": TG_PROXY, "https": TG_PROXY} if TG_PROXY else None
for attempt in range(1, 3):
try:
res = requests.post(url, data=data, files=files, timeout=35, proxies=proxies)
if res.status_code == 200:
return True
except Exception as e:
print(f"[TG ERROR] Попытка {attempt}/2: {e}")
time.sleep(2)
return False
def send_telegram_text(text: str):
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
payload = {
"chat_id": TELEGRAM_CHAT_ID,
"text": text,
"parse_mode": "HTML"
}
proxies = {"http": TG_PROXY, "https": TG_PROXY} if TG_PROXY else None
try:
requests.post(url, json=payload, timeout=15, proxies=proxies)
except Exception as e:
print(f"[TG ERROR] Сбой отправки текста: {e}")
def load_seen() -> set:
if os.path.exists(SEEN_FILE):
try:
with open(SEEN_FILE, "r", encoding="utf-8") as f:
return set(json.load(f))
except Exception:
return set()
return set()
def save_seen(seen_ids: set):
with open(SEEN_FILE, "w", encoding="utf-8") as f:
json.dump(list(seen_ids), f, ensure_ascii=False, indent=2)
def extract_price(text: str) -> int:
clean = re.sub(r"[^\d]", "", text)
return int(clean) if clean else 0
def parse_page(page, seen_ids: set, silent_mode: bool = False):
"""
silent_mode=True: тихое наполнение базы текущими объявлениями без отправки в TG.
silent_mode=False: боевой режим — отправка новых объявлений в TG со скриншотами.
"""
html_content = page.content()
soup = BeautifulSoup(html_content, "html.parser")
items = soup.find_all("div", {"data-marker": "item"})
print(f"\n[ИНФО] Найдено объявлений: {len(items)}")
new_found_count = 0
for item in items:
item_id = item.get("data-item-id")
if not item_id or item_id in seen_ids:
continue
# В тихом режиме просто заносим ID в базу и идем дальше
if silent_mode:
seen_ids.add(item_id)
save_seen(seen_ids)
new_found_count += 1
continue
# --- БОЕВОЙ РЕЖИМ (только для новых кепок) ---
title_el = item.find("a", {"data-marker": "item-title"})
if not title_el:
seen_ids.add(item_id)
save_seen(seen_ids)
continue
title = title_el.get_text(strip=True)
title_lower = title.lower()
link = "https://www.avito.ru" + title_el.get("href", "")
item_full_text = item.get_text(" ", strip=True).lower()
# 1. Фильтр бренда strictly в названии
has_brand_in_title = any(b in title_lower for b in BRAND_VARIANTS)
if not has_brand_in_title:
seen_ids.add(item_id)
save_seen(seen_ids)
continue
# 2. Стоп-слова
if any(sw in item_full_text for sw in STOP_WORDS):
seen_ids.add(item_id)
save_seen(seen_ids)
continue
# 3. Доставка
has_delivery = ("доставк" in item_full_text)
if REQUIRE_DELIVERY and not has_delivery:
print(f"[ПРОПУСК] {title} — нет доставки")
seen_ids.add(item_id)
save_seen(seen_ids)
continue
# 4. Цена
price_meta = item.find("meta", {"itemprop": "price"})
if price_meta and price_meta.get("content"):
price = int(price_meta["content"])
else:
price_box = item.find("strong")
price = extract_price(price_box.get_text()) if price_box else 0
is_negotiable = any(marker in item_full_text for marker in NEGOTIABLE_MARKERS) or (0 <= price <= 50)
is_regular_valid = (50 < price <= MAX_PRICE)
if is_regular_valid or is_negotiable:
price_str = f"💬 Договорная / В ЛС ({price} ₽)" if (is_negotiable and price <= 50) else f"{price} ₽"
caption = (
f"🧢 <b>НОВАЯ кепка Gutta на Авито!</b>\n\n"
f"📌 <b>Товар:</b> {title}\n"
f"💵 <b>Цена:</b> {price_str}\n"
f"🚚 <b>Доставка:</b> {'Есть' if has_delivery else 'Самовывоз'}"
)
print(f"[УСПЕХ] Найдена НОВАЯ кепка: {title} | {price_str}")
try:
card_locator = page.locator(f'div[data-marker="item"][data-item-id="{item_id}"]').first
card_locator.scroll_into_view_if_needed()
page.wait_for_timeout(800)
photo_bytes = card_locator.screenshot()
send_telegram_photo(caption, photo_bytes, link)
except Exception as e:
print(f"[SCREENSHOT ERROR] {e}")
send_telegram_text(f"{caption}\n🔗 <a href='{link}'>Открыть на Авито</a>")
time.sleep(TELEGRAM_SEND_DELAY)
seen_ids.add(item_id)
save_seen(seen_ids)
return new_found_count
def run_bot():
seen_ids = load_seen()
print(f"Запуск скрипта... В сохраненной базе: {len(seen_ids)} объявлений.")
with sync_playwright() as p:
context = p.chromium.launch_persistent_context(
user_data_dir=USER_DATA_DIR,
headless=False,
args=["--disable-blink-features=AutomationControlled"],
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
viewport={"width": 1366, "height": 768}
)
page = context.pages[0] if context.pages else context.new_page()
print("Загружаем страницу поиска Авито...")
page.goto(AVITO_URL, wait_until="domcontentloaded", timeout=45000)
page.wait_for_timeout(4000)
# ЕСЛИ БАЗА БЫЛА ПУСТАЯ (первый запуск) — ДЕЛАЕМ ТИХИЙ СКАН
if len(seen_ids) == 0:
print("[ТИХИЙ СТАРТ] База пустая. Сканируем текущую выдачу без отправки сообщений...")
scanned = parse_page(page, seen_ids, silent_mode=True)
msg = f"📦 <b>База наполнена!</b>\nЗапомнил {scanned} текущих объявлений. Теперь присылаю <u>только новые кепки</u>, которые только что выложили."
send_telegram_text(msg)
print(f"[ТИХИЙ СТАРТ] Записано {scanned} объявлений. Переходим в режим охоты за новыми!")
try:
while True:
# Случайный интервал ожидания перед следующей проверкой
sleep_interval = random.randint(MIN_DELAY, MAX_DELAY)
print(f"[OK] Ждем {sleep_interval} сек до следующей проверки...")
time.sleep(sleep_interval)
page.reload(wait_until="domcontentloaded", timeout=45000)
page.wait_for_timeout(3500)
# Теперь работает обычный боевой режим — реагирует только на свежак
parse_page(page, seen_ids, silent_mode=False)
finally:
save_seen(seen_ids)
print("[INFO] База просмотренных объявлений сохранена на диск!")
if __name__ == "__main__":
run_bot()
