203 lines
7.8 KiB
Python
203 lines
7.8 KiB
Python
"""
|
||
|
||
AUTHOR:
|
||
|
||
Khushal P Soonderji
|
||
|
||
DATE:
|
||
|
||
Friday, 29th Nov., 2024
|
||
|
||
OBJECTIVE:
|
||
|
||
To manage alert sending from one place. You may have various outlets for alerts, like Telegram, WhatsApp, etc.
|
||
Add endpoints for them here. Note that these alerts will be for the internal tech admins. It won't be for the
|
||
client-facing comms. Create a separate blueprint for that.
|
||
|
||
REFERENCES:
|
||
|
||
N/A
|
||
|
||
DOWNLOADS:
|
||
|
||
N/A
|
||
|
||
NOTES:
|
||
|
||
N/A
|
||
|
||
"""
|
||
|
||
|
||
# *****************************************************************************************************************
|
||
# ***** ****
|
||
# *** IMPORT ***
|
||
# ***** ****
|
||
# *****************************************************************************************************************
|
||
|
||
|
||
# To make sibling directories accessible for imports:
|
||
import sys
|
||
sys.path.append(".")
|
||
sys.path.append("..")
|
||
|
||
# For using Quart:
|
||
from quart import Blueprint, current_app
|
||
|
||
# My utils:
|
||
from utils_v2.string import json
|
||
from utils_v2.date_time import date_time
|
||
from utils_v2.api.codes import StatusCodes, HttpCodes
|
||
from utils_v2.api.response import ResponseModel
|
||
from utils_v2.api.async_quart import (
|
||
set_api_version,
|
||
read_input,
|
||
get_session_info,
|
||
log_request_to_mongo,
|
||
log_chain_to_mongo,
|
||
should_not_be_under_maintenance,
|
||
only_whitelisted_ips,
|
||
limit_rate,
|
||
validate_input,
|
||
handle_cancelled_request
|
||
)
|
||
|
||
# Data models:
|
||
from models.data.tech.alerts import (
|
||
ChatAlertRequestHeaders,
|
||
ChatAlertRequestData
|
||
)
|
||
|
||
# Common:
|
||
from shared import constants
|
||
|
||
# For asynchronous activities:
|
||
import asyncio
|
||
|
||
|
||
# *****************************************************************************************************************
|
||
# ***** ****
|
||
# *** MACROS / ONE-TIME INIT ***
|
||
# ***** ****
|
||
# *****************************************************************************************************************
|
||
|
||
|
||
# Related to Quart:
|
||
tech_chat_alert_bp = Blueprint("tech_chat_alert", __name__)
|
||
|
||
|
||
# *****************************************************************************************************************
|
||
# ***** ****
|
||
# *** VARIABLES ***
|
||
# ***** ****
|
||
# *****************************************************************************************************************
|
||
|
||
|
||
# --- Nothing Yet
|
||
|
||
|
||
# *****************************************************************************************************************
|
||
# ***** ****
|
||
# *** FUNCTIONS ***
|
||
# ***** ****
|
||
# *****************************************************************************************************************
|
||
|
||
|
||
@tech_chat_alert_bp.record_once
|
||
def init(blueprint_setup_state):
|
||
|
||
# This gets called when the blueprint is registered.
|
||
# Consider this to be a one-time setup for the whole blueprint:
|
||
pass
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------------------------------
|
||
|
||
|
||
@tech_chat_alert_bp.route("/chat/<source>", methods = ["POST"])
|
||
@set_api_version(api_version = "3.0.0")
|
||
@read_input(sanitize_headers = True, sanitize_data = True)
|
||
@get_session_info(key = "X-Session-Token", session_coro = "get_session")
|
||
@should_not_be_under_maintenance(attr_name = "is_under_maintenance")
|
||
@validate_input(
|
||
header_validator = lambda x: ChatAlertRequestHeaders(**x).model_dump(),
|
||
data_validator = lambda x: ChatAlertRequestData(**x)
|
||
)
|
||
@handle_cancelled_request()
|
||
async def user_login(
|
||
source: str = None,
|
||
inbound_headers: dict = None,
|
||
inbound_data: dict | ChatAlertRequestData = None,
|
||
inbound_files: dict = None,
|
||
**kwargs
|
||
):
|
||
|
||
"""
|
||
To relay an alert over some form of chat client like Telegram and WhatsApp.
|
||
:param source: The identifier of the source of the alert.
|
||
:param inbound_headers: auto-extracted by the decorators.
|
||
:param inbound_data: auto-extracted by the decorators.
|
||
:param inbound_files: auto-extracted by the decorators.
|
||
:param kwargs: Any number of extra inputs supplied by the decorators.
|
||
:return: A standard response structure.
|
||
"""
|
||
|
||
response = None
|
||
|
||
# ┳┳┓ ┏┓ •
|
||
# ┃┃┃┏┓┏┏┏┓┏┓┏┓ ┃ ┏┓┏┓┏╋┏┓┓┏┏╋┓┏┓┏┓
|
||
# ┛ ┗┗ ┛┛┗┻┗┫┗ ┗┛┗┛┛┗┛┗┛ ┗┻┗┗┗┗┛┛┗
|
||
# ┛
|
||
|
||
# Capture the event's time, and the person who caused the event:
|
||
event_dt = date_time.get_current_ist_date_time().strftime("%Y-%m-%d %I:%M:%S %p %Z")
|
||
if kwargs.get("session_info"): username = kwargs["session_info"].get("fullName")
|
||
else: username = None
|
||
|
||
# Construct the prefix:
|
||
if inbound_data.type == "warning": message_type = "⚠️ WARNING FROM"
|
||
elif inbound_data.type == "error": message_type = "🚨 ERROR IN"
|
||
else: message_type = "ℹ️ INFO FROM"
|
||
message_prefix = f"{message_type} *TCAOFF (Converse)*!\n{event_dt}\n\n"
|
||
message_prefix += f"*From:*\n`{username} (via '{source}')`\n\n"
|
||
|
||
# ┏┳┓ ┓
|
||
# ┃ ┏┓┃┏┓┏┓┏┓┏┓┏┳┓
|
||
# ┻ ┗ ┗┗ ┗┫┛ ┗┻┛┗┗
|
||
# ┛
|
||
|
||
if inbound_data.chatClient == "telegram":
|
||
response = await current_app.http_client.post(
|
||
url = current_app.script_data["telegram"]["connectors"]["nexcom"]["url"],
|
||
json = {
|
||
"appKey": current_app.script_data["telegram"]["connectors"]["nexcom"]["appKey"],
|
||
"chatId": inbound_data.chatId or current_app.script_data["telegram"]["chatIds"]["tcaoff"],
|
||
"message": message_prefix + inbound_data.message
|
||
}
|
||
)
|
||
|
||
# ┳┓
|
||
# ┣┫┏┓┏┏┓┏┓┏┓┏┏┓
|
||
# ┛┗┗ ┛┣┛┗┛┛┗┛┗
|
||
# ┛
|
||
|
||
# Construct and return the response:
|
||
response_json = response.json()
|
||
success = True if response_json.get("status", 0) == 1 else False
|
||
return ResponseModel(
|
||
status_code = StatusCodes.OK if success else StatusCodes.FAILED,
|
||
message = response_json.get("message", "unknown failure")
|
||
)
|
||
|
||
|
||
# *****************************************************************************************************************
|
||
# ***** ****
|
||
# *** MAIN PROGRAM ***
|
||
# ***** ****
|
||
# *****************************************************************************************************************
|
||
|
||
|
||
if __name__ == "__main__":
|
||
|
||
pass
|