yigit/SecretarX
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Multi User support

Browse Source
This commit is contained in:
Yigit Colakoglu 2024-09-10 13:57:34 +02:00
parent 468bf36f52
commit 259426df93
7 changed files with 483 additions and 158 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.gitignore vendored
View File

@ -4,6 +4,4 @@ __pycache__
.wsl
chromedriver.exe
chromedriver
config.py
default_config.py
data.db

6
config.py Normal file
View File

@ -0,0 +1,6 @@
import os
TELEGRAM_TOKEN = os.environ.get("SECRETARX_TG_TOKEN", None)
DB_PATH = "data.db"
assert TELEGRAM_TOKEN

5
main.py
View File

@ -1,7 +1,6 @@
from xclient import XClient
from tgbot import TelegramBot
import config
if __name__ == "__main__":
client = XClient("ycolakoglu", "mTJNhBZI8PBdIx0e0Lak")
telegram_bot = TelegramBot("7312187888:AAG9w2UhU8zu9CZTxRnh72ltSqa59friRs0")
telegram_bot = TelegramBot(config.TELEGRAM_TOKEN, config.DB_PATH)
telegram_bot.run()

4
requirements.txt Normal file
View File

@ -0,0 +1,4 @@
pysqlite3
selenium
requests
telebot

343
tgbot.py
View File

@ -3,25 +3,157 @@ from telebot.types import InlineKeyboardMarkup, InlineKeyboardButton
from datetime import datetime, timedelta, timezone
import time
from threading import Thread
import sqlite3
from xclient import XClient
class TelegramBot:
def __init__(self, bot_token, xclient):
def __init__(self, bot_token, db_path="bot_database.db"):
self.bot = telebot.TeleBot(bot_token)
self.xclient = xclient
self.xclients = {}
self.user_selected_slot = {} # Store user selected slot by chat ID
self.user_bookings = {} # Store user bookings by chat ID
self.watchlist = {} # Watchlist to store full slots by chat ID
self.db_path = db_path
self.init_db()
# Set up the command handlers
@self.bot.message_handler(commands=['start'])
@self.bot.message_handler(commands=["start"])
def send_welcome(message):
self.bot.reply_to(message, "Welcome! Use /make_booking to view and book available slots.")
self.handle_login(message)
@self.bot.message_handler(commands=['make_booking'])
@self.bot.message_handler(commands=["book"])
def make_booking(message):
self.make_booking(message)
@self.bot.callback_query_handler(
func=lambda call: call.data.startswith("book_")
)
def callback_booking(call):
self.callback_booking(call)
@self.bot.message_handler(commands=["cancel"])
def cancel_booking(message):
self.cancel_booking(message)
@self.bot.callback_query_handler(
func=lambda call: call.data.startswith("cancel_")
)
def callback_cancel_booking(call):
self.callback_cancel_booking(call)
@self.bot.message_handler(commands=["watchlist"])
def manage_watchlist(call):
self.manage_watchlist(call)
@self.bot.callback_query_handler(
func=lambda call: call.data.startswith("remove_")
)
def callback_remove_slot(call):
self.callback_remove_slot(call)
def init_db(self):
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
# Create tables
cursor.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
chat_id INTEGER UNIQUE,
username TEXT,
password TEXT
)
""")
conn.commit()
def get_xclient(self, chat_id):
if chat_id not in self.xclients:
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute(
"SELECT username, password FROM users WHERE chat_id = ?", (chat_id,)
)
result = cursor.fetchone()
if result:
username, password = result
self.xclients[chat_id] = XClient(username, password)
else:
self.xclients[chat_id] = None
return self.xclients[chat_id]
def handle_login(self, message):
chat_id = message.chat.id
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute("SELECT username FROM users WHERE chat_id = ?", (chat_id,))
result = cursor.fetchone()
if result:
self.bot.reply_to(message, "You are already logged in.")
else:
self.bot.reply_to(
message,
"Please provide your username and password in the format:\n`/login username password`",
)
self.bot.register_next_step_handler(message, self.process_login)
def process_login(self, message):
try:
command, username, password = message.text.split()
chat_id = message.chat.id
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute(
"SELECT * FROM users WHERE chat_id = ?",
(chat_id,),
)
user = cursor.fetchone()
if user:
# Update the row
cursor.execute(
"UPDATE users SET username = ?, password = ? WHERE chat_id = ?",
(username, password, chat_id),
)
self.bot.reply_to(message, f"Updated login details for {username}.")
else:
# Insert new row
cursor.execute(
"INSERT INTO users (chat_id, username, password) VALUES (?, ?, ?)",
(chat_id, username, password),
)
self.bot.reply_to(message, f"Registered new user {username}.")
except ValueError:
self.bot.reply_to(
message, "Invalid format. Please use `/login username password`."
)
def send_welcome(self, message):
self.bot.reply_to(
message, "Welcome! Use /make_booking to view and book available slots."
)
def make_booking(self, message):
try:
xclient = self.get_xclient(message.chat.id)
except Exception as e:
self.bot.reply_to(message, f"Error: {str(e)}")
return
if not xclient:
self.bot.reply_to(
message,
"You are not logged in. Please use `/login username password` to log in.",
)
return
start = datetime.now()
end = start + timedelta(days=1)
slots = self.xclient.list_slots(start, end)
slots = xclient.list_slots(start, end)
if slots:
markup = InlineKeyboardMarkup()
@ -31,16 +163,30 @@ class TelegramBot:
button_text = f"Slot {i + 1}: {slot.start_stamp} ({status})"
markup.add(InlineKeyboardButton(button_text, callback_data=f"book_{i}"))
self.bot.reply_to(message, "Select a slot to book or watch:", reply_markup=markup)
self.bot.reply_to(
message, "Select a slot to book or watch:", reply_markup=markup
)
self.user_selected_slot[message.chat.id] = slots
else:
self.bot.reply_to(message, "No slots found.")
@self.bot.callback_query_handler(func=lambda call: call.data.startswith("book_"))
def callback_booking(call):
def callback_booking(self, call):
try:
xclient = self.get_xclient(call.message.chat.id)
except Exception as e:
self.bot.reply_to(call.message, f"Error: {str(e)}")
return
if not xclient:
self.bot.reply_to(
call.message,
"You are not logged in. Please use `/login username password` to log in.",
)
return
try:
# Extract slot index from callback data
slot_index = int(call.data.split('_')[1])
slot_index = int(call.data.split("_")[1])
slots = self.user_selected_slot.get(call.message.chat.id, [])
if 0 <= slot_index < len(slots):
@ -49,55 +195,107 @@ class TelegramBot:
if selected_slot.available:
# Attempt to book the available slot
try:
self.xclient.make_booking(selected_slot)
self.bot.answer_callback_query(call.id, "Slot booked successfully!")
self.bot.send_message(call.message.chat.id, f"Booking confirmed for: {selected_slot.start_stamp}")
xclient.make_booking(selected_slot)
self.bot.answer_callback_query(
call.id, "Slot booked successfully!"
)
self.bot.send_message(
call.message.chat.id,
f"Booking confirmed for: {selected_slot.start_stamp}",
)
except Exception as e:
self.bot.answer_callback_query(call.id, "Failed to book slot.")
self.bot.send_message(call.message.chat.id, f"Error: {str(e)}")
else:
# Slot is full, add to watchlist
self.add_to_watchlist(call.message.chat.id, selected_slot)
self.bot.answer_callback_query(call.id, "Slot is full. Added to watchlist.")
self.bot.send_message(call.message.chat.id, f"Slot {selected_slot.start_stamp} is full. You will be notified when it becomes available.")
self.bot.answer_callback_query(
call.id, "Slot is full. Added to watchlist."
)
self.bot.send_message(
call.message.chat.id,
f"Slot {selected_slot.start_stamp} is full. You will be notified when it becomes available.",
)
else:
self.bot.answer_callback_query(call.id, "Invalid slot selection.")
except (IndexError, ValueError):
self.bot.answer_callback_query(call.id, "Invalid data.")
@self.bot.message_handler(commands=['cancel_booking'])
def cancel_booking(message):
## Remove the buttons from the previous message
self.bot.edit_message_reply_markup(
call.message.chat.id, call.message.message_id
)
def cancel_booking(self, message):
try:
xclient = self.get_xclient(message.chat.id)
except Exception as e:
self.bot.reply_to(message, f"Error: {str(e)}")
return
if not xclient:
self.bot.reply_to(
message,
"You are not logged in. Please use `/login username password` to log in.",
)
return
# Fetch user's current bookings
bookings = self.xclient.my_bookings(datetime.now(), datetime.now() + timedelta(days=31))
bookings = xclient.my_bookings(
datetime.now(), datetime.now() + timedelta(days=31)
)
if bookings:
markup = InlineKeyboardMarkup()
for i, booking in enumerate(bookings):
# Create a button for each booking
button_text = f"Booking {i + 1}: {booking.start_stamp}"
markup.add(InlineKeyboardButton(button_text, callback_data=f"cancel_{i}"))
markup.add(
InlineKeyboardButton(button_text, callback_data=f"cancel_{i}")
)
self.user_bookings[message.chat.id] = bookings
self.bot.reply_to(message, "Select a booking to cancel:", reply_markup=markup)
self.bot.reply_to(
message, "Select a booking to cancel:", reply_markup=markup
)
else:
self.bot.reply_to(message, "You have no bookings to cancel.")
@self.bot.callback_query_handler(func=lambda call: call.data.startswith("cancel_"))
def callback_cancel_booking(call):
def callback_cancel_booking(self, call):
try:
xclient = self.get_xclient(call.message.chat.id)
except Exception as e:
self.bot.reply_to(call.message, f"Error: {str(e)}")
return
if not xclient:
self.bot.reply_to(
call.message,
"You are not logged in. Please use `/login username password` to log in.",
)
return
try:
# Extract booking index from callback data
booking_index = int(call.data.split('_')[1])
booking_index = int(call.data.split("_")[1])
bookings = self.user_bookings.get(call.message.chat.id, [])
if 0 <= booking_index < len(bookings):
selected_booking = bookings[booking_index]
# Attempt to cancel the selected booking
try:
if self.xclient.cancel_booking(selected_booking):
self.bot.answer_callback_query(call.id, "Booking canceled successfully!")
self.bot.send_message(call.message.chat.id, f"Booking for {selected_booking.start_stamp} has been canceled.")
if xclient.cancel_booking(selected_booking):
self.bot.answer_callback_query(
call.id, "Booking canceled successfully!"
)
self.bot.send_message(
call.message.chat.id,
f"Booking for {selected_booking.start_stamp} has been canceled.",
)
else:
self.bot.answer_callback_query(call.id, "Failed to cancel booking.")
self.bot.answer_callback_query(
call.id, "Failed to cancel booking."
)
except Exception as e:
self.bot.answer_callback_query(call.id, "Failed to cancel booking.")
self.bot.send_message(call.message.chat.id, f"Error: {str(e)}")
@ -106,6 +304,62 @@ class TelegramBot:
except (IndexError, ValueError):
self.bot.answer_callback_query(call.id, "Invalid data.")
## Remove the buttons from the previous message
self.bot.edit_message_reply_markup(
call.message.chat.id, call.message.message_id
)
def manage_watchlist(self, message):
chat_id = message.chat.id
# Check if the user has any slots in the watchlist
if chat_id not in self.watchlist or len(self.watchlist[chat_id]) == 0:
self.bot.reply_to(message, "Your watchlist is empty.")
return
markup = InlineKeyboardMarkup()
# List the slots in the watchlist
for i, slot in enumerate(self.watchlist[chat_id]):
button_text = f"Slot {i + 1}: {slot.start_stamp} (FULL)"
markup.add(InlineKeyboardButton(button_text, callback_data=f"remove_{i}"))
# Send the list of slots to the user with remove buttons
self.bot.reply_to(message, "Your watchlist:", reply_markup=markup)
def callback_remove_slot(self, call):
try:
# Extract slot index from callback data
slot_index = int(call.data.split("_")[1])
slots = self.watchlist.get(call.message.chat.id, [])
if 0 <= slot_index < len(slots):
removed_slot = slots.pop(slot_index)
# Inform the user that the slot has been removed
self.bot.answer_callback_query(
call.id, f"Removed slot: {removed_slot.start_stamp}"
)
self.bot.send_message(
call.message.chat.id,
f"Slot {removed_slot.start_stamp} has been removed from your watchlist.",
)
# Update watchlist after removal
if not slots:
del self.watchlist[
call.message.chat.id
] # Remove the chat ID if the watchlist is empty
else:
self.bot.answer_callback_query(call.id, "Invalid slot selection.")
except (IndexError, ValueError):
self.bot.answer_callback_query(call.id, "Invalid data.")
# Remove the buttons from the previous message
self.bot.edit_message_reply_markup(
call.message.chat.id, call.message.message_id
)
def add_to_watchlist(self, chat_id, slot):
if chat_id not in self.watchlist:
self.watchlist[chat_id] = []
@ -114,23 +368,38 @@ class TelegramBot:
def check_watchlist(self):
now = datetime.now().replace(tzinfo=timezone.utc)
for chat_id, slots in list(self.watchlist.items()):
try:
xclient = self.get_xclient(chat_id)
except Exception as e:
print(f"Error polling watchlist: {str(e)}")
available_slots = []
for slot in slots:
if slot.start < now:
# If the slot has expired
self.bot.send_message(chat_id, f"Slot {slot.start_stamp} has expired and is no longer available.")
self.bot.send_message(
chat_id,
f"Slot {slot.start_stamp} has expired and is no longer available.",
)
available_slots.append(slot) # Mark it to remove from watchlist
elif self.xclient.check_booking_availability(slot):
elif xclient.check_booking_availability(slot):
# If the slot becomes available
try:
self.xclient.make_booking(slot)
self.bot.send_message(chat_id, f"Slot {slot.start_stamp} is now available and has been booked for you.")
xclient.make_booking(slot)
self.bot.send_message(
chat_id,
f"Slot {slot.start_stamp} is now available and has been booked for you.",
)
available_slots.append(slot) # Mark it to remove from watchlist
except Exception as e:
self.bot.send_message(chat_id, f"Error booking slot {slot.start_stamp}: {str(e)}")
self.bot.send_message(
chat_id, f"Error booking slot {slot.start_stamp}: {str(e)}"
)
# Remove the expired or booked slots from the watchlist
self.watchlist[chat_id] = [slot for slot in slots if slot not in available_slots]
self.watchlist[chat_id] = [
slot for slot in slots if slot not in available_slots
]
if not self.watchlist[chat_id]:
del self.watchlist[chat_id] # Remove chat_id from watchlist if empty
@ -157,10 +426,10 @@ class TelegramBot:
time.sleep(polling_interval) # Wait before checking the watchlist again
def run(self):
### start poll_periodically on seperate thread
thread = Thread(target = self.bot.polling, args = ())
thread = Thread(target=self.poll_periodically, args=())
thread.daemon = True
thread.start()
self.poll_periodically()
thread.kill()
self.bot.polling()

128
tgbot_multi.py
View File

@ -4,8 +4,9 @@ from datetime import datetime, timedelta, timezone
import time
from threading import Thread
class TelegramBot:
def __init__(self, bot_token, db_path='bot_database.db'):
def __init__(self, bot_token, db_path="bot_database.db"):
self.bot = telebot.TeleBot(bot_token)
self.db_path = db_path
self.user_selected_slot = {} # Store user selected slot by chat ID
@ -15,23 +16,27 @@ class TelegramBot:
self.init_db()
# Set up the command handlers
@self.bot.message_handler(commands=['start'])
@self.bot.message_handler(commands=["start"])
def send_welcome(message):
self.handle_login(message)
@self.bot.message_handler(commands=['make_booking'])
@self.bot.message_handler(commands=["make_booking"])
def make_booking(message):
self.handle_booking(message)
@self.bot.callback_query_handler(func=lambda call: call.data.startswith("book_"))
@self.bot.callback_query_handler(
func=lambda call: call.data.startswith("book_")
)
def callback_booking(call):
self.handle_callback_booking(call)
@self.bot.message_handler(commands=['cancel_booking'])
@self.bot.message_handler(commands=["cancel_booking"])
def cancel_booking(message):
self.handle_cancel_booking(message)
@self.bot.callback_query_handler(func=lambda call: call.data.startswith("cancel_"))
@self.bot.callback_query_handler(
func=lambda call: call.data.startswith("cancel_")
)
def callback_cancel_booking(call):
self.handle_callback_cancel_booking(call)
@ -39,33 +44,36 @@ class TelegramBot:
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
# Create tables
cursor.execute('''
cursor.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
chat_id INTEGER UNIQUE,
username TEXT,
password TEXT
)
''')
cursor.execute('''
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS chats (
chat_id INTEGER PRIMARY KEY,
username TEXT
)
''')
""")
conn.commit()
def handle_login(self, message):
chat_id = message.chat.id
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute('SELECT username FROM chats WHERE chat_id = ?', (chat_id,))
cursor.execute("SELECT username FROM chats WHERE chat_id = ?", (chat_id,))
result = cursor.fetchone()
if result:
self.bot.reply_to(message, "You are already logged in.")
else:
self.bot.reply_to(message, "Please provide your username and password in the format:\n`/login username password`")
self.bot.reply_to(
message,
"Please provide your username and password in the format:\n`/login username password`",
)
self.bot.register_next_step_handler(message, self.process_login)
def process_login(self, message):
@ -73,24 +81,33 @@ class TelegramBot:
command, username, password = message.text.split()
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute('SELECT * FROM users WHERE username = ? AND password = ?', (username, password))
cursor.execute(
"SELECT * FROM users WHERE username = ? AND password = ?",
(username, password),
)
user = cursor.fetchone()
if user:
cursor.execute('INSERT OR IGNORE INTO chats (chat_id, username) VALUES (?, ?)', (message.chat.id, username))
cursor.execute(
"INSERT OR IGNORE INTO chats (chat_id, username) VALUES (?, ?)",
(message.chat.id, username),
)
conn.commit()
self.bot.reply_to(message, "Login successful!")
else:
self.bot.reply_to(message, "Invalid credentials. Please try again.")
self.handle_login(message)
except ValueError:
self.bot.reply_to(message, "Invalid format. Please use `/login username password`.")
self.bot.reply_to(
message, "Invalid format. Please use `/login username password`."
)
def send_welcome(self, message):
self.bot.reply_to(
message, "Welcome! Use /make_booking to view and book available slots."
)
def send_welcome(self,message):
self.bot.reply_to(message, "Welcome! Use /make_booking to view and book available slots.")
def handle_booking(self,message):
def handle_booking(self, message):
start = datetime.now()
end = start + timedelta(days=1)
slots = self.xclient.list_slots(start, end)
@ -103,7 +120,9 @@ class TelegramBot:
button_text = f"Slot {i + 1}: {slot.start_stamp} ({status})"
markup.add(InlineKeyboardButton(button_text, callback_data=f"book_{i}"))
self.bot.reply_to(message, "Select a slot to book or watch:", reply_markup=markup)
self.bot.reply_to(
message, "Select a slot to book or watch:", reply_markup=markup
)
self.user_selected_slot[message.chat.id] = slots
else:
self.bot.reply_to(message, "No slots found.")
@ -111,7 +130,7 @@ class TelegramBot:
def handle_callback_booking(self, call):
try:
# Extract slot index from callback data
slot_index = int(call.data.split('_')[1])
slot_index = int(call.data.split("_")[1])
slots = self.user_selected_slot.get(call.message.chat.id, [])
if 0 <= slot_index < len(slots):
@ -121,16 +140,26 @@ class TelegramBot:
# Attempt to book the available slot
try:
self.xclient.make_booking(selected_slot)
self.bot.answer_callback_query(call.id, "Slot booked successfully!")
self.bot.send_message(call.message.chat.id, f"Booking confirmed for: {selected_slot.start_stamp}")
self.bot.answer_callback_query(
call.id, "Slot booked successfully!"
)
self.bot.send_message(
call.message.chat.id,
f"Booking confirmed for: {selected_slot.start_stamp}",
)
except Exception as e:
self.bot.answer_callback_query(call.id, "Failed to book slot.")
self.bot.send_message(call.message.chat.id, f"Error: {str(e)}")
else:
# Slot is full, add to watchlist
self.add_to_watchlist(call.message.chat.id, selected_slot)
self.bot.answer_callback_query(call.id, "Slot is full. Added to watchlist.")
self.bot.send_message(call.message.chat.id, f"Slot {selected_slot.start_stamp} is full. You will be notified when it becomes available.")
self.bot.answer_callback_query(
call.id, "Slot is full. Added to watchlist."
)
self.bot.send_message(
call.message.chat.id,
f"Slot {selected_slot.start_stamp} is full. You will be notified when it becomes available.",
)
else:
self.bot.answer_callback_query(call.id, "Invalid slot selection.")
except (IndexError, ValueError):
@ -138,24 +167,30 @@ class TelegramBot:
def handle_cancel_booking(self, message):
# Fetch user's current bookings
bookings = self.xclient.my_bookings(datetime.now(), datetime.now() + timedelta(days=31))
bookings = self.xclient.my_bookings(
datetime.now(), datetime.now() + timedelta(days=31)
)
if bookings:
markup = InlineKeyboardMarkup()
for i, booking in enumerate(bookings):
# Create a button for each booking
button_text = f"Booking {i + 1}: {booking.start_stamp}"
markup.add(InlineKeyboardButton(button_text, callback_data=f"cancel_{i}"))
markup.add(
InlineKeyboardButton(button_text, callback_data=f"cancel_{i}")
)
self.user_bookings[message.chat.id] = bookings
self.bot.reply_to(message, "Select a booking to cancel:", reply_markup=markup)
self.bot.reply_to(
message, "Select a booking to cancel:", reply_markup=markup
)
else:
self.bot.reply_to(message, "You have no bookings to cancel.")
def handle_callback_cancel_booking(self, call):
try:
# Extract booking index from callback data
booking_index = int(call.data.split('_')[1])
booking_index = int(call.data.split("_")[1])
bookings = self.user_bookings.get(call.message.chat.id, [])
if 0 <= booking_index < len(bookings):
@ -163,10 +198,17 @@ class TelegramBot:
# Attempt to cancel the selected booking
try:
if self.xclient.cancel_booking(selected_booking):
self.bot.answer_callback_query(call.id, "Booking canceled successfully!")
self.bot.send_message(call.message.chat.id, f"Booking for {selected_booking.start_stamp} has been canceled.")
self.bot.answer_callback_query(
call.id, "Booking canceled successfully!"
)
self.bot.send_message(
call.message.chat.id,
f"Booking for {selected_booking.start_stamp} has been canceled.",
)
else:
self.bot.answer_callback_query(call.id, "Failed to cancel booking.")
self.bot.answer_callback_query(
call.id, "Failed to cancel booking."
)
except Exception as e:
self.bot.answer_callback_query(call.id, "Failed to cancel booking.")
self.bot.send_message(call.message.chat.id, f"Error: {str(e)}")
@ -187,19 +229,29 @@ class TelegramBot:
for slot in slots:
if slot.start < now:
# If the slot has expired
self.bot.send_message(chat_id, f"Slot {slot.start_stamp} has expired and is no longer available.")
self.bot.send_message(
chat_id,
f"Slot {slot.start_stamp} has expired and is no longer available.",
)
available_slots.append(slot) # Mark it to remove from watchlist
elif self.xclient.check_booking_availability(slot):
# If the slot becomes available
try:
self.xclient.make_booking(slot)
self.bot.send_message(chat_id, f"Slot {slot.start_stamp} is now available and has been booked for you.")
self.bot.send_message(
chat_id,
f"Slot {slot.start_stamp} is now available and has been booked for you.",
)
available_slots.append(slot) # Mark it to remove from watchlist
except Exception as e:
self.bot.send_message(chat_id, f"Error booking slot {slot.start_stamp}: {str(e)}")
self.bot.send_message(
chat_id, f"Error booking slot {slot.start_stamp}: {str(e)}"
)
# Remove the expired or booked slots from the watchlist
self.watchlist[chat_id] = [slot for slot in slots if slot not in available_slots]
self.watchlist[chat_id] = [
slot for slot in slots if slot not in available_slots
]
if not self.watchlist[chat_id]:
del self.watchlist[chat_id] # Remove chat_id from watchlist if empty
@ -226,11 +278,11 @@ class TelegramBot:
time.sleep(polling_interval) # Wait before checking the watchlist again
def run(self):
### start poll_periodically on seperate thread
thread = Thread(target = self.poll_periodically, args = ())
thread = Thread(target=self.poll_periodically, args=())
thread.daemon = True
thread.start()
self.bot.polling()
thread.kill()

