Fixed file processing and standardization logic

app/api/routers/file_upload.py

@@ -6,6 +6,8 @@ import asyncio
 import os
 import csv
 import codecs
+from io import StringIO
+import pandas as pd
 
 from app.engine.postgresdb import get_db_session
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -31,31 +33,65 @@ async def create_file(input_file: UploadFile, db: AsyncSession = Depends(get_db_
         if not os.path.exists(file_upload_directory_path):
             os.makedirs(file_upload_directory_path)
 
-        input_csv_file = open(output_csv_file_path, "a")
+        # input_csv_file = open(output_csv_file_path, "a")
 
         try:
             if input_file.filename.endswith(".csv"):
-                read_csv = await input_file.read()
-                # result["output"] = read_csv
-
+                input_contents = await input_file.read()
+                buffer = StringIO(input_contents.decode('utf-8'))
+                read_csv = csv.reader(buffer)
+            
                 # Write items of .csv filte to directory
-                with open(output_csv_file_path, 'w', encoding="utf-8") as csv_file:
-                    result["processed_file"] = read_csv
-                    [csv_file.write(" ".join(row)+'\n') for row in read_csv.decode("utf-8").strip().splitlines()]
+                with codecs.open(output_csv_file_path, 'w+', encoding='utf-8') as csv_file:
+                    [csv_file.write(" ".join(row)+'\n') for row in read_csv]
                 csv_file.close()
+
+                new_output_csv_file_path = os.path.join(file_upload_directory_path, f"new_{input_file.filename}")
                 
-                with open(output_csv_file_path, 'r', encoding="utf-8") as csv_file:
-                    result["output"] = csv_file.read()
+                with codecs.open(output_csv_file_path, 'r', encoding='utf-8') as csv_file:
+                    with codecs.open(new_output_csv_file_path, 'w+', encoding='utf-8') as new_csv_file:
+                        csvreader = csv.reader(csv_file)
+                        header = next(csvreader)[0]
+                        new_header = header.replace(" ", ",")
+                        # print(f"header: \n{header} headerItem: {header[0]} new_header: {new_header}\n")
+                        new_csv_file.write(" ".join(new_header)+'\n')
+                        for row in csvreader:
+                           new_row_items = row[0].split(' ')
+                           new_row = ""
+                           description = ""
+                           for idx, element in enumerate(new_row_items):
+                            # Since our string has been delimited by spaces, we want to get all the words of the description which comes after the 'date' index (0) and 
+                            # before the second to last 'type' index
+                            if idx > 0 and idx < len(new_row_items) - 2:
+                                description = description + f" {element}"
+                            else:
+                                delim = ", " if idx > 0 else ""
+                                item = element if len(description) == 0 else f"{description}, {element}"
+                                new_row = new_row + delim + item
+                                description = ""
+
+                        #    print(f"row: {row} rowType: \n{type(row)} new_row: {new_row}\n")
+                           new_csv_file.write(" ".join(new_row)+'\n')
+                        new_csv_file.close()
                 csv_file.close()
+
+                df = pd.read_csv(new_output_csv_file_path, names=['transaction_date', 'name_description', 'type', 'amount'], header=0, dtype=str)
+                # df = df.reindex(columns=['transaction_date', 'name_description', 'type', 'amount'])
+                # transaction_date_index = df.columns.get_loc('transaction_date')
+                result["output"] = df
+                # print(f"column: \n{df['transaction_date']} \n")
                 
-                print(f"create_file result: {result}")
 
                 # With the newly created file and it's path, process and save it for embedding
-                processed_file = await process_file(output_csv_file_path)
+                # processed_file = await process_file(new_output_csv_file_path)
+                processed_file = await process_file(new_output_csv_file_path, df)
                 result["processed_file"] = processed_file
                 result = await asyncio.gather(processed_file)
                 result["result"] = result
                 await save_results(db, result)
+
+                buffer.close()
+                input_file.close()
                    
         except Exception as e:
             result["error"] = str(e)

app/categorization/categorizer.py

@@ -1,4 +1,3 @@
-# Standard library imports
 import re
 import ast
 import json
@@ -6,7 +5,6 @@ import logging
 import os
 from typing import Any, List, Tuple, Optional, Dict, Union
 
-# Third-party library imports
 import numpy as np
 import pandas as pd
 import asyncio
@@ -14,7 +12,8 @@ from rapidfuzz import process
 from tenacity import retry, wait_random_exponential, stop_after_attempt
 from pydantic import ValidationError
 
-# Local application/library specific imports
+from fastapi import HTTPException
+
 from langchain_openai import ChatOpenAI
 from langchain.chains import LLMChain
 from langchain.output_parsers import PydanticOutputParser, OutputFixingParser
@@ -44,16 +43,23 @@ def fuzzy_match_list_categorizer(
     Returns:
         str or None: Category of the matched description, or None if no match found.
     """
-
-    # Fuzzy-match this description against the reference descriptions
-    match_results = process.extractOne(
-        description, descriptions, score_cutoff=threshold)
-
-    # If a match is found, return the category of the matched description
-    if match_results:
-        return description_category_pairs.at[match_results[2], 'category']
-
-    return None
+    result = {"description": description, "descriptions": descriptions, "description_category_pairs": description_category_pairs, "match_results": "", "error": ""}
+    try:
+        # Fuzzy-match this description against the reference descriptions
+        match_results = process.extractOne(
+            description, descriptions, score_cutoff=threshold)
+        
+        result["match_results"] = match_results
+
+        # If a match is found, return the category of the matched description
+        if match_results:
+            return description_category_pairs.at[match_results[2], 'category']
+
+        return None
+        
+    except Exception as e:
+        result["error"] = str(e)
+        raise HTTPException(status_code = 500, detail=f"fuzzy_match_list_categorizer result: {result}")
 
 
 async def llm_list_categorizer(tx_list: pd.DataFrame) -> pd.DataFrame:
@@ -68,31 +74,42 @@ async def llm_list_categorizer(tx_list: pd.DataFrame) -> pd.DataFrame:
     Returns:
         pd.DataFrame: DataFrame mapping transaction descriptions to their inferred categories.
     """
+    
+    result = {"tx_list": tx_list, "results": "", "valid results": "", "valid outputs": "", "error": ""}
 
-    # Initialize language model and prompt
-    openai_api_key = os.environ['OPENAI_API_KEY']
-    llm = ChatOpenAI(temperature=0, model="gpt-3.5-turbo-0125",
-                     api_key=openai_api_key)
-    prompt = PromptTemplate.from_template(template=CATEGORY_TEMPLATE)
-    chain = LLMChain(llm=llm, prompt=prompt)
-
-    # Iterate over the DataFrame in batches of TX_PER_LLM_RUN transactions
-    tasks = [llm_sublist_categorizer(tx_list.attrs['file_name'], chain=chain, tx_descriptions="\n".join(chunk['name_description']).strip())
-             for chunk in np.array_split(tx_list, tx_list.shape[0] // TX_PER_LLM_RUN + 1)]
-
-    # Gather results and extract (valid) outputs
-    # The results variable is a list of 'results', each 'result' being the output of a single LLM run
-    results = await asyncio.gather(*tasks)
-
-    # Extract valid results (each valid result is a list of description-category pairs)
-    valid_results = [result['output'] for result in results if result['valid']]
-
-    # Flatten the list of valid results to obtain a single list of description-category pairs
-    valid_outputs = [
-        output for valid_result in valid_results for output in valid_result]
+    try:
+        # Initialize language model and prompt
+        openai_api_key = os.environ['OPENAI_API_KEY']
+        llm = ChatOpenAI(temperature=0, model="gpt-3.5-turbo-0125",
+                        api_key=openai_api_key)
+        prompt = PromptTemplate.from_template(template=CATEGORY_TEMPLATE)
+        chain = LLMChain(llm=llm, prompt=prompt)
+
+        # Iterate over the DataFrame in batches of TX_PER_LLM_RUN transactions
+        tasks = [llm_sublist_categorizer(tx_list.attrs['file_name'], chain=chain, tx_descriptions="\n".join(chunk['name_description']).strip())
+                for chunk in np.array_split(tx_list, tx_list.shape[0] // TX_PER_LLM_RUN + 1)]
+
+        # Gather results and extract (valid) outputs
+        # The results variable is a list of 'results', each 'result' being the output of a single LLM run
+        results = await asyncio.gather(*tasks)
+        result["results"] = results
+
+        # Extract valid results (each valid result is a list of description-category pairs)
+        valid_results = [result['output'] for result in results if result['valid']]
+        result["valid results"] = valid_results
+
+        # Flatten the list of valid results to obtain a single list of description-category pairs
+        valid_outputs = [
+            output for valid_result in valid_results for output in valid_result]
+        result["valid outputs"] = valid_outputs
+
+        # Return a DataFrame with the valid outputs
+        return pd.DataFrame(valid_outputs, columns=['name_description', 'category'])
+    
+    except Exception as e:
+        result["error"] = str(e)
+        raise HTTPException(status_code = 500, detail=f"llm_list_categorizer result: {result}")
 
-    # Return a DataFrame with the valid outputs
-    return pd.DataFrame(valid_outputs, columns=['name_description', 'category'])
 
 
 @retry(wait=wait_random_exponential(min=1, max=20), stop=stop_after_attempt(6))
@@ -143,5 +160,6 @@ async def llm_sublist_categorizer(
         logging.log(
             logging.ERROR, f"| File: {file_name} | Unexpected Error: {e}\nRaw Result: {raw_result}")
         result['valid'] = False
+        raise HTTPException(status_code = 500, detail=f"llm_sublist_categorizer result: {result}")
 
     return result

app/categorization/categorizer_list.py

@@ -24,7 +24,7 @@ async def categorize_list(tx_list: pd.DataFrame) -> pd.DataFrame:
     Returns:
         pd.DataFrame: The original DataFrame with an additional column for the category.
     """
-    result = {"tx_list": tx_list, "description_category_pairs": "", "uncategorized_descriptions": "", "categorized_descriptions": "", "error": ""}
+    result = {"tx_list": pd.DataFrame(), "description_category": "", "description_category_pairs": "", "uncategorized_descriptions": "", "categorized_descriptions": "", "error": ""}
 
     try:
         if not os.path.exists(CATEGORY_REFERENCE_OUTPUT_FILE):
@@ -33,17 +33,22 @@ async def categorize_list(tx_list: pd.DataFrame) -> pd.DataFrame:
         if os.path.exists(CATEGORY_REFERENCE_OUTPUT_FILE):
             # Read description-category pairs from the reference file
             description_category_pairs = pd.read_csv(
-                CATEGORY_REFERENCE_OUTPUT_FILE, header=None, names=['name_description', 'category']
+                CATEGORY_REFERENCE_OUTPUT_FILE, sep=r'\s*,\s*', header=None, names=['name_description', 'category'], encoding='utf-8',
+                engine='python'
             )
-            result['description_category_pairs'] = description_category_pairs
-
+            
             # Extract only descriptions for faster matching
-            descriptions = description_category_pairs['name_description'].values
+            description_category_pairs.columns = description_category_pairs.columns.str.strip()
+            # descriptions = description_category_pairs['name_description'].values
+            descriptions = tx_list['name_description'].values
+
+            result['description_category'] = descriptions
+            result['tx_list'] = tx_list['name_description']
 
             # Use fuzzy matching to find similar descriptions and assign the category
             tx_list['category'] = tx_list['name_description'].apply(
                 fuzzy_match_list_categorizer,
-                args=(descriptions, description_category_pairs),
+                args=(descriptions, description_category_pairs)
             )
 
             # Filter out uncategorized transactions, deduplicate, and sort by description

app/categorization/file_processing.py

@@ -20,7 +20,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 
 # Read file and process it (e.g. categorize transactions)
-async def process_file(file_path: str) -> Dict[str, Union[str, pd.DataFrame]]:
+async def process_file(file_path: str, df: pd.DataFrame) -> Dict[str, Union[str, pd.DataFrame]]:
     """
     Process the input file by reading, cleaning, standardizing, and categorizing the transactions.
 
@@ -32,11 +32,11 @@ async def process_file(file_path: str) -> Dict[str, Union[str, pd.DataFrame]]:
     """
 
     file_name = os.path.basename(file_path)
-    result = {"file_name": file_name, "output": pd.DataFrame(), "error": ""}
+    result = {"file_path": file_path, "output": pd.DataFrame(), "columns": [], "error": ""}
     try:
         # Read file into standardized tx format: transaction_date, name_description, type, amount, category, source
-        tx_list = standardize_csv_file(file_path)
-        print(f"process_file tx_list columns: \n {tx_list.columns.tolist()}")
+        tx_list = standardize_csv_file(file_path, df)
+        result["columns"] = tx_list.columns.tolist()
 
         # Categorize transactions
         result["output"] = await categorize_list(tx_list)
@@ -45,14 +45,13 @@ async def process_file(file_path: str) -> Dict[str, Union[str, pd.DataFrame]]:
     except Exception as e:
         # Return an error indicator and exception info
         logging.debug(logging.ERROR, f"| File: {file_name} | Unexpected Error: {e}")
-        print(f"ERROR processing file {file_name}: {e} errorType: {type(e)}")
         result["error"] = str(e)
-        raise HTTPException(status_code = 500, detail=f"process_file result: {result}")
+        raise HTTPException(status_code = 500, detail=f"\n\n process_file result: \n{result}")
 
     return result
 
 
-def standardize_csv_file(file_path: str) -> pd.DataFrame:
+def standardize_csv_file(file_path: str, d: pd.DataFrame) -> pd.DataFrame:
     """
     Read and prepare the data from the input file.
 
@@ -62,46 +61,60 @@ def standardize_csv_file(file_path: str) -> pd.DataFrame:
     Returns:
         pd.DataFrame: Prepared transaction data.
     """
-    result = {"csv_file": "", "file_path": file_path, "tx_list_columns": {}, "transaction_date": "", "tx_list_new_columns": "", "error": ""}
+    result = {"csv_file": "", "file_path": file_path, "new_csv_file": "", "transaction_date": "", "tx_list_new_columns": "", "error": ""}
+    
     try:
+        #1
         # result["csv_file"] = csv_file.read()
         # reader = csv.reader(csv_file)
         # data = list(reader)
-        # tx_list = pd.DataFrame(data, columns=data[0])
-
-        tx_list = pd.read_csv(file_path, encoding="utf-8")
-        print(f"standardize_csv_file tx_list: {tx_list} \n")
-        tx_list.loc[:, "category"] = ""
-        result["csv_file"] = tx_list
-            
-        # result["tx_list_columns"] = tx_list
-        tx_list.attrs["file_name"] = file_path
-        tx_list.columns = tx_list.columns.str.lower().str.strip()
+        # df = pd.DataFrame(data, columns=data[0])
+
+        #2
+        # df = pd.read_csv(file_path)
+
+        #3
+        # with open(file_path, 'r', encoding='utf-8') as file:
+        #     csv_reader = csv.reader(file)
+        #     data = list(csv_reader)
         
-        # # Standardize dates to YYYY/MM/DD format
-        # result["transaction_date"] = pd.to_datetime(tx_list["transaction_date"])
-        # print(f"transaction_date: {tx_list['transaction_date']}")
-        # tx_list["transaction_date"] = pd.to_datetime(tx_list["transaction_date"]).dt.strftime("%Y/%m/%d")
+        # df = pd.DataFrame(data)
+
+        headers = d.columns
+        df = pd.DataFrame(d.values[1:], columns=headers)
+
+        result["csv_file"] = df
+        df.attrs['file_name'] = file_path
+        # print(f"\n\n standardize_csv_file columns: \n{df} \n date:\n{df['transaction_date']} \n")
+        # tx_list.columns = ['transaction_date', 'name_description', 'type', 'amount', 'category']
 
+        # # Standardize dates to YYYY/MM/DD format
+        warnings.filterwarnings('ignore', 'Parsing dates', category=UserWarning)
+        # df['transaction_date'] =  pd.to_datetime(df['transaction_date']).dt.strftime('%d/%m/%Y')
+        
         # Add source and reindex to desired tx format; category column is new and therefore empty
-        tx_list.loc[:, "source"] = os.path.basename(file_path)
-        # tx_list = tx_list.reindex(columns=["transaction_date", "name_description", "type", "amount", "category"])
-        # result["tx_list_new_columns"] = tx_list.columns
-
-        # print(f"standardize_csv_file result: {result}")
-            
-        # with open(file_path, 'r') as csv_file:
-        # return tx_list
+        # tx_list.insert(0, "source", [os.path.basename(file_path) for i in range(len(tx_list.columns))])
+        # tx_list.insert(0, "category", [])
+
+        # df = df.reindex(columns=['transaction_date', 'name_description', 'type', 'amount', 'category'])
+        # tx_list['transaction_date'] = pd.to_datetime(tx_list['transaction_date']).dt.strftime('%Y/%m/%d')
+        # tx_list.set_index(['transaction_date', 'name_description', 'type', 'amount', 'category'])
+        df['source'] = pd.Series([os.path.basename(file_path)] * len(df.index))
+        df['category'] = pd.Series([])
+        result["new_csv_file"] = df
+        print(f"\n\n standardize_csv_file result: \n{df} \n")
+       
+        return df
     
     except Exception as e:
         # Return an error indicator and exception info
-        # print(f"standardize_csv_file exception: {e}")
         logging.debug("standardize_csv_file Error: {e}")
-        result["error"] = str(e)
+        if hasattr(e, 'message'):
+            result["error"] = str(e.message)
+        else:
+            result["error"] = str(e)
         raise HTTPException(status_code = 401, detail=f"standardize_csv_file exception: {result}")
 
-    return tx_list
-
 
 async def save_results(db: AsyncSession, results: List) -> None:
     """
@@ -168,4 +181,4 @@ async def save_results(db: AsyncSession, results: List) -> None:
         # print(f"standardize_csv_file exception: {e}")
         logging.debug("save_results Error: {e}")
         result["error"] = str(e)
-        raise HTTPException(status_code = 401, detail=f"save_results exception: {result}")
+        raise HTTPException(status_code = 401, detail=f"save_results exception: {error_messages}")

migration/versions/e8e988ebb7c7_.py

@@ -0,0 +1,30 @@
+"""
+
+Revision ID: e8e988ebb7c7
+Revises: 4e76691ab103
+Create Date: 2024-06-08 18:13:46.160432
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'e8e988ebb7c7'
+down_revision: Union[str, None] = '4e76691ab103'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    pass
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    pass
+    # ### end Alembic commands ###

transactions_2024.csv

@@ -1,29 +1,29 @@
 transaction_date,name_description,type,amount
-2023-12-30,Comcast Internet,Expense,9.96
-2023-12-30,Lemonade Home Insurance,Expense,17.53
-2023-12-30,Monthly Appartment Rent,Expense,2000.0
-2023-12-30,Staples Office Supplies,Expense,12.46
-2023-12-29,Selling Paintings,Income,13.63
-2023-12-29,Spotify,Expense,12.19
-2023-12-23,Target,Expense,27.08
-2023-12-22,IT Consulting,Income,541.57
-2023-12-22,Phone,Expense,10.7
-2023-12-20,ML Consulting,Income,2641.93
-2023-12-19,Chipotle,Expense,18.9
-2023-12-18,Cold Brew Coffee,Expense,17.67
-2023-12-18,Gas,Expense,8.80
-2023-12-18,CA Property Tax,Expense,1670.34
-2022-11-26,Salary,Income,4000.36
-2022-11-26,Cellphone,Expense,19.27
-2022-11-26,Gym Membership,Expense,24.71
-2022-11-25,Wholefoods,Expense,17.35
-2022-11-24,Freelancing,Income,2409.55
-2022-11-19,Spotify,Expense,20.76
-2022-10-25,Blogging,Income,4044.27
-2022-10-24,Uber Taxi,Expense,18.9
-2022-10-23,Uber Taxi,Expense,27.54
-2022-10-22,Apple Services,Expense,41.25
-2022-10-21,Netflix,Expense,22.8
-2022-01-16,Amazon Lux,Expense,24.11
-2022-01-15,Burger King,Expense,30.08
-2022-01-14,Amazon,Expense,11.0
+20/11/2023,Comcast Internet,Expense,9.96
+21/04/2023,Lemonade Home Insurance,Expense,17.53
+22/05/2023,Monthly Appartment Rent,Expense,2000.0
+14/06/2023,Staples Office Supplies,Expense,12.46
+17/07/2023,Selling Paintings,Income,13.63
+19/12/2023,Spotify,Expense,12.19
+05/10/2023,Target,Expense,27.08
+03/08/2023,IT Consulting,Income,541.57
+06/04/2023,Phone,Expense,10.7
+02/01/2023,ML Consulting,Income,2641.93
+05/04/2023,Chipotle,Expense,18.9
+18/05/2023,Cold Brew Coffee,Expense,17.67
+13/03/2023,Gas,Expense,8.80
+19/07/2023,CA Property Tax,Expense,1670.34
+20/08/2023,Salary,Income,4000.36
+20/09/2023,Cellphone,Expense,19.27
+06/10/2023,Gym Membership,Expense,24.71
+02/01/2023,Wholefoods,Expense,17.35
+11/06/2023,Freelancing,Income,2409.55
+12/11/2023,Spotify,Expense,20.76
+20/03/2023,Blogging,Income,4044.27
+12/02/2023,Uber Taxi,Expense,18.9
+23/07/2023,Uber Taxi,Expense,27.54
+01/09/2023,Apple Services,Expense,41.25
+07/05/2023,Netflix,Expense,22.8
+09/08/2023,Amazon Lux,Expense,24.11
+17/01/2023,Burger King,Expense,30.08
+14/01/2023,Amazon,Expense,11.0