Add logging to show encoding/decoding issue

app/api/routers/file_upload.py

@@ -5,6 +5,7 @@ from app.schema.index import FileUploadCreate
 import asyncio
 import os
 import csv
+import codecs
 
 from app.engine.postgresdb import get_db_session
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -38,19 +39,19 @@ async def create_file(input_file: UploadFile, db: AsyncSession = Depends(get_db_
                 # result["output"] = read_csv
 
                 # Write items of .csv filte to directory
-                with open(output_csv_file_path, 'w') as csv_file:
+                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").splitlines()]
+                    [csv_file.write(" ".join(row)+'\n') for row in read_csv.decode("utf-8").strip().splitlines()]
                 csv_file.close()
                 
-                # with open(output_csv_file_path, 'r') as csv_file:
-                #     result["output"] = csv_file.read()
-                # csv_file.close()
+                with open(output_csv_file_path, 'r', encoding="utf-8") as csv_file:
+                    result["output"] = csv_file.read()
+                csv_file.close()
                 
-                # print(f"create_file result: {result}")
+                print(f"create_file result: {result}")
 
                 # With the newly created file and it's path, process and save it for embedding
-                processed_file = process_file(output_csv_file_path)
+                processed_file = await process_file(output_csv_file_path)
                 result["processed_file"] = processed_file
                 result = await asyncio.gather(processed_file)
                 result["result"] = result

app/categorization/categorizer.py

@@ -77,7 +77,7 @@ async def llm_list_categorizer(tx_list: pd.DataFrame) -> pd.DataFrame:
     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())
+    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
@@ -92,7 +92,7 @@ async def llm_list_categorizer(tx_list: pd.DataFrame) -> pd.DataFrame:
         output for valid_result in valid_results for output in valid_result]
 
     # Return a DataFrame with the valid outputs
-    return pd.DataFrame(valid_outputs, columns=['name/description', 'category'])
+    return pd.DataFrame(valid_outputs, columns=['name_description', 'category'])
 
 
 @retry(wait=wait_random_exponential(min=1, max=20), stop=stop_after_attempt(6))

app/categorization/categorizer_list.py

@@ -1,11 +1,10 @@
-# Standard library imports
 import os
 from datetime import datetime
 
-# Third-party library imports
+from fastapi import HTTPException
+
 import pandas as pd
 
-# Local application/library specific imports
 from app.categorization.config import CATEGORY_REFERENCE_OUTPUT_FILE
 from app.categorization.categorizer import llm_list_categorizer, fuzzy_match_list_categorizer
 
@@ -25,46 +24,60 @@ 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": ""}
 
-    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']
-        )
-
-        # Extract only descriptions for faster matching
-        descriptions = description_category_pairs['name_description'].values
-
-        # 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),
-        )
-
-    # Filter out uncategorized transactions, deduplicate, and sort by description
-    uncategorized_descriptions = (
-        tx_list[tx_list['category'].isnull()]
-        .drop_duplicates(subset=['name_description'])
-        .sort_values(by=['name_description'])
-    )
-
-    # Ask the language model to categorize the remaining descriptions
-    if not uncategorized_descriptions.empty:
-        categorized_descriptions = await llm_list_categorizer(
-            uncategorized_descriptions[['name_description', 'category']]
-        )
-
-        categorized_descriptions.dropna(inplace=True)
-
-        # Update the category for uncategorized transactions based on the language model results
-        if not categorized_descriptions.empty:
-            tx_list['category'] = tx_list['category'].fillna(
-                tx_list['name_description'].map(
-                    categorized_descriptions.set_index('name_description')['category']
-                )
+    try:
+        if not os.path.exists(CATEGORY_REFERENCE_OUTPUT_FILE):
+            os.makedirs(CATEGORY_REFERENCE_OUTPUT_FILE)
+
+        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']
+            )
+            result['description_category_pairs'] = description_category_pairs
+
+            # Extract only descriptions for faster matching
+            descriptions = description_category_pairs['name_description'].values
+
+            # 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),
             )
-        
-        # Fill remaining NaN values in 'category' with 'Other'
-        tx_list['category'] = tx_list['category'].fillna('Other')
 
