Как создать автоматизированный инструмент для поиска бесплатных пробных консультаций в Telegram на Python

В сфере эзотерических услуг в Telegram существует огромное количество закрытых чатов, каналов и групп, где практикующие маги, тарологи и астрологи предлагают бесплатные пробные консультации (так называемые «пробники») для набора портфолио или отзывов. Вручную отслеживать такие предложения в десятках чатов невозможно. В этой статье мы подробно разберем, как написать собственный автоматизированный инструмент — связку Userbot-парсера (для сбора данных) и Telegram-бота (для удобного интерфейса) — на Python.

Архитектура решения: Два компонента

Согласно лучшим практикам парсинга Telegram (подтвержденным актуальными обзорами 2025–2026 гг.), надежная система состоит из двух раздельных частей:

• Userbot (Клиентский аккаунт / Telethon): Подключается к вашему личному аккаунту Telegram. Он «слушает» целевые чаты/каналы в реальном времени или парсит историю. Только Userbot имеет доступ к закрытым группам и может обходить ограничения Bot API.
• Bot (Ботовый аккаунт / Aiogram или python-telegram-bot): Интерфейс для вас (заказчика). Принимает команды, настройки фильтров, выдает найденные консультации, ведет базу данных.

Взаимодействие: Userbot находит сообщение → Анализирует его (ключевые слова, LLM) → Сохраняет в БД → Отправляет сигнал Боту → Бот уведомляет вас.

Этап 1: Подготовка окружения и API

1.1 Получение API ID и API Hash

1. Зайдите на my.telegram.org под своим номером телефона.
2. Перейдите в API development tools → Create new application.
3. Сохраните api_id (число) и api_hash (строка). Они понадобятся для Userbot.

1.2 Создание Бота через @BotFather

1. Напишите @BotFather → /newbot.
2. Укажите имя и username (должен оканчиваться на bot).
3. Сохраните BOT_TOKEN (например, 123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11).

1;3 Установка библиотек

pip install telethon aiogram python-dotenv aiosqlite "openai>=1.0.0" apscheduler

• Telethon — стандарт де-факто для Userbot’ов (парсинг, прослушка событий).
• Aiogram 3.x — современный асинхронный фреймворк для бота.
• OpenAI SDK — для умной фильтрации «мусора» и извлечение сути консультации (опционально, но рекомендуется).
• Aiosqlite / SQLAlchemy — локальная БД для хранения найденных предложений.

Этап 2: Настройка конфигурации (.env)

Никогда не хардкодьте секреты. Создайте файл .env в корне проекта:

# Userbot (Ваш личный аккаунт)
API_ID=1234567
API_HASH='your_api_hash_string'
SESSION_NAME='mag_parser_session' # Файл сессии создастся автоматически

BOT_TOKEN='123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11'
ADMIN_ID=987654321 # Ваш Telegram User ID (узнать у @userinfobot)

OPENAI_API_KEY='sk-...'
OPENAI_BASE_URL='https://api.openai.com/v1' # Или http://localhost:11434/v1 для Ollama
LLM_MODEL='gpt-4o-mini' # Дешевая и быстрая модель для классификации

TARGET_CHATS=-1001234567890,-1009876543210,@public_magic_chat

Этап 3: Userbot — «Уши и Глаза» системы (Telethon)

Этот скрипт работает постоянно в фоне. Он использует Event Handlers для реакции на новые сообщения мгновенно.

3.1 Основной модуль userbot/listener.py

import os
import re
import asyncio
import logging
from telethon import TelegramClient, events
from telethon.tl.types import Message
from dotenv import load_dotenv

from database import save_consultation
from llm_filter import analyze_message
from bot_notifier import notify_admin

load_dotenv

API_ID = int(os.getenv("API_ID"))
API_HASH = os.getenv("API_HASH")
SESSION = os.getenv("SESSION_NAME")
TARGET_CHATS_RAW = os.getenv("TARGET_CHATS", "")

TARGET_ENTITIES = []
for item in TARGET_CHATS_RAW.split(','):
 item = item.strip
 if not item: continue
 try:
 TARGET_ENTITIES.append(int(item))
 except ValueError:
 TARGET_ENTITIES.append(item) # username

client = TelegramClient(SESSION, API_ID, API_HASH)

MAGIC_KEYWORDS = [
 "бесплатн", "пробн", "подарок", "за отзыв", "на отзыв", 
 "реклама", "бартер", "обмен", "набор в группу", "набор клиентов",
 "таролог", "маг", "астролог", "нумеролог", "руны", "таро", "ритуал",
 "консультация", "разбор", "расклад", "прогноз", "энергетик"
]

KEYWORD_PATTERN = re.compile('|'.join(MAGIC_KEYWORDS), re.IGNORECASE)

