diff --git a/api/blueprints/message/chat/auth.py b/api/blueprints/message/chat/auth.py index 3948181..c5455ca 100644 --- a/api/blueprints/message/chat/auth.py +++ b/api/blueprints/message/chat/auth.py @@ -248,6 +248,37 @@ async def callback_test( "senderId": inbound_data.auth.senderId }, display_name = inbound_data.auth.senderId, + display_picture = None, + session_token = inbound_headers["X-Session-Token"] + ) + + # ┏┓ ┏┳┓ ┓ + # ┣ ┏┓┏┓ ┃ ┏┓┃┏┓┏┓┏┓┏┓┏┳┓ + # ┻ ┗┛┛ ┻ ┗ ┗┗ ┗┫┛ ┗┻┛┗┗ + # ┛ + + if inbound_data.chatClient == "telegram": + success = await current_app.telegram_controller.set_token_direct( + sql_conn = current_app.sql_writer, + mongo_data_conn = current_app.data_mongo, + auth_token = CoreAuthTokenModel( + serviceType = "chat", + client = inbound_data.chatClient, + authType = "auth", + auth = inbound_data.auth.model_dump(), + user = kwargs.get("session_info"), + clientUserId = { + "botToken": inbound_data.auth.botToken + }, + status = "active", + syncFreq = 60 + ), + token_notes = { + "botId": inbound_data.auth.botId, + "botName": inbound_data.auth.botName + }, + display_name = inbound_data.auth.botName, + display_picture = None, session_token = inbound_headers["X-Session-Token"] ) diff --git a/api/main.py b/api/main.py index a6dc531..843a629 100644 --- a/api/main.py +++ b/api/main.py @@ -88,6 +88,7 @@ from controllers_v2.message.mail.gmail import GmailController # --- from controllers_v2.message.chat.all_chat import AllChatController from controllers_v2.message.chat.whatsapp_nimbus import WhatsAppNimbusController +from controllers_v2.message.chat.telegram import TelegramController # --- from controllers_v2.finstitutions.trading.all_trading import AllTradingController from controllers_v2.finstitutions.trading.zerodha_kite import ZerodhaKiteTradingController @@ -506,6 +507,12 @@ async def app_startup(**kwargs): alert_url = current_app.script_data["alerts"]["url"], debug = enable_debugging ) + current_app.telegram_controller = TelegramController( + cache = current_app.module_cache, + http_client = current_app.http_client, + alert_url = current_app.script_data["alerts"]["url"], + debug = enable_debugging + ) current_app.printer("Message/Chat (C) ready.") # Finstitutions / Trading Controllers: diff --git a/background/finstitutions/trading/strategy_bhandari_0.py b/background/finstitutions/trading/strategy_bhandari_0.py index 9111c9a..aa33cc4 100644 --- a/background/finstitutions/trading/strategy_bhandari_0.py +++ b/background/finstitutions/trading/strategy_bhandari_0.py @@ -233,10 +233,11 @@ async def call_is_active( tick_key = tick_ref["tickKey"] redis_key = tick_ref["redisKey"] + # Check if the call is active: if ACTIVE_CALLS.get(redis_key): return True - if await redis_cache.get(redis_key): - ACTIVE_CALLS[redis_key] = tick_ref - return True + # if await redis_cache.get(redis_key): + # ACTIVE_CALLS[redis_key] = tick_ref + # return True return False @@ -324,9 +325,9 @@ async def test_one_tick(tick: dict) -> None: "Message: `Failed to mark active call.`\n\n" f"Tick Key; `{tick_key}`\n" f"Redis Key; `{redis_key}`\n" - f"Entry: `{tick_reference['entry']}`\n\n" - f"Rate: `{tick_reference['rate']}`\n\n" - f"LTP: `{tick['ltp']}`\n\n" + f"Entry: `{tick_reference['entry']}`\n" + f"Rate: `{tick_reference['rate']}`\n" + f"LTP: `{tick['ltp']}`" ), message_type = "error" ) @@ -336,9 +337,9 @@ async def test_one_tick(tick: dict) -> None: "Message: `Marked active call.`\n\n" f"Tick Key; `{tick_key}`\n" f"Redis Key; `{redis_key}`\n" - f"Entry: `{tick_reference['entry']}`\n\n" - f"Rate: `{tick_reference['rate']}`\n\n" - f"LTP: `{tick['ltp']}`\n\n" + f"Entry: `{tick_reference['entry']}`\n" + f"Rate: `{tick_reference['rate']}`\n" + f"LTP: `{tick['ltp']}`" ), message_type = "info" ) @@ -655,7 +656,7 @@ async def refresh_strategy_reference() -> bool: # Save the new reference in the global variable: async with strategy_reference_lock: STRATEGY_REFERENCE = formatted_reference - print(json.to_string(list(STRATEGY_REFERENCE))) + # print(json.to_string(list(STRATEGY_REFERENCE))) # Done here: return True @@ -710,12 +711,12 @@ async def main( # Run the heartbeat task and the infinite tick-reading loop: tasks = [ - # ticks_from_kafka( - # consumer = kafka_consumer, - # fetch_count = 500, - # fetch_timeout = 2.5 - # ), - heartbeat(interval_seconds = heartbeat_interval) + ticks_from_kafka( + consumer = kafka_consumer, + fetch_count = 500, + fetch_timeout = 2.5 + ), + # heartbeat(interval_seconds = heartbeat_interval) ] await asyncio.gather(*tasks) diff --git a/controllers_v2/message/chat/telegram.py b/controllers_v2/message/chat/telegram.py new file mode 100644 index 0000000..5788b1e --- /dev/null +++ b/controllers_v2/message/chat/telegram.py @@ -0,0 +1,215 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Tuesday, 28th Jan., 2025. + + OBJECTIVE: + + To handle all Telegram-related behaviour from one place. + + REFERENCES: + + N/A + + DOWNLOADS: + + N/A + +""" + + +# ***************************************************************************************************************** +# ***** **** +# *** IMPORT *** +# ***** **** +# ***************************************************************************************************************** + + +# To make sibling directories accessible for imports: +import sys +sys.path.append(".") +sys.path.append("..") + +# My async utils: +from utils_v2.date_time import date_time +from utils_v2.database.async_mysql_v2 import AsyncMySQL +from utils_v2.database.async_mongo_v2 import AsyncMongo +from utils_v2.cache.async_redis_cache_v2 import AsyncRedisCache + +# Controllers: +from controllers_v2.message.chat.base import ChatController + +# Models: +from models.core.auth_token import CoreAuthTokenModel +from models.core.message import CoreMessageModel +from models.message.chat.send import ( + NimbusWhatsAppMessage, + ChatSendOneResult, + ChatSendManyResults +) + +# Chat clients: +from utils_v2.whatsapp.nimbus.controllers.async_nimbus_whatsapp import AsyncNimbusWhatsapp + +# To work with datatypes: +from typing import List, Any + +# To make HTTP requests: +import httpx + +# For asynchronous activities: +import asyncio + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** CLASSES *** +# ***** **** +# ***************************************************************************************************************** + + +class TelegramController(ChatController): + + # ┏┓┓ ┓┏ + # ┃ ┃┏┓┏┏ ┃┃┏┓┏┓┏ + # ┗┛┗┗┻┛┛ ┗┛┗┻┛ ┛ + + CLIENT_NAME = "telegram" + + # ┏┓ + # ┃ ┏┓┏┓┏╋┏┓┓┏┏╋┏┓┏┓ + # ┗┛┗┛┛┗┛┗┛ ┗┻┗┗┗┛┛ + + def __init__( + self, + cache: AsyncRedisCache = None, + http_client: httpx.AsyncClient = None, + alert_url: str = None, + debug: bool = True, + debug_prefix: str = "Telegram (C) | ", + debug_only_errors: bool = True + ): + + """ + This is the controller for Telegram. + :param cache: The object to use for caching results from database calls. + :param http_client: The HTTP client + :param debug: Whether, or not, you would like to print debugging messages: + :param debug_prefix: The prefix to print with the debugging messages. + :param debug_only_errors: Whether you would like to print only error messages or all messages. + :return: None. + """ + + # Invoke the parent's constructor: + super().__init__( + cache = cache, + alert_url = alert_url, + http_client = http_client, + base_filter = {"client": self.CLIENT_NAME}, + debug = debug, + debug_prefix = debug_prefix, + debug_only_errors = debug_only_errors + ) + + # Init a variable in a parent: + self._client = self.CLIENT_NAME + + # ┏┓ ┓ ┳┳┓ + # ┗┓┏┓┏┓┏┫ ┃┃┃┏┓┏┏┏┓┏┓┏┓┏ + # ┗┛┗ ┛┗┗┻ ┛ ┗┗ ┛┛┗┻┗┫┗ ┛ + # ┛ + + async def send_one_message( + self, + sql_conn: AsyncMySQL, + mongo_data_conn: AsyncMongo, + auth_token: CoreAuthTokenModel, + client: AsyncNimbusWhatsapp, + message: NimbusWhatsAppMessage, + tags: List[Any] + ) -> ChatSendOneResult: + + """ + To send one message from the third-party client. + :param sql_conn: The connection to the database to use for this operation. + :param mongo_data_conn: The connection to the database to use for this operation. + :param auth_token: The auth-token model for the account from which the message has to be sent. + :param client: The connection/instance of the third-party client to use to perform this operation. + :param message: The message that you want to send to the recipient. + :param tags: Any tags that you would like to attach to the message. To be used later for internal filtering. + :return: The structured response model to describe the operation. + """ + + raise NotImplementedError + + async def send_many_messages( + self, + sql_conn: AsyncMySQL, + mongo_data_conn: AsyncMongo, + http_client: httpx.AsyncClient, + auth_token: CoreAuthTokenModel, + client: AsyncNimbusWhatsapp | None, + messages: List[NimbusWhatsAppMessage], + tags: List[Any] + ) -> ChatSendManyResults: + + """ + To send many chat messages in one go. + :param sql_conn: The connection to the database to use for this operation. + :param mongo_data_conn: The connection to the database to use for this operation. + :param http_client: An HTTP client to use to make API calls through the third-party client's class. + :param auth_token: The auth-token model for the account from which the message has to be sent. + :param client: The connection/instance of the third-party client to use to perform this operation. + :param messages: The messages that you want to send to the recipients. + :param tags: Any tags that you would like to attach to the message. To be used later for internal filtering. + :return: The structured response model to describe the operation. + """ + + raise NotImplementedError + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + pass diff --git a/models/message/chat/auth.py b/models/message/chat/auth.py index 176426c..ca3b170 100644 --- a/models/message/chat/auth.py +++ b/models/message/chat/auth.py @@ -77,6 +77,18 @@ REGEX_SESSION_TOKEN = r"^[a-f0-9]{8}-[a-f0-9]{4}-[1-5][a-f0-9]{3}-[89ab][a-f0-9] class TelegramAuth(BaseModel): + botId: str = Field( + description = "The id of the bot (that can be invoked with '@').", + min_length = 1, + frozen = True + ) + + botName: str = Field( + description = "The display name of the bot.", + min_length = 1, + frozen = True + ) + botToken: str = Field( description = "the token granted by BotFather", min_length = 1,