diff --git a/api/blueprints/common/otp/timed_otp.py b/api/blueprints/common/otp/timed_otp.py index e0bb4f8..91e40e4 100644 --- a/api/blueprints/common/otp/timed_otp.py +++ b/api/blueprints/common/otp/timed_otp.py @@ -291,7 +291,7 @@ async def generate_otp( @handle_cancelled_request() async def verify_otp( inbound_headers: dict = None, - inbound_data: dict | VerifyTimedOTP = None, + inbound_data: dict | VerifyTimedOTPRequestData = None, inbound_files: dict = None, **kwargs ): @@ -350,35 +350,65 @@ async def verify_otp( # ┣┫┏┫┏┫ ┃ ┃┓┏┓┏┓╋ ┃┃┃┓╋┣┓ ┃┃┏┓╋┏┓┏ # ┛┗┗┻┗┻ ┗┛┗┗┗ ┛┗┗ ┗┻┛┗┗┛┗ ┛┗┗┛┗┗ ┛ - # Fix the phone no. to have the country code: + # Fix the phone no. to have the country code: phone_no = regex.replace( text = inbound_data.phoneNo, pattern = r"[^\d]", substitute_text = "" ) - if not phone_no.startswith("254"): phone_no = "254" + phone_no - phone_no = "+" + phone_no + if phone_no.startswith("254"): phone_no = phone_no[3:] + if phone_no.startswith("0"): phone_no = phone_no[1:] + phone_no = "+254" + phone_no - # hit Omkar's API: - api_response = await current_app.http_client.post( - url = "https://api.thecaoffice.com/client/add/with/notes", - headers = {"X-Session-Token": inbound_headers["X-Session-Token"]}, - json = { - "email": inbound_data.email, - "clientName": inbound_data.username, - "address": None, - "phoneNo": phone_no, - "city": None, - "pincode": None, - "panCard": None, - "gst": None, - "entity": None, - "country": None, - "startDate": None, - "period": None, - "amount": None - } - ) + # Do this when the user is trying a new sign-up: + # Hit Omkar's API: + if inbound_data.isSignup: + api_response = await current_app.http_client.post( + url = "https://api.thecaoffice.com/client/add/with/notes", + headers = {"X-Session-Token": inbound_headers["X-Session-Token"]}, + json = { + "email": inbound_data.email, + "clientName": inbound_data.username, + "password": inbound_data.password, + "address": None, + "phoneNo": phone_no, + "city": None, + "pincode": None, + "panCard": None, + "gst": None, + "entity": None, + "country": None, + "startDate": None, + "period": None, + "amount": None + } + ) + + # When this is not a new sign-up. it is a password reset request: + # Hit Omkar's API: + else: + api_response = await current_app.http_client.post( + url = "https://api.thecaoffice.com/client/update/with/notes", + headers = {"X-Session-Token": inbound_headers["X-Session-Token"]}, + json = { + "idClient": inbound_data.idClient, + "email": inbound_data.email, + "clientName": inbound_data.clientName, + "username": inbound_data.username, + "password": inbound_data.password, + "address": inbound_data.address, + "phoneNo": phone_no, + "city": inbound_data.city, + "pincode": inbound_data.pincode, + "panCard": inbound_data.panCard, + "gst": inbound_data.gst, + "entity": inbound_data.entity, + "country": inbound_data.country, + "startDate": inbound_data.startDate, + "period": inbound_data.period, + "amount": inbound_data.amount + } + ) # ┳┓ # ┣┫┏┓┏┏┓┏┓┏┓┏┏┓ @@ -389,7 +419,7 @@ async def verify_otp( success = api_response.is_success return ResponseModel( status_code = StatusCodes.OK if success else StatusCodes.FAILED, - http_code = HttpCodes.SUCCESS if success else HttpCodes.UNAUTHORIZED + http_code = HttpCodes.SUCCESS if success else HttpCodes.UNAUTHORIZED, ) diff --git a/api/blueprints/message/mail/retrieve/list_v2.py b/api/blueprints/message/mail/retrieve/list_v2.py index 9237181..310d2cd 100644 --- a/api/blueprints/message/mail/retrieve/list_v2.py +++ b/api/blueprints/message/mail/retrieve/list_v2.py @@ -193,8 +193,17 @@ async def list_mails( if inbound_data.tags: additional_filter["tags"] = {"$in": inbound_data.tags} additional_filter = additional_filter or None - # Get the mails: - mails_list = await current_app.mail_controller.list_mails( + # # Get the mails: + # mails_list = await current_app.mail_controller.list_mails( + # mongo_data_conn = current_app.data_mongo, + # token_ids = token_ids, + # limit = inbound_data.count, + # skip = inbound_data.fromCount, + # additional_filter = additional_filter + # ) + + # Because someone wanted all sorts of messages in one place so fucking be it: + mails_list = await current_app.core_message_controller.get_message_previews( mongo_data_conn = current_app.data_mongo, token_ids = token_ids, limit = inbound_data.count, diff --git a/api/main.py b/api/main.py index 7e6460b..724eb4a 100644 --- a/api/main.py +++ b/api/main.py @@ -66,7 +66,7 @@ from utils_v2.goog.controllers.gmail.gmail_client import AsyncGmailClient from utils_v2.whatsapp.nimbus.controllers.async_nimbus_whatsapp import AsyncNimbusWhatsapp # Core Controller Models: -from controllers.core.message import CoreMessageController +# from controllers.core.message import CoreMessageController # from controllers.core.auth_token import CoreAuthTokenController from controllers.core.ai.llm import CoreLLMController from controllers.core.payment import CorePaymentController @@ -78,6 +78,7 @@ from controllers.core.payment import CorePaymentController # Controllers V2: from controllers_v2.core.auth_token import CoreAuthTokenController +from controllers_v2.core.message import CoreMessageController # --- from controllers_v2.message.sms.all_sms import AllSMSController from controllers_v2.message.sms.nimbus_sms_india import NimbusSMSIndiaController @@ -465,14 +466,20 @@ async def app_startup(**kwargs): # ┃ ┏┓┏┓╋┏┓┏┓┃┃┏┓┏┓┏ ┃┃┏┛ # ┗┛┗┛┛┗┗┛ ┗┛┗┗┗ ┛ ┛ ┗┛┗━ - # Auth-Token Controller(s): + # Core Controller(s): current_app.core_auth_token_controller = CoreAuthTokenController( 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("Auth-Token (C) ready.") + current_app.core_message_controller = CoreMessageController( + 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("Core (C) ready.") # Messages / SMS Controllers: current_app.sms_controller = AllSMSController( diff --git a/background/finstitutions/trading/tick_in_stateful.py b/background/finstitutions/trading/tick_in_stateful.py index 0ebc8c8..2442b19 100644 --- a/background/finstitutions/trading/tick_in_stateful.py +++ b/background/finstitutions/trading/tick_in_stateful.py @@ -58,6 +58,9 @@ import socket import datetime import time +# To work with tabulated data: +import pandas as pd + # Controllers: from controllers_v2.finstitutions.trading.all_trading import AllTradingController @@ -236,6 +239,12 @@ def setup_zerodha_kite_feed( :return: True if successful, else False. """ + # Ensure that we filter out duplicate records: + total_instruments_df = pd.DataFrame(total_instruments) + # print(total_instruments_df.to_string()) + total_instruments_df.drop_duplicates(subset = "broker_token", keep = "first", inplace = True) + total_instruments = total_instruments_df.to_dict(orient = "records") + # Declare the required global variables: global ZERODHA_INSTRUMENT_TOKENS global ZERODHA_INSTRUMENT_LOOKUP @@ -271,10 +280,34 @@ def setup_zerodha_kite_feed( instruments = [TradingSymbol.from_zerodha_kite(i) for i in instruments] # Create the lookup: + # invalid_broken_token_count = 0 + # print("INVALID BROKEN TOKENS:\n") + all_temp_instr = {} + pop_count = 0 for i in instruments: if str(i.brokerToken) in valid_broker_tokens: ZERODHA_INSTRUMENT_TOKENS.append(i.brokerToken) ZERODHA_INSTRUMENT_LOOKUP[i.brokerToken] = i.model_dump() + all_temp_instr[i.brokerToken] = i + pop_count += 1 + + print(f"Populated: {pop_count}") + invalid_broken_token_count = 0 + not_found_broker_tokens = [] + for broker_token, broker_symbol in zip(valid_broker_tokens, valid_broker_symbols): + broker_token = int(broker_token) + if ZERODHA_INSTRUMENT_LOOKUP.get(broker_token) is None: + not_found_broker_tokens.append(str(broker_token)) + try: print(f"FAILED: {broker_token: ^15} | {all_temp_instr[broker_token].symbol: ^30} | {all_temp_instr[broker_token].exchange}") + except: print(f"FAILED: {broker_token: ^15} | {broker_symbol: ^30} | ") + invalid_broken_token_count += 1 + print("\nTOTAL:", invalid_broken_token_count) + not_found_df = total_instruments_df[total_instruments_df["broker_token"].isin(not_found_broker_tokens)] + not_found_df['expiry_date'] = pd.to_datetime(not_found_df['expiry_date']) + not_found_df['expiry_date'] = not_found_df['expiry_date'].dt.strftime('%Y-%m-%d') + # print(not_found_df.to_string()) + # while True: pass + printer("Zerodha instruments loaded.") # If there are no symbols or too many symbols, we return with failure: diff --git a/controllers_v2/core/message.py b/controllers_v2/core/message.py index f2d44fa..68f46e3 100644 --- a/controllers_v2/core/message.py +++ b/controllers_v2/core/message.py @@ -34,7 +34,8 @@ import sys sys.path.append(".") sys.path.append("..") -# My async utils: +# My utils: +from utils_v2.string import json from utils_v2.database.async_mongo_v2 import AsyncMongo from utils_v2.cache.async_redis_cache_v2 import AsyncRedisCache diff --git a/controllers_v2/software/mikrotik/mikrotik_hostpot_1000.py b/controllers_v2/software/mikrotik/mikrotik_hostpot_1000.py index 418ae02..7d54e81 100644 --- a/controllers_v2/software/mikrotik/mikrotik_hostpot_1000.py +++ b/controllers_v2/software/mikrotik/mikrotik_hostpot_1000.py @@ -735,7 +735,9 @@ class MikroTikHotspot1000Controller(MikroTikController): mikrotik_client.new_action_chain(action_chain) # Try to configure the device: + roll_back_response = await self.quick_roll_back(mikrotik_client, use_https = False) config_response = await self.quick_config(mikrotik_client, mikrotik_auth, use_https = False) + config_response = roll_back_response + config_response config_response.actionChain = mikrotik_client.action_chain return config_response diff --git a/controllers_v2/software/mikrotik/mikrotik_pppoe_1000_v2.py b/controllers_v2/software/mikrotik/mikrotik_pppoe_1000_v2.py new file mode 100644 index 0000000..581bb74 --- /dev/null +++ b/controllers_v2/software/mikrotik/mikrotik_pppoe_1000_v2.py @@ -0,0 +1,532 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Monday, 3rd Mar., 2025. + + OBJECTIVE: + + To handle configuration for MikroTik servers such that they work in PPPoE mode with support for 1,000 clients. + + 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.string import json +from utils_v2.date_time import date_time +from utils_v2.mikrotik.controllers.async_mikrotik import AsyncMikroTik +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.software.mikrotik.base import MikroTikController + +# To make very controlled API calls: +from utils_v2.rest.controllers.async_base import AsyncREST +from utils_v2.rest.models.api_call import ApiResponse + +# Models: +from models.software.mikrotik.auth import ( + MikroTikPPPoE1000Auth, + MikroTikHotspot1000Auth, + MikroTikAuthResponse +) +from models.software.mikrotik.configure import MikroTikConfigAttemptResponse + +# To work with datatypes: +from typing import List, Any + +# To make HTTP requests: +import httpx + +# To work with IP addresses: +import ipaddress + +# to work with MongoDB: +from bson.objectid import ObjectId + +# To make abstract classes: +from abc import ABC, abstractmethod + +# For asynchronous activities: +import asyncio + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** CLASSES *** +# ***** **** +# ***************************************************************************************************************** + + +class MikroTikPPPoE1000Controller(MikroTikController): + + # ┏┓┓ ┓┏ + # ┃ ┃┏┓┏┏ ┃┃┏┓┏┓┏ + # ┗┛┗┗┻┛┛ ┗┛┗┻┛ ┛ + + CLIENT_NAME = "mikrotikPPPoE1000" + + # For automated configuration, and identification of automated configuration: + NAME_PREFIX = "easyfi" + CREATED_BY_NAME = NAME_PREFIX + "-pppoe-1000" + HW_INTERFACE_NAME = NAME_PREFIX + "-pppoe" + VLAN_NAME = NAME_PREFIX + "-vlan-{}" # .................... Substitute the VLAN's id here. + PRIVATE_IP_POOL_NAME = NAME_PREFIX + "-pppoe-pool-{}" # ... Substitute the VLAN's id here. + PPP_PROFILE_NAME = NAME_PREFIX + "-pppoe-prf-{}" # ........ Substitute the VLAN's id here. + PPP_SERVER_NAME = NAME_PREFIX + "-pppoe-srv-{}" # ......... Substitute the VLAN's id here. + RADIUS_SERVER_NAME = NAME_PREFIX + "-radius" # ............ Substitute the VLAN's id here. + NAT_RULE_NAME = NAME_PREFIX + "-pppoe-nat-{}" # ........... Substitute rule no. here. + SNMP_COMMUNITY_NAME = NAME_PREFIX + "-snmp" + + # ┏┓ + # ┃ ┏┓┏┓┏╋┏┓┓┏┏╋┏┓┏┓ + # ┗┛┗┛┛┗┛┗┛ ┗┻┗┗┗┛┛ + + def __init__( + self, + cache: AsyncRedisCache = None, + http_client: httpx.AsyncClient = None, + alert_url: str = None, + debug: bool = True, + debug_prefix: str = "MTik. PPPoE 1K (C) | ", + debug_only_errors: bool = True + ): + + """ + This is a special class that specifically handles configuration for MikroTik servers that work in PPPoE mode and + have support for upto 1,000 active clients. + :param cache: The object to use for caching results from database calls. + :param http_client: The HTTP client to use to make REST-ful API calls. + :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: + MikroTikController.__init__( + self, + 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 quick_config( + self, + mikrotik_client: AsyncMikroTik, + mikrotik_auth: MikroTikHotspot1000Auth, + use_https: bool = True + ) -> MikroTikConfigAttemptResponse: + + """ + To very quickly set up the MikroTik with just one VLAN. + :param mikrotik_client: The client to use. + :param mikrotik_auth: The set of credentials as received from the UI/API. + :param use_https: Whether to use HTTPS, or HTTP. + :return: A structured response to indicate what happened during the process. + """ + + # Start by assuming values: + config_result = MikroTikConfigAttemptResponse() + all_success = True + all_messages = ["STARTING QUICK CONFIG."] + + # Prepare the needed variables: + PPP_VLAN_ID = mikrotik_auth.vlanIds[0] + PPP_INTERFACE_NAME = self.HW_INTERFACE_NAME + PPP_VLAN_NAME = self.VLAN_NAME.format(PPP_VLAN_ID) + PPP_IP_POOL_NAME = self.PRIVATE_IP_POOL_NAME.format(PPP_VLAN_ID) + PPP_PROFILE_NAME = self.PPP_PROFILE_NAME.format(PPP_VLAN_ID) + PPP_SERVER_NAME = self.PPP_SERVER_NAME.format(PPP_VLAN_ID) + PPP_DHCP_SERVER_NAME = self.PRIVATE_IP_POOL_NAME.format(PPP_VLAN_ID) + PPP_SNMP_COMMUNITY_NAME = mikrotik_auth.snmpCommunity + # --- + PPP_PUBLIC_IP_SUBNET = mikrotik_auth.publicIpPool + PPP_FIRST_PUBLIC_IP = mikrotik_auth.firstPublicIp + PPP_LAST_PUBLIC_IP = mikrotik_auth.lastPublicIp + # --- + PPP_PRIVATE_IP_SUBNET = str(next(ipaddress.summarize_address_range( + ipaddress.IPv4Address(mikrotik_auth.firstPrivateIp), + ipaddress.IPv4Address(mikrotik_auth.lastPrivateIp) + ))) + PPP_FIRST_PRIVATE_IP = mikrotik_auth.firstPrivateIp + PPP_LAST_PRIVATE_IP = mikrotik_auth.lastPrivateIp + PPP_GATEWAY_PRIVATE_IP = str(ipaddress.IPv4Address(mikrotik_auth.firstPrivateIp) + 1) + # --- + PPP_RADIUS_SERVER_IP = mikrotik_auth.radius + PPP_RADIUS_SERVER_SECRET = mikrotik_auth.secret + # --- + PPP_COMMENT = mikrotik_client.create_comment_json() + + # STEP 1: + # Set up the interface: + api_response = await mikrotik_client.list_interfaces(use_https = use_https) + target_if_dot_id = None + for interface in api_response.data or []: + if interface["type"] == "ether": + target_if_dot_id = interface[".id"] + break + api_response = await mikrotik_client.update_interface( + dot_id = target_if_dot_id, + json_payload = { + "name": PPP_INTERFACE_NAME, + "comment": PPP_COMMENT + }, + use_https = use_https + ) + all_messages.append(f"STEP 1 ({api_response.action}): {api_response.success}") + if not api_response.success: all_success = False + + # STEP 2: + # Create the VLAN: + api_response = await mikrotik_client.add_vlan( + json_payload = { + "name": PPP_VLAN_NAME, + "interface": PPP_INTERFACE_NAME, + "vlan-id": PPP_VLAN_ID, + "disabled": "false", + "comment": PPP_COMMENT + } + ) + all_messages.append(f"STEP 2 ({api_response.action}): {api_response.success}") + if not api_response.success: all_success = False + + # STEP 3: + # Add a new IP Pool for the PPPoE users: + api_response = await mikrotik_client.add_ip_pool( + json_payload = { + "name": PPP_IP_POOL_NAME, + "ranges": PPP_PRIVATE_IP_SUBNET, + "comment": PPP_COMMENT + } + ) + all_messages.append(f"STEP 3 ({api_response.action}): {api_response.success}") + if not api_response.success: all_success = False + + # STEP 4: + # Add a new PPPoE Profile: + api_response = await mikrotik_client.add_ppp_profile( + json_payload = { + "name": PPP_PROFILE_NAME, + "local-address": PPP_GATEWAY_PRIVATE_IP, + "remote-address": PPP_IP_POOL_NAME, + "comment": PPP_COMMENT + } + ) + all_messages.append(f"STEP 4 ({api_response.action}): {api_response.success}") + if not api_response.success: all_success = False + + # STEP 5: + # Add a new PPPoE Server: + api_response = await mikrotik_client.add_ppp_server( + json_payload = { + "interface": PPP_VLAN_ID, + "profile": PPP_PROFILE_NAME, + "service-name": PPP_SERVER_NAME, + "comment": PPP_COMMENT + } + ) + all_messages.append(f"STEP 5 ({api_response.action}): {api_response.success}") + if not api_response.success: all_success = False + + # STEP 6: + # Set up PPPoE's AAA such that it uses RADIUS: + api_response = await mikrotik_client.set_ppp_aaa( + json_payload = { + "accounting": "true", + "interim-update": "3m", + "use-circuit-id-in-nas-port-id": "false", + "use-radius": "true" + } + ) + all_messages.append(f"STEP 6 ({api_response.action}): {api_response.success}") + if not api_response.success: all_success = False + + # STEP 7: + # Add a new RADIUS Server: + api_response = await mikrotik_client.add_radius_server( + json_payload = { + "accounting-port": "1813", + "address": PPP_RADIUS_SERVER_IP, + "authentication-port": "1812", + "disabled": "false", + "protocol": "udp", + "secret": PPP_RADIUS_SERVER_SECRET, + "service": "ppp,login,hotspot,dhcp", + "timeout": "300ms", + "comment": PPP_COMMENT + } + ) + all_messages.append(f"STEP 7 ({api_response.action}): {api_response.success}") + if not api_response.success: all_success = False + + # STEP 8: + # Allow incoming traffic from RADIUS: + api_response = await mikrotik_client.set_radius_incoming( + json_payload = { + "accept": "yes", + "port": "3799", + # "comment": HS_COMMENT # ... NOT SUPPORTED! + } + ) + all_messages.append(f"STEP 8 ({api_response.action}): {api_response.success}") + if not api_response.success: all_success = False + + async def quick_roll_back( + self, + mikrotik_client: AsyncMikroTik, + use_https: bool = True + ) -> MikroTikConfigAttemptResponse: + + """ + To very quickly roll back the setup done with the quick config method. + :param mikrotik_client: The client to use. + :param use_https: Whether to use HTTPS, or HTTP. + :return: A structured response to indicate what happened during the process. + """ + + # Start by assuming values: + config_result = MikroTikConfigAttemptResponse() + all_success = True + all_messages = ["STARTING QUICK ROLL-BACK."] + + # ┏┓ ┓ + # ┣┫┓┏╋┣┓ + # ┛┗┗┻┗┛┗ + + async def save_auth( + self, + sql_conn: AsyncMySQL, + mongo_data_conn: AsyncMongo, + mikrotik_auth: MikroTikPPPoE1000Auth, + action_chain: str | int = None + ) -> MikroTikAuthResponse: + + """ + Checks if a particular set of incoming credentials give access to a valid server and then stores the + credentials. + :param sql_conn: The database connection to use to perform this task. + :param mongo_data_conn: The database connection to use to perform this task. + :param mikrotik_auth: The set of credentials as received from the UI/API. + :param action_chain: A custom action chain to apply for logging the steps. + :return: A structured response to indicate what happened during authorization. + """ + + # Start with a blank response: + auth_response = MikroTikAuthResponse() + + # Create a MikroTik client: + mikrotik_client = AsyncMikroTik( + config_by = self.CREATED_BY_NAME, + mikrotik_ip = mikrotik_auth.nasIp, + username = mikrotik_auth.username, + password = mikrotik_auth.password, + port = mikrotik_auth.nasPort, + use_https = True, + http_client = None, + action_log_conn = mongo_data_conn + ) + mikrotik_client.new_action_chain(action_chain) + + # Try to configure the device: + # roll_back_response = await self.quick_roll_back(mikrotik_client, use_https = False) + config_response = await self.quick_config(mikrotik_client, mikrotik_auth, use_https = False) + # config_response = roll_back_response + config_response + config_response.actionChain = mikrotik_client.action_chain + return config_response + + # ┏┓ ┏• + # ┃ ┏┓┏┓╋┓┏┓ + # ┗┛┗┛┛┗┛┗┗┫ + # ┛ + + async def roll_back( + self, + mikrotik_client: AsyncMikroTik, + mikrotik_auth: MikroTikPPPoE1000Auth + ) -> MikroTikConfigAttemptResponse: + + """ + The rolling-back to the original state (as best as possible) in case the configurations fails midway after + completing some no. of steps. + :param mikrotik_client: The client to use. + :param mikrotik_auth: The set of credentials as received from the UI/API. + :return: A structured response to indicate what happened during the configuration attempt. + """ + + # Start with some variables: + keep_going = True + all_messages = ["STARTING ROLLBACK."] + roll_back_response = MikroTikConfigAttemptResponse() + + # Next, we roll back all the PPPoE Servers: + if keep_going: + step_response = await self.roll_back_ppp_servers(mikrotik_client, use_https = False) + all_messages.append(step_response.message) + keep_going = step_response.success + + # Next, we roll back all the PPPoE Profiles: + if keep_going: + step_response = await self.roll_back_ppp_profiles(mikrotik_client, use_https = False) + all_messages.append(step_response.message) + keep_going = step_response.success + + # Next, we roll back all the IP pools: + if keep_going: + step_response = await self.roll_back_ip_pools(mikrotik_client, use_https = False) + all_messages.append(step_response.message) + keep_going = step_response.success + + # Next we roll back all the VLANs: + if keep_going: + step_response = await self.roll_back_vlans(mikrotik_client, use_https = False) + all_messages.append(step_response.message) + keep_going = step_response.success + + # Next, we free-up the interface: + if keep_going: + step_response = await self.roll_back_interface(mikrotik_client, use_https = False) + all_messages.append(step_response.message) + keep_going = step_response.success + + # Done here: + roll_back_response.success = keep_going + roll_back_response.message = " -> ".join(all_messages) + return roll_back_response + + async def configure( + self, + mikrotik_client: AsyncMikroTik, + mikrotik_auth: MikroTikPPPoE1000Auth + ) -> MikroTikConfigAttemptResponse: + + """ + Run the configuration steps for the system. + :param mikrotik_client: The client to use. + :param mikrotik_auth: The set of credentials as received from the UI/API. + :return: A structured response to indicate what happened during the configuration attempt. + """ + + # Start with some variables: + keep_going = True + all_messages = ["STARTING CONFIG."] + config_response = MikroTikConfigAttemptResponse(actionChain = mikrotik_client.action_chain) + + # First, we arrange an interface: + if keep_going: + step_response = await self.set_up_interface(mikrotik_client, dot_id = "*3", use_https = False) + all_messages.append(step_response.message) + keep_going = step_response.success + + # Next, we create the needed VLANs: + if keep_going: + await asyncio.sleep(0.25) + step_response = await self.set_up_vlans(mikrotik_client, mikrotik_auth, use_https = False) + all_messages.append(step_response.message) + keep_going = step_response.success + + # Next, we create the Private IP Pools: + if keep_going: + await asyncio.sleep(0.25) + step_response = await self.set_up_ip_pools(mikrotik_client, mikrotik_auth, use_https = False) + all_messages.append(step_response.message) + keep_going = step_response.success + + # Next, we create the PPPoE Profiles: + if keep_going: + await asyncio.sleep(0.25) + step_response = await self.set_up_ppp_profiles(mikrotik_client, mikrotik_auth, use_https = False) + all_messages.append(step_response.message) + keep_going = step_response.success + + # Next, we create the PPPoE Servers: + if keep_going: + await asyncio.sleep(0.25) + step_response = await self.set_up_ppp_servers(mikrotik_client, mikrotik_auth, use_https = False) + all_messages.append(step_response.message) + keep_going = step_response.success + + # Done here: + config_response.success = keep_going + config_response.message = " -> ".join(all_messages) + return config_response + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + pass diff --git a/cron/finstitutions/trading/latest_ohlc_from_ticks.py b/cron/finstitutions/trading/latest_ohlc_from_ticks.py index 7362530..aefc17d 100644 --- a/cron/finstitutions/trading/latest_ohlc_from_ticks.py +++ b/cron/finstitutions/trading/latest_ohlc_from_ticks.py @@ -10,7 +10,7 @@ OBJECTIVE: - To infer the day's latest EoD + To infer the day's latest OHLC values from the ticks stored in the database. REFERENCES: @@ -93,8 +93,8 @@ http_client = httpx.AsyncClient( ) # For debugging: -printer = IceCreamDebugger(prefix = "EoD (Ticks) | ", includeContext = True) -no_context_printer = IceCreamDebugger(prefix = "EoD (Ticks) | ", includeContext = False) +printer = IceCreamDebugger(prefix = "1D OHLC (Ticks) | ", includeContext = True) +no_context_printer = IceCreamDebugger(prefix = "1D OHLC (Ticks) | ", includeContext = False) # ***************************************************************************************************************** @@ -170,7 +170,7 @@ async def init( SCRIPT_DATA = response.json().get("data") # Done with this step: - printer("Cred and Data loaded.") + no_context_printer("Cred and Data loaded.") # ┳┳┓ • ┳┓┳┓ # ┃┃┃┏┓┏┓┓┏┓┃┃┣┫ @@ -187,7 +187,7 @@ async def init( print("FATAL: MARIA-DB CONNECTION FAILED!") return False - printer("MariaDB connected.") + no_context_printer("MariaDB connected.") # ┳┳┓ # ┃┃┃┏┓┏┓┏┓┏┓ @@ -211,7 +211,7 @@ async def init( # ┻┛┗┛┛┗┗ # If everything went well, we return with success: - printer("Initialization done.") + no_context_printer("Initialization done.") return True @@ -259,21 +259,21 @@ async def send_telegram( # --------------------------------------------------------------------------------------------------------------------- -async def get_latest_eod_data(target_date: datetime.datetime = None) -> dict: +async def get_latest_ohlc_data(target_date: datetime.datetime = None) -> dict: """ - To fetch the latest EoD (daily candle) data from the ticks database. + To fetch the latest 1D OHLC (daily candle) data from the ticks database. :param target_date: The date (UTC) whose EoD ticks are desired. :return: The inferred daily candle data. """ - no_context_printer("Getting EoD data from ticks.") + no_context_printer("Getting 1D OHLC data from ticks.") # Prepare the inputs needed for the aggregation: if not isinstance(target_date, datetime.datetime): target_date = date_time.get_current_utc_date_time() else: target_date = date_time.to_timezone(target_date, date_time.TIMEZONE_UTC) - start_ts = target_date.replace(hour = 3, minute = 45, second = 0, microsecond = 0) - end_ts = target_date.replace(hour = 10, minute = 0, second = 0, microsecond = 0) + start_ts = target_date.replace(hour = 0, minute = 0, second = 0, microsecond = 0) + end_ts = target_date.replace(hour = 23, minute = 59, second = 59, microsecond = 999) # Construct the aggregation pipeline: # Consider only the target date's ticks: @@ -289,77 +289,39 @@ async def get_latest_eod_data(target_date: datetime.datetime = None) -> dict: # Add a field that has the rounded timestamp. # We round it to one day for EoD data: stage_1 = { - "$addFields": { - "roundTs": { - "$dateTrunc": { - "date": "$tradeTs", - "unit": "day", - "binSize": 1 - } - } + "$sort": { + "tradeTs": -1 } } # Now we convert tick to candlesticks: stage_2 = { "$group": { - "_id": { - "roundTs": "$roundTs", - "symbol": "$symbol" - }, - "roundTs": {"$last": "$roundTs"}, - "symbol": {"$last": "$symbol"}, - "name": {"$last": "$name"}, - "exchange": {"$last": "$exchange"}, - "segment": {"$last": "$segment"}, - "type": {"$last": "$type"}, - "expiry": {"$last": "$expiry"}, - "strike": {"$last": "$strike"}, - "open": {"$first": "$ltp"}, - "high": {"$max": "$ltp"}, - "low": {"$min": "$ltp"}, - "close": {"$last": "$ltp"}, - "vwap": {"$last": "$vwap"}, - "chg": {"$last": "$chg"}, - "pChg": {"$last": "$pChg"}, - "volume": {"$last": "$totVol"}, - "dayHigh": {"$last": "$h"}, - "dayLow": {"$last": "$l"}, - "ticks": {"$sum": 1}, - "broker": {"$last": "$broker"}, - "brokerToken": {"$last": "$brokerToken"}, - } - } - - # Finally we organize and present the data: - stage_3 = { - "$sort": { - "symbol": 1, - "roundTs": 1 - } - } - stage_4 = { - "$project": { - "_id": False + "_id": "$symbol", + "latestTick": {"$first": "$$ROOT"} } } # Now we run the aggregation: - eod_data = await data_mongo.aggregate( + agg_ohlc_data = await data_mongo.aggregate( collection = "__hot_zerodhaTicks", pipeline = [ stage_0, stage_1, - stage_2, - stage_3, - stage_4 + stage_2 ], limit = None, raise_exception = False ) + # Format the aggregation now: + ohlc_data = { + item["_id"]: item["latestTick"] + for item in agg_ohlc_data or [] + } + # Done here: - return eod_data + return ohlc_data # --------------------------------------------------------------------------------------------------------------------- @@ -369,7 +331,7 @@ async def save_latest_eod_data(eod_data: dict) -> bool: """ To save the loaded data to the SQL database. - :param eod_data: The dict of the EoD data received from the ticks database. + :param eod_data: The dict of the latest 1D OHLC data received from the ticks database. :return: True if successful, else False """ @@ -384,10 +346,10 @@ async def save_latest_eod_data(eod_data: dict) -> bool: "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s);" ) query_data = [] - for data in eod_data: - total_cash = data["vwap"] * data["volume"] - target_date = data["roundTs"].strftime("%Y-%m-%d") - curr_close = data["close"] + for symbol, data in eod_data.items(): + total_cash = data["vwap"] * data["totVol"] + target_date = data["tradeTs"].strftime("%Y-%m-%d") + curr_close = data["ltp"] prev_close = curr_close - data["chg"] one_query_data = ( data["exchange"], # ......................... exchange @@ -399,28 +361,40 @@ async def save_latest_eod_data(eod_data: dict) -> bool: data["expiry"], # ........................... expiry data["strike"], # ........................... strike prev_close, # ............................... prev_close - data["open"], # ............................. open - data["high"], # ............................. high - data["low"], # .............................. low + data["o"], # ................................ open + data["h"], # ................................ high + data["l"], # ................................ low curr_close, # ............................... close curr_close, # ............................... ltp data["vwap"], # ............................. vwap - data["volume"], # ........................... tot_vol + data["totVol"], # ........................... tot_vol total_cash, # ............................... tot_cash None, # ..................................... delivery_vol None, # ..................................... delivery_pct None, # ..................................... oi None, # ..................................... oi_chg target_date, # .............................. date - data["roundTs"].replace(tzinfo = None), # ... ts + data["tradeTs"].replace(tzinfo = None), # ... ts "Asia/Kolkata", # ........................... tz scrape_ts.replace(tzinfo = None), # ......... scrape_ts ) one_query_data = [None if pd.isna(d) else d for d in one_query_data] query_data.append(one_query_data) + # If the query data is empty: + if not query_data: + await send_telegram( + message = ( + f"*1D OHLC From Ticks:*\n\n" + "Message: `Did NOT find any 1D OHLC data from ticks.`\n\n" + ), + message_type = "warning" + ) + return False + # Run the commands: no_context_printer("Saving data to SQL DB.") + no_context_printer(len(query_data)) rows_affected, db_response, db_exception = await sql_writer.execute_many( query = query_str, data = query_data, @@ -428,7 +402,7 @@ async def save_latest_eod_data(eod_data: dict) -> bool: ) if db_exception: await send_telegram( message = ( - f"*EoD From Ticks:*\n\n" + f"*1D OHLC From Ticks:*\n\n" "Message: `SQL database threw an exception.`\n\n" f"Exception: `{db_exception}`" ), @@ -452,38 +426,39 @@ async def run_once() -> bool: """ # Get the data from the ticks database: - eod_data = await get_latest_eod_data() - if not eod_data: - await send_telegram( - message = ( - f"*EoD From Ticks:*\n\n" - "Message: `Failed to get EoD candles from ticks.`" - ), - message_type = "error" - ) - return False + ohlc_data = await get_latest_ohlc_data() # If there is no data to save: - if not eod_data: + if not ohlc_data: + + # Send out an alert: await send_telegram( message = ( - f"*EoD From Ticks:*\n\n" - "Message: `No EoD data to save to SQL.`" + f"*1D OHLC From Ticks:*\n\n" + "Message: `Failed to get 1D OHLC candles from ticks.`" ), message_type = "error" ) + + # Return with failure: return False # Save the data to the SQL database: - success = await save_latest_eod_data(eod_data) + success = await save_latest_eod_data(ohlc_data) + + # If the data was not saved: if not success: + + # Send out an alert: await send_telegram( message = ( - f"*EoD From Ticks:*\n\n" + f"*1D OHLC From Ticks:*\n\n" "Message: `Failed to save EoD data to SQL.`" ), message_type = "error" ) + + # Return with failure: return False # If both th steps succeeded, we are good to go: @@ -493,39 +468,60 @@ async def run_once() -> bool: # --------------------------------------------------------------------------------------------------------------------- -async def main( - start_time: datetime.datetime, - end_time: datetime.datetime, - interval_seconds: int = 300, -): +async def main(script_args) -> None: """ 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 script_args: The args received from the command line. :return: None. """ + # ┏┳ ┓ ┏┳┓• + # ┃┏┓┣┓ ┃ ┓┏┳┓┏┓┏╋┏┓┏┳┓┏┓┏ + # ┗┛┗┛┗┛ ┻ ┗┛┗┗┗ ┛┗┗┻┛┗┗┣┛┛ + # ┛ + + # Figure out the system's timezone so that cron activities can run as per it: + system_tz = date_time.get_system_timezone(as_string = False) + # Start configuring the scheduler: - printer("Configuring the schedule-manager.") + no_context_printer("Configuring the schedule-manager.") schedule_manager = Scheduler() - # Create all the timestamps at which the job must be done: - all_job_ts = [] + # Get the current date-time and parse the open and close time values as UTC. + # The input string does NOT have the date value. By default, python will take a date from way back in the past. + # When you translate from UTC to the local machine's timezone, if the date value is from way back in the past, an + # accidental historical timezone may get applied. Fo example, there was a time before the adoption of IST when + # "Asia/Kolkata" meant an offset of +05:53. Ensure to replace the date values to today's values to avoid ending up + # with old offsets: + now = date_time.get_current_utc_date_time(as_string = False) + now = date_time.to_timezone(now, timezone = system_tz) + # --- + jobs_start_time = datetime.datetime.strptime(script_args.start_time, "%H:%M:%S") + jobs_start_time = jobs_start_time.replace(year = now.year, month = now.month, day = now.day) + jobs_start_time = date_time.as_if_timezone(jobs_start_time, timezone = date_time.TIMEZONE_UTC) + jobs_start_time = date_time.to_timezone(jobs_start_time, timezone = system_tz) + # --- + jobs_end_time = datetime.datetime.strptime(script_args.end_time, "%H:%M:%S") + jobs_end_time = jobs_end_time.replace(year = now.year, month = now.month, day = now.day) + jobs_end_time = date_time.as_if_timezone(jobs_end_time, timezone = date_time.TIMEZONE_UTC) + jobs_end_time = date_time.to_timezone(jobs_end_time, timezone = system_tz) + + # Create all the timestamps at which the job(s) must be done: + all_jobs_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 + ts = jobs_start_time + datetime.timedelta(seconds = offset_seconds) + if ts > jobs_end_time: break + all_jobs_ts.append(date_time.to_timezone(ts, timezone = system_tz)) + offset_seconds += script_args.interval # Add the jobs: - for ts in all_job_ts: schedule_manager.daily(ts, run_once) - printer(len(all_job_ts)) + for ts in all_jobs_ts: schedule_manager.daily(ts.time(), run_once) + no_context_printer(len(all_jobs_ts)) # Infinite loop to keep doing the tasks: - printer("Schedule-manager ready.") + no_context_printer("Schedule-manager ready.") while True: await asyncio.sleep(3_600) @@ -544,7 +540,8 @@ if __name__ == "__main__": # Get the config. from the command-line: parser = argparse.ArgumentParser( description = ( - "To periodically infer EoD data from tick-by-tick data and feed it into the SQL database." + "To periodically infer the current day's latest 1D OHLC data from tick-by-tick data and feed it into the " + "SQL database." ) ) parser.add_argument( @@ -555,13 +552,15 @@ if __name__ == "__main__": ) parser.add_argument( "--start-time", + dest = "start_time", type = str, - help = "The 24-hr time of the day (in 'HH:MM:SS' format) from which the data can be refreshed." + help = "The UTC 24-hr time of the day (in 'HH:MM:SS' format) from which the data can be refreshed." ) parser.add_argument( "--end-time", + dest = "end_time", type = str, - help = "The 24-hr time of the day (in 'HH:MM:SS' format) till which the data must be refreshed." + help = "The UTC 24-hr time of the day (in 'HH:MM:SS' format) till which the data must be refreshed." ) parser.add_argument( "--interval", @@ -588,14 +587,14 @@ if __name__ == "__main__": # Initialize and run the main code: if await init( script_id = args.script_id, - debug = args.debug - ): await main( - start_time = start_time, - end_time = end_time, - interval_seconds = args.interval, + debug = False + ): await main(args) + else: await send_telegram( + message = "Failed to initialize DB connectivity for 1D OHLC from Ticks!", + message_type = "error" ) - # Disconnect from the database: + # Disconnect from the database(s): disconnected = await sql_writer.disconnect() # disconnected = await data_mongo.disconnect() diff --git a/models/api/common/otp/timed_otp.py b/models/api/common/otp/timed_otp.py index 9a8885f..3a26761 100644 --- a/models/api/common/otp/timed_otp.py +++ b/models/api/common/otp/timed_otp.py @@ -189,11 +189,26 @@ class VerifyTimedOTPRequestData(BaseModel): id: str | Dict | List otp: str + isSignup: bool + username: str password: str email: EmailStr phoneNo: str + idClient: int | None = Field(default = None) + clientName: str | None = Field(default = None) + address: str | None = Field(default = None) + city: str | None = Field(default = None) + pincode: str | None = Field(default = None) + country: str | None = Field(default = None) + panCard: str | None = Field(default = None) + gst: str | None = Field(default = None) + entity: str | None = Field(default = None) + startDate: str | None = Field(default = None) + period: str | None = Field(default = None) + amount: float | int | None = Field(default = None) + class Config: extra = "forbid" diff --git a/models/finstitutions/trading/ticks.py b/models/finstitutions/trading/ticks.py index 56dd884..aa6a4ac 100644 --- a/models/finstitutions/trading/ticks.py +++ b/models/finstitutions/trading/ticks.py @@ -388,7 +388,7 @@ class TradingTick(BaseModel): "vwap": self.vwap, "totVol": self.totVol, "rcvdTs": self.rcvdTs.timestamp(), - "tradeTs": self.tradeTs.timestamp(), + "tradeTs": None if self.tradeTs is None else self.tradeTs.timestamp(), "tradeTz": self.tradeTz, "exchgTs": self.exchgTs.timestamp(), "exchgTz": self.exchgTz diff --git a/to_git.sh b/to_git.sh index 5c57f26..a28ea7c 100644 --- a/to_git.sh +++ b/to_git.sh @@ -18,7 +18,7 @@ echo "Files added." # Make the commit: git commit -m "$COMMENT" -echo "Commit done." +echo "Commit done."get # Push th commit to git: git push -u https://wtt.ditscentre.in/ditscentre/api_utils_converse_v2.git "$BRANCH"