@client.on(events.NewMessage(chats=TARGET_ENTITIES))
async def handler(event: events.NewMessage.Event):
 message: Message = event.message
 text = message.text or message.message or ""
 
 # 1; Быстрый фильтр по ключевым словам
 if not KEYWORD_PATTERN.search(text):
 return

 logging.info(f"[Userbot] Найден кандидат в чате {message.chat_id}: {text[:50]}...")

 # 2. Умный анализ через LLM (проверяем: это точно бесплатная консультация от мага?)
 analysis = await analyze_message(text, message.chat_id, message.id)
 
 if analysis and analysis.get("is_free_consultation"):
 # 3. Сохраняем в БД
 data = {
 "chat_id": message.chat_id,
 "chat_title": getattr(message.chat, 'title', 'Unknown'),
 "msg_id": message.id,
 "sender_id": message.sender_id,
 "text": text,
 "date": message.date.isoformat,
 "summary": analysis.get("summary"),
 "contact_info": analysis.get("contact"), # username, ссылка, инструкция
 "category": analysis.get("category") # таро, руны, ритуалы и т.д.
 }
 await save_consultation(data)
 
 # 4. Уведомляем админа через основного Бота
 await notify_admin(data)
 logging.info(f"[Userbot] Консультация сохранена и отправлена админу: {message.id}")

async def main:
 logging.basicConfig(level=logging.INFO, format='%(asctime)s ⎯ %(name)s ⸺ %(levelname)s ⸺ %(message)s')
 logging.info("Userbot запущен. Слушаю целевые чаты;..")
 await client.start
 await client.run_until_disconnected

if __name__ == '__main__':
 asyncio.run(main)

Этап 4: Умная фильтрация с LLL (OpenAI / Ollama)

Простых ключевых слов недостаточно: в чатах полно спама, продаж и просто слов «бесплатно» в другом контексте. LLM отлично понимает контекст: «маг делает бесплатный расклад на отзыв» vs «купите мой курс бесплатно не могу».

4.1 Модуль llm_filter.py

import json
import os
from openai import AsyncOpenAI
from dotenv import load_dotenv

load_dotenv
client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"), base_url=os.getenv("OPENAI_BASE_URL"))
MODEL = os.getenv("LLM_MODEL", "gpt-4o-mini")

SYSTEM_PROMPT = """
Ты — классификатор сообщений из эзотерических чатов Telegram.
Твоя задача: определить, является ли сообщение ПРЕДЛОЖЕНИЕМ БЕСПЛАТНОЙ КОНСУЛЬТАЦИИ от практикующего мага/таролога/астролога.
Игнорируй: продажи курсов, платные услуги, просьбы о помощи, спам, отзывы клиентов, рекламу каналов.

Отвечай СТРОГО в JSON формате:
{
 "is_free_consultation": boolean,
 "summary": "Краткая суть (1-2 предложения): что предлагают, на каких условиях",
 "contact": "Как связаться (username @..., ссылка t.me/..., 'в лс', 'пиши в комментариях')",
 "category": "Категория: таро / руны / астрология / нумерология / ритуалы / магия / другое"
}
"""

