Update utility.py

utility.py

@@ -6,11 +6,11 @@ from datetime import datetime, timezone
 from typing import List, Dict, Union, Optional, Any
 from google.cloud import firestore
 import pandas as pd
-import inflect  # For robust pluralization
-from thefuzz import process as fuzzy_process  # For fuzzy string matching
+import inflect
+from thefuzz import process as fuzzy_process
 from pandasai import SmartDatalake
 from pandasai.responses.response_parser import ResponseParser
-from pandasai.exceptions import NoCodeFoundError # For specific error handling
+from pandasai.exceptions import NoCodeFoundError
 from langchain_google_genai import ChatGoogleGenerativeAI
 import google.generativeai as genai
 import re
@@ -108,27 +108,23 @@ def generateResponse(prompt: str) -> str:
 
     system_prompt = """
     Analyze the user's request for business transaction management. Your goal is to extract structured information about one or more transactions and output it as a valid JSON list.
-
-**1. Output Format:**
-You MUST output your response as a valid JSON list `[]` containing one or more transaction objects `{}`.
-
-**2. Transaction Object Structure:**
-Each transaction object MUST have the following keys:
-- `"intent"`: The user's goal (e.g., "create", "read", "update", "delete").
-- `"transaction_type"`: The category of the transaction (e.g., "sale", "purchase", "inventory", "expense", "asset", "liability", "query", "service_offering").
-- `"details"`: An object containing key-value pairs extracted from the request.
-
-**3. Key Naming Conventions for the `details` Object:**
-- **For Expenses:** Use `"amount"`, `"description"`, and `"category"`.
-- **For Assets:** Use `"value"` for the monetary worth and `"name"` for the item's name.
-- **For Liabilities:** Use `"amount"` and `"creditor"`.
-- **For Sales/Inventory:** Use `"item"`, `"quantity"`, and `"price"`.
-- **For all financial transactions:** If a currency symbol or code is present (e.g., $, £, €, ZAR, R), include a `"currency"` key.
-
-**4. Important Rules:**
-- **Rule for Queries:** For "read" intents or general questions, set `transaction_type` to "query" and the `details` object MUST contain a single key `"query"` with the user's full, original question as the value.
-- **Rule for Multiple Items:** If the user's request contains multiple distinct transactions (e.g., recording an expense AND an asset), create a separate JSON object for each one within the main list.
-- **Rule for Expense Normalization:** For "create" intents with `transaction_type` "expense", analyze the `description`. If it contains common keywords, normalize it to a single word. For example, if the description is "paid for fuel for the delivery truck", the normalized `description` in the JSON should be "fuel". If it's "office electricity bill", normalize it to "electricity".
+    **1. Output Format:**
+    You MUST output your response as a valid JSON list `[]` containing one or more transaction objects `{}`.
+    **2. Transaction Object Structure:**
+    Each transaction object MUST have the following keys:
+    - `"intent"`: The user's goal (e.g., "create", "read", "update", "delete").
+    - `"transaction_type"`: The category of the transaction (e.g., "sale", "purchase", "inventory", "expense", "asset", "liability", "query", "service_offering").
+    - `"details"`: An object containing key-value pairs extracted from the request.
+    **3. Key Naming Conventions for the `details` Object:**
+    - **For Expenses:** Use `"amount"`, `"description"`, and `"category"`.
+    - **For Assets:** Use `"value"` for the monetary worth and `"name"` for the item's name.
+    - **For Liabilities:** Use `"amount"` and `"creditor"`.
+    - **For Sales/Inventory:** Use `"item"`, `"quantity"`, and `"price"`. For services sold by time, use `"hours"`.
+    - **For all financial transactions:** If a currency symbol or code is present (e.g., $, £, €, ZAR, R), include a `"currency"` key.
+    **4. Important Rules:**
+    - **Rule for Queries:** For "read" intents or general questions, set `transaction_type` to "query" and the `details` object MUST contain a single key `"query"` with the user's full, original question as the value.
+    - **Rule for Multiple Items:** If the user's request contains multiple distinct transactions, create a separate JSON object for each one.
+    - **Rule for Expense Normalization:** For "create" intents with `transaction_type` "expense", analyze the `description`. If it contains common keywords, normalize it to a single word. For example, "paid for fuel for the delivery truck" becomes "fuel".
     """
     try:
         full_prompt = [system_prompt, prompt]
