Feature flask s3 end to end process

neo_dolfin/ai/savings/SavingPredAI.py

@@ -6,12 +6,12 @@
 import matplotlib.pyplot as plt
 from scipy.stats import norm
 from operator import itemgetter
-import  csv
+import csv
 import itertools
 import datetime
 
 #read input file 
-transfile = "neo_dolfin/ai/savings/modified_transactions_data.csv"
+transfile = "neo_dolfin/static/modified_transactions_data.csv"
 df = pd.read_csv(transfile)
 print("File read into df")
 print(df.head(5))
@@ -121,8 +121,6 @@ def modelfit(pdqfit, spdqfit):
     print(results.summary().tables[1])
     return results
 
-
-
 def drawPredictions():
     results = modelfit(pdqfit,spdqfit)  
     current = datetime.date.today() 
@@ -134,13 +132,6 @@ def drawPredictions():
     #print(pred.predicted_mean)
     pred_mean = pd.DataFrame(pred.predicted_mean)
     pred_ci['mean_balance'] = pred_mean
-    predtrans_file = "neo_dolfin/ai/savings/Predicted_Balances.csv"
+    predtrans_file = "neo_dolfin/static/Predicted_Balances.csv"
     pred_ci.to_csv(predtrans_file, sep='\t')
-    return pred
-
-def plotPredictedBal():
-    pred = drawPredictions()
-    pred.predicted_mean.plot( label='One-step ahead Forecast', alpha=.7)
-    plt.legend()
-    plt.show()
-plotPredictedBal()
+    return pred

neo_dolfin/app.py

@@ -17,17 +17,19 @@
 import ssl 
 import nltk
 #import certifi
+import requests
+import datetime
+from services.basiq_service import BasiqService
+from io import StringIO
 
 #import dash
 #import dash_core_components as dcc#
 #import dash_html_components as html
 #test
 
 load_dotenv()  # Load environment variables from .env
-
 from classes import *
 from functions import * 
-# from ai.chatbot.chatbot_logic import predict_class, get_response, determine_sentiment, listen_to_user, initialize_chatbot_logic
 from ai.chatbot import chatbot_logic
 
 # Chatbot Logic req files for VENV
@@ -47,26 +49,32 @@
 nltk.data.path.append(nltk_data_path)
 nltk.download('punkt', download_dir=nltk_data_path)
 nltk.download('wordnet', download_dir=nltk_data_path)
+from services.basiq_service import BasiqService
+from services.s3_service import S3Service
 
 app = Flask(__name__)
 app.config['SECRET_KEY'] = secrets.token_hex(16)  # Replace with a secure random key
 app.static_folder = 'static'
 
-df = pd.read_csv('static/transaction_ut.csv')
+# Default S3 storage values
+bucket_name = 'neodolfin-transaction-data-storage-01'
+dummy_csv_filename = 'static/data/transaction_ut.csv'
+
+df1 = pd.read_csv('static/data/transaction_ut.csv')
+df2 = pd.read_csv('static/data/modified_transactions_data.csv')
+df3 = pd.read_csv('static/data/Predicted_Balances.csv')
 
 # AWS STUFF
 AWS_REGION = os.environ.get('AWS_REGION')
 AWS_COGNITO_USER_POOL_ID = os.environ.get('AWS_COGNITO_USER_POOL_ID')
 AWS_COGNITO_APP_CLIENT_ID = os.environ.get('AWS_COGNITO_APP_CLIENT_ID')#
 AWS_COGNITO_CLIENT_SECRET = os.environ.get('AWS_COGNITO_CLIENT_SECRET')
 
+## AWS S3 Service + Basiq API 
 client = boto3.client('cognito-idp', region_name=AWS_REGION)
-
-# DASH APP
-# Initialize Dash app
-#dash_app = dash.Dash(__name__, server=app, url_base_pathname='/dash/')  # Set the route to '/dash/'
-# Define the Dash layout
-#dash_app.layout = dash_layout#
+s3_client = boto3.client('s3')
+basiq_service = BasiqService()
+s3_service = S3Service()
 
 # ROUTING
 
@@ -309,17 +317,11 @@ def signupmfadevice():
 
 ## APPLICATION HOME PAGE - REQUIRES USER TO BE SIGNED IN TO ACCESS
 @app.route('/home/')
-def auth_home(): 
+async def auth_home(): 
     if not is_token_valid():