async def analyze_message(text: str, chat_id: int, msg_id: int) -> dict | None:
 if not text or len(text) < 10:
 return None
 
 try:
 response = await client.chat.completions.create(
 model=MODEL,
 messages=[
 {"role": "system", "content": SYSTEM_PROMPT},
 {"role": "user", "content": f"Chat ID: {chat_id}, Msg ID: {msg_id}
Text:
{text[:3000]}"}
 ],
 response_format={"type": "json_object"},
 temperature=0.1,
 max_tokens=300
 ) result = json.loads(response.choices[0].message.content)
 return result
 except Exception as e:
 print(f"[LLM Error] {e}")
 return None

Этап 5: База данных (SQLite + aiosqlite)

5.1 Модуль database.py

import aiosqlite
import os
from dotenv import load_dotenv

load_dotenv
DB_PATH = os.getenv("DB_PATH", "consultations.db")

async def init_db:
 async with aiosqlite.connect(DB_PATH) as db:
 await db.execute("""
 CREATE TABLE IF NOT EXISTS consultations (
 id INTEGER PRIMARY KEY AUTOINCREMENT,
 chat_id INTEGER NOT NULL,
 chat_title TEXT,
 msg_id INTEGER NOT NULL,
 sender_id INTEGER,
 text TEXT NOT NULL,
 date TEXT NOT NULL,
 contact_info TEXT,
 category TEXT,
 status TEXT DEFAULT 'new', -- new, viewed, contacted, done
 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
 UNIQUE(chat_id, msg_id)
 )
 """)
 await db.commit

async def save_consultation(data: dict):
 async with aiosqlite.connect(DB_PATH) as db:
 try:
 await db.execute("""
 INSERT OR IGNORE INTO consultations 
 (chat_id, chat_title, msg_id, sender_id, text, date, summary, contact_info, category)
 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
 """, (
 data["chat_id"], data["chat_title"], data["msg_id"], data["sender_id"],
 data["text"], data["date"], data["summary"], data["contact_info"], data["category"]
 ))
 await db.commit
 except Exception as e:
 print(f"[DB Error] {e}")

async def get_consultations(status: str = 'new', limit: int = 10):
 async with aiosqlite.connect(DB_PATH) as db:
 db.row_factory = aiosqlite.Row
 cursor = await db.execute(
 "SELECT * FROM consultations WHERE status = ? ORDER BY date DESC LIMIT ?", 
 (status, limit)
 ) return await cursor.fetchall

async def update_status(msg_id: int, chat_id: int, new_status: str):
 async with aiosqlite.connect(DB_PATH) as db:
 await db.execute("UPDATE consultations SET status = ? WHERE msg_id = ? AND chat_id = ?", (new_status, msg_id, chat_id))
 await db.commit

Этап 6: Бот-интерфейс (Aiogram 3.x)

Бот нужен для удобного просмотра найденных вариантов, управления статусами и настройки парсера без перезапуска кода.

6.1 Основной файл bot/main.py

import asyncio
import logging
import os
from aiogram import Bot, Dispatcher, types, F
from aiogram.filters import Command
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton
from aiogram.enums import ParseMode
from dotenv import load_dotenv
from database import init_db, get_consultations, update_status
from userbot.listener import main as run_userbot # Импорт функции запуска юзербота

load_dotenv
BOT_TOKEN = os.getenv("BOT_TOKEN")
ADMIN_ID = int(os.getenv("ADMIN_ID"))
dp = Dispatcher

def get_consult_kb(chat_id: int, msg_id: int, status: str):
 buttons = []
 if status == 'new':
 buttons.append([InlineKeyboardButton(text="✅ Взял в работу", callback_data=f"status_{chat_id}_{msg_id}_contacted")])
 if status != 'done':
 buttons.append([InlineKeyboardButton(text="🏁 Завершено", callback_data=f"status_{chat_id}_{msg_id}_done")])
 buttons.append([InlineKeyboardButton(text="🗑 Скрыть", callback_data=f"status_{chat_id}_{msg_id}_hidden")])
 return InlineKeyboardMarkup(inline_keyboard=buttons)

@dp.message(Command("start"))
async def cmd_start(message: types.Message):
 if message.from_user.id != ADMIN_ID:
 return await message.answer("Доступ запрещен.")
 await message.answer("🔮 Маг-Бот запущен.
Используй /list для просмотра новых консультаций.")

@dp.message(Command("list"))
async def cmd_list(message: types.Message):
 if message.from_user.id != ADMIN_ID: return
 items = await get_consultations('new', limit=5)
 if not items:
 return await message.answer("Новых бесплатных консультаций не найдено 🎉")
 
 for item in items:
 text = (
 f"📢 Источник: {item['chat_title']} ({item['chat_id']})
"
 f"🔮 Категория: {item['category'] or 'не определена'}
"
 f"📝 Суть: {item['summary'] or 'нет резюме'}
" f"📞 Контакт: {item['contact_info'] or 'не указан'}
"
 f"📅 Дата: {item['date']}
"
 f"🔗 Ссылка: Перейти к сообщению"
 )
 await message.answer(text, reply_markup=get_consult_kb(item['chat_id'], item['msg_id'], item['status']), disable_web_page_preview=True)

@dp.callback_query(F.data.startswith("status_"))
async def cb_status(callback: types.CallbackQuery):
 if callback.from_user.id != ADMIN_ID: return
 _, chat_id, msg_id, new_status = callback.data.split("_")
 await update_status(int(msg_id), int(chat_id), new_status)
 await callback.answer(f"Статус обновлен: {new_status}")
 # Обновляем клавиатуру или удаляем сообщение
 if new_status in ['done', 'hidden']:
 await callback.message.delete
 else:
 await callback.message.edit_reply_markup(reply_markup=get_consult_kb(int(chat_id), int(msg_id), new_status))

async def notify_admin(data: dict):
 """Вызывается из userbot/listener.py"""
 text = (
 f"🚨 НОВАЯ БЕСПЛАТНАЯ КОНСУЛЬТАЦИЯ!

"
 f"📢 Чат: {data['chat_title']}
"
 f"🔮 Тип: {data['category']}
"
 f"📝 Суть: {data['summary']}
"
 f"📞 Как связаться: {data['contact_info']}
"
 f"🔗 Открыть оригинал"
 )
 try:
 await bot.send_message(ADMIN_ID, text, reply_markup=get_consult_kb(data['chat_id'], data['msg_id'], 'new'), disable_web_page_preview=True)
 except Exception as e:
 logging.error(f"[Bot Notify Error] {e}")

async def main:
 await init_db
 logging.basicConfig(level=logging.INFO)
 
 # Запускаем Userbot как фоновую задачу
 asyncio.create_task(run_userbot)
 
 # Запускаем поллинг Бота
 print("Bot started polling...")
 await dp.start_polling(bot)

if __name__ == "__main__":
 try:
 asyncio.run(main)
 except KeyboardInterrupt:
 print("Stopped")

Важно: В файле userbot/listener.py импорт from bot_notifier import notify_admin должен указывать на функцию notify_admin из bot/main.py. Лучше вынести эту функцию в отдельный модуль notifier.py, чтобы избежать циклических импортов.

Этап 7: Развертывание на сервере (VPS / Cloud)

Скрипты должны работать 24/7. Варианты:

1. Systemd (Linux VPS) — Классика: Создайте сервис /etc/systemd/system/magic_bot.service.
2. Docker / Docker Compose — Рекомендуемый стандарт 2026: Изолирует окружение, упрощает обновления.
3. Облачные парсеры (Webbee.ai и аналоги): Если не хотите возиться с сервером, можно задеплоить только логику парсинга в облако, а бота держать локально или на бесплатном хостинге (Railway, Render, Fly.io имеют бесплатные тиры).

Пример docker-compose.yml

version: '3.8'
services:
 magic-bot:
 build: .
 container_name: magic_bot_app
 restart: unless-stopped
 env_file: ;env
 volumes:
 ⎯ ./data:/app/data # Для БД и сессии Telethon
 # Если используете локальный LLM (Ollama), раскомментируйте:
 # extra_hosts:
 # ⸺ "host.docker.internal:host-gateway"

Dockerfile

FROM python:3.11-slim
WORKDIR /app
RUN apt-get update && apt-get install -y --no-install-recommends gcc libpq-dev && rm -rf /var/lib/apt/lists/*
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["python", "bot/main.py"]

Этап 8: Поиск целевых чатов (Откуда парсить?)

Бот бесполезен без источников. Как найти чаты, где маги дают бесплатные консультации?

• Поиск в Telegram: Ключевые фразы: «бесплатный расклад на отзыв», «маги чат», «тарологи общение», «пробная консультация магия», «бартер эзотерика».
• Каталоги каналов: TgStat, Telemetr, Telega.in — ищите по тегам #эзотерика, #таро, #магия. Смотрите чаты-клубы, а не только каналы (в каналах нет диалогов).
• Парсинг участников: Найдите 5-10 крупных каналов тарологов. Спарсьте их участников (Telethon позволяет это сделать). Посмотрите, в каких чатах состоят эти люди (через общие чаты) — так найдете закрытые комьюнити.
• Конкурентный разум: Добавьте в целевые чаты ботов-парсеров конкурентов (если они публичны) или просто мониторьте чаты, где рекламируют «набор в группу на бесплатную диагностику».

⚠️ Критически важные нюансы безопасности и этики

Чек-лист запуска (Summary)

1. [ ] Куплен/подготовлен отдельный номер для Userbot-аккаунта (не основной!).
2. [ ] Получены API_ID, API_HASH на my.telegram.org.
3. [ ] Создан Бот через @BotFather, получен BOT_TOKEN.
4. [ ] Установлен Python 3.10+, библиотеки из requirements.txt.
5. [ ] Настроен .env со всеми ключами и TARGET_CHATS.
6. [ ] Запущен python bot/main.py (или docker compose up -d).
7. [ ] Userbot авторизовался (в консоли введите код из Telegram при первом запуске).
8. [ ] Проверьте лог: [Userbot] Userbot запущен. Слушаю целевые чаты....
9. [ ] Напишите в бот /list — должен ответить «Новых... не найдено» (пока нет новых сообщений).
10. [ ] Напишите тестовое сообщение в один из целевых чатов от второго аккаунта: «Даю бесплатный расклад Таро на отзыв, пиши в ЛС @test_mage». Бот должен прислать уведомление.

Создание собственного бота для поиска бесплатных консультаций магов, это классическая задача на стыке OSINT (Open Source Intelligence) и автоматизации. Ключевой стек — Telethon (Userbot) + Aiogram (Bot) + LLM-фильтрация — позволяет отсеять 99% шума и получать только релевантные лиды в реальном времени.

Главное правило: начинайте малого. Подключите 5 самых активных чатов, настройте промпт LLM под свои критерии «качественной консультации», и только потом масштабируйтесь. Удачной охоты за пробниками! 🔮✨