@@ -172,33 +168,25 @@ def _get_canonical_info(user_phone: str, item_name: str) -> Dict[str, Any]:
     Finds the canonical version of an item using fuzzy matching for existing items
     and inflect for new ones.
     """
-    # --- CHANGE 1: Fuzzy Search and Robust Pluralization ---
     inventory_ref = db.collection("users").document(user_phone).collection("inventory_and_services")
     name_lower = item_name.lower().strip()
 
-    # 1. Fetch all existing item names for fuzzy matching
     all_item_docs = list(inventory_ref.stream())
     all_item_names = [doc.id for doc in all_item_docs]
 
     if all_item_names:
-        # 2. Find the best match using fuzzy logic
         best_match = fuzzy_process.extractOne(name_lower, all_item_names)
-        
-        # 3. Apply a strict threshold
         if best_match and best_match[1] >= 90:
             matched_name = best_match[0]
-            # Find the corresponding document
             for doc in all_item_docs:
                 if doc.id == matched_name:
                     return {'doc': doc, 'name': matched_name}
 
-    # 4. If no good match is found, create a clean singular name for a new item
     singular = p.singular_noun(name_lower)
     if not singular:
         singular = name_lower
     
     return {'doc': None, 'name': singular}
-    # --- END OF CHANGE 1 ---
 
 
 def create_or_update_inventory_or_service_offering(user_phone: str, transaction_data: List[Dict]) -> tuple[bool, str]:
@@ -232,7 +220,7 @@ def create_or_update_inventory_or_service_offering(user_phone: str, transaction_
         doc_ref = inventory_ref.document(canonical_name)
         doc_data = {
             'details': {**details, change_key: firestore.Increment(change_amount)},
-            'type': 'service' if 'service_name' in details else 'good',
+            'type': 'service' if 'service_name' in details or 'hours' in details else 'good',
             'last_updated': datetime.now(timezone.utc).isoformat(),
         }
         
@@ -253,7 +241,7 @@ def create_or_update_inventory_or_service_offering(user_phone: str, transaction_
 
 def create_sale(user_phone: str, transaction_data: List[Dict]) -> tuple[bool, str]:
     """
