https://pastein.ru/t/Dhg
Загрузка данных
import os
import re
import json
import time
import random
import requests
import winsound
from bs4 import BeautifulSoup
from playwright.sync_api import sync_playwright
# ==================== ДАННЫЕ TELEGRAM ====================
TELEGRAM_BOT_TOKEN = "9294467928:AAH2h5voX9sRZQBhyiNHTLYtllLVENyC5Rg"
TELEGRAM_CHAT_ID = "6777945678"
TG_PROXY = None
# ==================== ДАННЫЕ DISCORD ====================
DISCORD_WEBHOOK_URL = "https://discord.com/api/webhooks/1566143801162997931/leweU23-IIsHzo7YpijMJnS7nPxH-jmOH0vkVORrMBC7jiUrQ0znar_aI0tBfk0S9LVA"
# ==================== НАСТРОЙКИ ПОИСКА И ЦЕН ====================
MAX_PRICE = 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 = ["в лс", "в личк", "договорн", "предлож", "обмен"]
STOP_WORDS = ["type", "тайп", "колготк", "чулк", "носк", "лосин", "белье", "трусы"]
MIN_DELAY = 40
MAX_DELAY = 75
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")
SOUND_FILE = os.path.join(BASE_DIR, "notify.wav")
def play_sound():
"""Воспроизведение приятного звука на компьютере."""
try:
if os.path.exists(SOUND_FILE):
# Проигрываем notify.wav в фоновом режиме (SND_ASYNC)
winsound.PlaySound(SOUND_FILE, winsound.SND_FILENAME | winsound.SND_ASYNC)
else:
# Если файл еще не закинули — звучит системный колокольчик Windows
winsound.MessageBeep(winsound.MB_ICONASTERISK)
except Exception as e:
print(f"[SOUND ERROR] Не удалось воспроизвести звук: {e}")
def send_discord_start():
"""Стартовое уведомление в Discord."""
data = {
"embeds": [{
"title": "🚀 Бот мониторинга Gutta запущен!",
"description": "Мониторинг кепок до 3 000 ₽ активен. Ожидаем новые лоты.",
"color": 0x2ECC71 # Зеленый цвет
}]
}
try:
requests.post(DISCORD_WEBHOOK_URL, json=data, timeout=10)
except Exception as e:
print(f"[DISCORD ERROR] {e}")
def send_discord_photo(title: str, price_str: str, delivery_str: str, item_url: str, photo_bytes: bytes):
"""Отправка найденного лота в Discord со скриншотом."""
embed = {
"title": f"🧢 Найдена кепка: {title}",
"url": item_url,
"color": 0x3498DB, # Синий цвет
"fields": [
{"name": "💵 Цена", "value": price_str, "inline": True},
{"name": "🚚 Доставка", "value": delivery_str, "inline": True}
],
"image": {"url": "attachment://cap.png"}
}
payload = {"embeds": [embed]}
files = {
"payload_json": (None, json.dumps(payload), "application/json"),
"file": ("cap.png", photo_bytes, "image/png")
}
try:
requests.post(DISCORD_WEBHOOK_URL, files=files, timeout=20)
except Exception as e:
print(f"[DISCORD ERROR] Не удалось отправить лот: {e}")
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. Стоп-слова
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
delivery_str = "Есть (Авито Доставка)" if has_delivery else "Только самовывоз"
# 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> {delivery_str}"
)
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()
# 1. Проигрываем звук на ПК
play_sound()
# 2. Отправляем в Telegram
send_telegram_photo(caption, photo_bytes, link)
# 3. Отправляем в Discord
send_discord_photo(title, price_str, delivery_str, link, photo_bytes)
except Exception as e:
print(f"[SCREENSHOT ERROR] {e}")
send_telegram_text(f"{caption}\n🔗 <a href='{link}'>Открыть на Авито</a>")
time.sleep(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)} объявлений.")
# Оповещение о старте бота в оба канала
send_telegram_text("🚀 <i>Бот запущен. Мониторинг Gutta активен (TG + Discord)!</i>")
send_discord_start()
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} текущих объявлений. Теперь присылаю только новые кепки."
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()
