- Сообщения
- 254
- Реакции
- 1.164
- Продажи
- 14
- Кешбек
- 24.58$
Здравствуйте. Редко участвую в конкурсах, однако вот.Решил расписать вам технологию изготовления мной, специального решения автопродаж. Написано с нуля, и решает все основные вопросы, которые нужны для подобного типа решений.Это будут бот(ы) для телеграм и удобная веб админка для управления всем функционалом.
Присутствует защита от блокировок. А также рассылка от ботов, и от юзеров,чтобы сохранять клиентов всегда. Оплата принимается ботов на автомате с помощью usdt trc-20.
И другие плюшки!
Глава 1. Инструментарий.
В качестве основного языка взят, идеальный на мой взгляд язык – python. Он сочетает в себе простоту синтаксиса, мощь открытых библиотек и скорость разработки. На самом деле на нем очень хорошо писать веб решения так, как он интерпретируем и имеет множество идеальных заготовок.
Библиотеки flask sqlalchemy, jinja2, pyTelegramBotAPI, telethon.
Основные файлы:
- app.py - Flask веб-приложение с админ-панелью
- bot_handlers.py - Обработчики Telegram-бота
- models.py - Модели базы данных SQLAlchemy
- run.py - Точка входа для запуска приложения
- tron_payment_checker.py - Проверка USDT платежей
- user_broadcast.py - Система массовых рассылок
Конфигурация:
- requirements.txt - Зависимости Python
- .env.example - Пример файла окружения
Глава 2. Программируем.
В качестве основного языка разработки был использован python, и его микрофреймворк flask. Внутри также разработка на шаблонах jinja2 (тот же html,только с удобными тегами автоматизации) и далее css,js для красивостей.
Для базы данных стоит по умолчанию sqlite, но так как используется система sqlalchemy, то можно просто в конфиге поменять и на более мощные при желании базы mysql,postgresql.
Сначала нужно инициализировать проект.
# Создание виртуального окружения
Код:
python3 -m venv .
source bin/activate # Linux/Mac
# или
# Scripts\activate # Windows
# Установка зависимостей
pip install Flask==2.3.3 Flask-SQLAlchemy==3.0.5 Flask-Migrate==4.0.5 pyTelegramBotAPI==4.14.0 python-dotenv==1.0.0 qrcode==7.4.2 Pillow==10.0.1 Telethon==1.34.0Создать файл конфиг
Код:
# Секретный ключ для Flask
SECRET_KEY=your-very-secret-key-here-change-this-in-production
# URL базы данных
DATABASE_URL=sqlite:///bot.db
# Режим Flask
FLASK_ENV=development
# Порт и хост
FLASK_PORT=5000
FLASK_HOST=0.0.0.0Создать модели базы данных. Написано коротко, не все написал чтобы не засорять статью. Пример юзер модели добавил.
Код:
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime
import json
# Создаем объект db
db = SQLAlchemy()
class User(db.Model):
"""Модель для хранения информации о пользователях"""
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
telegram_id = db.Column(db.BigInteger, nullable=False)
bot_id = db.Column(db.Integer, db.ForeignKey('bots.id'), nullable=False)
username = db.Column(db.String(100), nullable=True)
first_name = db.Column(db.String(100), nullable=True)
last_name = db.Column(db.String(100), nullable=True)
phone_number = db.Column(db.String(20), nullable=True)
language_code = db.Column(db.String(10), nullable=True)
is_active = db.Column(db.Boolean, default=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
last_activity = db.Column(db.DateTime, default=datetime.utcnow)
# Уникальность по telegram_id и bot_id
__table_args__ = (db.UniqueConstraint('telegram_id', 'bot_id', name='unique_user_bot'),)
def __repr__(self):
return f'<User {self.telegram_id}>'
Код:
from flask import Flask, render_template, request, jsonify, redirect, url_for, flash, session
from flask_migrate import Migrate
from functools import wraps
import os
from datetime import datetime
import threading
import telebot
import json
import logging
import hashlib
# Настройка логирования
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = Flask(__name__)
app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY', 'your-secret-key-here')
app.config['SQLALCHEMY_DATABASE_URI'] = os.environ.get('DATABASE_URL', 'sqlite:///bot.db')
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# Импорт db из models
from models import db, Bot, User, Setting, Address
# Инициализация db с приложением
db.init_app(app)
migrate = Migrate(app, db)
# Словарь активных ботов
active_bots = {}
# Функции аутентификации
def get_admin_password():
"""Получить пароль администратора из настроек"""
setting = Setting.query.filter_by(key='admin_password').first()
return setting.value if setting else ''
def hash_password(password):
"""Хеширование пароля"""
return hashlib.sha256(password.encode()).hexdigest()
def check_password(password):
"""Проверка пароля"""
stored_password = get_admin_password()
if not stored_password:
return password == ''
return hash_password(password) == stored_password
def login_required(f):
"""Декоратор для проверки авторизации"""
@wraps(f)
def decorated_function(*args, **kwargs):
if not session.get('authenticated'):
return redirect(url_for('login'))
return f(*args, **kwargs)
return decorated_function
@app.route('/login', methods=['GET', 'POST'])
def login():
"""Страница входа"""
if request.method == 'POST':
password = request.form.get('password', '')
if check_password(password):
session['authenticated'] = True
flash('Успешный вход в систему!', 'success')
return redirect(url_for('index'))
else:
flash('Неверный пароль!', 'error')
return render_template('login.html')
@app.route('/logout')
def logout():
"""Выход из системы"""
session.pop('authenticated', None)
flash('Вы вышли из системы', 'info')
return redirect(url_for('login'))
@app.route('/')
@login_required
def index():
"""Главная страница админ-панели"""
bots = Bot.query.all()
users_count = User.query.count()
active_users_count = User.query.filter_by(is_active=True).count()
return render_template('index.html',
bots=bots,
users_count=users_count,
active_users_count=active_users_count)
@app.route('/bots')
@login_required
def bots():
"""Страница управления ботами"""
bots = Bot.query.all()
return render_template('bots.html', bots=bots)
@app.route('/users')
@login_required
def users():
"""Страница управления пользователями"""
users = User.query.all()
return render_template('users.html', users=users)
if __name__ == '__main__':
app.run(debug=True)
Код:
#!/usr/bin/env python3
import os
from dotenv import load_dotenv
# Загружаем переменные окружения
load_dotenv()
from app import app, db
if __name__ == '__main__':
# Создаем таблицы базы данных
with app.app_context():
db.create_all()
print("База данных инициализирована")
# Получаем настройки
host = os.environ.get('FLASK_HOST', '0.0.0.0')
port = int(os.environ.get('FLASK_PORT', 5000))
debug = os.environ.get('FLASK_ENV', 'production') == 'development'
print(f"Запуск Bot на {host}:{port}")
print(f"Веб-админка: http://{host}:{port}")
# Запускаем приложение
app.run(host=host, port=port, debug=debug)
[CODE]
[CENTER][ATTACH type="full" width="828px" size="1199x499"]2271599[/ATTACH]
Основная страница админки
[/CENTER]
Каждый бот запускается в отдельном потоке,что удобно и безопасно, и позволяет управлять целой армией ботов.
[CODE]
def run_bot():
try:
telegram_bot.polling(none_stop=True)
except Exception as e:
logger.error(f"Ошибка в потоке бота {bot_id}: {e}")
bot_thread = threading.Thread(target=run_bot)
bot_thread.daemon = True
bot_thread.start()
Учтите при,что все операции с базой необходимо выполнять в контексте app.
with app.app_context():
user = get_or_create_user(message, bot_id)
# работа с базой данныхДля безопасности было реализовано,что админка доступна лишь с локалхоста, то есть с тор домена. И защищена паролем.
Также бот помнит на каком этапе находится каждый из пользователей благодаря системе сессий.
Также сам бот было решено создать многоязычным поэтому реализована система переводов внутри русский и грузинский, вместе с любыми параметрами. В коде существует словарь TEXTS с переводами на русский и грузинский языки, содержащий более 100 текстовых строк. Это не просто набор фраз — это продуманная система, где каждое сообщение имеет свой ключ и может содержать параметры для подстановки.
'order_created': '
Заказ #{order_id} создан!\n\n Сумма: ${amount} USDT\n Адрес для оплаты:\n\n`{address}`'Рассылка через ботов также возможна
Реализация обработчиков бота в телеграм.
Код:
from telebot import types
from telebot.types import InlineKeyboardMarkup, InlineKeyboardButton
import json
import os
from datetime import datetime
from models import db, User, Bot, Setting
import logging
logger = logging.getLogger(__name__)
# Словари для локализации
TEXTS = {
'ru': {
'welcome': ' Добро пожаловать! Выберите язык:',
'main_menu': ' Главное меню',
'language_selected': '✅ Язык выбран: Русский',
'contact_operator': ' Связаться с оператором',
'catalog': '️ Каталог товаров',
'my_orders': ' Мои заказы',
'settings': '⚙️ Настройки',
'change_language': ' Изменить язык',
},
'en': {
'welcome': ' Welcome! Choose your language:',
'main_menu': ' Main menu',
'language_selected': '✅ Language selected: English',
'contact_operator': ' Contact operator',
'catalog': '️ Product catalog',
'my_orders': ' My orders',
'settings': '⚙️ Settings',
'change_language': ' Change language',
}
}
def get_text(user_lang, key):
"""Получить текст на языке пользователя"""
return TEXTS.get(user_lang, TEXTS['ru']).get(key, key)
def create_main_menu_keyboard(user_lang='ru'):
"""Создать главное меню"""
keyboard = InlineKeyboardMarkup(row_width=2)
buttons = [
InlineKeyboardButton(get_text(user_lang, 'catalog'), callback_data='catalog'),
InlineKeyboardButton(get_text(user_lang, 'my_orders'), callback_data='my_orders'),
InlineKeyboardButton(get_text(user_lang, 'contact_operator'), callback_data='contact_operator'),
InlineKeyboardButton(get_text(user_lang, 'settings'), callback_data='settings'),
]
keyboard.add(*buttons)
return keyboard
def create_language_keyboard():
"""Создать клавиатуру выбора языка"""
keyboard = InlineKeyboardMarkup(row_width=2)
keyboard.add(
InlineKeyboardButton(' Русский', callback_data='lang_ru'),
InlineKeyboardButton(' English', callback_data='lang_en')
)
return keyboard
def create_bot_handlers(bot, bot_model):
"""Создать обработчики для бота"""
@bot.message_handler(commands=['start'])
def handle_start(message):
"""Обработчик команды /start"""
try:
# Получаем или создаем пользователя
user = User.query.filter_by(
telegram_id=message.from_user.id,
bot_id=bot_model.id
).first()
if not user:
user = User(
telegram_id=message.from_user.id,
bot_id=bot_model.id,
username=message.from_user.username,
first_name=message.from_user.first_name,
last_name=message.from_user.last_name,
language_code=message.from_user.language_code or 'ru'
)
db.session.add(user)
db.session.commit()
# Обновляем активность
user.last_activity = datetime.utcnow()
db.session.commit()
# Отправляем приветствие
bot.send_message(
message.chat.id,
get_text(user.language_code, 'welcome'),
reply_markup=create_language_keyboard()
)
except Exception as e:
logger.error(f"Ошибка в handle_start: {e}")
bot.send_message(message.chat.id, "Произошла ошибка. Попробуйте позже.")
@bot.callback_query_handler(func=lambda call: call.data.startswith('lang_'))
def handle_language_selection(call):
"""Обработчик выбора языка"""
try:
lang = call.data.split('_')[1]
# Обновляем язык пользователя
user = User.query.filter_by(
telegram_id=call.from_user.id,
bot_id=bot_model.id
).first()
if user:
user.language_code = lang
db.session.commit()
# Отправляем главное меню
bot.edit_message_text(
get_text(lang, 'language_selected'),
call.message.chat.id,
call.message.message_id
)
bot.send_message(
call.message.chat.id,
get_text(lang, 'main_menu'),
reply_markup=create_main_menu_keyboard(lang)
)
except Exception as e:
logger.error(f"Ошибка в handle_language_selection: {e}")
@bot.callback_query_handler(func=lambda call: True)
def handle_callback_query(call):
"""Обработчик callback запросов"""
try:
user = User.query.filter_by(
telegram_id=call.from_user.id,
bot_id=bot_model.id
).first()
if not user:
return
user_lang = user.language_code or 'ru'
if call.data == 'catalog':
bot.answer_callback_query(call.id, "Каталог в разработке")
elif call.data == 'my_orders':
bot.answer_callback_query(call.id, "Заказы в разработке")
elif call.data == 'contact_operator':
bot.answer_callback_query(call.id, "Связь с оператором в разработке")
elif call.data == 'settings':
bot.edit_message_text(
get_text(user_lang, 'settings'),
call.message.chat.id,
call.message.message_id,
reply_markup=create_language_keyboard()
)
except Exception as e:
logger.error(f"Ошибка в handle_callback_query: {e}")
return botНиже приложил пример реализации самого интерфейса бота в тг.
Глава 3. Тестируем, размышляем, дополняем.
Для защит от различных блокировок конкурентов, переборов и прочих нюансов было решено придумать удобные решения. Путем анализа других ботов и проверки своего создали некоторые моменты, которые решают большинство проблем.
А именно:
- Задержки между сообщениями:
Код:
# Небольшая задержка между отправками
await asyncio.sleep(1)- Ограничения в интерфейсе веб-админки
Код:
<li class="mb-2">
<i class="fas fa-check text-success"></i>
<small>Не отправляйте более 30 сообщений в минуту</small>
</li>- Обработка ошибок
Код:
try:
await client.send_message(entity, message)
sent_count += 1
except Exception as e:
failed_count += 1
errors.append(f"{target}: {str(e)}")- Система капчи для проверки при входе в бота юзера
Код:
def create_captcha_for_user(user_id, bot_id):
# Генерируем случайные эмодзи
all_emojis = ['', '', '', '⭐', '', '', '', '']
target_emoji = random.choice(all_emojis)
# Создаем варианты ответов
emoji_options = [target_emoji]
while len(emoji_options) < 6:
emoji = random.choice(all_emojis)
If emoji not in emoji_options:
emoji_options.append(emoji)
random.shuffle(emoji_options)- Возможность запуска множества ботов одновременно. В админке можно по нажатию клавиш и форм создать множество ботов и включать их в разных потоках независимо.
Код:
# Словарь активных ботов
active_bots = {}
@app.route('/start_bot/<int:bot_id>')
def start_bot(bot_id):
# Каждый бот запускается в отдельном потоке
bot_thread = threading.Thread(target=run_bot)
bot_thread.daemon = True
bot_thread.start()
active_bots[bot_id] = {
'bot': telegram_bot,
'thread': bot_thread
}- Самая мощная функция — рассылки через пользовательские аккаунты с помощью telethon. Это обходит многие ограничения обычного апи тг. Нужно для того,чтобы когда заблокируют одного бота можно было разослать старым пользователям через обычного юзера оповещение о новых ботах или каналах,форумах. Таким образом ваши клиенты всегда при вас. Отправа идет по username, или номеру если у пользователя нет ника.
Код:
class UserBroadcastManager:
async def send_broadcast(self, phone, message, targets, api_id, api_hash):
client = TelegramClient(session_file, api_id, api_hash)
for target in targets:
try:
# Определяем тип цели (username или номер)
if target.startswith('@'):
entity = await client.get_entity(target)
elif target.startswith('+') or target.isdigit():
entity = await client.get_entity(target)
await client.send_message(entity, message)
sent_count += 1
# Задержка между отправками
await asyncio.sleep(1)
except Exception as e:
failed_count += 1
Можно легко добавлять аккаунты пользователей(для рассылки) в админке и получать их сессии вписав код с тг.
function startAuth() {
// Двухэтапная авторизация
// 1. Отправка кода на телефон
// 2. Подтверждение кода (+ 2FA если нужно)
fetch('/user_broadcast/auth', {
method: 'POST',
body: formData
})
.then(response => response.json())
.then(data => {
if (data.success) {
// Переход ко второму шагу
showCodeInput();
}
});
}- Особенно элегантно решена задача с QR-кодами для платежей. Функция generate_qr_code() не только создает QR-код с настраиваемым дизайном (цвета, размер, логотип), но и schedule_qr_deletion() автоматически удаляет его через заданное время. Это повышает безопасность для ваших тормознутых клиентов.
- Колесо фортуны. Функцию работающая через сайт и выдающая коды бонусы и скидки.
- Одна из самых интересных функций get_smart_stash(). Это не просто выбор случайного товара со склада. Система анализирует доступные клады и выбирает оптимальный по нескольким критериям:
Приоритет по весу и расположению — сначала ищет точное совпадение
Временные факторы — предпочитает более старые клады
Это обеспечивает эффективную ротацию товаров и оптимизацию логистики. Очень удобно при автоматических продажах
Код:
def get_smart_stash(product_id, weight):
"""
Получить клад с умной логикой распределения.
Выбирает клады с разницей по загрузке минимум в 5 позиций.
Логика: 1-й, 6-й, 11-й, 16-й, потом 2-й, 7-й, 12-й, 3-й, 8-й и т.д.
"""
try:
# Получаем все доступные клады для данного продукта и веса, отсортированные по ID (порядок загрузки)
available_stashes = Stash.query.filter_by(
product_id=product_id,
weight=weight,
is_sold=False,
order_item_id=None
).order_by(Stash.id).all()
if not available_stashes:
return None
# Если кладов меньше 5, просто берем первый
if len(available_stashes) < 5:
return available_stashes[0]
# Получаем ключ для отслеживания текущего паттерна выбора
# Используем комбинацию product_id и weight как ключ
pattern_key = f"stash_pattern_{product_id}_{weight}"
# Получаем или создаем настройку для отслеживания паттерна
pattern_setting = Setting.query.filter_by(key=pattern_key).first()
if not pattern_setting:
pattern_setting = Setting(key=pattern_key, value="0,0") # current_group, position_in_group
db.session.add(pattern_setting)
db.session.commit()
# Парсим текущее состояние паттерна
try:
current_group, position_in_group = map(int, pattern_setting.value.split(','))
except:
current_group, position_in_group = 0, 0
# Вычисляем индекс клада для выбора
# Группы: 0 = позиции 0,5,10,15... | 1 = позиции 1,6,11,16... | и т.д.
stash_index = current_group + (position_in_group * 5)
# Если индекс выходит за границы, переходим к следующей группе
if stash_index >= len(available_stashes):
current_group += 1
position_in_group = 0
# Если все группы пройдены, начинаем сначала
if current_group >= 5:
current_group = 0
position_in_group = 0
stash_index = current_group + (position_in_group * 5)
# Если все еще выходит за границы, берем первый доступный
if stash_index >= len(available_stashes):
stash_index = 0
current_group = 0
position_in_group = 0
# Выбираем клад
selected_stash = available_stashes[stash_index]
# Обновляем паттерн для следующего выбора
position_in_group += 1
# Если достигли конца текущей группы, переходим к следующей
max_positions_in_group = (len(available_stashes) - current_group + 4) // 5 # Округление вверх
if position_in_group >= max_positions_in_group:
current_group += 1
position_in_group = 0
# Если все группы пройдены, начинаем сначала
if current_group >= 5:
current_group = 0
# Сохраняем обновленный паттерн
pattern_setting.value = f"{current_group},{position_in_group}"
db.session.commit()
logger.info(f"Выбран клад ID={selected_stash.id} (индекс {stash_index} из {len(available_stashes)}) для продукта {product_id}, вес {weight}")
return selected_stash
except Exception as e:
logger.error(f"Ошибка в get_smart_stash: {str(e)}")
# В случае ошибки возвращаем первый доступный клад
return Stash.query.filter_by(
product_id=product_id,
weight=weight,
is_sold=False,
order_item_id=None
).first()Послесловие:
Как вы видите нет ничего сложного в своем собственном решении для автопродаж.
При прямых руках и желании вы можете создать себе независимый от сторонних программных решений автоматизированный бизнес с ботами и админкой.
Если будут дополнительные вопросы по реализации подобных решений или конкретным интересным идеям, пишите в вопросы к статье, помогу чем смогу.
Спасибо за внимание!!! Процветания вашим проектам и бизнесам!
Как вы видите нет ничего сложного в своем собственном решении для автопродаж.
При прямых руках и желании вы можете создать себе независимый от сторонних программных решений автоматизированный бизнес с ботами и админкой.
Если будут дополнительные вопросы по реализации подобных решений или конкретным интересным идеям, пишите в вопросы к статье, помогу чем смогу.
Спасибо за внимание!!! Процветания вашим проектам и бизнесам!
