""" AUTHOR: Khushal P Soonderji DATE: Thursday, 28th Nov., 2024 OBJECTIVE: To provide a base class for common behaviour of NSE's APIs. REFERENCES: N/A DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys sys.path.append(".") sys.path.append("..") # System-level activities: import io # My utils: from utils_v2.date_time import date_time from utils_v2.nse.models.api_call import NSEApiResponse # To make API calls: import httpx # To work with date and time: import datetime # For working with datatypes: from typing import Literal, List # For debugging: from icecream import IceCreamDebugger import inspect # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** class AsyncNSEBase: def __init__( self, base_url: str, data_url: str, http_client: httpx.AsyncClient, cookies_refresh_interval: int | float = 300, debug = True, debug_prefix = "NSE | ", debug_only_errors = True ): """ To be used as a base class for all sorts of NSE scraping. :param base_url: The base URL (that you open in your browser) for your targeted module. :param data_url: The internal URL (that NSE's script calls by itself) for your targeted module. :param http_client: An HTTP client to use to make API calls. :param cookies_refresh_interval: The no. of seconds after which the cookies should get refreshed. :param debug: Whether, or not, you want to show debugging messages. :param debug_prefix: The prefix string to use to recognize the module that is printing the debugging text. :param debug_only_errors: Whether you want to show all debugging messages, or just error messages. """ # Prepare the debugging utility: self._debug_prefix = debug_prefix self._printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True) if not debug: self._printer.disable() self._debug_only_errors = debug_only_errors # Accept the input configuration: self._base_url = base_url self._data_url = data_url self._http_client = http_client # Make some provisions for later: self._cookies = None self._cookies_refresh_interval = cookies_refresh_interval self._last_cookies_refresh = datetime.datetime.fromtimestamp(0, tz = date_time.TIMEZONE_UTC) def enable_debug(self): self._printer.enable() def disable_debug(self): self._printer.disable() def debug_only_errors(self): self._debug_only_errors = True def debug_everything(self): self._debug_only_errors = False # ┏┓ ┓ • ┳┳┓ ┓ • # ┃ ┏┓┏┓┃┏┓┏┓┏ ┃┃┃┏┓┃┏┓┏┓┏┓ # ┗┛┗┛┗┛┛┗┗┗ ┛ ┛ ┗┗┻┛┗┗┛┗┗┫ # ┛ @property def seconds_since_cookies_refreshed(self): return (date_time.get_current_utc_date_time() - self._last_cookies_refresh).total_seconds() async def refresh_cookies(self) -> bool: """ To get cookies for the given base URL. NSE has some sort of strict cookies and origin policy that is beyond my current understanding. But I have noticed that if you furnish the right base URL and then use that, NSE will respond just fine. USAGE: Call this right after you initialize your class and then call it every so often after that. :return: The cookies in the 'data' field of the """ # Start by assuming failure: success = False # Make the API call: api_response = await self.get( url = self._base_url, headers = { "Accept": "*/*", "Accept-Encoding": "gzip, deflate, br", "Accept-Language": "en-US,en;q=0.9", "Connection": "keep-alive", "Upgrade-Insecure-Requests": "1", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:120.0) Gecko/20100101 Firefox/120.0" }, refresh_cookies = False ) # If the API call succeeds: if api_response.httpCode in [200]: self._last_cookies_refresh = date_time.get_current_utc_date_time(as_string = False) self._cookies = api_response.response.cookies # If it failed, alert on the terminal: else: self._printer("Cookies Refresh FAILED!") # Done here: return success # ┏┓┏┓┳ ┏┓ ┓┓• # ┣┫┃┃┃ ┃ ┏┓┃┃┓┏┓┏┓ # ┛┗┣┛┻ ┗┛┗┻┗┗┗┛┗┗┫ # ┛ async def get( self, url: str = None, headers: dict = None, params: dict = None, cookies: httpx.Cookies | dict = None, refresh_cookies: bool = True ) -> NSEApiResponse: """ To call an API using the GET method. :param url: A custom URL to hit. If not specified, the data URL will be hit. :param headers: Custom headers to use. If not specified, default values will be used. :param params: The params to send in the query string itself. :param cookies: Custom cookies to send, else the ones fetched from the base URL will be used. :param refresh_cookies: Set this to False is you would like to block the auto refreshing of cookies. Remember that the refreshing is controlled through the interval specified in the constructor. :return: A structured response that includes the raw response, the exception (if any), and so on. """ # Select and refresh the cookies as needed: if cookies is None and refresh_cookies: if self.seconds_since_cookies_refreshed > self._cookies_refresh_interval: await self.refresh_cookies() use_cookies = self._cookies else: use_cookies = cookies # Prepare the structure of the response: api_response = NSEApiResponse( action = inspect.stack()[1].function, url = url or self._data_url, method = "GET" ) try: # Make the API call: response = await self._http_client.get( url = url or self._data_url, headers = headers or { "Accept": "*/*", "Accept-Encoding": "gzip, deflate, br", "Accept-Language": "en-US,en;q=0.9", "Connection": "keep-alive", "Host": "www.nseindia.com", "Referer": self._base_url, "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-origin", "Sec-GPC": "1", "Upgrade-Insecure-Requests": "1", "TE": "trailers", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:120.0) Gecko/20100101 Firefox/120.0" }, params = params, cookies = use_cookies ) # Note down the results: api_response.response = response api_response.httpCode = response.status_code api_response.message = response.reason_phrase # If something goes wrong: except Exception as exception: api_response.exception = exception api_response.message = str(exception) self._printer(exception, api_response.url, api_response.method, headers, params) # Done here: return api_response # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass