nsedt.utils

  1"""
  2utils for nsedt
  3"""
  4
  5import io
  6import json
  7import datetime
  8import zipfile
  9
 10from io import BytesIO
 11
 12from datetime import datetime, date
 13from warnings import warn
 14
 15import requests
 16
 17import pandas as pd
 18
 19from fake_http_header import FakeHttpHeader
 20from nsedt.resources import constants as cns
 21
 22
 23
 24def format_df(df):
 25    """
 26    Arg:\n
 27        - df: pandas df
 28    Reuturn:
 29        - formatted df column 
 30    """
 31    df.columns = df.columns.str.lower().str.replace(' ','_').str.replace('\t','')
 32    return df
 33
 34def format_date(input_string: str, date_format: str):
 35    """
 36    Args:\n
 37        - input_string : str date format for a format to check
 38        - format : type of string to format
 39    Returns:\n
 40        - str: date format in input string
 41    """
 42    try:
 43        return datetime.strptime(input_string, "%d-%m-%Y").strftime(date_format)
 44    except ValueError:
 45        return None
 46
 47
 48def get_headers():
 49    """
 50    Args:
 51        ---\n
 52    Returns:\n
 53        Json: json containing nse header
 54    """
 55
 56    return FakeHttpHeader().as_header_dict()
 57
 58
 59def get_cookies():
 60    """
 61    Args:
 62        ---
 63
 64    Returns:
 65
 66        Json: json containing nse cookies
 67    """
 68
 69    response = requests.get(cns.BASE_URL, timeout=30, headers=get_headers())
 70    if response.status_code != 200:
 71        raise ValueError("Retry again in a minute.")
 72    return response.cookies.get_dict()
 73
 74
 75def fetch_url(url, cookies, key=None, response_type="panda_df"):
 76    """
 77    Args:
 78
 79        url (str): URL to fetch
 80        cookies (str): NSE cookies
 81        key (str, Optional):
 82
 83    Returns:
 84
 85        Pandas DataFrame: df containing url data
 86
 87    """
 88
 89    response = requests.get(
 90        url=url,
 91        timeout=30,
 92        headers=get_headers(),
 93        cookies=cookies,
 94    )
 95
 96    if response.status_code == 200:
 97        json_response = json.loads(response.content)
 98
 99        if response_type != "panda_df":
100            return json_response
101        if key is None:
102            return pd.DataFrame.from_dict(json_response)
103
104        return pd.DataFrame.from_dict(json_response[key])
105
106    raise ValueError("Please try again in a minute.")
107
108
109def get_symbol(symbol: str, get_key: str) -> str:
110    """_summary_
111
112    Args:
113        symbol (str): _description_
114        get_key (str): _description_
115
116    Returns:
117        str: _description_
118    """
119
120    symbol_map = cns.SYMBOL_MAP
121    val = None
122    for item in symbol_map:
123        key_list = item["keys"]
124        if symbol in key_list:
125            val = item[get_key]
126
127    return val or symbol
128
129
130def check_nd_convert(start_date: str, end_date: str) -> datetime:
131    """
132    The function `check_nd_convert` takes two date strings in the format "%d-%m-%Y" and 
133    converts them to datetime objects if they are not already in that format.
134
135    :param start_date: The `start_date` parameter is the starting date of a period,
136     specified as a string in the format "%d-%m-%Y"
137    :type start_date: str
138    :param end_date: The `end_date` parameter is a string representing the end date in the format
139    "%d-%m-%Y"
140    :type end_date: str
141    :return: the start_date and end_date as datetime objects.
142    """
143
144    if isinstance(start_date, date) and isinstance(end_date, date):
145        warn(
146            """Passing start_date, end_date in date is deprecated
147now pass in str '%d-%m-%Y' format""",
148            DeprecationWarning,
149            stacklevel=2,
150        )
151
152    elif isinstance(start_date, str) and isinstance(end_date, str):
153        start_date = datetime.strptime(start_date, "%d-%m-%Y")
154        end_date = datetime.strptime(end_date, "%d-%m-%Y")
155
156    else:
157        raise ValueError("Input is of an unknown type")
158
159    return start_date, end_date
160
161
162def fetch_csv(url, cookies, response_type="panda_df", skip_rows=None):
163    """
164    Args:
165
166        url (str): URL to fetch
167        cookies (str): NSE cookies
168        key (str, Optional):
169
170    Returns:
171
172        Pandas DataFrame: df generated from csv
173        OR
174        Json: json output of the csv
175        OR
176        String: raw content for files where it cannot be processed into Json or 
177                Pandas df
178
179    """
180
181    response = requests.get(
182        url=url, timeout=30, headers=get_headers(), cookies=cookies )
183    if response.status_code == 200:
184        if response_type == "raw":
185            return response.content
186        csv_content = response.content.decode('utf-8')
187        df = pd.read_csv(io.StringIO(csv_content), skiprows=skip_rows)
188        df = format_df(df)
189        return df.to_json(orient='records') if response_type == "json" else df
190    raise ValueError("Please try again in a minute.")
191
192
193def fetch_zip(url, cookies, file_name, response_type="panda_df", skip_rows=None):
194    """
195    Args:
196
197        url (str): URL to fetch
198        cookies (str): NSE cookies
199        key (str, Optional):
200    Returns:
201
202        Pandas DataFrame: df generated from csv
203        OR
204        Json: json output of the csv
205        OR
206        Pandas DF:  Pandas df of the csv file
207    """
208
209    if not file_name:
210        raise ValueError("Please give file name to return")
211
212    response = requests.get(
213        url=url, timeout=30, headers=get_headers(), cookies=cookies )
214    if response.status_code == 200:
215        zip_content = BytesIO(response.content)
216        # Open the zip file in memory
217        with zipfile.ZipFile(zip_content, 'r') as zip_ref:
218            # Retrieve the list of file names in the zip file
219            try:
220                csv_content = zip_ref.read(file_name)
221            except Exception as e:
222                raise ValueError("File not found in the zip folder.") from e
223
224            df = pd.read_csv(BytesIO(csv_content), skiprows=skip_rows)
225            df = format_df(df)
226            return df.to_json(orient='records') if response_type == "json" else df
227    raise ValueError("File might not be available this time or check your params")