-        # INSERT BOTO3 REQUEST HERE
-        # CHECK IF USERNAME.FOLDER EXISTS 
-            # IF NOT, MAKE A DIRECTORY AND UPLOAD THE DUMMY DATA
-                # df should be set to dummy data. 
-            # IF YES, GET LIST OF CSV OBJECTS IN USER DIRECTORY 
-                # LOAD LAST CSV OBJECT INTO df VAR.
         return redirect('/signin')  # Redirect to sign-in page if the token is expired
     if is_token_valid():
-        # print(request)
+        #s3connection(bucket_name, s3_client, basiq_service, s3_service)
         return render_template("home.html")
 
 @app.route('/dash/')
@@ -332,7 +334,7 @@ def auth_dash():
 
 ## APPLICATION NEWS PAGE - REQUIRES USER TO BE SIGNED IN TO ACCESS    
 @app.route('/news/')
-def auth_news(): 
+def auth_news():
     if not is_token_valid():
         return redirect('/signin')
     if is_token_valid():

neo_dolfin/functions.py

@@ -14,6 +14,9 @@
 import base64
 import qrcode
 import logging 
+import datetime
+from services.basiq_service import BasiqService
+from io import StringIO
 
 #FUNCTIONS
 def is_token_valid(): #Confirm whether access token exists and has not expired, token provided by AWS Cognito, stored in local session
@@ -35,4 +38,62 @@ def calculate_secret_hash(client_id, client_secret, username): #Calculate secret
     message = username + client_id
     dig = hmac.new(str(client_secret).encode('utf-8'),
                    msg=message.encode('utf-8'), digestmod=hashlib.sha256).digest()
-    return base64.b64encode(dig).decode()
+    return base64.b64encode(dig).decode()
+
+async def s3connection(bucket_name, s3_client, basiq_service, s3_service):
+    success = True
+    current_time = datetime.datetime.now().strftime("%m%d%Y%H%M%S")
+
+        # Create default bucket if not exist
+    try:
+        resp = s3_client.head_bucket(Bucket=bucket_name)
+        if resp['ResponseMetadata']['HTTPStatusCode'] == 200:
+            print("Bucket found")
+        else:
+            raise Exception("Get bucket failed")
+    except Exception:
+        print("Default bucket does not exist. Creating bucket")
+        s3_client.create_bucket(Bucket = bucket_name, CreateBucketConfiguration={'LocationConstraint': 'ap-southeast-2'})
+        s3_client.create_bucket(Bucket = bucket_name + '-processed', CreateBucketConfiguration={'LocationConstraint': 'ap-southeast-2'})
+
+        # Check if user has a directory in the S3 bucket
+    try:
+        df_csv = s3_client.list_objects(Bucket = bucket_name, Prefix = 'raw_data_' + session.get('username'))
+
+    except Exception:
+        print("No folder exists for user: " + session.get('username'))
+        success = False
+
+            # If the user directory does not exist, create that object in the S3 bucket and load the dummy data as a dataframe
+    if not success:
+        print("Creating object")
+
+            # Get access token for the Basiq API
+        access_token = basiq_service.get_access_token()
+
+            # Get all transaction data from Basiq for user and convert to dataframe. Currently returning dummy user data
+        user_transaction_data = basiq_service.get_all_transaction_data_for_user(access_token)
+        Transactions = pd.json_normalize(user_transaction_data, record_path=['data'])
+        df = pd.DataFrame(Transactions)
+
+            # Prepare dataframe data to be set as object in s3 bucket
+        csv_buffer = StringIO()
+        df.to_csv(csv_buffer)
+        testresp = await s3_service.set_object(bucket_name, "raw_data_" + session.get('username') + current_time + ".csv", csv_buffer.getvalue())
+
+            # modified from https://stackoverflow.com/questions/45375999/how-to-download-the-latest-file-of-an-s3-bucket-using-boto3
+        get_latest_object = lambda obj: int(obj['LastModified'].strftime('%S'))
+            # Get latest object with specific prefix from the processed bucket and set to df2 for savings model
+        objects = s3_client.list_objects(Bucket = bucket_name + "-processed", Prefix = "raw_data_" + session.get('username' ))['Contents']  
+        latest_object = [obj['Key'] for obj in sorted(objects, key = get_latest_object)][0]
+        df2 = pd.read_csv(s3_client.get_object(Bucket = bucket_name + '-processed', Key = latest_object).get('Body'))
+
+        #If success, get the latest object and read it into a dataframe
+    if success:
+            # modified from https://stackoverflow.com/questions/45375999/how-to-download-the-latest-file-of-an-s3-bucket-using-boto3
+        get_latest_object = lambda obj: int(obj['LastModified'].strftime('%S'))
+        objects = s3_client.list_objects(Bucket = bucket_name, Prefix = "raw_data_" + session.get('username' ))['Contents']    
+        latest_object = [obj['Key'] for obj in sorted(objects, key = get_latest_object)][0]
+        df1 = pd.read_csv(s3_client.get_object(Bucket = bucket_name, Key = latest_object).get('Body'))
+        df2 = s3_client.get_object(Bucket = bucket_name + '-processed', Key = latest_object).get('Body')
+        df2 = pd.read_csv(df2)