-    Process sales with fuzzy name matching, user price override, and service bypass.
+    Process sales with fuzzy name matching, user price override, and on-the-fly service creation.
     """
     feedback_messages = []
     any_success = False
@@ -269,10 +257,7 @@ def create_sale(user_phone: str, transaction_data: List[Dict]) -> tuple[bool, st
         try:
             canonical_info = _get_canonical_info(user_phone, item_name)
             canonical_name = canonical_info['name']
-            inventory_doc = canonical_info['doc']
-            inventory_data = inventory_doc.to_dict() if inventory_doc else None
-            item_type = inventory_data.get('type') if inventory_data else ('service' if 'service_name' in details else 'good')
-
+            
             last_selling_price = None
             sales_ref = db.collection("users").document(user_phone).collection("sales")
             all_sales_query = sales_ref.where('details.item', '==', canonical_name)
@@ -285,14 +270,16 @@ def create_sale(user_phone: str, transaction_data: List[Dict]) -> tuple[bool, st
 
             @firestore.transactional
             def process_one_sale(transaction, sale_details):
+                is_new_item = canonical_info['doc'] is None
+                # Infer type for new items based on details like 'hours'
+                item_type = 'service' if is_new_item and 'hours' in sale_details else 'good'
+                
                 user_price = sale_details.get('price') or sale_details.get('unit_price')
                 
                 if user_price is not None:
                     selling_price = user_price
-                    logger.info(f"Using user-specified price for '{canonical_name}': {selling_price}")
                 elif last_selling_price is not None:
                     selling_price = last_selling_price
-                    logger.info(f"Using last known price for '{canonical_name}': {selling_price}")
                 else:
                     return f"Sale failed for new item '{canonical_name}': You must specify a price for the first sale."
 
@@ -325,12 +312,17 @@ def create_sale(user_phone: str, transaction_data: List[Dict]) -> tuple[bool, st
                         if current_stock < quantity_sold:
                             return f"Sale failed for '{canonical_name}': Insufficient stock (Have: {current_stock}, Need: {quantity_sold})."
                         
-                        transaction.update(item_doc_ref, {
-                            f'details.{stock_key}': firestore.Increment(-quantity_sold),
-                            'last_updated': datetime.now(timezone.utc).isoformat()
-                        })
+                        transaction.update(item_doc_ref, {f'details.{stock_key}': firestore.Increment(-quantity_sold)})
                 elif item_type == 'good':
                     return f"Sale failed for '{canonical_name}': Item not found in inventory. Please add it first."
+                elif is_new_item and item_type == 'service':
+                    logger.info(f"Creating new service '{canonical_name}' during sale.")
+                    service_record = {
+                        'details': {'item': canonical_name, 'price': selling_price},
+                        'type': 'service',
+                        'last_updated': datetime.now(timezone.utc).isoformat()
+                    }
+                    transaction.set(item_doc_ref, service_record)
 
                 sale_doc_ref = sales_ref.document()
                 sale_record = {
@@ -355,7 +347,6 @@ def create_sale(user_phone: str, transaction_data: List[Dict]) -> tuple[bool, st
 
 
 def create_expense(user_phone: str, transaction_data: List[Dict]) -> tuple[bool, str]:
-    """Create new expense records in Firestore."""
     batch = db.batch()
     expenses_ref = db.collection("users").document(user_phone).collection("expenses")
     success_count = 0
@@ -390,7 +381,6 @@ def create_expense(user_phone: str, transaction_data: List[Dict]) -> tuple[bool,
 
 
 def create_asset(user_phone: str, transaction_data: List[Dict]) -> tuple[bool, str]:
-    """Create new asset records in Firestore with data stored in a 'details' sub-document."""
     batch = db.batch()
     assets_ref = db.collection("users").document(user_phone).collection("assets")
     success_count = 0
@@ -425,7 +415,6 @@ def create_asset(user_phone: str, transaction_data: List[Dict]) -> tuple[bool, s
 
 
 def create_liability(user_phone: str, transaction_data: List[Dict]) -> tuple[bool, str]:
-    """Create new liability records in Firestore."""
     batch = db.batch()
     liabilities_ref = db.collection("users").document(user_phone).collection("liabilities")
     success_count = 0
@@ -469,43 +458,33 @@ def _validate_dataframe(df: pd.DataFrame) -> pd.DataFrame:
     if df.empty:
         return df
 
-    # --- CHANGE 2: Robust Data Validation ---
-    # 1. Validate and convert timestamp columns to a consistent UTC format
     for col in ['timestamp', 'created_at', 'last_updated', 'acquisition_date', 'due_date']:
         if col in df.columns:
-            # The key fix: utc=True handles mixed timezone-aware/naive data
             df[col] = pd.to_datetime(df[col], errors='coerce', utc=True)
 
-    # 2. Validate and convert numeric columns
     numeric_cols = ['price', 'unit_price', 'quantity', 'amount', 'value', 'cost', 'hours', 'units_available']
     for col in numeric_cols:
         if col in df.columns:
             df[col] = pd.to_numeric(df[col], errors='coerce').fillna(0)
 
-    # 3. Validate and fill missing categorical/object columns
     for col in df.select_dtypes(include=['object']).columns:
         df[col] = df[col].fillna('Unknown')
     
     return df
-    # --- END OF CHANGE 2 ---
 
 
 def _fetch_all_collections_as_dfs(user_phone: str) -> List[pd.DataFrame]:
     """
