Merge commit '8bf92c04ed0e9fe72699763a3a65f4a27d93a4b5' as 'utils_v2'
This commit is contained in:
@@ -0,0 +1,281 @@
|
||||
"""
|
||||
|
||||
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.data.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
|
||||
Reference in New Issue
Block a user