diff --git a/.gitignore b/.gitignore index 68bc17f..b01b83b 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,5 @@ +data/ + # Byte-compiled / optimized / DLL files __pycache__/ *.py[cod] diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..cd62eb5 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,8 @@ +FROM python:3.12 + +WORKDIR /app +COPY . . + +RUN pip install -r requirements.txt + +CMD ["python", "src/main.py"] \ No newline at end of file diff --git a/README.md b/README.md index 85007bf..133217d 100644 --- a/README.md +++ b/README.md @@ -1 +1,45 @@ -# capt-laundry-bot \ No newline at end of file +# CAPT Laundry Bot + +Laundry bot for the management of + +## Setup + +### Requirements + +- Python 3.12 +- Docker (Only for deployment) + +### Telegram Bot Setup (Local) + +1. Create your own Telegram bot by following [BotFather](https://t.me/BotFather) instructions +2. Copy the `API_KEY` (keep this key secret) + +### Running the Bot + +1. Copy this repository + +``` +git clone https://github.com/jloh02/capt-laundry-bot +``` + +2. Create a `.env` file in root folder with the following content and update Telegram bot API key + +``` + +``` + +3. Install Packages + +``` +pip install -r requirements.txt +``` + +4. Run Bot + +``` +python src/main.py +``` + +## Design Considerations + +Instead of a DB, we opted for a local JSON file to allow for ease of deployment and logetivity of the project as this project will be managed at an individual basis outside of the management of CAPT diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..b840352 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,10 @@ +version: "3" +services: + laundry-bot: + build: + context: . + dockerfile: Dockerfile + volumes: + - static-volume:./data +volumes: + static-volume: diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..90ffc66 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +python-dateutil==2.9.0 +python-dotenv==1.0.1 +python-telegram-bot==21.1.1 +python-telegram-bot[job-queue]==21.1.1 +pytz==2024.1 \ No newline at end of file diff --git a/src/config.py b/src/config.py new file mode 100644 index 0000000..eba61f1 --- /dev/null +++ b/src/config.py @@ -0,0 +1,16 @@ +import os +from dotenv import load_dotenv + +config = {} + +def read_dotenv(): + global config + load_dotenv() + config.update( + { + "TELEGRAM_BOT_API_KEY": os.getenv("TELEGRAM_BOT_API_KEY"), + "WEBHOOK_URL": os.getenv("WEBHOOK_URL"), + "PRODUCTION": os.getenv("PRODUCTION") == "True", + "BASE_PATH": ("/app/static" if config.get("PRODUCTION") else "./data"), + } + ) diff --git a/src/machine.py b/src/machine.py new file mode 100644 index 0000000..6523ab6 --- /dev/null +++ b/src/machine.py @@ -0,0 +1,60 @@ +from abc import ABC +import datetime +import pytz +import utils +import storage + +SGT_TIMEZONE = pytz.timezone("Asia/Singapore") + + +class Machine(ABC): + # constant value which stores total time required for start (IN SECONDS) + name = None + time_to_complete = None + + def __init__(self, new_time_to_complete, new_name): + self.time_to_complete = new_time_to_complete + self.name = new_name + + def get_name(self): + return self.name + + def get_time_to_complete(self): + return self.time_to_complete + + def status(self): + curr_user, end_time = storage.get_laundry_timer(self.name) + if utils.is_available(end_time): + reply = f"AVAILABLE \U00002705" + if curr_user: + reply += f', last used by @{curr_user} ({end_time.astimezone(SGT_TIMEZONE).strftime("%d/%m/%Y %I:%M%p")})' + return reply + else: + time_delta = end_time - datetime.datetime.now() + time_in_min = time_delta.seconds // 60 + time_in_sec = time_delta.seconds % 60 + return f"UNAVAILABLE \U0000274C for {time_in_min}mins and {time_in_sec}s by @{curr_user}" + + def time_left_mins(self): + return self.time_to_complete // 60 + + def time_left_secs(self): + return self.time_to_complete % 60 + + def total_time(self): + return f"{self.time_left_mins()}mins" + + def start_machine(self, new_user): + _, end_time = storage.get_laundry_timer(self.name) + if not utils.is_available(end_time): + return False + else: + new_end_time = datetime.datetime.now() + datetime.timedelta( + seconds=self.time_to_complete + ) + new_curr_user = new_user + storage.set_laundry_timer(self.name, new_curr_user, new_end_time) + return True + + def alarm(self): + return "Fuyohhhhhh!! Your clothes are ready for collection! Please collect them now so that others may use it" diff --git a/src/main.py b/src/main.py new file mode 100644 index 0000000..1ca36a8 --- /dev/null +++ b/src/main.py @@ -0,0 +1,283 @@ +import os +import strings +import storage +from telegram import ( + Bot, + Update, + InlineKeyboardButton, + InlineKeyboardMarkup, +) +from telegram.ext import ( + Application, + CommandHandler, + CallbackContext, + CallbackQueryHandler, + ConversationHandler, +) +from machine import Machine +from config import config, read_dotenv + +read_dotenv() +storage.read() + +MENU = 1 + +TBOT = Bot(config.get("TELEGRAM_BOT_API_KEY")) + +WASHER_TIMER = 32 * 60 +DRYER_TIMER = 32 * 60 +DRYER_ONE = Machine(DRYER_TIMER, "DRYER ONE") +DRYER_TWO = Machine(DRYER_TIMER, "DRYER TWO") +WASHER_ONE = Machine(WASHER_TIMER, "WASHER ONE") +WASHER_TWO = Machine(WASHER_TIMER, "WASHER TWO") + + +COMMANDS_DICT = { + "start": "Display help page and version", + "select": strings.SELECT_COMMAND_DESCRIPTION, + "status": strings.STATUS_COMMAND_DESCRIPTION, +} + +TBOT.set_my_commands(COMMANDS_DICT.items()) + + +def main(): + application = ( + Application.builder().token(config.get("TELEGRAM_BOT_API_KEY")).build() + ) + + # Use the pattern parameter to pass CallbackQueries with specific + # data pattern to the corresponding handlers. + # ^ means "start of line/string" + # $ means "end of line/string" + # So ^ABC$ will only allow 'ABC' + + ENTRY_POINT_DICT = { + "start": start, + "select": select, + "status": status, + } + FALLBACK_DICT = ENTRY_POINT_DICT + + MENU_DICT = { + "exit": cancel, + "exits": cancel, + "dryer_one": create_double_confirm_callback("dryer_one"), + "dryer_two": create_double_confirm_callback("dryer_two"), + "washer_one": create_double_confirm_callback("washer_one"), + "washer_two": create_double_confirm_callback("washer_two"), + "no_dryer_one": backtomenu, + "no_dryer_two": backtomenu, + "no_washer_one": backtomenu, + "no_washer_two": backtomenu, + "yes_dryer_one": set_timer_machine(DRYER_ONE), + "yes_dryer_two": set_timer_machine(DRYER_TWO), + "yes_washer_one": set_timer_machine(WASHER_ONE), + "yes_washer_two": set_timer_machine(WASHER_TWO), + } + + conv_handler = ConversationHandler( + entry_points=[CommandHandler(cmd, fn) for cmd, fn in ENTRY_POINT_DICT.items()], + states={ + MENU: [ + CallbackQueryHandler(fn, pattern=f"^{cmd}$") + for cmd, fn in MENU_DICT.items() + ] + }, + fallbacks=[CommandHandler(cmd, fn) for cmd, fn in FALLBACK_DICT.items()], + ) + + application.add_handler(conv_handler) + + if config.get("PRODUCTION"): + application.run_webhook( + listen="0.0.0.0", + port=os.environ.get("PORT", 8080), + webhook_url=config.get("WEBHOOK_URL"), + ) + else: + application.run_polling() + + +WELCOME_MESSAGE = f"Welcome to Dragon Laundry Bot ({os.environ.get('VERSION','dev')})!\n\nUse the following commands to use this bot:\n/select: {strings.SELECT_COMMAND_DESCRIPTION}\n/status: {strings.STATUS_COMMAND_DESCRIPTION}\n\nThank you for using the bot!\nCredit to: @Kaijudo" + +START_INLINE_KEYBOARD = InlineKeyboardMarkup( + [[InlineKeyboardButton("Exit", callback_data="exit")]] +) + + +async def start(update: Update, context: CallbackContext): + # Don't allow users to use /start command in group chats + if update.message.chat.type != "private": + TBOT.send_message( + chat_id=update.message.from_user.id, + text=f"Hi @{update.message.from_user.username},\n\nThanks for calling me in the groupchat. To prevent spamming in the group, please type /start to me privately in this chat instead!", + ) + return MENU + + if len(context.args) > 0: + return + + await update.message.reply_text( + WELCOME_MESSAGE, + reply_markup=START_INLINE_KEYBOARD, + ) + return MENU + + +SELECT_MACHINE_INLINE_KEYBOARD = InlineKeyboardMarkup( + [ + [ + InlineKeyboardButton("Dryer One", callback_data="dryer_one"), + InlineKeyboardButton("Dryer Two", callback_data="dryer_two"), + ], + [ + InlineKeyboardButton("Washer One", callback_data="washer_one"), + InlineKeyboardButton("Washer Two", callback_data="washer_two"), + ], + [InlineKeyboardButton("Exit", callback_data="exit")], + ] +) + + +async def select(update: Update, context: CallbackContext): + # Don't allow users to use /select command in group chats + if update.message.chat.type != "private": + return MENU + + await update.message.reply_text( + "\U0001F606\U0001F923 Please choose a service: \U0001F606\U0001F923", + reply_markup=SELECT_MACHINE_INLINE_KEYBOARD, + ) + return MENU + + +async def cancel(update: Update, context: CallbackContext): + """ + Returns `ConversationHandler.END`, which tells the ConversationHandler that the conversation is over + """ + query = update.callback_query + await query.answer() + await query.edit_message_text( + text="Haiyaaa then you call me for what\n\nUse /start again to call me" + ) + return ConversationHandler.END + + +def create_inline_for_callback(machine_name): + markup = InlineKeyboardMarkup( + [ + [ + InlineKeyboardButton("Yes", callback_data=f"yes_{machine_name}"), + ], + [InlineKeyboardButton("No", callback_data=f"no_{machine_name}")], + ] + ) + text = f"Timer for {machine_name.upper().replace('_',' ')} will begin?" + return (text, markup) + + +def create_double_confirm_callback(machine_name: str): + text, markup = create_inline_for_callback(machine_name) + + async def callback(update: Update, _: CallbackContext) -> int: + query = update.callback_query + await query.answer() + await query.edit_message_text(text=text, reply_markup=markup) + return MENU + + return callback + + +EXIT_INLINE_KEYBOARD = InlineKeyboardMarkup( + [[InlineKeyboardButton("Exit", callback_data="exits")]] +) + + +async def backtomenu(update: Update, context: CallbackContext): + query = update.callback_query + await query.answer() + await query.edit_message_text( + WELCOME_MESSAGE, + reply_markup=EXIT_INLINE_KEYBOARD, + ) + + +def remove_job_if_exists(name: str, context: CallbackContext) -> bool: + """Remove job with given name. Returns whether job was removed.""" + current_jobs = context.job_queue.get_jobs_by_name(name) + if not current_jobs: + return False + for job in current_jobs: + job.schedule_removal() + return True + + +def alarm(context: CallbackContext, machine) -> None: + """Send the alarm message.""" + job = context.job + context.bot.send_message( + job.context, + text="Fuyohhhhhh!! Your clothes are ready for collection! Please collect them now so that others may use it", + ) + + +def set_timer_machine(machine): + async def set_timer(update, context): + machine_name = machine.get_name() + upper_name = machine_name.upper() + underscore_name = machine_name.lower().replace(" ", "_") + + """Add a job to the queue.""" + chat_id = update.effective_message.chat_id + query = update.callback_query + await query.answer() + + job_removed = remove_job_if_exists(str(chat_id), context) + + if not (machine.start_machine(update.effective_message.chat.username)): + text = f"{upper_name} is currently in use. Please come back again later!" + await query.edit_message_text(text=text) + else: + context.job_queue.run_once( + lambda context: alarm(context, machine), + machine.get_time_to_complete(), + chat_id=chat_id, + name=underscore_name, + ) + text = f"Timer Set for {machine.time_left_mins()}mins for {upper_name}. Please come back again!" + await query.edit_message_text(text=text) + + return MENU + + return set_timer + + +async def status(update: Update, context: CallbackContext): + DRYER_ONE_TIMER = DRYER_ONE.status() + DRYER_TWO_TIMER = DRYER_TWO.status() + WASHER_ONE_TIMER = WASHER_ONE.status() + WASHER_TWO_TIMER = WASHER_TWO.status() + + reply_text = f"""Status of Laundry Machines: + +Dryer One: {DRYER_ONE_TIMER} + +Dryer Two: {DRYER_TWO_TIMER} + +Washer One: {WASHER_ONE_TIMER} + +Washer Two: {WASHER_TWO_TIMER}""" + + # Don't allow users to use /status command in group chats + if update.message.chat.type != "private": + TBOT.send_message( + chat_id=update.message.from_user.id, + text=f"""Hi @{update.message.from_user.username} ,thanks for calling me in the groupchat. \n\nTo prevent spamming in the group, I have sent you a private message instead!\n\n{reply_text}""", + ) + return MENU + await update.message.reply_text(reply_text) + + +if __name__ == "__main__": + main() diff --git a/src/storage.py b/src/storage.py new file mode 100644 index 0000000..e78504d --- /dev/null +++ b/src/storage.py @@ -0,0 +1,38 @@ +import os +import json +from config import config +import datetime + +data_cache = {} + +def get_timer_path(): + return f"{config.get("BASE_PATH")}/timers.json" + +def read(): + global data_cache + + if not os.path.isfile(get_timer_path()): + return + + data_cache.clear() + with open(get_timer_path(), "r") as f: + data_cache.update(json.load(f)) + +def write(): + global data_cache + if not os.path.isfile(get_timer_path()): + os.makedirs(os.path.dirname(get_timer_path())) + + with open(get_timer_path(), "w") as f: + json.dump(data_cache, f) + +def set_laundry_timer(name: str, curr_user: str, end_time: datetime.datetime): + global data_cache + data_cache.update({name:{"currUser": curr_user, "endTime": int(end_time.timestamp())}}) + write() + +def get_laundry_timer(name: str) -> tuple[str, datetime.datetime]: + data = data_cache.get(name) + if data and data.get("currUser") and data.get("endTime"): + return (data.get("currUser"), datetime.datetime.fromtimestamp(data.get("endTime"))) + return ("", None) diff --git a/src/strings.py b/src/strings.py new file mode 100644 index 0000000..86415a3 --- /dev/null +++ b/src/strings.py @@ -0,0 +1,2 @@ +SELECT_COMMAND_DESCRIPTION = "Select the washer/dryer that you want to use" +STATUS_COMMAND_DESCRIPTION = "Check the status of Washers and Dryers" \ No newline at end of file diff --git a/src/utils.py b/src/utils.py new file mode 100644 index 0000000..b1544bf --- /dev/null +++ b/src/utils.py @@ -0,0 +1,7 @@ +import datetime + +# Returns True if end_time is None because timer not set --> not being used +def is_available(end_time): + if not end_time: + return True + return end_time < datetime.datetime.now()