Telegram Bot Analytics - v0 by Vercel

Telegram Bot Analytics - v0 by VercelTelegram Bot Analytics - v0 by Vercel

You need to write a program code in Python to analyze the following information in a telegram bot:

Analytics - how many people visited the telegram bot
Number of registered people in the Telegram bot
How many people paid
How many have not paid
Total attendance of telegram bot
Who was disconnected from the Telegram Bot at the end of the monthly subscription.
Subscription disconnections for non-payment of the next month

Translated with DeepL.com (free version)

379d ago

I'll create a Python script that helps you track analytics for your Telegram bot, including visitor counts, registrations, payments, and subscription management.

Telegram Bot Analytics

bot_analytics.py

How to Use This Code

Setup:
- Install required packages: pip install python-telegram-bot
- Replace "YOUR_BOT_TOKEN" with your actual Telegram bot token
- Set your admin user ID in the stats function
Features Implemented:
- Tracks unique visitors and total visits
- Registers new users
- Handles subscription payments and expiration
- Provides analytics on paid/unpaid users
- Identifies expired subscriptions
- Tracks users who didn't renew subscriptions
Commands:
- /start - Registers a user and welcomes them
- /stats - Shows analytics (admin only)
- /pay - Simulates a payment (in a real bot, connect to a payment processor)
Database:
- Uses SQLite to store user data and visit history
- Automatically creates tables on first run

For a production bot, you would want to connect this to a real payment processor and possibly use a more robust database like PostgreSQL for larger user bases.

379d ago

import sqlite3
from datetime import datetime, timedelta
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters, CallbackContext

# Database setup
def setup_database():
    conn = sqlite3.connect('telegram_bot.db')
    cursor = conn.cursor()
    
    # Create users table
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS users (
        user_id INTEGER PRIMARY KEY,
        username TEXT,
        first_name TEXT,
        last_name TEXT,
        registration_date TEXT,
        last_visit TEXT,
        is_paid BOOLEAN DEFAULT 0,
        subscription_end_date TEXT
    )
    ''')
    
    # Create visits table
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS visits (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        user_id INTEGER,
        visit_date TEXT,
        FOREIGN KEY (user_id) REFERENCES users (user_id)
    )
    ''')
    
    conn.commit()
    conn.close()

# User tracking functions
async def track_visit(user_id):
    conn = sqlite3.connect('telegram_bot.db')
    cursor = conn.cursor()
    
    current_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    
    # Update last visit in users table
    cursor.execute('''
    UPDATE users SET last_visit = ? WHERE user_id = ?
    ''', (current_date, user_id))
    
    # Add entry to visits table
    cursor.execute('''
    INSERT INTO visits (user_id, visit_date) VALUES (?, ?)
    ''', (user_id, current_date))
    
    conn.commit()
    conn.close()

async def register_user(user_id, username, first_name, last_name):
    conn = sqlite3.connect('telegram_bot.db')
    cursor = conn.cursor()
    
    # Check if user already exists
    cursor.execute('SELECT user_id FROM users WHERE user_id = ?', (user_id,))
    if cursor.fetchone() is None:
        current_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        cursor.execute('''
        INSERT INTO users (user_id, username, first_name, last_name, registration_date, last_visit)
        VALUES (?, ?, ?, ?, ?, ?)
        ''', (user_id, username, first_name, last_name, current_date, current_date))
    
    conn.commit()
    conn.close()

async def process_payment(user_id, months=1):
    conn = sqlite3.connect('telegram_bot.db')
    cursor = conn.cursor()
    
    # Get current subscription end date
    cursor.execute('SELECT subscription_end_date FROM users WHERE user_id = ?', (user_id,))
    result = cursor.fetchone()
    
    if result and result[0]:
        current_end_date = datetime.strptime(result[0], '%Y-%m-%d %H:%M:%S')
        # If subscription is still active, extend from current end date
        if current_end_date > datetime.now():
            new_end_date = current_end_date + timedelta(days=30*months)
        else:
            # If expired, start from now
            new_end_date = datetime.now() + timedelta(days=30*months)
    else:
        # First time subscription
        new_end_date = datetime.now() + timedelta(days=30*months)
    
    # Update user as paid with new subscription end date
    cursor.execute('''
    UPDATE users SET is_paid = 1, subscription_end_date = ? WHERE user_id = ?
    ''', (new_end_date.strftime('%Y-%m-%d %H:%M:%S'), user_id))
    
    conn.commit()
    conn.close()

