import re import os from datetime import datetime import openai from google.cloud import firestore from dotenv import load_dotenv # Make API connection load_dotenv() # gemini_api_key = os.environ['Gemini'] o_api_key = os.getenv("openai_api_key") openai.api_key = o_api_key # Authenticate to Firesotre with the JSON account key db = firestore.Client.from_service_account_json("firestore-key.json") # Generate AI response from user input def generateResponse(prompt): #----- Call API to classify and extract relevant transaction information # These templates help provide a unified response format for use as context clues when # parsing the AI generated response into a structured data format relevant_info_template = """ Intent: The CRUD operation Transaction Type: The type of transaction Details: as a sublist of the key details like name of item, amount, description, among other details you are able to extract. """ sample_response_template = """ The information provided indicates that you want to **create/record** a new transaction. **Extracted Information**: **Intent**: Create Transaction 1: **Transaction Type**: Purchase **Details**: - Item: Car - Purpose: Business - Amount: 1000 - Tax: 200 - Note: A new car for business Transaction 2: **Transaction Type**: Expense **Details**: - Item: Office Chair - Amount: 300 USD - Category: Furniture """ response = openai.chat.completions.create( model="gpt-4o", messages=[ {"role": "system", "content": f"You are a helpful assistant that classifies transactions written in natural language into CRUD operations (Create, Read, Update, and Delete) and extracts relevant information. Format the relevant information extracted from the transaction text in this format: {relevant_info_template}. You can use markdown syntax to present a nicely formated and readable response to the user, but make sure the user does not see the markdown keyword. Keywords and field names must be in bold face. A sample response could look like this: {sample_response_template}. Delineate multiple transactions with the label 'Transaction 1' before the start of the relevant information for each transaction. There should be only one intent even in the case of multiple transactions."}, {"role": "user", "content": prompt} ] ) #----- Process response try: response = response.choices[0].message.content except Exception as e: print(f'An error occurred: {str(e)}') response = None return response # TODO Handle multiple multiple transactions from one user input. def parse_ai_response(response_text): # Initialize the structured data dictionary data = { "intent": None, "transaction_type": None, "details": {}, "created_at": datetime.now().isoformat() # Add current date and time } # Extract the intent intent_match = re.search(r"\*\*Intent\*\*:\s*(\w+)", response_text) if intent_match: data["intent"] = intent_match.group(1) # Extract the transaction type transaction_type_match = re.search(r"\*\*Transaction Type\*\*:\s*(\w+)", response_text) if transaction_type_match: data["transaction_type"] = transaction_type_match.group(1) # Extract details details = {} detail_matches = re.findall(r"-\s*([\w\s]+):\s*([\d\w\s,.]+(?:\sUSD)?)", response_text) for field, value in detail_matches: # Clean up the field name and value field = field.strip().lower().replace(" ", "_") # Convert to snake_case value = value.strip() # Convert numeric values where possible if "USD" in value: value = float(value.replace(" USD", "").replace(",", "")) elif value.isdigit(): value = int(value) details[field] = value # Store details in the structured data data["details"] = details return data def parse_multiple_transactions(response_text): transactions = [] # Split the response into transaction sections based on keyword 'Transaction #' transaction_sections = re.split(r"Transaction \d+:", response_text, flags=re.IGNORECASE) # Adjust regex to handle variations in "Transaction X:" # transaction_sections = re.split(r"(?i)(?<=\n)transaction\s+\d+:", response_text) transaction_sections = [section.strip() for section in transaction_sections if section.strip()] # Remove the first section if it's not a valid transaction if not re.search(r"\*\*Transaction Type\*\*", transaction_sections[0], re.IGNORECASE): transaction_sections.pop(0) print(len(transaction_sections)) print(transaction_sections) # Extract intent: with support for a single intent per user prompt intent_match = re.search(r"\*\*Intent\*\*:\s*(\w+)", response_text) if intent_match: intent = intent_match.group(1) for section in transaction_sections: # Initialize transaction data transaction_data = { "intent": intent, # global intent "transaction_type": None, "details": {}, "created_at": datetime.now().isoformat() } # transaction_data["intent"] = intent # Extract transaction type transaction_type_match = re.search(r"\*\*Transaction Type\*\*:\s*(\w+)", section) if transaction_type_match: transaction_data["transaction_type"] = transaction_type_match.group(1) # Extract details details = {} detail_matches = re.findall(r"-\s*([\w\s]+):\s*([\d\w\s,.]+(?:\sUSD)?)", section) for field, value in detail_matches: field = field.strip().lower().replace(" ", "_") # Convert to snake_case value = value.strip() # Convert numeric values where possible if "USD" in value: value = float(value.replace(" USD", "").replace(",", "")) elif value.isdigit(): value = int(value) details[field] = value transaction_data["details"] = details transactions.append(transaction_data) return transactions def create_transaction(user_phone, transaction_data): for transaction in transaction_data: doc_ref = db.collection("users").document(user_phone).collection("transactions").document() # transaction_data['transaction_id'] # let's default to random generated document ids doc_ref.set(transaction) # print("Transaction created successfully!") return True # Update logic: # - def update_transaction(user_phone, transaction_id, update_data): doc_ref = db.collection("users").document(user_phone).collection("transactions").document(transaction_id) doc_ref.update(update_data) # print("Transaction updated successfully!") return True def fetch_transaction(user_phone, transaction_id=None): if transaction_id: doc_ref = db.collection("users").document(user_phone).collection("transactions").document(transaction_id) transaction = doc_ref.get() if transaction.exists: return transaction.to_dict() else: # print("Transaction not found.") return None else: collection_ref = db.collection("users").document(user_phone).collection("transactions") transactions = [doc.to_dict() for doc in collection_ref.stream()] return transactions # Delete fields or an entire transaction document. # Delete specific fields def delete_transaction_fields(user_phone, transaction_id, fields_to_delete): doc_ref = db.collection("users").document(user_phone).collection("transactions").document(transaction_id) updates = {field: firestore.DELETE_FIELD for field in fields_to_delete} doc_ref.update(updates) print("Fields deleted successfully!") # Delete an entire transaction def delete_transaction(user_phone, transaction_id): doc_ref = db.collection("users").document(user_phone).collection("transactions").document(transaction_id) doc_ref.delete() print("Transaction deleted successfully!") # Example usage # response_text = """ # The information provided indicates that you want to **create/record** a new transaction. # **Extracted Information**: # **Intent**: Create # **Transaction Type**: Purchase # **Details**: # - Item: Car # - Purpose: Business # - Amount: 100000 USD # - Tax: 1000 USD # """ # parsed_data = parse_ai_response(response_text) # print(parsed_data)