HoseinA05/DB-Management-Telegram-Bot
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

added authentication to the bot. added bot commnads. added create button for when the db is empty or no data is available to show.

Browse Source
This commit is contained in:
Hosein
2026-01-05 23:40:13 +03:30
parent 4b5b24c97c
commit caac6c3a08
18 changed files with 369 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bot/callbacks/categories.py
+12 -3
View File
@@ -1,5 +1,5 @@
from telebot import types, TeleBot
from database.models import Categories
from database.models import Categories, Admins
from bot.handlers.start import startMarkup
from bot.utils.crud_helpers import create_entity_markup
@@ -38,7 +38,7 @@ def register(bot: TeleBot):
<b>Description:</b> \n{category[2]}
"""
details.strip()
markup = create_entity_markup("category", category_id)
markup = create_entity_markup("category", category_id, True)
bot.send_message(call.message.chat.id, details,
reply_markup=markup, parse_mode="HTML")
@@ -91,9 +91,13 @@ def register(bot: TeleBot):
message.chat.id, "❌ Failed to create category.", reply_markup=startMarkup())
# Editing a Category Flow
@bot.callback_query_handler(func=lambda call: call.data.startswith('edit_category_'))
def start_category_editing(call):
if not Admins.is_authenticated(call.from_user.id):
bot.answer_callback_query(call.id, "⛔ Unauthorized access!")
bot.send_message(call.message.chat.id, "Please /login first.")
return
category_id = call.data.split('_')[2]
category = Categories.getCategorieById(category_id)
@@ -150,6 +154,11 @@ def register(bot: TeleBot):
# Deleting a Category
@bot.callback_query_handler(func=lambda call: call.data.startswith('delete_category_'))
def delete_category(call):
if not Admins.is_authenticated(call.from_user.id):
bot.answer_callback_query(call.id, "⛔ Unauthorized access!")
bot.send_message(call.message.chat.id, "Please /login first.")
return
category_id = call.data.split('_')[2]
if (Categories.deleteCategory(category_id)):
bot.send_message(call.message.chat.id, "✅ Category deleted.",
bot/callbacks/courses.py
+12 -2
View File
@@ -1,5 +1,5 @@
from telebot import TeleBot, types
from database.models import Courses
from database.models import Courses, Admins
from bot.utils.crud_helpers import create_entity_markup
from bot.utils.formatters import format_course_info
from bot.handlers.start import startMarkup
@@ -38,7 +38,7 @@ def register(bot: TeleBot):
return
details = format_course_info(course)
markup = create_entity_markup("course", course_id)
markup = create_entity_markup("course", course_id, True)
markup.add(types.InlineKeyboardButton(
f"👨‍🏫 Teacher's Info", callback_data=f"teacher_{course[3]}"))
@@ -93,6 +93,11 @@ def register(bot: TeleBot):
# Editing a Course Flow
@bot.callback_query_handler(func=lambda call: call.data.startswith('edit_course_'))
def start_course_editing(call):
if not Admins.is_authenticated(call.from_user.id):
bot.answer_callback_query(call.id, "⛔ Unauthorized access!")
bot.send_message(call.message.chat.id, "Please /login first.")
return
course_id = call.data.split('_')[2]
course = Courses.getCourseById(course_id)
@@ -149,6 +154,11 @@ def register(bot: TeleBot):
# Deleting a Teacher
@bot.callback_query_handler(func=lambda call: call.data.startswith('delete_course_'))
def delete_course(call):
if not Admins.is_authenticated(call.from_user.id):
bot.answer_callback_query(call.id, "⛔ Unauthorized access!")
bot.send_message(call.message.chat.id, "Please /login first.")
return
course_id = call.data.split('_')[2]
if (Courses.deleteCourse(course_id)):
bot.send_message(call.message.chat.id, "✅ Course deleted.",
bot/callbacks/students.py
+22 -2
View File
@@ -1,5 +1,5 @@
from telebot import types, TeleBot
from database.models import Students
from database.models import Students, Admins
from bot.utils.formatters import format_student_info
from bot.utils.crud_helpers import create_entity_markup
from bot.handlers.start import startMarkup
@@ -30,12 +30,17 @@ def register(bot: TeleBot):
# Showing details of a Student
@bot.callback_query_handler(func=lambda call: call.data.startswith('student_'))
def show_student_details(call):
if not Admins.is_authenticated(call.from_user.id):
bot.answer_callback_query(call.id, "⛔ Unauthorized access!")
bot.send_message(call.message.chat.id, "Please /login first.")
return
student_id = call.data.split('_')[1]
student = Students.getStudentById(student_id)
if student:
details = format_student_info(student)
markup = create_entity_markup("student", student_id)
markup = create_entity_markup("student", student_id, True)
bot.send_message(call.message.chat.id, details,
reply_markup=markup, parse_mode="HTML")
@@ -47,6 +52,11 @@ def register(bot: TeleBot):
# Creating a Student Flow
@bot.callback_query_handler(func=lambda call: call.data == 'create_student')
def create_student(call):
if not Admins.is_authenticated(call.from_user.id):
bot.answer_callback_query(call.id, "⛔ Unauthorized access!")
bot.send_message(call.message.chat.id, "Please /login first.")
return
msg = bot.send_message(call.message.chat.id,
"Please enter following data: (enter any key to start)",
reply_markup=cancelMarkup)
@@ -90,6 +100,11 @@ def register(bot: TeleBot):
# Editing a Student Flow
@bot.callback_query_handler(func=lambda call: call.data.startswith('edit_student_'))
def start_student_editing(call):
if not Admins.is_authenticated(call.from_user.id):
bot.answer_callback_query(call.id, "⛔ Unauthorized access!")
bot.send_message(call.message.chat.id, "Please /login first.")
return
student_id = call.data.split('_')[2]
student = Students.getStudentById(student_id)
@@ -146,6 +161,11 @@ def register(bot: TeleBot):
# Deleting a Teacher
@bot.callback_query_handler(func=lambda call: call.data.startswith('delete_student_'))
def delete_student(call):
if not Admins.is_authenticated(call.from_user.id):
bot.answer_callback_query(call.id, "⛔ Unauthorized access!")
bot.send_message(call.message.chat.id, "Please /login first.")
return
student_id = call.data.split('_')[2]
if (Students.deleteStudent(student_id)):
bot.send_message(call.message.chat.id, "✅ Student deleted.",
bot/callbacks/tags.py
+12 -3
View File
@@ -1,5 +1,5 @@
from telebot import types, TeleBot
from database.models import Tags
from database.models import Tags, Admins
from bot.handlers.start import startMarkup
from bot.utils.crud_helpers import create_entity_markup
@@ -34,7 +34,7 @@ def register(bot: TeleBot):
<b>slug:</b> {tag[2]}
"""
details.strip()
markup = create_entity_markup("tag", tag_id)
markup = create_entity_markup("tag", tag_id, True)
bot.send_message(call.message.chat.id, details,
reply_markup=markup, parse_mode="HTML")
@@ -87,9 +87,13 @@ def register(bot: TeleBot):
message.chat.id, "❌ Failed to create tag.", reply_markup=startMarkup())
# Editing a Tag Flow
@bot.callback_query_handler(func=lambda call: call.data.startswith('edit_tag_'))
def start_tag_editing(call):
if not Admins.is_authenticated(call.from_user.id):
bot.answer_callback_query(call.id, "⛔ Unauthorized access!")
bot.send_message(call.message.chat.id, "Please /login first.")
return
tag_id = call.data.split('_')[2]
tag = Tags.getTagById(tag_id)
@@ -146,6 +150,11 @@ def register(bot: TeleBot):
# Deleting a Tag
@bot.callback_query_handler(func=lambda call: call.data.startswith('delete_tag_'))
def delete_tag(call):
if not Admins.is_authenticated(call.from_user.id):
bot.answer_callback_query(call.id, "⛔ Unauthorized access!")
bot.send_message(call.message.chat.id, "Please /login first.")
return
tag_id = call.data.split('_')[2]
if (Tags.deleteTag(tag_id)):
bot.send_message(call.message.chat.id, "✅ Tag deleted.",
bot/callbacks/teachers.py
+18 -4
View File
@@ -1,12 +1,11 @@
from telebot import types, TeleBot
from database.models import Teachers
from database.models import Teachers, Admins
from bot.utils.formatters import format_teacher_info
from bot.handlers.start import startMarkup
from bot.utils.crud_helpers import create_entity_markup
# TODO: Add Buttons for skipping Optional Fields (Add to all entities).
# TODO: Add Option for handling all updates at once.
# TODO: Add Authentication for sensitive actions and info.
# TODO: Add Option for handling all updates at once.\
# TODO: Add Option for seeing courses taught by a teacher in Teacher Details.
@@ -45,7 +44,7 @@ def register(bot: TeleBot):
if teacher:
details = format_teacher_info(teacher)
markup = create_entity_markup("teacher", teacher_id)
markup = create_entity_markup("teacher", teacher_id, True)
bot.send_message(call.message.chat.id, details,
reply_markup=markup, parse_mode="HTML")
@@ -57,6 +56,11 @@ def register(bot: TeleBot):
# Creating a Teacher Flow
@bot.callback_query_handler(func=lambda call: call.data == 'create_teacher')
def start_teacher_creation(call):
if not Admins.is_authenticated(call.from_user.id):
bot.answer_callback_query(call.id, "⛔ Unauthorized access!")
bot.send_message(call.message.chat.id, "Please /login first.")
return
msg = bot.send_message(call.message.chat.id,
"Please enter following data: (enter any key to start)",
reply_markup=cancelMarkup)
@@ -100,6 +104,11 @@ def register(bot: TeleBot):
# Editing a Teacher Flow
@bot.callback_query_handler(func=lambda call: call.data.startswith('edit_teacher_'))
def start_teacher_editing(call):
if not Admins.is_authenticated(call.from_user.id):
bot.answer_callback_query(call.id, "⛔ Unauthorized access!")
bot.send_message(call.message.chat.id, "Please /login first.")
return
teacher_id = call.data.split('_')[2]
teacher = Teachers.getTeacherById(teacher_id)
@@ -156,6 +165,11 @@ def register(bot: TeleBot):
# Deleting a Teacher
@bot.callback_query_handler(func=lambda call: call.data.startswith('delete_teacher_'))
def delete_teacher(call):
if not Admins.is_authenticated(call.from_user.id):
bot.answer_callback_query(call.id, "⛔ Unauthorized access!")
bot.send_message(call.message.chat.id, "Please /login first.")
return
teacher_id = call.data.split('_')[2]
if (Teachers.deleteTeacher(teacher_id)):
bot.send_message(call.message.chat.id, "✅ Teacher deleted.",
bot/handlers/categories.py
+5 -1
View File
@@ -25,7 +25,11 @@ def register(bot):
bot.send_message(
message.chat.id, "Here is the data:", reply_markup=markup)
else:
bot.reply_to(message, "No data found.")
markup = types.InlineKeyboardMarkup(row_width=1)
markup.add(types.InlineKeyboardButton(
"➕ Create New Category", callback_data="create_category"))
bot.reply_to(message, "No data found.", reply_markup=markup)
except Exception as e:
logger.error(f"Error in 'Show Categories' handler: {e}")
bot.reply_to(message, "Sorry, an error occurred.")
bot/handlers/common.py
+6 -1
View File
@@ -1,4 +1,9 @@
from telebot import types
from bot.handlers.start import startMessage
def register(bot):
@bot.message_handler(func=lambda message: True)
def handle_unknown(message):
bot.reply_to(message, f"You said: {message.text}")
bot.reply_to(
message, f"You said: {message.text}. send /start to get started.")
bot/handlers/courses.py
+4 -1
View File
@@ -24,7 +24,10 @@ def register(bot):
bot.send_message(
message.chat.id, "Here is the data:", reply_markup=markup)
else:
bot.reply_to(message, "No data found.")
markup = types.InlineKeyboardMarkup(row_width=1)
markup.add(types.InlineKeyboardButton(
"➕ Create New Course", callback_data="create_course"))
bot.reply_to(message, "No data found.", reply_markup=markup)
except Exception as e:
logger.error(f"Error in (Show Courses) handler: {e}")
bot.reply_to(message, "Sorry, an error occurred.")
bot/handlers/init.py
+15 -1
View File
@@ -1,11 +1,25 @@
from . import start, students, teachers, courses, tags, categories, common
from . import start, students, teachers, courses, tags, categories, common, login
from bot.callbacks.init import register_all_callbacks
from telebot import types
def register_all_handlers(bot):
"""Register all handlers and callbacks"""
# Set bot commands
commands = [
types.BotCommand(command="start", description="Start the bot"),
types.BotCommand(command="login", description="Login to your account"),
types.BotCommand(command="logout",
description="Logout of your account"),
types.BotCommand(command="session",
description="Info about your session"),
]
bot.set_my_commands(commands)
# Message handlers
start.register(bot)
login.register(bot)
students.register(bot)
teachers.register(bot)
courses.register(bot)
bot/handlers/login.py
+110
View File
@@ -0,0 +1,110 @@
from telebot import types, TeleBot
from database.models import Admins
from bot.handlers.start import startMarkup
def register(bot: TeleBot):
@bot.message_handler(commands=['login'])
def start_login(message):
# Check if already authenticated
if Admins.is_authenticated(message.from_user.id):
session_info = Admins.get_session_info(message.from_user.id)
bot.send_message(
message.chat.id,
f"✅ You are already logged in!\n\n"
f"Session expires in: {session_info['time_remaining']}"
)
return
msg = bot.send_message(
message.chat.id,
"🔐 Please enter your username:"
)
bot.register_next_step_handler(msg, get_password)
def get_password(message):
"""Get username and ask for password"""
username = message.text.strip()
try:
bot.delete_message(message.chat.id, message.message_id)
except:
pass
msg = bot.send_message(
message.chat.id,
f"Username: {username}\n\n🔑 Now enter your password:"
)
bot.register_next_step_handler(msg, verify_and_login, username)
def verify_and_login(message, username):
"""Verify credentials and create session"""
password = message.text.strip()
telegram_id = message.from_user.id
try:
bot.delete_message(message.chat.id, message.message_id)
except:
pass
# Attempt login (this creates the session)
if Admins.login(username, password, telegram_id):
bot.send_message(
message.chat.id,
f"✅ Login successful!\n\n"
f"Welcome back, {username}!\n"
f"Session valid for: {Admins.SESSION_TIMEOUT_HOURS} hours\n"
f"Use /session for info about your session, or /logout to logout.",
reply_markup=startMarkup()
)
else:
bot.send_message(
message.chat.id,
"❌ Invalid credentials or unauthorized Telegram account.\n\n"
"Please check your username and password.\nContact @HoseinA05 for support."
)
@bot.message_handler(commands=['logout'])
def logout(message):
"""Logout user manually"""
if not Admins.is_authenticated(message.from_user.id):
bot.send_message(message.chat.id, "You are not logged in.")
return
if Admins.logout(message.from_user.id):
bot.send_message(
message.chat.id,
"✅ You have been logged out successfully.\n\n"
"Use /login to access admin features again."
)
else:
bot.send_message(
message.chat.id, "❌ Logout failed. Please try again later.")
@bot.message_handler(commands=['session'])
def check_session(message):
"""Check current session status"""
session_info = Admins.get_session_info(message.from_user.id)
if not session_info:
bot.send_message(
message.chat.id, "❌ You don't have an admin account.")
return
if session_info['is_active']:
bot.send_message(
message.chat.id,
f"✅ Active Session\n\n"
f"Username: {session_info['username']}\n"
f"Last Login: {session_info['last_login']}\n"
f"Time Remaining: {session_info['time_remaining']}"
)
else:
bot.send_message(
message.chat.id,
f"⏱️ Session Expired\n\n"
f"Username: {session_info['username']}\n"
f"Please use /login to continue."
)
bot/handlers/start.py
+10 -10
View File
@@ -1,5 +1,4 @@
from telebot import types
from bot.utils.auth import is_authenticated
# TODO: Add Pagincation to Showing rows of entities (To all entities)
@@ -7,16 +6,17 @@ from bot.utils.auth import is_authenticated
def register(bot):
@bot.message_handler(commands=["start"])
def start_handler(message):
if is_authenticated(message.from_user.id):
bot.send_message(
message.chat.id, "authorized user! Welcome.")
else:
bot.send_message(
message.chat.id, "Welcome! Use buttons to fetch from database.")
startMessage(bot, message=message)
markup = startMarkup()
bot.reply_to(message, "Use buttons to fetch from database.",
reply_markup=markup)
def startMessage(bot, message):
# if is_authenticated(message.from_user.id):
# bot.send_message(
# message.chat.id, "authorized user! Welcome.")
markup = startMarkup()
bot.reply_to(message, "Use buttons to fetch from database.",
reply_markup=markup)
def startMarkup():
bot/handlers/students.py
+13 -5
View File
@@ -1,13 +1,18 @@
import logging
from telebot import types
from database.models import Students
from telebot import types, TeleBot
from database.models import Students, Admins
logger = logging.getLogger(__name__)
def register(bot):
def register(bot: TeleBot):
@bot.message_handler(func=lambda message: message.text == "Show Students")
def get_students(message):
if not Admins.is_authenticated(message.from_user.id):
bot.send_message(
message.chat.id, "⛔ Unauthorized access!\nPlease /login first.")
return
try:
data = Students.getAllStudents()
if data:
@@ -20,12 +25,15 @@ def register(bot):
# add button for creating a new student
markup.add(types.InlineKeyboardButton(
"➕ Create New Student", callback_data="create_student"))
"➕ Create New Student 🔒", callback_data="create_student"))
bot.send_message(
message.chat.id, "Here is the data:", reply_markup=markup)
else:
bot.reply_to(message, "No data found.")
markup = types.InlineKeyboardMarkup(row_width=1)
markup.add(types.InlineKeyboardButton(
"➕ Create New Student 🔒", callback_data="create_student"))
bot.reply_to(message, "No data found.", reply_markup=markup)
except Exception as e:
logger.error(f"Error in get_students handler: {e}")
bot.reply_to(message, "Sorry, an error occurred.")
bot/handlers/tags.py
+2 -2
View File
@@ -25,11 +25,11 @@ def register(bot: TeleBot):
bot.send_message(
message.chat.id, "Here is the data:", reply_markup=markup)
else:
markup = types.InlineKeyboardMarkup(row_width=2)
markup = types.InlineKeyboardMarkup(row_width=1)
markup.add(types.InlineKeyboardButton(
"➕ Create New Tag", callback_data="create_tag"))
bot.reply_to(message, "No data found.", markup=markup)
bot.reply_to(message, "No data found.", reply_markup=markup)
except Exception as e:
logger.error(f"Error in 'Show Tags' handler: {e}")
bot.reply_to(message, "Sorry, an error occurred.")
bot/handlers/teachers.py
+5 -2
View File
@@ -19,12 +19,15 @@ def register(bot):
markup.add(btn)
# add button for creating a new teacher
markup.add(types.InlineKeyboardButton(
"➕ Create New Teacher", callback_data="create_teacher"))
"➕ Create New Teacher 🔒", callback_data="create_teacher"))
bot.send_message(
message.chat.id, "Here is the data:", reply_markup=markup)
else:
bot.reply_to(message, "No data found.")
markup = types.InlineKeyboardMarkup(row_width=1)
markup.add(types.InlineKeyboardButton(
"➕ Create New Teacher 🔒", callback_data="create_teacher"))
bot.reply_to(message, "No data found.", reply_markup=markup)
except Exception as e:
logger.error(f"Error in get_teachers handler: {e}")
bot.reply_to(message, "Sorry, an error occurred.")
bot/utils/auth.py
-4
View File
@@ -1,4 +0,0 @@
authenticated_users = {2032936226}
def is_authenticated(user_id):
return user_id in authenticated_users
bot/utils/crud_helpers.py
+3 -3
View File
@@ -1,11 +1,11 @@
from telebot import types
def create_entity_markup(entity_name, enitity_id):
def create_entity_markup(entity_name, enitity_id, is_auth_needed=False):
markup = types.InlineKeyboardMarkup()
markup.add(types.InlineKeyboardButton(
f"✏️ Edit {entity_name}", callback_data=f"edit_{entity_name}_{enitity_id}"))
f"✏️ Edit {entity_name} " + ("🔒" if is_auth_needed else ""), callback_data=f"edit_{entity_name}_{enitity_id}"))
markup.add(types.InlineKeyboardButton(
f"🗑️ Delete {entity_name}", callback_data=f"delete_{entity_name}_{enitity_id}"))
f"🗑️ Delete {entity_name}" + ("🔒" if is_auth_needed else ""), callback_data=f"delete_{entity_name}_{enitity_id}"))
return markup
database/models.py
+116
View File
@@ -1,6 +1,9 @@
import logging
import database.connection as connection
import hashlib
from datetime import datetime, timedelta
logger = logging.getLogger(__name__)
@@ -55,6 +58,119 @@ class BaseRepository:
connection.release_db_connection(conn)
class Admins(BaseRepository):
SESSION_TIMEOUT_HOURS = 1
@staticmethod
def hash_password(password: str) -> str:
"""Hash password using SHA256"""
return hashlib.sha256(password.encode()).hexdigest()
@staticmethod
def login(username: str, password: str, telegram_id: int) -> bool:
"""
Verify credentials and create a session.
Returns True if login successful, False otherwise.
"""
def query(cursor):
password_hash = Admins.hash_password(password)
cursor.execute(
"SELECT id FROM admins WHERE username = %s AND password_hash = %s AND telegram_id = %s",
(username, password_hash, telegram_id)
)
result = cursor.fetchone()
return result is not None
result = Admins._execute_query(query, "login")
if result:
# Create session by updating last_login and session_expires_at
def mutation(cursor, conn):
expires_at = datetime.now() + timedelta(hours=Admins.SESSION_TIMEOUT_HOURS)
cursor.execute(
"""UPDATE admins
SET last_login = CURRENT_TIMESTAMP,
session_expires_at = %s
WHERE telegram_id = %s""",
(expires_at, telegram_id)
)
return Admins._execute_mutation(mutation, "login")
return False
@staticmethod
def is_authenticated(telegram_id: int) -> bool:
"""Check if user has a valid active session"""
def query(cursor):
cursor.execute(
"SELECT session_expires_at FROM admins WHERE telegram_id = %s",
(telegram_id,)
)
result = cursor.fetchone()
if not result or result[0] is None:
return False
# Check if session has expired
expires_at = result[0]
if isinstance(expires_at, str):
expires_at = datetime.fromisoformat(expires_at)
return datetime.now() < expires_at
result = Admins._execute_query(query, "is_authenticated")
return result if result is not None else False
@staticmethod
def logout(telegram_id: int) -> bool:
"""Logout admin by clearing session"""
def mutation(cursor, conn):
cursor.execute(
"UPDATE admins SET session_expires_at = NULL WHERE telegram_id = %s",
(telegram_id,)
)
return Admins._execute_mutation(mutation, "logout")
@staticmethod
def get_session_info(telegram_id: int) -> dict:
"""Get session info for the admin"""
def query(cursor):
cursor.execute(
"SELECT username, last_login, session_expires_at FROM admins WHERE telegram_id = %s",
(telegram_id,)
)
result = cursor.fetchone()
return result
result = Admins._execute_query(query, "get_session_info")
if not result:
return None
session_info = {
'username': result[0],
'last_login': result[1],
'expires_at': result[2],
'is_active': False
}
if result[2]:
expires_at = result[2]
if isinstance(expires_at, str):
expires_at = datetime.fromisoformat(expires_at)
session_info['is_active'] = datetime.now() < expires_at
session_info['time_remaining'] = str(
expires_at - datetime.now()).split('.')[0]
return session_info
class Students(BaseRepository):
table_name = "students"
dev.py
+4 -3
View File
@@ -7,6 +7,7 @@ load_dotenv()
BOT_TOKEN = os.environ.get("BOT_TOKEN")
def ensure_no_webhook():
"""Remove webhook before starting polling"""
response = requests.post(
@@ -15,13 +16,13 @@ def ensure_no_webhook():
if response.json()['ok']:
print("✅ Webhook removed, starting polling...")
return True
print("⚠️ Warning: Could not remove webhook")
return False
if __name__ == "__main__":
if ensure_no_webhook():
bot = get_bot()
print("🤖 Bot is polling locally...")
bot.polling(non_stop=True)
bot.polling(non_stop=True)