[credit scoring] Credit Score Script #credit-score #csv #json

get_credit_score.py

import requests
from requests.auth import HTTPBasicAuth
import json
import csv
import logging
import sys
from logging.handlers import TimedRotatingFileHandler
FORMATTER = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
LOG_FILE = 'my_app.log'

def get_console_handler():
   console_handler = logging.StreamHandler(sys.stdout)
   console_handler.setFormatter(FORMATTER)
   return console_handler
def get_file_handler():
   file_handler = TimedRotatingFileHandler(LOG_FILE, when='h')
   file_handler.setFormatter(FORMATTER)
   return file_handler
def get_logger(logger_name):
   logger = logging.getLogger(logger_name)
   logger.setLevel(logging.DEBUG) # better to have too much log than not enough
   logger.addHandler(get_console_handler())
   logger.addHandler(get_file_handler())
   # with this pattern, it's rarely necessary to propagate the error up to parent
   logger.propagate = False
   return logger

my_logger = get_logger("__name__")

token = ''

#logging.basicConfig(filename='app.log', filemode='w', format='%(name)s - %(levelname)s - %(message)s', level=logging.DEBUG)

def get_auth_token(token):
  endpoint='https://test-gateway.tulaa.io/uaa-server/oauth/token'
  
  payload = {
    'username':'<my-email>',
    'password':'<my-password>',
    'grant_type':'password'
  } 
  
  response = requests.post(endpoint,auth=HTTPBasicAuth('tulaa_web_client','<password>'),data=payload)
  
  if(response.ok):
    
    jData = json.loads(response.content)
    
    for key in jData:
      if key == 'access_token' in jData:
        token='Bearer'+' '+str(jData[key])
        # print(token)
        
        #logging.debug('debug')
        my_logger.debug('a debug message')
        
      return token
  else:
    my_logger.debug('a debug message')
    raise Exception('Cannot fetch access token:{}'.format(response.status_code)) 
  
  
def get_user_score(token):
  
  token = get_auth_token(token)
  
  headers ={
    'Authorization':token    
  }
  
  csvFile1 = 'idNumbers.csv'
  csvFile2 = 'users-with-crb-score.csv'
  csvFile3 = 'farmers-with-crb-score.csv'
  csvFile4 = 'farmers_identification.csv'
  
  with open(csvFile4,'r') as readFile:
    
    reader = csv.reader(readFile,delimiter=',')
    
    next(reader) # skip header
    
    for _ in range(914): # skip the first 194 rows
      next(reader)
    
    for row in reader:
      params = {
        'identityNumber':row[4],
        'loanAmount':500,
        'reason':1
        }      
     
      endpoint = 'https://test-cogency-service.tulaa.io/credit-info?'
      
      response = requests.get(endpoint,headers=headers,params=params)
      
      if(response.ok):
        
        jsonObjectInfo = json.loads(response.content)
        
        # header = ['metropol_credit_score','id_number']
        
        with open(csvFile3,'a',newline='') as writeFile:
          
          filewriter = csv.writer(writeFile,delimiter=',')      
            
          credit_score = jsonObjectInfo['data']['credit_score']
          id_number = jsonObjectInfo['data']['identity_number']
            
          eachJsonObject = [credit_score,id_number]
             
          print(eachJsonObject)      
          # filewriter.writerow(i for i in header)          
          filewriter.writerow(eachJsonObject)
          
          #logging.debug('debug') 
          my_logger.debug('a debug message')                 
      
        writeFile.close()
          
    else:
      # logging.debug('debug')
      my_logger.debug('a debug message') 
      response.raise_for_status() 
      
  readFile.close()             
  
if __name__ == "__main__":
  # get_auth_token(token)
  get_user_score(token)