Create app.py

app.py

@@ -0,0 +1,327 @@
+import streamlit as st
+import requests
+import base64
+import uuid
+import os
+from dotenv import load_dotenv
+from datetime import datetime, date
+import plaid
+from plaid import Client as PlaidClient
+import stripe
+
+# Load environment variables from .env file
+load_dotenv()
+
+# 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")
+
+# Modern Treasury Configuration
+st.sidebar.subheader("Modern Treasury")
+mt_api_key = st.sidebar.text_input("Modern Treasury API Key", type="password", value=os.getenv('MT_API_KEY'))
+mt_organization_id = st.sidebar.text_input("Organization ID", value=os.getenv('MT_ORG_ID'))
+
+# Citibank API Configuration
+st.sidebar.subheader("Citibank API")
+citibank_client_id = st.sidebar.text_input("Citibank Client ID", value=os.getenv('CITIBANK_CLIENT_ID'))
+citibank_client_secret = st.sidebar.text_input("Citibank Client Secret", type="password", value=os.getenv('CITIBANK_CLIENT_SECRET'))
+citibank_api_key = st.sidebar.text_input("Citibank API Key", value=os.getenv('CITIBANK_API_KEY'))
+
+# Plaid API Configuration
+st.sidebar.subheader("Plaid API")
+plaid_client_id = st.sidebar.text_input("Plaid Client ID", value=os.getenv('PLAID_CLIENT_ID'))
+plaid_secret = st.sidebar.text_input("Plaid Secret", type="password", value=os.getenv('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=os.getenv('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):
+    # Combine Organization ID and API Key, separated by a colon
+    credentials = f"{org_id}:{api_key}"
+    # Encode the credentials in base64
+    base64_credentials = base64.b64encode(credentials.encode()).decode('utf-8')
+    # Return the full Authorization header
+    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):
+        env = getattr(plaid.Environment, environment.capitalize())
+        self.client = PlaidClient(
+            client_id=client_id,
+            secret=secret,
+            environment=env
+        )
+
+    def create_link_token(self):
+        configs = {
+            'user': {
+                'client_user_id': str(uuid.uuid4())
+            },
+            'client_name': 'TransactPro',
+            'products': ['auth', 'transactions'],
+            'country_codes': ['US'],
+            'language': 'en',
+        }
+        response = self.client.LinkToken.create(configs)
+        return response['link_token']
+
+    def exchange_public_token(self, public_token):
+        response = self.client.Item.public_token.exchange(public_token)
+        return response['access_token']
+
+    def get_accounts(self, access_token):
+        response = self.client.Accounts.get(access_token)
+        return response['accounts']
+
+    def get_transactions(self, access_token, start_date, end_date):
+        response = self.client.Transactions.get(access_token, start_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.")