# Analytics functions
def get_analytics():
    conn = sqlite3.connect('telegram_bot.db')
    cursor = conn.cursor()
    
    # Total visitors (unique users who visited)
    cursor.execute('SELECT COUNT(DISTINCT user_id) FROM visits')
    total_visitors = cursor.fetchone()[0]
    
    # Total registered users
    cursor.execute('SELECT COUNT(*) FROM users')
    registered_users = cursor.fetchone()[0]
    
    # Paid users
    cursor.execute('SELECT COUNT(*) FROM users WHERE is_paid = 1')
    paid_users = cursor.fetchone()[0]
    
    # Unpaid users
    unpaid_users = registered_users - paid_users
    
    # Total visits
    cursor.execute('SELECT COUNT(*) FROM visits')
    total_visits = cursor.fetchone()[0]
    
    # Users with expired subscriptions
    current_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    cursor.execute('''
    SELECT COUNT(*) FROM users 
    WHERE is_paid = 1 AND subscription_end_date < ?
    ''', (current_date,))
    expired_subscriptions = cursor.fetchone()[0]
    
    # Users who didn't renew last month
    one_month_ago = (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d %H:%M:%S')
    cursor.execute('''
    SELECT COUNT(*) FROM users 
    WHERE is_paid = 0 AND subscription_end_date BETWEEN ? AND ?
    ''', (one_month_ago, current_date))
    non_renewed = cursor.fetchone()[0]
    
    conn.close()
    
    return {
        "total_visitors": total_visitors,
        "registered_users": registered_users,
        "paid_users": paid_users,
        "unpaid_users": unpaid_users,
        "total_visits": total_visits,
        "expired_subscriptions": expired_subscriptions,
        "non_renewed_subscriptions": non_renewed
    }

# Check for expired subscriptions daily
async def check_expired_subscriptions(context: CallbackContext):
    conn = sqlite3.connect('telegram_bot.db')
    cursor = conn.cursor()
    
    current_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    
    # Find users with expired subscriptions
    cursor.execute('''
    SELECT user_id FROM users 
    WHERE is_paid = 1 AND subscription_end_date < ?
    ''', (current_date,))
    
    expired_users = cursor.fetchall()
    
    # Mark users as unpaid
    for user in expired_users:
        user_id = user[0]
        cursor.execute('''
        UPDATE users SET is_paid = 0 WHERE user_id = ?
        ''', (user_id,))
        
        # Notify user about expiration (in a real bot)
        try:
            await context.bot.send_message(
                chat_id=user_id, 
                text="Your subscription has expired. Please renew to continue using premium features."
            )
        except Exception as e:
            print(f"Failed to notify user {user_id}: {e}")
    
    conn.commit()
    conn.close()

# Command handlers
async def start(update: Update, context: CallbackContext):
    user = update.effective_user
    await register_user(user.id, user.username, user.first_name, user.last_name)
    await track_visit(user.id)
    
    await update.message.reply_text(f'Hello {user.first_name}! Welcome to our bot.')

async def stats(update: Update, context: CallbackContext):
    # Only allow admin to see stats (you would need to define admin_user_id)
    admin_user_id = 12345  # Replace with your admin user ID
    if update.effective_user.id != admin_user_id:
        await update.message.reply_text("You don't have permission to view stats.")
        return
    
    analytics = get_analytics()
    
    stats_message = f"""
📊 *Bot Analytics*

👥 *Users*:
- Visitors: {analytics['total_visitors']}
- Registered: {analytics['registered_users']}

💰 *Payments*:
- Paid users: {analytics['paid_users']}
- Unpaid users: {analytics['unpaid_users']}

📈 *Activity*:
- Total visits: {analytics['total_visits']}

🔄 *Subscriptions*:
- Expired subscriptions: {analytics['expired_subscriptions']}
- Non-renewed last month: {analytics['non_renewed_subscriptions']}
"""
    
    await update.message.reply_text(stats_message, parse_mode='Markdown')

async def handle_message(update: Update, context: CallbackContext):
    user_id = update.effective_user.id
    await track_visit(user_id)
    
    # Process the message (your bot logic here)
    await update.message.reply_text("I received your message!")

# For demonstration: command to simulate payment
async def pay(update: Update, context: CallbackContext):
    user_id = update.effective_user.id
    
    # In a real bot, this would be connected to an actual payment processor
    await process_payment(user_id)
    
    await update.message.reply_text("Payment processed successfully! Your subscription is now active.")

def main():
    # Set up the database
    setup_database()
    
    # Create the application
    application = Application.builder().token("YOUR_BOT_TOKEN").build()
    
    # Add handlers
    application.add_handler(CommandHandler("start", start))
    application.add_handler(CommandHandler("stats", stats))
    application.add_handler(CommandHandler("pay", pay))
    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
    
    # Schedule daily check for expired subscriptions
    job_queue = application.job_queue
    job_queue.run_daily(check_expired_subscriptions, time=datetime.time(hour=0, minute=0))
    
    # Start the Bot
    application.run_polling()

if __name__ == '__main__':
    main()

No Output

Run the code to generate an output.

import sqlite3
from datetime import datetime, timedelta
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters, CallbackContext

# Database setup
def setup_database():
    conn = sqlite3.connect('telegram_bot.db')
    cursor = conn.cursor()
    
    # Create users table
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS users (
        user_id INTEGER PRIMARY KEY,
        username TEXT,
        first_name TEXT,
        last_name TEXT,
        registration_date TEXT,
        last_visit TEXT,
        is_paid BOOLEAN DEFAULT 0,
        subscription_end_date TEXT
    )
    ''')
    
    # Create visits table
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS visits (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        user_id INTEGER,
        visit_date TEXT,
        FOREIGN KEY (user_id) REFERENCES users (user_id)
    )
    ''')
    
    conn.commit()
    conn.close()

