khushalps/api_utils_converse_v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

(20250123) Cron script for work reminders is ready.

Browse Source
This commit is contained in:
khushal
2025-01-23 14:56:35 +05:30
parent 44ff7719f3
commit 8c2572d464
1 changed files with 130 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cron/misc/reminders_for_work.py
+129 -78
View File
@@ -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)
# ---------------------------------------------------------------------------------------------------------------------
@@ -545,13 +513,63 @@ async def send_reminders_once() -> None:
reminders_df = await get_reminders_to_send()
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())