neo_dolfin/requirements.txt

@@ -3,9 +3,12 @@ appnope==0.1.3
 asttokens==2.2.1
 astunparse==1.6.3
 backcall==0.2.0
+aioboto3
+asgiref==3.7.2
+asyncio==3.4.3
 blinker==1.6.2
-boto3==1.28.30
-botocore==1.31.30
+boto3
+#botocore==1.31.30
 cachetools==5.3.1
 certifi==2023.7.22
 charset-normalizer==3.2.0
@@ -19,7 +22,7 @@ decorator==5.1.1
 dnspython==2.4.2
 email-validator==2.0.0.post2
 executing==1.2.0
-Flask==2.3.2
+Flask[async]==2.3.2
 Flask-WTF==1.1.1
 flatbuffers==23.5.26
 fonttools==4.42.1
@@ -79,6 +82,7 @@ python-dotenv==1.0.0
 pytz==2023.3
 pyzmq==25.1.1
 qrcode==7.4.2
+requests==2.31.0
 regex==2023.8.8
 requests==2.31.0
 requests-oauthlib==1.3.1

neo_dolfin/services/Interfaces/ibasiq_service.py

@@ -0,0 +1,18 @@
+from abc import ABC, abstractmethod
+
+class IBasiqService(ABC):
+    @abstractmethod
+    def get_access_token(self):
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_all_transaction_data_for_user(self, access_token):
+        raise NotImplementedError
+
+    @abstractmethod
+    def create_user_transaction_data_object(self, data, username, bucket_name, file_name):
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_headers(self, access_token):
+        raise NotImplementedError

neo_dolfin/services/Interfaces/is3_service.py

@@ -0,0 +1,22 @@
+from abc import ABC, abstractmethod
+
+class IS3Service(ABC):
+    @abstractmethod
+    def set_object(self, bucket_name, object_name, object_bytes):
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_specified_object(bucket_name, object_name):
+        raise NotImplementedError
+
+    @abstractmethod
+    def get_latest_object(bucket_name, username):
+        raise NotImplementedError
+
+    @abstractmethod
+    def create_bucket(bucket_name, configuration_json = None):
+        raise NotImplementedError
+
+    @abstractmethod
+    def delete_bucket(bucket_name):
+        raise NotImplementedError

neo_dolfin/services/basiq_service.py

@@ -0,0 +1,56 @@
+import boto3 as boto3
+from dotenv import load_dotenv
+import os
+from services.Interfaces.ibasiq_service import IBasiqService
+import requests
+import json
+import datetime
+
+# Methods adapted from Trimester 1 Dolfin code to suit our use case
+load_dotenv()  # Load environment variables from .env
+BASE_URL = os.environ.get('BASE_URL')
+BASIQ_ID = os.environ.get('BASIQ_ID')
+API_KEY = os.environ.get('API_KEY')
+
+s3 = boto3.client('s3')
+
+url = f"{BASE_URL}users/{BASIQ_ID}/transactions"
+
+class BasiqService(IBasiqService):
+    def __init__(self):
+        self._data_source = []
+
+    def get_access_token(self):
+            url = BASE_URL + "token"
+            payload = "scope=SERVER_ACCESS"
+            headers = {
+                "accept": "application/json",
+                "basiq-version": "3.0",
+                "content-type": "application/x-www-form-urlencoded",
+                "Authorization": f"Basic {API_KEY}"
+            }
+
+            response = requests.post(url, data=payload, headers=headers)
+            json_response = response.json()
+            access_token = "Bearer " + json_response["access_token"]
+            return access_token
+
+    def create_user_transaction_data_object(data, username, bucket_name, file_name):
+        #Add exception handling
+        current_time = datetime.datetime.now().strftime("%m%d%Y%H%M%S")
+
+        return s3.put_object(Body = data, Bucket = bucket_name, Key = username + "/" + file_name + current_time)
+
+    def get_headers(self, access_token):
+        headers = {
+                "accept": "application/json",
+                "authorization": access_token,
+        }
+
+        return headers
+
+    def get_all_transaction_data_for_user(self, access_token):
+        headers = self.get_headers(access_token)
+
+        response = requests.get(url, headers=headers)
+        return json.loads(response.text)