kiteext_pro_open.py

## WARNING: This is not meant to be used, this is just for demonstration purpose. Get official kite API access from Zerodha.
## USAGE:
# from kiteext_pro_open import KiteExt
# kite = KiteExt()
# kite.login_with_credentials(user_id, password, twofa_secret)
# print(kite.profile())
# print(kite.ltp(['NSE:TCS']))
# print(kite.quote(['NSE:TCS']))

# known issue: history api not working.

import pyotp
import requests
from urllib.parse import urlparse, parse_qs

from kiteconnect import KiteConnect

endpoints = {
    'app_login': 'https://kite.zerodha.com/connect/login',
    'login': 'https://kite.zerodha.com/api/login',
    'twofa': 'https://kite.zerodha.com/api/twofa',
    'app_authorize': 'https://kite.zerodha.com/connect/finish',
    'streak_login': 'https://api-op.streak.tech/login/',
    'streak_session': 'https://api-op.streak.tech/get_session_status/'
}
streak_api_key = '2b0bx36bnep2vslf'

class KiteExt(KiteConnect):
    def __init__(self):
        super().__init__(api_key=streak_api_key)

    def login_with_credentials(self, user_id, password, twofa_secret):
        s = requests.session()
        s.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36', 
            'Accept-Encoding': 'gzip, deflate', 'Accept': '*/*', 'Connection': 'keep-alive', 'Referer': 'https://www.streak.tech/'
        }
        r1 = s.get(endpoints['app_login'], params={'api_key': streak_api_key})
        sess_id = parse_qs(urlparse(r1.url).query)['sess_id']
        r2 = s.post(
            endpoints['login'],
            data = {'user_id':user_id, 'password':password},
            allow_redirects=True
        )
        request_id = r2.json()['data']['request_id']
        r3 = s.post(
            endpoints['twofa'],
            data={'user_id':user_id, 'request_id':request_id, 'twofa_value':pyotp.TOTP(twofa_secret).now(), 'skip_session':True},
            allow_redirects=True
        )
        r4 = s.get(
            endpoints['app_login'],
            params={'api_key':streak_api_key, 'sess_id':sess_id, 'skip_session':True},
            allow_redirects=True
        )
        request_token = parse_qs(urlparse(r4.url).query)['request_token'][0]
        r5 = s.post(
            endpoints['streak_login'],
            data = {'action':'login', 'type':'login', 'status':'success','request_token':request_token}
        )
        csrf = r5.json()['csrf']
        sessionid = r5.json()['sessionid']
        s.headers.update({'Authorization': f"csrftoken={csrf};csrfmiddlewaretoken={csrf};sessionid={sessionid}"})
        r6 = s.get(endpoints['streak_session'])
        access_token = r6.json()['access_token']
        self.set_access_token(access_token)
        return

@mr-easy Where did you get streak_api_key ?