utility.py

import re
import os
from datetime import datetime
import openai
from google.cloud import firestore
from dotenv import load_dotenv
from pandasai import SmartDatalake
from pandasai import Agent
from pandasai.responses.response_parser import  ResponseParser
import pandas as pd
from pandasai.llm import OpenAI
#from langchain.llms.sambanova import Sambaverse
#from langchain_community.llms.sambanova import Sambaverse
from langchain_community.chat_models.sambanova import ChatSambaNovaCloud


# 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")

client = openai.OpenAI(
    api_key=os.environ.get("sambanova_api_key"),
    base_url="https://api.sambanova.ai/v1",
)




sambaverse_api_key = os.environ.get("SAMBAVERSE_API_KEY")


llm = ChatSambaNovaCloud(
    model="Meta-Llama-3.1-70B-Instruct",
    max_tokens=1024,
    temperature=0.7,
    top_k=1,
    top_p=0.01,
)


print(llm.invoke("Why should I use open source models?"))



class FlaskResponse(ResponseParser):
  def __init__(self, context) -> None:
      super().__init__(context)

  def format_dataframe(self, result):
      return result['value'].to_html()

  def format_plot(self, result):
        # Save the plot using savefig
      try:
          
          img_path = result['value']
      
        
      except ValueError:
          img_path = str(result['value'])
          print("value error!", img_path)
        
      print("response_class_path:", img_path)
      return img_path

  def format_other(self, result):
      return str(result['value'])
      
# Generate AI response from user input
def generateResponse(prompt,model='Meta-Llama-3.1-70B-Instruct'):
  #----- 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, one of create, read, update, or delete
          Transaction Type: The type of transaction such as purchases, sales
          Details: as a sublist of the key details like name of item, quantity, cost price, currency, unit for the quantity, description, among other details you are able to extract. 
      """
    sample_single_transaction_template = """
      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
              - Quantity: 1
              - Unit: None
              - Cost: 10000
              - Tax: 200
              - Currency: USD
              - Note: A new car for business
              """
    
    sample_multi_transaction_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
              - Quantity: 1
              - Unit: None
              - Cost: 10000
              - Tax: 200
              - Currency: USD
              - Note: A new car for business

      Transaction 2:

      **Transaction Type**: Expense  
      **Details**:  
            - Item: Office Chair  
            - Quantity: 2
            - Unit: None
            - Cost: 300 
            - Currency: USD  
            - Category: Furniture  
      """
    response = client.chat.completions.create(
        model = model,
        # 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. You should be able to recognize the currency being used and any quantity units into separate fields. Format the relevant information extracted from the transaction text in this format: {relevant_info_template}. 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 for a single transaction could look like this: {sample_single_transaction_template}, while multiple transactions could look like this: {sample_multi_transaction_template}. 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


def parse_value(value):
    """
    Parses a value string into the appropriate data type and detects currency.
    Handles various currencies, percentages, numbers, and text.
    """
    value = value.strip()
    try:
        # Match currency codes or symbols dynamically
        currency_match = re.search(r"([A-Z]{3}|\$|€|£)", value)
        currency = currency_match.group(1) if currency_match else None

        # Remove currency symbols or codes for numeric conversion
        cleaned_value = re.sub(r"[A-Z]{3}|\$|€|£", "", value).replace(",", "").strip()

        # Handle percentages
        if "%" in cleaned_value:
            return float(cleaned_value.replace("%", "")), currency
        # Handle plain numbers (integers or floats)
        elif cleaned_value.replace(".", "", 1).isdigit():
            return float(cleaned_value) if "." in cleaned_value else int(cleaned_value), currency
        # Return as text if no numeric parsing is possible
        return value, currency
    except ValueError:
        # Fallback to original value if parsing fails
        return value, None

def extract_transaction_details(text):
    """
    Extracts transaction details from a given text input.
    Handles both bold (**field**) and non-bold (field) formats.
    """
    details = {}
    transaction_currency = None  # Default currency field

    # Regex to match key-value pairs
    detail_matches = re.findall(
        r"-\s*\*{0,2}([\w\s]+)\*{0,2}:\s*([\w\s,.$%-]+?)(?:\s*[\n]|$)",  # Stop matching before newline or end of string
        text,
        re.DOTALL
    )
    # print("Detail matches:", detail_matches)  # Debugging

    for field, value in detail_matches:
        # Standardize the field name (convert to snake_case)
        field = field.strip().lower().replace(" ", "_")
        
        # Parse the value and dynamically detect currency
        parsed_value, detected_currency = parse_value(value)
        if detected_currency and not transaction_currency:
            transaction_currency = detected_currency  # Set the transaction-level currency if not already set
        
        details[field] = parsed_value

    # Add currency as a separate field
    if transaction_currency:
        details["currency"] = transaction_currency

    return details


# Parsing single transactions
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)
    
    # Store details in the structured data
    data["details"] = extract_transaction_details(response_text)
    
    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)
    
    # 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()
        }

        # 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
        transaction_data["details"] = extract_transaction_details(section)

        transactions.append(transaction_data)

    return transactions

def read_datalake(user_phone, user_question):
    inventory_ref = db.collection("users").document(user_phone).collection("inventory")

    sales_ref = db.collection("users").document(user_phone).collection('sales')

    inventory_list = [doc.to_dict() for doc in inventory_ref.stream()]
    sales_list = [doc.to_dict() for doc in sales_ref.stream()]

    inventory_df = pd.DataFrame(inventory_list)
    sales_df = pd.DataFrame(sales_list)
    

    lake = SmartDatalake([inventory_df, sales_df], config={"llm": llm, "response_parser": FlaskResponse, "enable_cache": False, "save_logs": False})
    response = lake.chat(user_question)

    return response

def create_inventory(user_phone, transaction_data):
    for transaction in transaction_data:
        item_name = transaction['details']['item'] # assumes unique item name
        doc_ref = db.collection("users").document(user_phone).collection("inventory").document(item_name)
        # transaction_data['transaction_id'] # let's default to random generated document ids
        doc_ref.set(transaction)
        # print("Transaction created successfully!")
    return True

def create_sale(user_phone, transaction_data):
    
    for transaction in transaction_data:
        item_name = transaction['details']['item'] # assumes item names are unique per transactions
        # fetch the inventory
        print(item_name, '\n')
        print(transaction)
        inventory = fetch_transaction(user_phone, item_name)
        
        # Do the sales calculations
        new_stock_level = inventory['details']['quantity'] - transaction['details']['quantity']
        inventory['details']['quantity'] = new_stock_level

        # update the inventory
        doc_ref = db.collection("users").document(user_phone).collection("inventory").document(item_name)
        doc_ref.set(inventory)
        
        # Create the sale
        doc_ref = db.collection("users").document(user_phone).collection("sales").document(item_name)
        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("inventory").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("inventory")
        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)