mirror of
https://github.com/x1xhlol/system-prompts-and-models-of-ai-tools.git
synced 2026-08-16 21:24:09 +00:00
477 lines
17 KiB
Python
477 lines
17 KiB
Python
import requests
|
|
import os
|
|
import json
|
|
|
|
|
|
def get_token():
|
|
user_id = os.getenv('USER_ID')
|
|
password = os.getenv('PASSWORD')
|
|
api_endpoint = os.getenv('API_ENDPOINT')
|
|
# print(f"API end point: {api_endpoint}")
|
|
|
|
url = f'{api_endpoint}/api/v1/token'
|
|
credentials = {
|
|
"email": user_id,
|
|
"password": password
|
|
}
|
|
headers = {
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
try:
|
|
response = requests.post(url, json=credentials, headers=headers)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if 'token' in data:
|
|
token = data['token']
|
|
os.environ["AUTH_TOKEN"] = token
|
|
return token
|
|
else:
|
|
return "Token not found in response."
|
|
else:
|
|
return f"Failed to get token. Status code: {response.status_code}, Error: {response.text}"
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
return f"An error occurred: {e}"
|
|
|
|
|
|
def get_role_id(entity_id):
|
|
api_endpoint = os.getenv('API_ENDPOINT')
|
|
auth_token = os.getenv('AUTH_TOKEN')
|
|
|
|
url = f'{api_endpoint}/api/v1/role_details?entityId={entity_id}'
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'Authorization': auth_token,
|
|
}
|
|
|
|
try:
|
|
print(f"*** Calling get role id... for entity {entity_id}")
|
|
response = requests.get(url, headers=headers)
|
|
response_data = response.json()
|
|
if response.status_code == 200 or response.status_code == 201:
|
|
role_id = response_data["data"]["_id"]
|
|
print(f"Roleid set to {role_id} in the environment.")
|
|
return {"data": role_id, "status_code": 200}
|
|
else:
|
|
print(f"Failed to get data. Status code: {response.status_code}")
|
|
print(f"Error: {response.text}")
|
|
return {"error": f"Failed to get data. Status code: {response.status_code}",
|
|
"status_code": response.status_code, "details": response.text}
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
return {"error": f"An error occurred: {e}", "status_code": 500}
|
|
|
|
|
|
def get_onboarding_status(onboarding_id):
|
|
api_endpoint = os.getenv('API_ENDPOINT')
|
|
auth_token = os.getenv('AUTH_TOKEN')
|
|
# role_id = os.getenv('ROLE_ID')
|
|
|
|
# url = f'{api_endpoint}/api/v1/entity/onboarding_status?id={onboarding_id}'
|
|
# headers = {
|
|
# 'Content-Type': 'application/json',
|
|
# 'Authorization': auth_token,
|
|
# # 'X-Entity-Role': role_id
|
|
# }
|
|
url = f'{api_endpoint}/api/v1/entity/onboarding_status?id={onboarding_id}'
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'Authorization': auth_token,
|
|
# 'X-Entity-Role': role_id
|
|
}
|
|
|
|
try:
|
|
response = requests.get(url, headers=headers)
|
|
|
|
if response.status_code == 200:
|
|
# print(response.json())
|
|
return {"data": response.json(), "status_code": 200}
|
|
else:
|
|
print(f"Failed to get data. Status code: {response.status_code}")
|
|
print(f"Error: {response.text}")
|
|
return {"error": f"Failed to get data. Status code: {response.status_code}",
|
|
"status_code": response.status_code, "details": response.text}
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
return {"error": f"An error occurred: {e}", "status_code": 500}
|
|
|
|
|
|
def update_onboarding_status(step_number, status, error_message, summary):
|
|
api_endpoint = os.getenv('API_ENDPOINT')
|
|
auth_token = os.getenv('AUTH_TOKEN')
|
|
role_id = os.getenv('ROLE_ID')
|
|
onboarding_id = os.getenv('ONBOARDING_ID')
|
|
print(f"Using onboardinid: {onboarding_id}, roleid:{role_id} to update onBoardingStatus.")
|
|
|
|
url = f"{api_endpoint}/api/v1/entity/onboarding_status?id={onboarding_id}"
|
|
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'Authorization': auth_token,
|
|
'X-Entity-Role': role_id
|
|
}
|
|
|
|
try:
|
|
# -1 to initialize the steps array
|
|
if step_number == -1:
|
|
data = {
|
|
"id": onboarding_id,
|
|
"steps": [
|
|
{"number": 1, "stepName": "Validate uploaded files", "status": "NOT-STARTED", "errorMessage": None,
|
|
"summary": None, "completedAt": None},
|
|
{"number": 2, "stepName": "Load fund details", "status": "NOT-STARTED", "errorMessage": None,
|
|
"summary": None, "completedAt": None},
|
|
{"number": 3, "stepName": "Load partners", "status": "NOT-STARTED", "errorMessage": None,
|
|
"summary": None, "completedAt": None},
|
|
{"number": 4, "stepName": "Load portfolio (financials) information", "status": "NOT-STARTED",
|
|
"errorMessage": None, "summary": None, "completedAt": None},
|
|
{"number": 5, "stepName": "Load bank transations", "status": "NOT-STARTED", "errorMessage": None,
|
|
"summary": None, "completedAt": None},
|
|
{"number": 6, "stepName": "Load journals", "status": "NOT-STARTED", "errorMessage": None,
|
|
"summary": None, "completedAt": None},
|
|
{"number": 7, "stepName": "Load fund performance", "status": "NOT-STARTED", "errorMessage": None,
|
|
"summary": None, "completedAt": None}
|
|
]
|
|
}
|
|
else:
|
|
response = get_onboarding_status(onboarding_id)
|
|
onboarding_status = response["data"]["data"]
|
|
steps = onboarding_status["steps"]
|
|
steps[step_number]["status"] = status
|
|
steps[step_number]["errorMessage"] = error_message
|
|
steps[step_number]["summary"] = summary
|
|
data = {
|
|
"id": onboarding_id,
|
|
"steps": steps
|
|
}
|
|
|
|
response = requests.put(url, headers=headers, data=json.dumps(data))
|
|
|
|
if response.status_code == 200:
|
|
return {"data": response.json(), "status_code": 200}
|
|
else:
|
|
return {"error": f"Failed to update onboarding status. Status code: {response.status_code}",
|
|
"status_code": response.status_code, "details": response.text}
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
return {"error": f"An error occurred: {e}", "status_code": 500}
|
|
|
|
|
|
def get_entity_details(entity_id):
|
|
api_endpoint = os.getenv('API_ENDPOINT')
|
|
auth_token = os.getenv('AUTH_TOKEN')
|
|
url = f'{api_endpoint}/api/v1/entity/entity_details?entityId={entity_id}'
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'Authorization': auth_token
|
|
}
|
|
response = requests.get(url, headers=headers)
|
|
|
|
if response.status_code == 200:
|
|
# print(response.json())
|
|
return response.json()
|
|
else:
|
|
# Print the status code and error message if the request failed
|
|
print(f"Failed to get data. Status code: {response.status_code}")
|
|
print(f"Error: {response.text}")
|
|
|
|
|
|
def add_user(user):
|
|
api_endpoint = os.getenv('API_ENDPOINT')
|
|
auth_token = os.getenv('AUTH_TOKEN')
|
|
# role_id = os.getenv('ROLE_ID')
|
|
|
|
url = f"{api_endpoint}/api/v1/user"
|
|
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'Authorization': auth_token,
|
|
}
|
|
|
|
try:
|
|
response = requests.post(url, headers=headers, data=json.dumps(user))
|
|
|
|
if response.status_code == 200:
|
|
return {"data": response.json(), "status_code": 200}
|
|
else:
|
|
return {"error": f"Failed to add user. Status code: {response.status_code}",
|
|
"status_code": response.status_code, "details": response.text}
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
return {"error": f"An error occurred: {e}", "status_code": 500}
|
|
|
|
|
|
def add_role(role_payload):
|
|
api_endpoint = os.getenv('API_ENDPOINT')
|
|
auth_token = os.getenv('AUTH_TOKEN')
|
|
role_id = os.getenv('ROLE_ID')
|
|
|
|
print(f"Role ID going in to add role API: {role_id}")
|
|
url = f"{api_endpoint}/api/v1/entity/role/partner"
|
|
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'Authorization': auth_token,
|
|
'X-Entity-Role': role_id
|
|
}
|
|
|
|
try:
|
|
response = requests.post(url, headers=headers, data=json.dumps(role_payload))
|
|
|
|
if response.status_code == 200 or response.status_code == 201:
|
|
return {"data": response.json(), "status_code": 200}
|
|
else:
|
|
return {"error": f"Failed to add role/user. Status code: {response.status_code}",
|
|
"status_code": response.status_code, "details": response.text}
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
return {"error": f"An error occurred: {e}", "status_code": 500}
|
|
|
|
|
|
def update_fund(fund_payload):
|
|
api_endpoint = os.getenv('API_ENDPOINT')
|
|
auth_token = os.getenv('AUTH_TOKEN')
|
|
role_id = os.getenv('ROLE_ID')
|
|
|
|
print(f"Update fund use role header: {role_id}")
|
|
url = f"{api_endpoint}/api/v1/entity/fund_info"
|
|
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'Authorization': auth_token,
|
|
'X-Entity-Role': role_id
|
|
}
|
|
|
|
try:
|
|
response = requests.put(url, headers=headers, data=json.dumps(fund_payload))
|
|
|
|
if response.status_code == 200 or response.status_code == 201:
|
|
return {"data": response.json(), "status_code": 200}
|
|
else:
|
|
return {"error": f"Failed to add role/user. Status code: {response.status_code}",
|
|
"status_code": response.status_code, "details": response.text}
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
return {"error": f"An error occurred: {e}", "status_code": 500}
|
|
|
|
|
|
def add_portfolio_company(payload):
|
|
api_endpoint = os.getenv('API_ENDPOINT')
|
|
auth_token = os.getenv('AUTH_TOKEN')
|
|
role_id = os.getenv('ROLE_ID')
|
|
|
|
print(f"Update fund use role header: {role_id}")
|
|
url = f"{api_endpoint}/api/v1/portfolio-company/single"
|
|
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'Authorization': auth_token,
|
|
'X-Entity-Role': role_id
|
|
}
|
|
|
|
try:
|
|
response = requests.post(url, headers=headers, data=json.dumps(payload))
|
|
|
|
if response.status_code == 200:
|
|
return {"data": response.json(), "status_code": 200}
|
|
else:
|
|
return {"error": f"Failed to add portfolio company. Status code: {response.status_code}",
|
|
"status_code": response.status_code, "details": response.text}
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
return {"error": f"An error occurred: {e}", "status_code": 500}
|
|
|
|
|
|
def add_portfolio_company_investment(payload):
|
|
api_endpoint = os.getenv('API_ENDPOINT')
|
|
auth_token = os.getenv('AUTH_TOKEN')
|
|
role_id = os.getenv('ROLE_ID')
|
|
|
|
print(f"Update fund use role header: {role_id}")
|
|
url = f"{api_endpoint}/api/v1/portfolio-company/investment"
|
|
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'Authorization': auth_token,
|
|
'X-Entity-Role': role_id
|
|
}
|
|
|
|
try:
|
|
response = requests.post(url, headers=headers, data=json.dumps(payload))
|
|
|
|
if response.status_code == 200:
|
|
return {"data": response.json(), "status_code": 200}
|
|
else:
|
|
return {"error": f"Failed to add portfolio company investment. Status code: {response.status_code}",
|
|
"status_code": response.status_code, "details": response.text}
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
return {"error": f"An error occurred: {e}", "status_code": 500}
|
|
|
|
|
|
def add_journal(payload):
|
|
api_endpoint = os.getenv('API_ENDPOINT')
|
|
auth_token = os.getenv('AUTH_TOKEN')
|
|
role_id = os.getenv('ROLE_ID')
|
|
|
|
print(f"Update fund use role header: {role_id}")
|
|
url = f"{api_endpoint}/api/v1/bank/journal"
|
|
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'Authorization': auth_token,
|
|
'X-Entity-Role': role_id
|
|
}
|
|
|
|
try:
|
|
response = requests.post(url, headers=headers, data=json.dumps(payload))
|
|
|
|
if response.status_code == 200:
|
|
return {"data": response.json(), "status_code": 200}
|
|
else:
|
|
return {"error": f"Failed to add journal. Status code: {response.status_code}",
|
|
"status_code": response.status_code, "details": response.text}
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
return {"error": f"An error occurred: {e}", "status_code": 500}
|
|
|
|
|
|
def add_journal_ledger(payload):
|
|
print(payload)
|
|
api_endpoint = os.getenv('API_ENDPOINT')
|
|
auth_token = os.getenv('AUTH_TOKEN')
|
|
role_id = os.getenv('ROLE_ID')
|
|
|
|
print(f"Update fund use role header: {role_id}")
|
|
url = f"{api_endpoint}/api/v1/bank/journal-ledger-single"
|
|
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'Authorization': auth_token,
|
|
'X-Entity-Role': role_id
|
|
}
|
|
|
|
try:
|
|
response = requests.post(url, headers=headers, data=json.dumps(payload))
|
|
|
|
if response.status_code == 200:
|
|
return {"data": response.json(), "status_code": 200}
|
|
else:
|
|
return {"error": f"Failed to add journal ledger. Status code: {response.status_code}",
|
|
"status_code": response.status_code, "details": response.text}
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
return {"error": f"An error occurred: {e}", "status_code": 500}
|
|
|
|
|
|
def add_performance_record(payload):
|
|
api_endpoint = os.getenv('API_ENDPOINT')
|
|
auth_token = os.getenv('AUTH_TOKEN')
|
|
role_id = os.getenv('ROLE_ID')
|
|
|
|
print(f"Update fund use role header: {role_id}")
|
|
url = f"{api_endpoint}/api/v1/transaction/name"
|
|
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'Authorization': auth_token,
|
|
'X-Entity-Role': role_id
|
|
}
|
|
|
|
try:
|
|
response = requests.post(url, headers=headers, data=json.dumps(payload))
|
|
|
|
if response.status_code == 200:
|
|
return {"data": response.json(), "status_code": 200}
|
|
else:
|
|
return {"error": f"Failed to add performance. Status code: {response.status_code}",
|
|
"status_code": response.status_code, "details": response.text}
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
return {"error": f"An error occurred: {e}", "status_code": 500}
|
|
|
|
|
|
def add_bank_transaction_records(payload):
|
|
api_endpoint = os.getenv('API_ENDPOINT')
|
|
auth_token = os.getenv('AUTH_TOKEN')
|
|
role_id = os.getenv('ROLE_ID')
|
|
|
|
print(f"Update fund use role header: {role_id}")
|
|
url = f"{api_endpoint}/api/v1/bank"
|
|
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'Authorization': auth_token,
|
|
'X-Entity-Role': role_id
|
|
}
|
|
|
|
try:
|
|
response = requests.post(url, headers=headers, data=json.dumps(payload))
|
|
|
|
if response.status_code == 200:
|
|
return {"data": response.json(), "status_code": 200}
|
|
else:
|
|
return {"error": f"Failed to add bank transaction. Status code: {response.status_code}",
|
|
"status_code": response.status_code, "details": response.text}
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
return {"error": f"An error occurred: {e}", "status_code": 500}
|
|
|
|
|
|
def get_all_role():
|
|
|
|
api_endpoint = os.getenv('API_ENDPOINT')
|
|
|
|
url = f"{api_endpoint}/api/v1/entity/role/all_roledata"
|
|
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'Authorization': get_token(),
|
|
'X-Entity-Role':os.getenv('ROLE_ID')
|
|
}
|
|
|
|
try:
|
|
print(f"*** Calling get role id. for all LPAs")
|
|
response = requests.get(url, headers=headers)
|
|
response_data = response.json()
|
|
if response.status_code == 200 or response.status_code == 201:
|
|
lpa_role_id_details = response_data["data"]["data"]
|
|
# print(f"LPA Roleid {lpa_role_id_details} in the environment.")
|
|
return lpa_role_id_details
|
|
else:
|
|
print(f"Failed to get data. Status code: {response.status_code}")
|
|
print(f"Error: {response.text}")
|
|
return {"error": f"Failed to get data. Status code: {response.status_code}",
|
|
"status_code": response.status_code, "details": response.text}
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
return {"error": f"An error occurred: {e}", "status_code": 500}
|
|
|
|
def load_partner_data(type,entity,role_id,data):
|
|
|
|
api_endpoint = os.getenv('API_ENDPOINT')
|
|
|
|
url = f"{api_endpoint}/api/v1/entity/role/partner_data"
|
|
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'Authorization': get_token(),
|
|
}
|
|
|
|
data = {
|
|
"entityId":f"{entity}",
|
|
"roleId":f"{role_id}",
|
|
"miscdata": json.loads(data),
|
|
"type":f"{type}"
|
|
}
|
|
|
|
try:
|
|
output = requests.post(url, data=json.dumps(data), headers=headers)
|
|
if output.status_code == 200 or output.status_code == 201:
|
|
print("Success posted the request")
|
|
return True
|
|
except Exception as e:
|
|
print(e)
|
|
return False |