Spaces:
Sleeping
Sleeping
| 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.") | |