Create utility.py

utility.py

@@ -0,0 +1,220 @@
+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)