import os import glob import shutil import logging import warnings from typing import Optional, Union, Dict, List from datetime import datetime import pandas as pd from dateparser import parse from fastapi import HTTPException import csv from app.categorization.categorizer_list import categorize_list from app.categorization.config import RESULT_OUTPUT_FILE, CATEGORY_REFERENCE_OUTPUT_FILE from app.model.transaction import Transaction from app.schema.index import TransactionCreate from sqlalchemy.ext.asyncio import AsyncSession # Read file and process it (e.g. categorize transactions) 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. Args: file_path (str): Path to the input file. df (dataframe): Dataframe representing input file Returns: Dict[str, Union[str, pd.DataFrame]]: Dictionary containing the file name, processed output, and error information if any """ file_name = os.path.basename(file_path) result = {'file_path': file_name, 'output': pd.DataFrame(), 'error': ''} try: # Read file into standardized tx format: transaction_date, name_description, type, amount, category, source tx_list = standardize_csv_file(file_path, df) result["columns"] = tx_list.columns.tolist() # Categorize transactions result["output"] = await categorize_list(tx_list) # print(f"File processed sucessfully: {file_name}") except Exception as e: # Return an error indicator and exception info logging.debug(logging.ERROR, f"| File: {file_name} | Unexpected Error: {e}") result["error"] = str(e) raise HTTPException(status_code = 500, detail=f"\n\n process_file result: \n{result}") return result def standardize_csv_file(file_path: str, d: pd.DataFrame) -> pd.DataFrame: """ Read and prepare the data from the input file. Args: file_path (str): Path to the input file. Returns: pd.DataFrame: Prepared transaction data. """ 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) # 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) # 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.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 logging.debug("standardize_csv_file Error: {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}") async def save_results(db: AsyncSession, results: List) -> Dict[str, Union[List[str], List[TransactionCreate], pd.DataFrame]]: """ Merge all interim results in the input folder and write the merged results to the output file. Args: in_folder (str): Path to the input folder containing interim results. out_file (str): Path to the output file. Returns: None """ result = {'ok_files': '', 'ko_files': '', 'error_messages': '', 'txn_list_to_save': '', 'old_ref_data': '', 'new_ref_data': ''} try: # Concatenate all (valid) results into a single DataFrame # Print errors to console ok_files = [] ko_files = [] error_messages = [] # print(f"save_results results: {results}") col_list = ["transaction_date", "name_description", "type", "amount", "category"] tx_list = pd.DataFrame(columns=col_list) # print(f"save_results tx_list: {tx_list}") for result in results: if not result["error"]: ok_files.append(result["file_name"]) print(f"save_results ok_files: {ok_files}\n") result_df = result["output"] print(f"save_results result_df: {result_df}\n") result_df.columns = col_list print(f"save_results result_df.columns: {result_df.columns}\n") tx_list = pd.concat([tx_list, result_df], ignore_index=True) print(f"save_results tx_list: {tx_list}\n") else: ko_files.append(result["file_name"]) print(f"save_results ko_files: {ko_files}\n") error_messages.append(f"{result['file_name']}: {result['error']}") print(f"save_results error_messages: {error_messages}\n") result['ok_files'] = ok_files result['ko_files'] = ko_files result['error_messages'] = error_messages # 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()] result['txn_list_to_save'] = txn_list_to_save print(f"save_results txn_list_to_save: {txn_list_to_save}\n") 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) print(f"save_results new_ref_data: {new_ref_data}\n") print(f"save_results old_ref_data: {old_ref_data}\n") # 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 ) result['old_ref_data'] = old_ref_data result['new_ref_data'] = new_ref_data print(f"\nsave_results result:\n {result}\n") # 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: {error_messages}") return result