-    return tx_list
+            # Filter out uncategorized transactions, deduplicate, and sort by description
+            uncategorized_descriptions = (
+                tx_list[tx_list['category'].isnull()]
+                .drop_duplicates(subset=['name_description'])
+                .sort_values(by=['name_description'])
+            )
+            result['uncategorized_descriptions'] = uncategorized_descriptions
+
+            # Ask the language model to categorize the remaining descriptions
+            if not uncategorized_descriptions.empty:
+                categorized_descriptions = await llm_list_categorizer(
+                    uncategorized_descriptions[['name_description', 'category']]
+                )
+
+                categorized_descriptions.dropna(inplace=True)
+                result['categorized_descriptions'] = categorized_descriptions
+
+                # Update the category for uncategorized transactions based on the language model results
+                if not categorized_descriptions.empty:
+                    tx_list['category'] = tx_list['category'].fillna(
+                        tx_list['name_description'].map(
+                            categorized_descriptions.set_index('name_description')['category']
+                        )
+                    )
+                
+                # Fill remaining NaN values in 'category' with 'Other'
+                tx_list['category'] = tx_list['category'].fillna('Other')
+
+        return tx_list
+    
+    except Exception as e:
+        # Return an error indicator and exception info
+        print(f"ERROR categorizer_list: {e} errorType: {type(e)}")
+        result["error"] = str(e)
+        raise HTTPException(status_code = 500, detail=f"categorize_list result: {result}")

app/categorization/file_processing.py

@@ -34,8 +34,9 @@ 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": ""}
     try:
-        # Read file into standardized tx format: source, date, type, category, description, amount
+        # 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()}")
 
         # Categorize transactions
         result["output"] = await categorize_list(tx_list)
@@ -44,7 +45,7 @@ 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}")
+        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}")
 
@@ -63,40 +64,43 @@ def standardize_csv_file(file_path: str) -> pd.DataFrame:
     """
     result = {"csv_file": "", "file_path": file_path, "tx_list_columns": {}, "transaction_date": "", "tx_list_new_columns": "", "error": ""}
     try:
-        with open(file_path, 'r') as csv_file:
-            # result["csv_file"] = csv_file.read()
-            # reader = csv.reader(csv_file)
-            # data = list(reader)
-            # tx_list = pd.DataFrame(data, columns=data[0])
-
-            # result["csv_file"] = csv_file.read()
-            tx_list = pd.read_csv(file_path)
-            # tx_list.dropna(inplace=True)
-
-            # result["tx_list_columns"] = tx_list
-            tx_list.attrs["file_name"] = file_path
-            tx_list.columns = tx_list.columns.str.lower().str.strip()
+        # 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()
         
-            # 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")
-
-            # 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", "type", "category", "name_description", "amount"])
-            # result["tx_list_new_columns"] = tx_list
+        # # 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")
+
+        # 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}")
             
-        return tx_list
+        # with open(file_path, 'r') as csv_file:
+        # return tx_list
     
     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)
-        raise HTTPException(status_code = 401, detail=f"standardize_csv_file result: {result}")
+        raise HTTPException(status_code = 401, detail=f"standardize_csv_file exception: {result}")
 
-    # return tx_list
+    return tx_list
 
 
 async def save_results(db: AsyncSession, results: List) -> None:
@@ -110,51 +114,58 @@ async def save_results(db: AsyncSession, results: List) -> None:
     Returns:
         None
     """