33
xclient.py
View File

@ -60,17 +60,13 @@ class XClient:
time.sleep(1)
delcom_auth = json.loads(
driver.execute_script(
"return window.localStorage.getItem('delcom_auth');"
)
driver.execute_script("return window.localStorage.getItem('delcom_auth');")
)
driver.quit()
access_token = delcom_auth["tokenResponse"].get("accessToken", None)
return access_token
def list_slots(self, start, end, tagIDs=[28]):
"""
URl:
@ -119,14 +115,16 @@ class XClient:
"$or": [
{"memberId": member_id}, # Fetch bookings based on the user ID
],
"booking.startDate": {"$gte": start.strftime("%Y-%m-%dT%H:%M:%S.%fZ")},
"booking.startDate": {
"$gte": start.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
},
"booking.endDate": {"$lte": end.strftime("%Y-%m-%dT%H:%M:%S.%fZ")},
},
),
"join": ["booking", "linkedProduct", "product"],
"fetchOptimizedCustomerDashboard": "true",
"fetchTicket": "false",
"sort": "booking.startDate,ASC"
"sort": "booking.startDate,ASC",
}
# Make the GET request to the API
@ -138,12 +136,13 @@ class XClient:
# Check if the request was successful
if response.status_code == 200:
bookings = [Booking(**(x | {"bookingId": x["id"], "bookableProductId": None})) for x in response.json()["data"]["bookings"]]
bookings = [
Booking(**(x | {"bookingId": x["id"], "bookableProductId": None}))
for x in response.json()["data"]["bookings"]
]
return bookings
elif response.status_code == 403:
raise Exception(
f"Failed to fetch bookings: {response.json()['message']}"
)
raise Exception(f"Failed to fetch bookings: {response.json()['message']}")
else:
raise Exception(
f"Failed to fetch bookings: {response.status_code}, {response.text}"
@ -186,9 +185,7 @@ class XClient:
if response.status_code == 201:
return response.json() # Return response if successful
elif response.status_code == 403:
raise Exception(
f"Failed to make booking: {response.json()['message']}"
)
raise Exception(f"Failed to make booking: {response.json()['message']}")
else:
raise Exception(
f"Failed to make booking: {response.status_code}, {response.text}"
@ -256,7 +253,9 @@ class XClient:
participation_id = i["id"]
break
assert participation_id is not None, "Booking not found in user's participations"
assert (
participation_id is not None
), "Booking not found in user's participations"
## Send a DELETE request to the URL https://backbone-web-api.production.delft.delcom.nl/participations/{id}
response = self.session.delete(
@ -267,9 +266,7 @@ class XClient:
if response.status_code == 200:
return True
elif response.status_code == 403:
raise Exception(
f"Failed to cancel booking: {response.json()['message']}"
)
raise Exception(f"Failed to cancel booking: {response.json()['message']}")
else:
raise Exception(
f"Failed to fetch bookings: {response.status_code}, {response.text}"