# User tracking functions
async def track_visit(user_id):
    conn = sqlite3.connect('telegram_bot.db')
    cursor = conn.cursor()
    
    current_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    
    # Update last visit in users table
    cursor.execute('''
    UPDATE users SET last_visit = ? WHERE user_id = ?
    ''', (current_date, user_id))
    
    # Add entry to visits table
    cursor.execute('''
    INSERT INTO visits (user_id, visit_date) VALUES (?, ?)
    ''', (user_id, current_date))
    
    conn.commit()
    conn.close()

async def register_user(user_id, username, first_name, last_name):
    conn = sqlite3.connect('telegram_bot.db')
    cursor = conn.cursor()
    
    # Check if user already exists
    cursor.execute('SELECT user_id FROM users WHERE user_id = ?', (user_id,))
    if cursor.fetchone() is None:
        current_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        cursor.execute('''
        INSERT INTO users (user_id, username, first_name, last_name, registration_date, last_visit)
        VALUES (?, ?, ?, ?, ?, ?)
        ''', (user_id, username, first_name, last_name, current_date, current_date))
    
    conn.commit()
    conn.close()

async def process_payment(user_id, months=1):
    conn = sqlite3.connect('telegram_bot.db')
    cursor = conn.cursor()
    
    # Get current subscription end date
    cursor.execute('SELECT subscription_end_date FROM users WHERE user_id = ?', (user_id,))
    result = cursor.fetchone()
    
    if result and result[0]:
        current_end_date = datetime.strptime(result[0], '%Y-%m-%d %H:%M:%S')
        # If subscription is still active, extend from current end date
        if current_end_date > datetime.now():
            new_end_date = current_end_date + timedelta(days=30*months)
        else:
            # If expired, start from now
            new_end_date = datetime.now() + timedelta(days=30*months)
    else:
        # First time subscription
        new_end_date = datetime.now() + timedelta(days=30*months)
    
    # Update user as paid with new subscription end date
    cursor.execute('''
    UPDATE users SET is_paid = 1, subscription_end_date = ? WHERE user_id = ?
    ''', (new_end_date.strftime('%Y-%m-%d %H:%M:%S'), user_id))
    
    conn.commit()
    conn.close()