+    try:
+        # Concatenate all (valid) results into a single DataFrame
+        # Print errors to console
+        ok_files = []
+        ko_files = []
+        error_messages = []
+
+        col_list = ["transaction_date", "name_description", "type", "amount", "category"]
+        tx_list = pd.DataFrame(columns=col_list)
+        for result in results:
+            print(f"save_results result: {result}")
+            if not result["error"]:
+                ok_files.append(result["file_name"])
+                result_df = result["output"]
+                result_df.columns = col_list
+                tx_list = pd.concat([tx_list, result_df], ignore_index=True)
+            else:
+                ko_files.append(result["file_name"])
+                error_messages.append(f"{result['file_name']}: {result['error']}")
+
+        # Save to database
+        # FIXME: get user_id from session
+        txn_list_to_save = [TransactionCreate(**row.to_dict(), user_id=1) for _, row in tx_list.iterrows()]
+        await Transaction.bulk_create(db, txn_list_to_save)
+
+        new_ref_data = tx_list[["name_description", "category"]]
+
+        if not os.path.exists(CATEGORY_REFERENCE_OUTPUT_FILE):
+            os.makedirs(CATEGORY_REFERENCE_OUTPUT_FILE)
+
+        if os.path.exists(CATEGORY_REFERENCE_OUTPUT_FILE):
+            # If it exists, add master file to interim results
+            old_ref_data = pd.read_csv(CATEGORY_REFERENCE_OUTPUT_FILE, names=["name_description", "category"], header=0)
+            new_ref_data = pd.concat([old_ref_data, new_ref_data], ignore_index=True)
+
+        # Drop duplicates, sort, and write to create new Master File
+        new_ref_data.drop_duplicates(subset=["name_description"]).sort_values(by=["name_description"]).to_csv(
+            CATEGORY_REFERENCE_OUTPUT_FILE, mode="w", index=False, header=True
+        )
+
+        # Summarize results
+        print(f"\nProcessed {len(results)} files: {len(ok_files)} successful, {len(ko_files)} with errors\n")
+        logging.debug(f"\nProcessed {len(results)} files: {len(ok_files)} successful, {len(ko_files)} with errors\n")
+        if len(ko_files):
+            print(f"Errors in the following files:")
+            for message in error_messages:
+                print(f"  {message}")
+            print("\n")
 
-    # Concatenate all (valid) results into a single DataFrame
-    # Print errors to console
-    ok_files = []
-    ko_files = []
-    error_messages = []
-
-    col_list = ["transaction_date", "type", "category", "name_description", "amount"]
-    tx_list = pd.DataFrame(columns=col_list)
-    for result in results:
-        print(f"save_results result: {result}")
-        if not result["error"]:
-            ok_files.append(result["file_name"])
-            result_df = result["output"]
-            result_df.columns = col_list
-            tx_list = pd.concat([tx_list, result_df], ignore_index=True)
-        else:
-            ko_files.append(result["file_name"])
-            error_messages.append(f"{result['file_name']}: {result['error']}")
-
-    # Save to database
-    # FIXME: get user_id from session
-    txn_list_to_save = [TransactionCreate(**row.to_dict(), user_id=1) for _, row in tx_list.iterrows()]
-    await Transaction.bulk_create(db, txn_list_to_save)
-
-    new_ref_data = tx_list[["name_description", "category"]]
-
-    if not os.path.exists(CATEGORY_REFERENCE_OUTPUT_FILE):
-        os.makedirs(CATEGORY_REFERENCE_OUTPUT_FILE)
-
-    if os.path.exists(CATEGORY_REFERENCE_OUTPUT_FILE):
-        # If it exists, add master file to interim results
-        old_ref_data = pd.read_csv(CATEGORY_REFERENCE_OUTPUT_FILE, names=["name_description", "category"], header=0)
-        new_ref_data = pd.concat([old_ref_data, new_ref_data], ignore_index=True)
-
-    # Drop duplicates, sort, and write to create new Master File
-    new_ref_data.drop_duplicates(subset=["name_description"]).sort_values(by=["name_description"]).to_csv(
-        CATEGORY_REFERENCE_OUTPUT_FILE, mode="w", index=False, header=True
-    )
-
-    # Summarize results
-    print(f"\nProcessed {len(results)} files: {len(ok_files)} successful, {len(ko_files)} with errors\n")
-    logging.debug(f"\nProcessed {len(results)} files: {len(ok_files)} successful, {len(ko_files)} with errors\n")
-    if len(ko_files):
-        print(f"Errors in the following files:")
-        for message in error_messages:
-            print(f"  {message}")
-        print("\n")
+    except Exception as e:
+        # Return an error indicator and exception info
+        # 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}")