https://pastein.ru/t/yPg
Загрузка данных
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 = "9294467928:AAH2h5voX9sRZQBhyiNHTLYtllLVENyC5Rg"
TELEGRAM_CHAT_ID = "6777945678"
TG_PROXY = None # Укажи прокси вида "http://127.0.0.1:1080", если в сети глушат Telegram
# ==================== НАСТРОЙКИ ПОИСКА И ЦЕН ====================
MAX_PRICE = 3000
# Твоя ссылка на поиск кепок Gutta до 3000 ₽:
AVITO_URL = "https://www.avito.ru/novosibirsk/odezhda_obuv_aksessuary/aksessuary/golovnye_ubory-ASgBAgICAkTeAtoL_MYNrNo_?cd=1&context=H4sIAAAAAAAA_wEmANn_YToxOntzOjE6InkiO3M6MTY6IjVyS1luV0N5SVhVNTRDaGwiO303Q_P7JgAAAA&f=ASgBAgECAkTeAtoL_MYNrNo_AUXGmgwUeyJmcm9tIjowLCJ0byI6MzAwMH0&localPriority=0&q=%D0%BA%D0%B5%D0%BF%D0%BA%D0%B0+gutta&s=104"
# Варианты названия бренда (проверяем строго в заголовке)
BRAND_VARIANTS = ["gutta", "гутта", "gatta", "гатта"]
# Договорная цена / в ЛС
NEGOTIABLE_MARKERS = ["в лс", "в личк", "договорн", "предлож", "обмен"]
# Минус-слова (отсекаем реплики "type" и лишние вещи)
STOP_WORDS = ["type", "тайп", "колготк", "чулк", "носк", "лосин", "белье", "трусы"]
# Задержка между регулярными проверками (в секундах)
MIN_DELAY = 40
MAX_DELAY = 75
TELEGRAM_SEND_DELAY = 3 # Пауза между отправками фото в Telegram
# Пути к файлам профиля и базы strictly в папке со скриптом
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):
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
# В тихом режиме просто запоминаем существующие кепки
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. Бренд строго в названии
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. Стоп-слова (реплики "type" и т.д.)
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
# 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",
"--window-position=-3000,-3000",
"--window-size=1366,768"
],
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()
