From 8c2572d46422ffd4a85e5556417f87f2dac1e314 Mon Sep 17 00:00:00 2001 From: khushal Date: Thu, 23 Jan 2025 14:56:35 +0530 Subject: [PATCH] (20250123) Cron script for work reminders is ready. --- cron/misc/reminders_for_work.py | 209 ++++++++++++++++++++------------ 1 file changed, 130 insertions(+), 79 deletions(-) diff --git a/cron/misc/reminders_for_work.py b/cron/misc/reminders_for_work.py index ea994f8..f4fbbb8 100644 --- a/cron/misc/reminders_for_work.py +++ b/cron/misc/reminders_for_work.py @@ -84,6 +84,9 @@ from typing import List, Literal # For random values: import random +# For scheduling and cron: +from scheduler.asyncio import Scheduler + # Debugging: from icecream import IceCreamDebugger @@ -284,6 +287,7 @@ async def get_reminders_to_send() -> pd.DataFrame | None: reminders_df = None db_status = None db_message = None + exception = None try: @@ -302,32 +306,35 @@ async def get_reminders_to_send() -> pd.DataFrame | None: db_message = db_json["message"] # Create the response DataFrame: - reminders_df = pd.DataFrame(db_json["data"]["rs0"]) - reminders_df.rename( - columns = { - "token_id": "tokenKey", - "recepient": "to", - "integration_type": "serviceType", - "content": "content" - }, - inplace = True - ) + if db_status == 1 and not exception: + reminders_df = pd.DataFrame(db_json["data"]["rs0"]) + reminders_df.rename( + columns = { + "token_id": "tokenKey", + "recepient": "to", + "integration_type": "serviceType", + "provider": "client", + "message": "content" + }, + inplace = True + ) # If something goes wrong: - except Exception as exception: + except Exception as e: + exception = e printer(exception) - # # Send out any needed alert: - # if exception or not db_status: - # await send_telegram( - # message = ( - # "*Reminders For Work*\n\n" - # f"Status: `{db_status}`\n\n" - # f"Message: `{db_message}`\n\n" - # f"Exception: `{exception}`" - # ), - # message_type = "error" - # ) + # Send out any needed alert: + if exception or not db_status: + await send_telegram( + message = ( + "*Reminders For Work*\n\n" + f"Status: `{db_status}`\n\n" + f"Message: `{db_message}`\n\n" + f"Exception: `{exception}`" + ), + message_type = "error" + ) # Done here: return reminders_df @@ -350,18 +357,19 @@ async def send_reminders_by_chat(chat_df: pd.DataFrame) -> None: # Get a list of unique token-keys: unique_token_keys = chat_df["tokenKey"].unique().tolist() + if len(unique_token_keys) > 0: printer("Sending reminders by chat.", len(chat_df)) # For every unique token key, we make an API call: for token_key in unique_token_keys: # Construct the input(s) needed for the API call: token_key_df = chat_df[chat_df["tokenKey"] == token_key] - messages = [ - { + messages = [] + for index, row in token_key_df.iterrows(): + if row["client"] == "whatsappNimbus": messages.append({ "recipientNo": row["to"], "message": row["content"] - } for index, row in token_key_df.iterrows() - ] + }) # We are now ready to make the API call: exception = None @@ -406,6 +414,13 @@ async def send_reminders_by_chat(chat_df: pd.DataFrame) -> None: async def send_one_reminder_by_mail(reminder_row) -> None: + """ + To send one mail out. Sending a mail is a time-consuming process so we use this function as an async task as many + times as we need to. + :param reminder_row: The single entry from the tabulated reminder data. + :return: None. + """ + # Start with blank variables: log_id = None api_message = None @@ -472,56 +487,9 @@ async def send_reminders_by_mail(mail_df: pd.DataFrame) -> None: # Iterate over all the rows and create mail-sending tasks: tasks = [send_one_reminder_by_mail(row) for index, row in mail_df.iterrows()] - await asyncio.gather(*tasks) - # for index, row in mail_df.iterrows(): - - # # Create the input(s) needed for making the API call: - # input_json = { - # "tokenKey": row["tokenKey"], - # "to": [row["to"]], - # "subject": f"Reminder for Work - {datetime.datetime.now().strftime('%d/%m/%Y')}", - # "body": [ - # { - # "type": "html", - # "part": { - # "content": row["content"] - # } - # } - # ] - # } - # - # # We are now ready to make the API call: - # exception = None - # try: - # - # # Make the API call: - # response = await http_client.post( - # url = r"https://api.thecaoffice.com/converse/mail", - # json = input_json - # ) - # - # # Get the log id from the response: - # log_id = response.json()["logId"] - # api_message = response.json()["message"] - # - # # If the API call failed: - # response.raise_for_status() - # - # # If the API call goes wrong: - # except Exception as e: - # exception = e - # printer(exception) - # - # # Send out an alert if needed: - # if exception is not None: - # await send_telegram( - # message = ( - # f"*Reminders For Work (Mail)*\n\nException: `{exception}`\n\n" - # f"Message: `{api_message}`\n\n" - # f"Log Id: `{log_id}`" - # ), - # message_type = "error" - # ) + if tasks: + printer("Sending reminders by mail.", len(mail_df)) + await asyncio.gather(*tasks) # --------------------------------------------------------------------------------------------------------------------- @@ -543,15 +511,65 @@ async def send_reminders_once() -> None: """ reminders_df = await get_reminders_to_send() - if reminders_df is not None: + if reminders_df is not None: tasks = [ - # send_reminders_by_chat(reminders_df[reminders_df["serviceType"] == "chat"]), + send_reminders_by_chat(reminders_df[reminders_df["serviceType"] == "chat"]), send_reminders_by_mail(reminders_df[reminders_df["serviceType"] == "mail"]), # send_reminders_by_sms(reminders_df[reminders_df["serviceType"] == "sms"]) ] await asyncio.gather(*tasks) +# --------------------------------------------------------------------------------------------------------------------- + + +async def main( + start_time: datetime.datetime, + end_time: datetime.datetime, + interval_seconds: int = 300, + heartbeat_seconds: int = 1_800 +): + + """ + The main scheduler that manages jobs. + :param start_time: The time of the day at which messages can start going out. + :param end_time: The time of the day after which new messages should not go out. + :param interval_seconds: The time (in seconds) between two reminder jobs. + :param heartbeat_seconds: The time (in seconds) after which heartbeat messages must be sent out. + :return: None. + """ + + # Start configuring the scheduler: + printer("Configuring the schedule-manager.") + schedule_manager = Scheduler() + + # Create all the timestamps at which the job must be done: + all_job_ts = [] + offset_seconds = 0 + while True: + ts = start_time + datetime.timedelta(seconds = offset_seconds) + if ts > end_time: break + all_job_ts.append(ts.time()) + offset_seconds += interval_seconds + + # Add the jobs: + for ts in all_job_ts: + schedule_manager.daily(ts, send_reminders_once) + printer(len(all_job_ts)) + + # Infinite loop to keep doing the tasks: + printer("Schedule-manager ready.") + while True: + await send_telegram( + message = ( + "*Reminders For Work (Heartbeat)*\n\n" + f"Heartbeat Interval: `{heartbeat_seconds:,}`" + ), + message_type = "info" + ) + await asyncio.sleep(heartbeat_seconds) + + # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** @@ -566,6 +584,28 @@ if __name__ == "__main__": # Get the config from the command-line: parser = argparse.ArgumentParser(description = f"To automatically sync. the mails for all users.") + parser.add_argument( + "--start-time", + type = str, + help = "The 24-hr time of the day (in 'HH:MM:SS' format) from which reminders can start going out." + ) + parser.add_argument( + "--end-time", + type = str, + help = "The 24-hr time of the day (in 'HH:MM:SS' format) till which reminders can keep going out." + ) + parser.add_argument( + "--interval", + type = int, + help = "The no. of seconds after which you would like to check for new reminders to send.", + default = 300 + ) + parser.add_argument( + "--heartbeat-interval", + type = int, + help = "The no. of seconds after which you would like to receive heartbeat messages.", + default = 1_800 + ) parser.add_argument( "-s", "--script-id", type = str, @@ -586,9 +626,20 @@ if __name__ == "__main__": printer.disable() async def runner(): + + # Parse the inputs: + start_time = datetime.datetime.strptime(args.start_time, "%H:%M:%S") + end_time = datetime.datetime.strptime(args.end_time, "%H:%M:%S") + + # Initialize and run the scheduler: if await init( script_id = args.script_id, debug = args.debug - ): await send_reminders_once() + ): await main( + start_time = start_time, + end_time = end_time, + interval_seconds = args.interval, + heartbeat_seconds = args.heartbeat_interval + ) asyncio.run(runner())