Joke Bot Upgrade
From Simple Lists to SQLite Database with User Interaction
Goal: Upgrade joke bot with database, user submissions, and ratings
Prerequisites: Basic Python knowledge, our existing joke bot
Time: 60 minutes to complete upgrade
What Is SQLite?
Serverless, self-contained SQL database perfect for Telegram bots
Why SQLite for Our Joke Bot?
1
No Server Needed: Database lives in a single file
2
Zero Configuration: Just import and use
3
Lightweight: Perfect for small to medium applications
4
SQL Standard: Uses standard SQL commands
5
Built-in Python: No extra installation needed
Think of it like this:
SQLite is like an Excel file that speaks SQL. It's a complete database in a single file that you can include with your application.
Perfect for Telegram bots, mobile apps, and desktop applications.
import sqlite3
conn = sqlite3.connect('joke_bot.db')
cursor = conn.cursor()
cursor.execute('''CREATE TABLE jokes (
id INTEGER PRIMARY KEY,
text TEXT NOT NULL,
added_by TEXT,
likes INTEGER DEFAULT 0,
dislikes INTEGER DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)''')
conn.commit()
conn.close()
Our Upgrade Goals
From static list to dynamic, user-powered joke database
Current Limitations:
1. Jokes are hardcoded in Python list
2. No way for users to add new jokes
3. No rating system to find the best jokes
4. Jokes are lost when bot restarts
New Features We'll Add:
1
SQLite Database:
Store jokes permanently in a database file
2
User Submissions:
/addjoke command for users to submit jokes
3
Rating System:
/like and /dislike commands to rate jokes
4
Top Jokes:
/top command to show most popular jokes
5
Random with Weight:
Better jokes appear more often
Step 1: Database Setup (10 minutes)
Create the SQLite database and connection functions
import sqlite3
from datetime import datetime
class JokeDatabase:
def __init__(self, db_name='joke_bot.db'):
self.db_name = db_name
self.init_database()
def get_connection(self):
conn = sqlite3.connect(self.db_name)
conn.row_factory = sqlite3.Row
return conn
def init_database(self):
conn = self.get_connection()
cursor = conn.cursor()
cursor.execute('''CREATE TABLE IF NOT EXISTS jokes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
joke_text TEXT NOT NULL,
user_id INTEGER,
username TEXT,
likes INTEGER DEFAULT 0,
dislikes INTEGER DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)''')
cursor.execute('''CREATE TABLE IF NOT EXISTS ratings (
user_id INTEGER,
joke_id INTEGER,
vote INTEGER,
PRIMARY KEY (user_id, joke_id),
FOREIGN KEY (joke_id) REFERENCES jokes(id)
)''')
conn.commit()
conn.close()
self.add_default_jokes()
def add_default_jokes(self):
conn = self.get_connection()
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM jokes")
count = cursor.fetchone()[0]
if count == 0:
default_jokes = [
("Why did the robot go to school? To recharge his brain! š", 0, "System"),
("Knock knock!\nWho's there?\nLettuce!\nLettuce who?\nLettuce in!", 0, "System"),
("Why don't eggs tell jokes? They'd crack each other up! š„", 0, "System")
]
cursor.executemany('''INSERT INTO jokes (joke_text, user_id, username)
VALUES (?, ?, ?)''', default_jokes)
conn.commit()
print(f"Added {len(default_jokes)} default jokes to database")
conn.close()
Database Schema Explained:
š
jokes table: Stores all jokes with ratings
ā
ratings table: Tracks who voted for which joke
š
Primary keys: Unique IDs for each record
š
Foreign key: Links ratings to jokes
Step 2: Database Operations (15 minutes)
Add CRUD (Create, Read, Update, Delete) functions
def add_joke(self, joke_text, user_id, username):
conn = self.get_connection()
cursor = conn.cursor()
cursor.execute('''INSERT INTO jokes (joke_text, user_id, username)
VALUES (?, ?, ?)''', (joke_text, user_id, username))
joke_id = cursor.lastrowid
conn.commit()
conn.close()
return joke_id
def get_random_joke(self):
conn = self.get_connection()
cursor = conn.cursor()
cursor.execute('''SELECT * FROM jokes
ORDER BY (likes - dislikes + 5) * RANDOM() DESC
LIMIT 1''')
joke = cursor.fetchone()
conn.close()
return dict(joke) if joke else None
def get_joke_by_id(self, joke_id):
conn = self.get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM jokes WHERE id = ?", (joke_id,))
joke = cursor.fetchone()
conn.close()
return dict(joke) if joke else None
def get_top_jokes(self, limit=5):
conn = self.get_connection()
cursor = conn.cursor()
cursor.execute('''SELECT * FROM jokes
ORDER BY (likes - dislikes) DESC, likes DESC
LIMIT ?''', (limit,))
jokes = [dict(row) for row in cursor.fetchall()]
conn.close()
return jokes
def get_total_jokes(self):
conn = self.get_connection()
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM jokes")
count = cursor.fetchone()[0]
conn.close()
return count
SQL Functions Explained:
š
Weighted Random: (likes - dislikes + 5) * RANDOM()
ā¬ļø
Top Jokes: Sorted by net score (likes - dislikes)
š¾
Parameterized Queries: Safe from SQL injection
š
Connection Management: Open/close connections properly
Step 3: Rating System (15 minutes)
Implement like/dislike functionality with vote tracking
def rate_joke(self, user_id, joke_id, vote):
conn = self.get_connection()
cursor = conn.cursor()
cursor.execute('''SELECT vote FROM ratings
WHERE user_id = ? AND joke_id = ?''', (user_id, joke_id))
existing_vote = cursor.fetchone()
if existing_vote:
old_vote = existing_vote[0]
if vote == 0:
cursor.execute('''DELETE FROM ratings
WHERE user_id = ? AND joke_id = ?''', (user_id, joke_id))
if old_vote == 1:
cursor.execute('''UPDATE jokes SET likes = likes - 1
WHERE id = ?''', (joke_id,))
elif old_vote == -1:
cursor.execute('''UPDATE jokes SET dislikes = dislikes - 1
WHERE id = ?''', (joke_id,))
elif old_vote != vote:
cursor.execute('''UPDATE ratings SET vote = ?
WHERE user_id = ? AND joke_id = ?''', (vote, user_id, joke_id))
if old_vote == 1 and vote == -1:
cursor.execute('''UPDATE jokes
SET likes = likes - 1, dislikes = dislikes + 1
WHERE id = ?''', (joke_id,))
elif old_vote == -1 and vote == 1:
cursor.execute('''UPDATE jokes
SET likes = likes + 1, dislikes = dislikes - 1
WHERE id = ?''', (joke_id,))
else:
if vote != 0:
cursor.execute('''INSERT INTO ratings (user_id, joke_id, vote)
VALUES (?, ?, ?)''', (user_id, joke_id, vote))
if vote == 1:
cursor.execute('''UPDATE jokes SET likes = likes + 1
WHERE id = ?''', (joke_id,))
elif vote == -1:
cursor.execute('''UPDATE jokes SET dislikes = dislikes + 1
WHERE id = ?''', (joke_id,))
conn.commit()
cursor.execute("SELECT likes, dislikes FROM jokes WHERE id = ?", (joke_id,))
result = cursor.fetchone()
conn.close()
return {
'likes': result[0] if result else 0,
'dislikes': result[1] if result else 0
}
def get_user_vote(self, user_id, joke_id):
conn = self.get_connection()
cursor = conn.cursor()
cursor.execute('''SELECT vote FROM ratings
WHERE user_id = ? AND joke_id = ?''', (user_id, joke_id))
result = cursor.fetchone()
conn.close()
return result[0] if result else 0
Rating Logic:
1
Prevent Double Voting: Users can only vote once per joke
2
Change Votes: Users can change their mind
3
Remove Votes: Users can remove their rating
4
Real-time Updates: Counts update immediately
Step 4: Telegram Bot Integration (15 minutes)
Update app.py to use the database
from telegram import Update, InlineKeyboardMarkup, InlineKeyboardButton
from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters, CallbackQueryHandler
from database import JokeDatabase
import random
db = JokeDatabase()
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
welcome_text = """š¤ Welcome to Joke Bot 2.0! š¤
Commands:
/joke - Get a random joke
/addjoke - Submit your own joke
/top - See top-rated jokes
/like - Like the last joke
/dislike - Dislike the last joke
/stats - Bot statistics
Now with database, user submissions, and ratings!"""
await update.message.reply_text(welcome_text)
async def send_joke(update: Update, context: ContextTypes.DEFAULT_TYPE):
joke = db.get_random_joke()
if not joke:
await update.message.reply_text("No jokes in database yet! Use /addjoke to add one.")
return
context.user_data['last_joke_id'] = joke['id']
rating_text = f"š {joke['likes']} š {joke['dislikes']}"
joke_text = f"{joke['joke_text']}\n\n{rating_text}"
keyboard = [
[
InlineKeyboardButton("š Like", callback_data=f"like_{joke['id']}"),
InlineKeyboardButton("š Dislike", callback_data=f"dislike_{joke['id']}")
]
]
reply_markup = InlineKeyboardMarkup(keyboard)
await update.message.reply_text(joke_text, reply_markup=reply_markup)
async def add_joke_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
if not context.args:
await update.message.reply_text(
"Please provide a joke after the command:\n"
"Example: /addjoke Why did the chicken cross the road?"
)
return
joke_text = " ".join(context.args)
user = update.effective_user
joke_id = db.add_joke(joke_text, user.id, user.username)
await update.message.reply_text(
f"ā
Joke added successfully! (ID: {joke_id})\n"
"Others can now rate it with /like and /dislike"
)
New Bot Features:
š¤
Inline Buttons: Quick like/dislike without commands
š¾
User Context: Remember last joke for rating
š
Real Stats: Display actual like/dislike counts
š¤
User Tracking: Record who submitted each joke
Step 5: Rating Commands (10 minutes)
Add like/dislike commands and callback handlers
async def like_joke(update: Update, context: ContextTypes.DEFAULT_TYPE):
user_id = update.effective_user.id
if context.args and context.args[0].isdigit():
joke_id = int(context.args[0])
elif 'last_joke_id' in context.user_data:
joke_id = context.user_data['last_joke_id']
else:
await update.message.reply_text(
"Please view a joke first with /joke, or specify a joke ID: /like 123"
)
return
result = db.rate_joke(user_id, joke_id, 1)
joke = db.get_joke_by_id(joke_id)
if joke:
await update.message.reply_text(
f"š Liked joke #{joke_id}!\n"
f"Current rating: {result['likes']} š {result['dislikes']} š"
)
else:
await update.message.reply_text("Joke not found!")
async def dislike_joke(update: Update, context: ContextTypes.DEFAULT_TYPE):
user_id = update.effective_user.id
if context.args and context.args[0].isdigit():
joke_id = int(context.args[0])
elif 'last_joke_id' in context.user_data:
joke_id = context.user_data['last_joke_id']
else:
await update.message.reply_text(
"Please view a joke first with /joke, or specify a joke ID: /dislike 123"
)
return
result = db.rate_joke(user_id, joke_id, -1)
joke = db.get_joke_by_id(joke_id)
if joke:
await update.message.reply_text(
f"š Disliked joke #{joke_id}!\n"
f"Current rating: {result['likes']} š {result['dislikes']} š"
)
else:
await update.message.reply_text("Joke not found!")
async def button_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
query = update.callback_query
await query.answer()
data = query.data
user_id = query.from_user.id
if data.startswith('like_'):
joke_id = int(data.split('_')[1])
result = db.rate_joke(user_id, joke_id, 1)
await query.edit_message_text(
text=query.message.text.split('\n\n')[0] +
f"\n\nš {result['likes']} š {result['dislikes']}",
reply_markup=query.message.reply_markup
)
elif data.startswith('dislike_'):
joke_id = int(data.split('_')[1])
result = db.rate_joke(user_id, joke_id, -1)
await query.edit_message_text(
text=query.message.text.split('\n\n')[0] +
f"\n\nš {result['likes']} š {result['dislikes']}",
reply_markup=query.message.reply_markup
)
Rating Features:
1
Multiple Methods: Command or inline button
2
Flexible Targeting: By ID or last viewed joke
3
Real-time Updates: Buttons update instantly
4
Feedback: Users see their vote was counted
Step 6: Additional Features (10 minutes)
Add statistics, top jokes, and admin commands
async def top_jokes(update: Update, context: ContextTypes.DEFAULT_TYPE):
limit = 5
if context.args and context.args[0].isdigit():
limit = min(int(context.args[0]), 20)
jokes = db.get_top_jokes(limit)
if not jokes:
await update.message.reply_text("No jokes in database yet!")
return
response = f"š TOP {len(jokes)} JOKES š\n\n"
for i, joke in enumerate(jokes, 1):
score = joke['likes'] - joke['dislikes']
response += f"{i}. ID {joke['id']}: Score {score} (š{joke['likes']} š{joke['dislikes']})\n"
response += f" \"{joke['joke_text'][:50]}...\"\n\n"
await update.message.reply_text(response)
async def bot_stats(update: Update, context: ContextTypes.DEFAULT_TYPE):
total_jokes = db.get_total_jokes()
conn = db.get_connection()
cursor = conn.cursor()
cursor.execute("SELECT SUM(likes), SUM(dislikes) FROM jokes")
result = cursor.fetchone()
total_likes = result[0] or 0
total_dislikes = result[1] or 0
conn.close()
stats_text = f"""š JOKE BOT STATISTICS š
Total Jokes: {total_jokes}
Total Ratings: {total_likes + total_dislikes}
š Total Likes: {total_likes}
š Total Dislikes: {total_dislikes}
š Approval Rate: {(total_likes/(total_likes+total_dislikes)*100 if (total_likes+total_dislikes) > 0 else 0):.1f}%
Commands Today:
/joke - Get random joke
/addjoke - Submit your joke
/top - See top jokes
/stats - View these stats"""
await update.message.reply_text(stats_text)
async def my_jokes(update: Update, context: ContextTypes.DEFAULT_TYPE):
user_id = update.effective_user.id
conn = db.get_connection()
cursor = conn.cursor()
cursor.execute('''SELECT id, joke_text, likes, dislikes
FROM jokes WHERE user_id = ?
ORDER BY created_at DESC LIMIT 10''', (user_id,))
user_jokes = [dict(row) for row in cursor.fetchall()]
conn.close()
if not user_jokes:
await update.message.reply_text("You haven't submitted any jokes yet! Use /addjoke")
return
response = f"š YOUR JOKES ({len(user_jokes)} total)\n\n"
for joke in user_jokes:
score = joke['likes'] - joke['dislikes']
response += f"ID {joke['id']}: Score {score} (š{joke['likes']} š{joke['dislikes']})\n"
response += f" \"{joke['joke_text'][:60]}...\"\n\n"
await update.message.reply_text(response)
Enhanced Features:
š
Statistics: Track bot usage and engagement
š
Leaderboard: Top jokes ranking system
š¤
Personal Stats: Users see their own contributions
š
Analytics: Approval rate and engagement metrics
Step 7: Complete Main Function (5 minutes)
Put it all together and run the upgraded bot
def main():
BOT_TOKEN = "YOUR_BOT_TOKEN_HERE"
app = Application.builder().token(BOT_TOKEN).build()
app.add_handler(CommandHandler("start", start))
app.add_handler(CommandHandler("joke", send_joke))
app.add_handler(CommandHandler("addjoke", add_joke_command))
app.add_handler(CommandHandler("like", like_joke))
app.add_handler(CommandHandler("dislike", dislike_joke))
app.add_handler(CommandHandler("top", top_jokes))
app.add_handler(CommandHandler("stats", bot_stats))
app.add_handler(CommandHandler("myjokes", my_jokes))
app.add_handler(CallbackQueryHandler(button_callback))
print("š¤ Joke Bot 2.0 is running...")
print("š Database initialized: joke_bot.db")
print("šÆ Features: User submissions, ratings, leaderboard")
print("Press Ctrl+C to stop.\n")
app.run_polling()
if __name__ == "__main__":
main()
Project Structure:
joke_bot_2.0/
āāā app.py
āāā database.py
āāā joke_bot.db
āāā requirements.txt
āāā README.md
1
Install: pip install python-telegram-bot==20.3
3
Test: Open Telegram, find your bot, type /start
4
Deploy: Can run on Raspberry Pi, VPS, or cloud
Database Administration Tools
Tools to view and manage your SQLite database
import sqlite3
import pandas as pd
from datetime import datetime
class JokeAdmin:
def __init__(self, db_name='joke_bot.db'):
self.db_name = db_name
def export_to_csv(self, filename='jokes_export.csv'):
conn = sqlite3.connect(self.db_name)
df = pd.read_sql_query("SELECT * FROM jokes", conn)
df.to_csv(filename, index=False, encoding='utf-8')
conn.close()
print(f"Exported {len(df)} jokes to {filename}")
def show_stats(self):
conn = sqlite3.connect(self.db_name)
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM jokes")
total_jokes = cursor.fetchone()[0]
cursor.execute("SELECT COUNT(DISTINCT user_id) FROM jokes WHERE user_id != 0")
unique_users = cursor.fetchone()[0]
cursor.execute("SELECT SUM(likes), SUM(dislikes) FROM jokes")
likes, dislikes = cursor.fetchone()
cursor.execute('''SELECT username, COUNT(*) as joke_count
FROM jokes WHERE user_id != 0
GROUP BY user_id ORDER BY joke_count DESC LIMIT 5''')
top_contributors = cursor.fetchall()
conn.close()
print(f"""š DATABASE STATISTICS š
Total Jokes: {total_jokes}
Unique Contributors: {unique_users}
Total Likes: {likes or 0}
Total Dislikes: {dislikes or 0}
š Top Contributors:""")
for username, count in top_contributors:
print(f" {username}: {count} jokes")
def backup_database(self):
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
backup_name = f"backup_joke_bot_{timestamp}.db"
import shutil
shutil.copy2(self.db_name, backup_name)
print(f"Backup created: {backup_name}")
if __name__ == "__main__":
admin = JokeAdmin()
admin.show_stats()
admin.export_to_csv()
admin.backup_database()
Admin Features:
š¾
Backups: Automatic database backups
š
Analytics: Detailed usage statistics
š¤
Export: CSV export for analysis
š
Leaderboards: Top contributors recognition
SQLite Browser Tool:
Download DB Browser for SQLite (free) to visually explore your database:
⢠View tables and data
⢠Run SQL queries
⢠Export/import data
⢠Available for Windows, Mac, Linux
Advanced Features (Optional)
Take your joke bot to the next level
def add_categories():
conn = sqlite3.connect('joke_bot.db')
cursor = conn.cursor()
try:
cursor.execute("ALTER TABLE jokes ADD COLUMN category TEXT DEFAULT 'general'")
except sqlite3.OperationalError:
pass
cursor.execute('''CREATE TABLE IF NOT EXISTS categories (
id INTEGER PRIMARY KEY,
name TEXT UNIQUE,
description TEXT
)''')
categories = [
('programming', 'Programming and tech jokes'),
('dad', 'Classic dad jokes'),
('animal', 'Animal jokes'),
('school', 'School and education jokes')
]
cursor.executemany('''INSERT OR IGNORE INTO categories (name, description)
VALUES (?, ?)''', categories)
conn.commit()
conn.close()
async def send_daily_joke(context: ContextTypes.DEFAULT_TYPE):
joke = db.get_random_joke()
if joke:
conn = db.get_connection()
cursor = conn.cursor()
cursor.execute("SELECT user_id FROM subscribers WHERE active = 1")
users = cursor.fetchall()
conn.close()
for user_id in users:
try:
await context.bot.send_message(
chat_id=user_id[0],
text=f"š
Daily Joke!\n\n{joke['joke_text']}"
)
except Exception as e:
print(f"Failed to send to user {user_id}: {e}")
def search_jokes(search_term, limit=10):
conn = sqlite3.connect('joke_bot.db')
cursor = conn.cursor()
cursor.execute('''SELECT * FROM jokes
WHERE joke_text LIKE ?
ORDER BY (likes - dislikes) DESC
LIMIT ?''', (f'%{search_term}%', limit))
jokes = [dict(row) for row in cursor.fetchall()]
conn.close()
return jokes
Optional Enhancements:
š·ļø
Categories: Organize jokes by type
š
Notifications: Daily joke subscriptions
š
Search: Find jokes by keyword
š¤
AI Integration: Generate jokes with GPT API
š
Web Dashboard: View stats in browser
Testing Your Upgraded Bot
Comprehensive testing guide
Test Commands Sequence:
1. /start - See welcome message with all commands
2. /joke - Get random joke with rating buttons
3. Click š and š buttons - Test inline rating
4. /addjoke Why did the Python cross the road? To get to the other side!
5. /joke again - See if new joke appears
6. /like and /dislike - Test command rating
7. /top 3 - See top jokes leaderboard
8. /stats - Check bot statistics
9. /myjokes - View your submissions
Expected Results:
ā
Database File: joke_bot.db created automatically
ā
Default Jokes: 3 jokes pre-loaded
ā
Rating Persistence: Votes saved between restarts
ā
User Tracking: Your username saved with submissions
ā
Weighted Random: Better jokes appear more often
import sqlite3
conn = sqlite3.connect('joke_bot.db')
cursor = conn.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
tables = cursor.fetchall()
print("Tables:", tables)
cursor.execute("SELECT COUNT(*) FROM jokes")
count = cursor.fetchone()[0]
print(f"Total jokes: {count}")
conn.close()
Complete Project Review
What you've accomplished in 60 minutes
š CONGRATULATIONS! š
You've successfully upgraded your simple joke bot to a full-featured application with:
Core Upgrades:
ā
SQLite Database: Persistent storage for jokes
ā
User Submissions: Community-driven content
ā
Rating System: Like/dislike with tracking
ā
Inline Buttons: Better user experience
ā
Statistics: Track bot usage and engagement
ā
Leaderboard: Top jokes ranking
Technical Skills Learned:
š
Python: Advanced Python programming
šļø
SQLite: Database design and queries
š¤
Telegram Bot API: Advanced bot features
š
Data Modeling: Table design and relationships
š§
Software Architecture: Modular, maintainable code
Next Steps:
1. Add joke categories (/programming, /dadjokes)
2. Implement daily joke notifications
3. Create a web dashboard for statistics
4. Add joke search functionality
5. Implement user profiles and achievements
6. Add multi-language support
Your Joke Bot is Now Production-Ready! š
Share it with friends, deploy it online, and watch your joke collection grow!