# Analytics functions
def get_analytics():
    conn = sqlite3.connect('telegram_bot.db')
    cursor = conn.cursor()
    
    # Total visitors (unique users who visited)
    cursor.execute('SELECT COUNT(DISTINCT user_id) FROM visits')
    total_visitors = cursor.fetchone()[0]
    
    # Total registered users
    cursor.execute('SELECT COUNT(*) FROM users')
    registered_users = cursor.fetchone()[0]
    
    # Paid users
    cursor.execute('SELECT COUNT(*) FROM users WHERE is_paid = 1')
    paid_users = cursor.fetchone()[0]
    
    # Unpaid users
    unpaid_users = registered_users - paid_users
    
    # Total visits
    cursor.execute('SELECT COUNT(*) FROM visits')
    total_visits = cursor.fetchone()[0]
    
    # Users with expired subscriptions
    current_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    cursor.execute('''
    SELECT COUNT(*) FROM users 
    WHERE is_paid = 1 AND subscription_end_date < ?
    ''', (current_date,))
    expired_subscriptions = cursor.fetchone()[0]
    
    # Users who didn't renew last month
    one_month_ago = (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d %H:%M:%S')
    cursor.execute('''
    SELECT COUNT(*) FROM users 
    WHERE is_paid = 0 AND subscription_end_date BETWEEN ? AND ?
    ''', (one_month_ago, current_date))
    non_renewed = cursor.fetchone()[0]
    
    conn.close()
    
    return {
        "total_visitors": total_visitors,
        "registered_users": registered_users,
        "paid_users": paid_users,
        "unpaid_users": unpaid_users,
        "total_visits": total_visits,
        "expired_subscriptions": expired_subscriptions,
        "non_renewed_subscriptions": non_renewed
    }

# Check for expired subscriptions daily
async def check_expired_subscriptions(context: CallbackContext):
    conn = sqlite3.connect('telegram_bot.db')
    cursor = conn.cursor()
    
    current_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    
    # Find users with expired subscriptions
    cursor.execute('''
    SELECT user_id FROM users 
    WHERE is_paid = 1 AND subscription_end_date < ?
    ''', (current_date,))
    
    expired_users = cursor.fetchall()
    
    # Mark users as unpaid
    for user in expired_users:
        user_id = user[0]
        cursor.execute('''
        UPDATE users SET is_paid = 0 WHERE user_id = ?
        ''', (user_id,))
        
        # Notify user about expiration (in a real bot)
        try:
            await context.bot.send_message(
                chat_id=user_id, 
                text="Your subscription has expired. Please renew to continue using premium features."
            )
        except Exception as e:
            print(f"Failed to notify user {user_id}: {e}")
    
    conn.commit()
    conn.close()

# Command handlers
async def start(update: Update, context: CallbackContext):
    user = update.effective_user
    await register_user(user.id, user.username, user.first_name, user.last_name)
    await track_visit(user.id)
    
    await update.message.reply_text(f'Hello {user.first_name}! Welcome to our bot.')

async def stats(update: Update, context: CallbackContext):
    # Only allow admin to see stats (you would need to define admin_user_id)
    admin_user_id = 12345  # Replace with your admin user ID
    if update.effective_user.id != admin_user_id:
        await update.message.reply_text("You don't have permission to view stats.")
        return
    
    analytics = get_analytics()
    
    stats_message = f"""
📊 *Bot Analytics*

👥 *Users*:
- Visitors: {analytics['total_visitors']}
- Registered: {analytics['registered_users']}

💰 *Payments*:
- Paid users: {analytics['paid_users']}
- Unpaid users: {analytics['unpaid_users']}

📈 *Activity*:
- Total visits: {analytics['total_visits']}

🔄 *Subscriptions*:
- Expired subscriptions: {analytics['expired_subscriptions']}
- Non-renewed last month: {analytics['non_renewed_subscriptions']}
"""
    
    await update.message.reply_text(stats_message, parse_mode='Markdown')

async def handle_message(update: Update, context: CallbackContext):
    user_id = update.effective_user.id
    await track_visit(user_id)
    
    # Process the message (your bot logic here)
    await update.message.reply_text("I received your message!")

# For demonstration: command to simulate payment
async def pay(update: Update, context: CallbackContext):
    user_id = update.effective_user.id
    
    # In a real bot, this would be connected to an actual payment processor
    await process_payment(user_id)
    
    await update.message.reply_text("Payment processed successfully! Your subscription is now active.")

def main():
    # Set up the database
    setup_database()
    
    # Create the application
    application = Application.builder().token("YOUR_BOT_TOKEN").build()
    
    # Add handlers
    application.add_handler(CommandHandler("start", start))
    application.add_handler(CommandHandler("stats", stats))
    application.add_handler(CommandHandler("pay", pay))
    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
    
    # Schedule daily check for expired subscriptions
    job_queue = application.job_queue
    job_queue.run_daily(check_expired_subscriptions, time=datetime.time(hour=0, minute=0))
    
    # Start the Bot
    application.run_polling()

if __name__ == '__main__':
    main()

No Output

Run the code to generate an output.