-    Fetches all user data, splits inventory/services, validates, and returns DataFrames.
+    Fetches all user data, splits/validates DataFrames, and engineers features.
     """
-    collections = ['sales', 'expenses', 'assets', 'liabilities']
     all_dfs = []
 
-    # Handle inventory and services separately
     inv_serv_docs = db.collection("users").document(user_phone).collection('inventory_and_services').stream()
-    inventory_data = []
-    services_data = []
+    inventory_data, services_data = [], []
     for doc in inv_serv_docs:
         doc_data = doc.to_dict()
         flat_data = {**doc_data, **doc_data.get('details', {})}
         if 'details' in flat_data: del flat_data['details']
-        
         if doc_data.get('type') == 'service':
             services_data.append(flat_data)
         else:
@@ -521,23 +500,28 @@ def _fetch_all_collections_as_dfs(user_phone: str) -> List[pd.DataFrame]:
         services_df.name = "services"
         all_dfs.append(_validate_dataframe(services_df))
 
-    # Handle other collections
-    for coll_name in collections:
+    collections_to_fetch = {'sales': 'sales', 'expenses': 'expenses', 'assets': 'assets', 'liabilities': 'liabilities'}
+    for df_name, coll_name in collections_to_fetch.items():
         docs = db.collection("users").document(user_phone).collection(coll_name).stream()
-        data = []
-        for doc in docs:
-            doc_data = doc.to_dict()
-            if 'details' in doc_data and isinstance(doc_data['details'], dict):
-                flat_data = {**doc_data, **doc_data['details']}
-                del flat_data['details']
-                data.append(flat_data)
-            else:
-                data.append(doc_data)
+        data = [doc.to_dict() for doc in docs]
         
         if data:
-            df = pd.DataFrame(data)
-            df.name = coll_name
-            all_dfs.append(_validate_dataframe(df))
+            flat_data_list = []
+            for item in data:
+                flat_item = {**item, **item.get('details', {})}
+                if 'details' in flat_item: del flat_item['details']
+                flat_data_list.append(flat_item)
+
+            df = pd.DataFrame(flat_data_list)
+            df.name = df_name
+            
+            validated_df = _validate_dataframe(df)
+            
+            if df_name == 'sales':
+                if 'price' in validated_df.columns and 'quantity' in validated_df.columns:
+                    validated_df['sale_total'] = validated_df['price'] * validated_df['quantity']
+            
+            all_dfs.append(validated_df)
             
     return all_dfs
 
@@ -587,7 +571,6 @@ def read_datalake(user_phone: str, query: str) -> str:
 
 
 def _find_document_by_details(user_phone: str, collection_name: str, details: Dict) -> Optional[Any]:
-    """Helper to find a single document in a collection matching specific details."""
     col_ref = db.collection("users").document(user_phone).collection(collection_name)
     
     if 'transaction_id' in details and details['transaction_id']:
@@ -624,7 +607,6 @@ def _find_document_by_details(user_phone: str, collection_name: str, details: Di
 
 
 def update_transaction(user_phone: str, transaction_data: List[Dict]) -> tuple[bool, str]:
-    """Update existing transaction(s) based on provided data."""
     feedback = []
     any_success = False
     for trans in transaction_data:
@@ -669,7 +651,6 @@ def update_transaction(user_phone: str, transaction_data: List[Dict]) -> tuple[b
 
 
 def delete_transaction(user_phone: str, transaction_data: List[Dict]) -> tuple[bool, str]:
-    """Delete specified transactions."""
     feedback = []
     any_success = False
     for trans in transaction_data:
@@ -706,7 +687,6 @@ def delete_transaction(user_phone: str, transaction_data: List[Dict]) -> tuple[b
 
 
 def persist_temporary_transaction(transactions: List[Dict], mobile: str) -> bool:
-    """Store transactions temporarily in Firestore before confirmation."""
     if not transactions: return False
     try:
         doc_ref = db.collection("users").document(mobile).collection("temp_transactions").document("pending")
@@ -719,7 +699,6 @@ def persist_temporary_transaction(transactions: List[Dict], mobile: str) -> bool
 CURRENCY_SYMBOL_REGEX = re.compile(r"^\s*[\$\£\€\¥\₹R]")
 
 def format_transaction_response(transactions: Union[List[Dict], Dict, None]) -> str:
-    """Format transaction data for user display, looking inside the 'details' field."""
     if not transactions: return "No transaction data to display."
     if isinstance(transactions, dict): transactions = [transactions]
 
@@ -757,7 +736,6 @@ def format_transaction_response(transactions: Union[List[Dict], Dict, None]) ->
 
 
 def fetch_transaction(user_phone: str, identifier: str, collection: str = "inventory_and_services"):
-    """Retrieve a specific transaction from Firestore."""
     try:
         doc = db.collection("users").document(user_phone).collection(collection).document(identifier).get()
         if doc.exists: return doc.to_dict()
@@ -768,9 +746,6 @@ def fetch_transaction(user_phone: str, identifier: str, collection: str = "inven
 
 
 def process_intent(parsed_trans_data: List[Dict], mobile: str) -> str:
-    """
-    Groups transactions by intent and type, processes each group, and returns a consolidated feedback report.
-    """
     if not parsed_trans_data:
         return "I couldn't understand the transaction details. Could you please try again?"