Login script for id.kyivcity.gov.ua

kyivcityapi.py

# -*- coding: utf-8 -*-
#!/usr/bin/env python

import requests
from lxml.html import fromstring
import time
import pickle
import re
import os

LOGIN_URL = 'https://id.kyivcity.gov.ua/login/phone'
HACK_URL  = 'https://my.kyivcity.gov.ua/api/openid/login?callback_url=https://my.kyivcity.gov.ua/auth/callback'
TOKEN_URL = 'https://my.kyivcity.gov.ua/api/tokens?code={}&state={}&callback_url=https://my.kyivcity.gov.ua/auth/callback'
CARDS_URL = 'https://my.kyivcity.gov.ua/api/me/travel-cards'
HISTORY_URL = 'https://my.kyivcity.gov.ua/api/travel-cards/{}/product-uses?page[size]={}&page[number]={}'
CSRF_URL = 'https://id.kyivcity.gov.ua/csrf'
PAYS_URL = 'https://my.kyivcity.gov.ua/api/travel-cards/{}/payments/history?page[size]={}&page[number]={}'

headers_login_page = {
         'Referer': 'https://my.kyivcity.gov.ua/',
         'Accept':'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
         'User-Agent':'Mozilla/5.0 (X11; Linux x86_64; rv:60.0) Gecko/20100101 Firefox/60.0',
         'Connection': 'keep-alive',
         'Upgrade-Insecure-Requests': '1',
         'Pragma': 'no-cache',
         'Cache-Control': 'no-cache',
         'Accept-Language': 'en-US,en;q=0.5',
         'Accept-Encoding': 'gzip, deflate, br',
         'Connection': 'keep-alive'
        }

headers_login_action = {
         'Referer': 'https://id.kyivcity.gov.ua/ui/login?login_opts=facebook%2Ceds%2Clinkedin%2Cphone%2Cemail%2Cgoogle%2Cpbbankid%2Cnbubankid',
         'Accept':'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
         'User-Agent':'Mozilla/5.0 (X11; Linux x86_64; rv:60.0) Gecko/20100101 Firefox/60.0',
         'Content-Type':'application/x-www-form-urlencoded',
         'Connection': 'keep-alive',
         'Upgrade-Insecure-Requests': '1',
         'Pragma': 'no-cache',
         'Cache-Control': 'no-cache',
         'Origin': 'https://id.kyivcity.gov.ua',
         'Accept-Language': 'en-US,en;q=0.5',
         'Accept-Encoding': 'gzip, deflate, br',
         'Connection': 'keep-alive'
        }

class KyivCityAPI(object):
    """docstring for KyivCityAPI"""
    def __init__(self, login, password, retry_count = 3):
        super(KyivCityAPI, self).__init__()

        self._login = login
        self._password = password
        self._session = requests.session()
        self._token = ''
        self._openid_auth = {}

        self._retry_count = retry_count

        self.history_total = 0

    def login_is_exist(self):
        return os.path.isfile('./login.p')

    def save_login(self):
        backup_data = {'token':self._token, 'session':self._session}
        pickle.dump( backup_data, open("./login.p", "wb"))
        print('Save login data')

    def load_login(self):
        print('Login exist. Load login data')
        backup_data = pickle.load(open("./login.p", "rb"))
        self._token = backup_data['token']
        #print type(self._token)
        if isinstance(self._token, str):
                self._session = backup_data['session']
        else:
            self.login(ignore=True)

    def login(self, ignore=False):
        if self.login_is_exist() and not ignore:
            self.load_login()
            return

        print('Login process')
        # Get login page
        try:
            r = self._session.get(HACK_URL, headers=headers_login_page)
        except:
            raise

        #root = fromstring(r.content)
        input_fields = {}
        #try:
        #    for field in root.cssselect("input"):
        #        input_fields[field.attrib['id']] = field.attrib['value']
        #except:
        #    raise
        #print input_filds

        try:
            r = self._session.get(CSRF_URL, headers=headers_login_page)
        except:
            raise
        print(r.content)
        input_fields['csrf_date'] = r.json()['token']
        # Post login data
        login_data = {'login':self._login,
                      'password':self._password,
                      '_csrf':input_fields['csrf_date']}
        print(login_data)
        try:
            r = self._session.post(LOGIN_URL, data = login_data, headers=headers_login_action)
        except:
            raise
        print(r.headers, r.url, r.content)
        #Make hack request for getting state and code
        #try:
        #    r = self._session.get(HACK_URL, headers=headers_login_action)
        #except:
        #    raise
        #print r.headers, r.url, r.content
        try:
            m = re.search('callback\?(\w*)=(.*)&(\w*)=(.*)', r.url)
            if hasattr(m, 'lastindex'):
                self._openid_auth[m.group(1)] = m.group(2)
                self._openid_auth[m.group(3)] = m.group(4)
            else:
                raise Exception('Can\'t get login code and state')
        except:
            raise Exception('Can\'t get login code and state')

        #Get token
        try:
            r = self._session.get(
                TOKEN_URL.format(
                    self._openid_auth['code'], 
                    self._openid_auth['state']), 
                    headers=headers_login_action
                )
        except:
            raise

        try:
            self._token = r.json()
        except:
            raise Exception('Can\'t parse token data')  
        print('Save login data_')
        self.save_login()

    def autorization(self): 
        if isinstance(self._token, dict):
            print(self._token)
            self.login(ignore=True)
        return 'Bearer ' + self._token

    def _api_call(self, url, type = list, retry = 0):
        try:
            r = self._session.get(url, headers={'Authorization': self.autorization(), 'x-total-result-count-include':   
'true'})
            data = r.json()
            if 'X-Total-Result-Count' in r.headers:
                self.history_total = int(r.headers['X-Total-Result-Count'])

            if isinstance(data, type):
                return data
            else:
                if retry < self._retry_count:
                    print('Retry login ', retry)
                    self.login(ignore=True)
                    return self._api_call(url, type, retry+1)
                else:
                    print('Retry login failed')
                    raise Exception('Can\'t make api call') 
        except:
            raise    

    def get_cards(self):
        return self._api_call(CARDS_URL)

    def get_history(self, card, size = 1, page = 1):
        return self._api_call(HISTORY_URL.format(card, size, page))

    def get_pays(self, card, size = 1, page = 1):
        return self._api_call(PAYS_URL.format(card, size, page))

if __name__ == "__main__":
    import sys
    arg = sys.argv
    if len(arg) < 2:
        raise Exception('Usage: python file.py login pass')
    api = KyivCityAPI(arg[1], arg[2])
    api.login()
    cards = api.get_cards()
    STEP = 20
    d = api.get_history(cards[0]['id'], size = STEP)
    page = 1

    while page*STEP < api.history_total:
        page+=1
        d += api.get_history(cards[0]['id'], size = STEP , page = page)
        print(page, page*STEP, api.history_total)
    travels = [a['travelType'] for a in d]
    occurence = {t:travels.count(t) for t in set(travels)}
    out = sorted(list(occurence.items()), key = 
             lambda kv:(kv[1], kv[0]), reverse=True)
    format_line = '*Всего поездок:* ' + str(api.history_total)
    for o in out:
        print(float(o[1])/api.history_total*100)
        #format_line += '*' + o[0] + '*:' + str(o[1])/api.history_total) + '%\n'
    print(format_line)

requirements.txt

requests
lxml