import streamlit as st import requests import base64 import uuid import os from dotenv import load_dotenv, dotenv_values from datetime import datetime, date from plaid.model.products import Products from plaid.model.country_code import CountryCode from plaid.model.link_token_create_request import LinkTokenCreateRequest from plaid.configuration import Configuration from plaid.api_client import ApiClient from plaid.api.plaid_api import PlaidApi import stripe # Set page config st.set_page_config( page_title="TransactPro: Unified Financial Platform", page_icon="💳", layout="wide", ) # Title of the app st.title("TransactPro: Unified Financial Platform") st.subheader("Made by Citibank Demo Business Inc.") # Sidebar for API keys and configuration st.sidebar.header("API Configuration") # File uploader for .env file st.sidebar.subheader("Upload .env File") uploaded_env = st.sidebar.file_uploader("Choose a .env file", type="env") if uploaded_env: env_data = dotenv_values(stream=uploaded_env) st.sidebar.success("Environment variables loaded!") else: st.sidebar.info("You can upload your .env file to automatically load environment variables.") # Load environment variables either from uploaded .env or default system def get_env_var(key, default=""): return env_data.get(key) if uploaded_env else os.getenv(key, default) # Modern Treasury Configuration st.sidebar.subheader("Modern Treasury") mt_api_key = st.sidebar.text_input("Modern Treasury API Key", type="password", value=get_env_var('MT_API_KEY')) mt_organization_id = st.sidebar.text_input("Organization ID", value=get_env_var('MT_ORG_ID')) # Citibank API Configuration st.sidebar.subheader("Citibank API") citibank_client_id = st.sidebar.text_input("Citibank Client ID", value=get_env_var('CITIBANK_CLIENT_ID')) citibank_client_secret = st.sidebar.text_input("Citibank Client Secret", type="password", value=get_env_var('CITIBANK_CLIENT_SECRET')) citibank_api_key = st.sidebar.text_input("Citibank API Key", value=get_env_var('CITIBANK_API_KEY')) # Plaid API Configuration st.sidebar.subheader("Plaid API") plaid_client_id = st.sidebar.text_input("Plaid Client ID", value=get_env_var('PLAID_CLIENT_ID')) plaid_secret = st.sidebar.text_input("Plaid Secret", type="password", value=get_env_var('PLAID_SECRET')) plaid_env = st.sidebar.selectbox("Plaid Environment", ["sandbox", "development", "production"], index=0) # Stripe API Configuration st.sidebar.subheader("Stripe API") stripe_api_key = st.sidebar.text_input("Stripe API Key", type="password", value=get_env_var('STRIPE_API_KEY')) if stripe_api_key: stripe.api_key = stripe_api_key # Main application endpoint_choice = st.selectbox("Choose Service", [ "Citibank API", "Plaid API", "Stripe API", "Modern Treasury", ]) # Helper functions def create_basic_auth_header(org_id, api_key): credentials = f"{org_id}:{api_key}" base64_credentials = base64.b64encode(credentials.encode()).decode('utf-8') return f"Basic {base64_credentials}" def fetch_modern_treasury_data(api_key, org_id, endpoint): base_url = "https://app.moderntreasury.com/api/" endpoints = { "Payment Orders": "payment_orders", "Expected Payments": "expected_payments", "Returns": "returns", "Incoming Payment Details": "incoming_payment_details", "Counterparties": "counterparties", "Internal Accounts": "internal_accounts", "Ledgers": "ledgers", "Ledger Accounts": "ledger_accounts" } url = base_url + endpoints[endpoint] headers = { "Authorization": create_basic_auth_header(org_id, api_key) } response = requests.get(url, headers=headers) if response.status_code == 200: return response.json() else: st.error(f"Error: {response.status_code} - {response.text}") return None # Citibank Service Class class CitibankService: def __init__(self, client_id, client_secret, api_key): self.client_id = client_id self.client_secret = client_secret self.api_key = api_key self.token_url = "https://sandbox.apihub.citi.com/gcb/api/clientCredentials/oauth2/token/us/gcb" self.base_url = "https://sandbox.apihub.citi.com/gcb/api/v1" def get_access_token(self): payload = { "grant_type": "client_credentials", "scope": "accounts_details_transactions" } headers = { "Authorization": f"Basic {self._encode_credentials()}", "Content-Type": "application/x-www-form-urlencoded" } response = requests.post(self.token_url, data=payload, headers=headers) if response.status_code == 200: return response.json()["access_token"] else: st.error(f"Error obtaining access token: {response.text}") return None def get_accounts(self, access_token): headers = { "Authorization": f"Bearer {access_token}", "client_id": self.client_id, "uuid": str(uuid.uuid4()), "Accept": "application/json" } url = f"{self.base_url}/accounts" response = requests.get(url, headers=headers) if response.status_code == 200: return response.json() else: st.error(f"Error fetching accounts: {response.text}") return None def get_transactions(self, access_token, account_id, from_date, to_date): headers = { "Authorization": f"Bearer {access_token}", "client_id": self.client_id, "uuid": str(uuid.uuid4()), "Accept": "application/json" } params = { "transactionFromDate": from_date, "transactionToDate": to_date } url = f"{self.base_url}/accounts/{account_id}/transactions" response = requests.get(url, headers=headers, params=params) if response.status_code == 200: return response.json() else: st.error(f"Error fetching transactions: {response.text}") return None def _encode_credentials(self): credentials = f"{self.client_id}:{self.client_secret}" return base64.b64encode(credentials.encode()).decode() # Plaid Service Class class PlaidService: def __init__(self, client_id, secret, environment): configuration = Configuration( host=f"https://{environment}.plaid.com", api_key={ 'clientId': client_id, 'secret': secret, } ) api_client = ApiClient(configuration) self.client = PlaidApi(api_client) def create_link_token(self): request = LinkTokenCreateRequest( products=[Products('auth')], client_name='TransactPro', country_codes=[CountryCode('US')], user={'client_user_id': str(uuid.uuid4())}, language='en' ) response = self.client.link_token_create(request) return response.link_token def exchange_public_token(self, public_token): response = self.client.item_public_token_exchange({'public_token': public_token}) return response.access_token def get_accounts(self, access_token): response = self.client.accounts_get({'access_token': access_token}) return response.accounts def get_transactions(self, access_token, start_date, end_date): response = self.client.transactions_get({ 'access_token': access_token, 'start_date': start_date, 'end_date': end_date }) return response.transactions # Main application logic if endpoint_choice == "Citibank API": st.header("Citibank API Integration") if not citibank_client_id or not citibank_client_secret or not citibank_api_key: st.error("Please provide Citibank API credentials in the sidebar.") else: citibank_service = CitibankService(citibank_client_id, citibank_client_secret, citibank_api_key) access_token = citibank_service.get_access_token() if access_token: citibank_action = st.selectbox("Choose an action", [ "Retrieve Account Details", "Retrieve Transactions" ]) if citibank_action == "Retrieve Account Details": data = citibank_service.get_accounts(access_token) if data: st.json(data) elif citibank_action == "Retrieve Transactions": account_id = st.text_input("Enter Account ID") from_date = st.date_input("From Date", value=date.today()) to_date = st.date_input("To Date", value=date.today()) if account_id and from_date and to_date: data = citibank_service.get_transactions( access_token, account_id, from_date.strftime("%Y-%m-%d"), to_date.strftime("%Y-%m-%d") ) if data: st.json(data) else: st.info("Please enter Account ID and select From and To dates.") elif endpoint_choice == "Plaid API": st.header("Plaid API Integration") if not plaid_client_id or not plaid_secret: st.error("Please provide Plaid API credentials in the sidebar.") else: plaid_service = PlaidService(plaid_client_id, plaid_secret, plaid_env) st.write("Initializing Plaid Link...") link_token = plaid_service.create_link_token() st.write("Link Token created. Use this token in your frontend to initialize Plaid Link.") st.code(link_token, language='text') public_token = st.text_input("Enter Public Token (from Plaid Link)") if public_token: access_token_response = plaid_service.exchange_public_token(public_token) access_token = access_token_response['access_token'] st.write("Access Token obtained.") st.code(access_token, language='text') plaid_action = st.selectbox("Choose an action", [ "Get Accounts", "Get Transactions" ]) if plaid_action == "Get Accounts": accounts = plaid_service.get_accounts(access_token) st.json(accounts) elif plaid_action == "Get Transactions": start_date = st.date_input("Start Date", value=date.today()) end_date = st.date_input("End Date", value=date.today()) if start_date and end_date: transactions = plaid_service.get_transactions( access_token, start_date.strftime("%Y-%m-%d"), end_date.strftime("%Y-%m-%d") ) st.json(transactions) else: st.info("Please select Start and End dates.") elif endpoint_choice == "Stripe API": st.header("Stripe API Integration") if not stripe_api_key: st.error("Please provide the Stripe API Key in the sidebar.") else: stripe_action = st.selectbox("Choose an action", [ "Create Account Link", "Create Account Session" ]) if stripe_action == "Create Account Link": account_id = st.text_input("Enter Connected Account ID") if account_id: try: account_link = stripe.AccountLink.create( account=account_id, refresh_url="https://example.com/reauth", return_url="https://example.com/return", type="account_onboarding", ) st.write("Account Link created:") st.write(account_link.url) except Exception as e: st.error(f"Error: {e}") else: st.info("Please enter a Connected Account ID.") elif stripe_action == "Create Account Session": account_id = st.text_input("Enter Connected Account ID") if account_id: try: account_session = stripe.AccountSession.create( account=account_id, components={ "account_onboarding": {"enabled": True}, "payments": {"enabled": True}, "payouts": {"enabled": True}, "balances": {"enabled": True}, } ) st.write("Account Session created:") st.write("Client Secret:") st.code(account_session.client_secret, language='text') except Exception as e: st.error(f"Error: {e}") else: st.info("Please enter a Connected Account ID.") elif endpoint_choice == "Modern Treasury": st.header("Modern Treasury API Integration") endpoint = st.selectbox("Choose Endpoint", [ "Payment Orders", "Expected Payments", "Returns", "Incoming Payment Details", "Counterparties", "Internal Accounts", "Ledgers", "Ledger Accounts" ]) if st.button("Fetch Data"): if mt_api_key and mt_organization_id: data = fetch_modern_treasury_data(mt_api_key, mt_organization_id, endpoint) if data: st.json(data) else: st.error("Please provide both a valid API key and Organization ID.") # Display footer st.sidebar.markdown("Powered by Streamlit") st.sidebar.markdown("Made by Citibank Demo Business Inc.")