Notes

import requests
import os
import pandas as pd
from Config.common_data import prod_constants
from Utils.login import login_update_tokens

def make_api_call(url, post_body, headers):
r = requests.post(url, json=post_body, headers=headers)
print(f"API call to {url} -> {r.status_code}")
return r.json()

def get_client_summary():
# Load authorization data
with open("..DataHeadersAuthorization.txt", "r") as f:
auth = f.read().strip()
with open("..DataHeadersx-sp-authorization-jwt.txt", "r") as f:
jwt = f.read().strip()

headers = {
"Authorization": auth,
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/113.0.0.0 Safari/537.36",
"Content-Type": "application/json",
"Accept": "application/json",
"x-sp-authorization-jwt": jwt,
"Origin": "https://stage-cpa.taxcaddy.com",
"Referer": "https://stage-cpa.taxcaddy.com/",
"Accept-Language": "en-US,en;q=0.9"
}

# Initialize data collection dictionary
data_collected = {}

# Define API endpoints and post bodies
endpoints = [
{
"url": prod_constants['url']['client_summary_url'],
"post_body": {
"taxYear": 2022,
"templateID": 1,
"specificID": 9301,
"pagination": {"pageSize": 10000, "pageNumber": 1},
"sort": {"columnName": "TaxPayerFirstName", "sortDirection": 1},
"filters": [{"filterID": 22, "filterName": "Search Text", "filterValues": [{"text": ""}]}]
}
},
{
"url": 'https://api.maddy.com/api/v1/clientmanagement/071/client/list',
"post_body": {
"taxYear": 2022,
"templateID": 3,
"specificID": 7314,
"pagination": {"pageSize": 10000, "pageNumber": 1},
"sort": {"columnName": "TaxPayerFirstName", "sortDirection": 1},
"filters": [{"filterID": 22, "filterName": "Search Text", "filterValues": [{"text": ""}]}]
}
}
]

# Iterate over API endpoints
for endpoint in endpoints:
url = endpoint['url']
post_body = endpoint['post_body']

nextPageexists = True
pageNo = 1

# Loop for pagination
while nextPageexists:
# Update page number in post body
post_body['pagination']['pageNumber'] = pageNo

# Make API call
response = make_api_call(url, post_body, headers)

# Process response
if "error" in response or "Message" in response:
if "error" in response:
print(response["error"])
if "Message" in response:
print(response["Message"])
if "Authorization has been denied" in response.get("error", "") or "Authorization has been denied" in response.get("Message", ""):
login_update_tokens()
exit()
nextPageexists = False
else:
if "response" in response:
response = response["response"]
clientCount = response.get("clientCount", 0)
print(f"Total : {clientCount}")

listDTOClientDetails = response.get("listDTOClientDetails", [])
for listDTOClientDetail in listDTOClientDetails:
id = listDTOClientDetail.get("id", "")
userId = listDTOClientDetail.get("userId", "")
status = listDTOClientDetail.get("status", "")
taxPayerEmail = listDTOClientDetail.get("taxPayerEmail", "")
isCheckbox_enabled = listDTOClientDetail.get("isCheckbox_enabled", "")
taxPayerFirstName = listDTOClientDetail.get("taxPayerFirstName", "")
taxPayerLastName = listDTOClientDetail.get("taxPayerLastName", "")
ownerFirstName = listDTOClientDetail.get("ownerFirstName", "")
data_collected[id] = {
"userId": userId,
"taxPayerEmail": taxPayerEmail,
"taxPayerFirstName": taxPayerFirstName,
"taxPayerLastName": taxPayerLastName,
"ownerFirstName": ownerFirstName,
"invite-status": status,
"invite-isCheckbox_enabled": isCheckbox_enabled
}

# Move to next page if it exists
nextPageexists = clientCount >= post_body['pagination']['pageSize']
if nextPageexists:
pageNo += 1

# Create DataFrame from collected data
df = pd.DataFrame.from_dict(data_collected, orient='index')

# Write DataFrame to Excel file
client_id_excel_path = prod_constants['file_paths']['client_overview_output_file_path']
df.to_excel(client_id_excel_path, index=True)

# Call the function
get_client_summary()