Update utility.py

utility.py

@@ -2,7 +2,7 @@
 import json
 import os
 import logging
-from datetime import datetime, timezone
+from datetime import datetime, timezone, timedelta
 from typing import List, Dict, Union, Optional, Any, Tuple
 from google.cloud import firestore
 import pandas as pd
@@ -62,7 +62,6 @@ def render_df_as_image(df: pd.DataFrame) -> Optional[str]:
         return "The data requested is empty."
     try:
         img_path = os.path.join(user_defined_path, f"report_{uuid.uuid4()}.png")
-        # Use a larger figure size for better readability on mobile
         dfi.export(df, img_path, table_conversion='matplotlib', dpi=200)
         return img_path
     except Exception as e:
@@ -75,7 +74,7 @@ class FlaskResponse(ResponseParser):
         super().__init__(context)
 
     def format_dataframe(self, result):
-        # SURGICALLY MODIFIED: Intercept the dataframe and convert it to an image
+        # MODIFIED: Universal DataFrame to image rendering pipeline
         df = result['value']
         return render_df_as_image(df)
 
@@ -118,13 +117,56 @@ except Exception as e:
     llm = None
 
 
+# --- NEW FEATURE: Programmatic Report Engine ---
+class ReportEngine:
+    def __init__(self, dfs_with_names: List[Tuple[str, pd.DataFrame]]):
+        self.dfs = {name: df for name, df in dfs_with_names}
+        self.now = datetime.now(timezone.utc)
+        self.kpis = {}
+
+    def _get_time_boundaries(self):
+        today_start = self.now.replace(hour=0, minute=0, second=0, microsecond=0)
+        week_start = today_start - timedelta(days=self.now.weekday())
+        month_start = today_start.replace(day=1)
+        return {"today": today_start, "this_week": week_start, "this_month": month_start}
+
+    def calculate_all_kpis(self):
+        boundaries = self._get_time_boundaries()
+        sales_df = self.dfs.get('sales')
+
+        if sales_df is not None and not sales_df.empty:
+            sales_df['timestamp'] = pd.to_datetime(sales_df['timestamp'], errors='coerce', utc=True)
+            
+            # Overall KPIs
+            self.kpis['overall_revenue'] = sales_df['sale_total'].sum()
+            self.kpis['overall_sales_count'] = len(sales_df)
+            
+            # Time-based KPIs
+            for period, start_date in boundaries.items():
+                mask = sales_df['timestamp'] >= start_date
+                period_sales = sales_df.loc[mask]
+                self.kpis[f'revenue_{period}'] = period_sales['sale_total'].sum()
+                self.kpis[f'sales_count_{period}'] = len(period_sales)
+        return self
+
+    def format_kpis_for_llm(self) -> str:
+        if not self.kpis:
+            return "No sales data available to generate KPIs."
+        
+        return f"""
+        *Business KPIs (as of {self.now.strftime('%Y-%m-%d')})*
+        - *Revenue Today:* ${self.kpis.get('revenue_today', 0):.2f} from {self.kpis.get('sales_count_today', 0)} sales.
+        - *Revenue This Week:* ${self.kpis.get('revenue_this_week', 0):.2f} from {self.kpis.get('sales_count_this_week', 0)} sales.
+        - *Revenue This Month:* ${self.kpis.get('revenue_this_month', 0):.2f} from {self.kpis.get('sales_count_this_month', 0)} sales.
+        - *Total Revenue All-Time:* ${self.kpis.get('overall_revenue', 0):.2f} from {self.kpis.get('overall_sales_count', 0)} sales.
+        """
+
+# --- End of Report Engine ---
+
 def generateResponse(prompt: str) -> str:
     """Generate structured JSON response from user input using Generative AI."""
     if not model:
         return '{"error": "Model not available"}'
-
-    # --- CORRECTED SYSTEM PROMPT ---
-    # This is the restored, comprehensive prompt that correctly classifies multiple transaction types.
     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.
 
@@ -242,13 +284,11 @@ def _get_canonical_info(user_phone: str, item_name: str) -> Dict[str, Any]:
     all_item_names = [doc.id for doc in all_item_docs]
 
     if all_item_names:
-        # 1. Exact Match First (to protect good data)
         if name_lower in all_item_names:
             for doc in all_item_docs:
                 if doc.id == name_lower:
                     return {'doc': doc, 'name': name_lower}
 
-        # 2. Fuzzy Match as a Fallback (for typos and history)
         best_match = fuzzy_process.extractOne(name_lower, all_item_names)
         if best_match and best_match[1] >= 90:
             matched_name = best_match[0]
@@ -256,7 +296,6 @@ def _get_canonical_info(user_phone: str, item_name: str) -> Dict[str, Any]:
                 if doc.id == matched_name:
                     return {'doc': doc, 'name': matched_name}
 
-    # 3. Create New as a Last Resort
     singular = p.singular_noun(name_lower)
     if not singular:
         singular = name_lower
@@ -265,47 +304,37 @@ def _get_canonical_info(user_phone: str, item_name: str) -> Dict[str, Any]:
 
 
 def create_or_update_inventory_or_service_offering(user_phone: str, transaction_data: List[Dict]) -> tuple[bool, str]:
-    """Create/update inventory items or service offerings with name normalization."""
     batch = db.batch()
     inventory_ref = db.collection("users").document(user_phone).collection("inventory_and_services")
     feedback_messages = []
     success_count = 0
-
     for transaction in transaction_data:
         details = transaction.get('details', {})
         item_name = details.get('item') or details.get('service_name')
-        
         if not item_name:
             feedback_messages.append("Skipped: An inventory/service update was missing an item or service name.")
             continue
-
         canonical_info = _get_canonical_info(user_phone, item_name)
         canonical_name = canonical_info['name']
-        
         if 'item' in details: details['item'] = canonical_name
         if 'service_name' in details: details['service_name'] = canonical_name
-
         try:
             change_key = 'quantity' if 'quantity' in details else 'units_available'
             change_amount = int(details.get(change_key, 0))
         except (ValueError, TypeError):
             feedback_messages.append(f"Skipped '{canonical_name}': Invalid quantity or units format.")
             continue
-
         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',
             'last_updated': datetime.now(timezone.utc).isoformat(),
         }
-        
         batch.set(doc_ref, doc_data, merge=True)
         feedback_messages.append(f"Processed '{canonical_name}': change of {change_amount}.")
         success_count += 1
-
     if success_count == 0:
         return False, "\n".join(feedback_messages) if feedback_messages else "No valid inventory/service items to process."
-
     try:
         batch.commit()
         return True, "\n".join(feedback_messages)
@@ -315,44 +344,31 @@ 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 on-the-fly service creation.
-    """
     feedback_messages = []
     any_success = False
-
     for t in transaction_data:
         details = t.get('details', {})
         item_name = details.get('item') or details.get('service_name')
-        
         if not item_name:
             feedback_messages.append("Sale failed: Missing item or service name.")
             continue
-
         try:
             canonical_info = _get_canonical_info(user_phone, item_name)
             canonical_name = canonical_info['name']
-            
             last_selling_price = None
             sales_ref = db.collection("users").document(user_phone).collection("sales")
             all_sales_query = sales_ref.where('details.item', '==', canonical_name)
             all_sales_docs = list(all_sales_query.stream())
-
             if all_sales_docs:
                 all_sales_docs.sort(key=lambda doc: doc.to_dict().get('timestamp', ''), reverse=True)
                 last_sale_data = all_sales_docs[0].to_dict()
                 last_selling_price = last_sale_data.get('details', {}).get('price')
-
             @firestore.transactional
             def process_one_sale(transaction, sale_details):
                 is_new_item = canonical_info['doc'] is None
-                
-                # Infer type based on the original transaction type from the LLM
                 original_trans_type = t.get('transaction_type')
                 item_type = 'service' if original_trans_type == 'service_offering' else 'good'
-
                 user_price = sale_details.get('price') or sale_details.get('unit_price')
-                
                 if user_price is not None:
                     selling_price = user_price
                 elif last_selling_price is not None:
@@ -362,35 +378,28 @@ def create_sale(user_phone: str, transaction_data: List[Dict]) -> tuple[bool, st
                         return f"Sale failed for new item '{canonical_name}': You must specify a price for the first sale."
                     else:
                         selling_price = 0
-
                 if not isinstance(selling_price, (int, float)): selling_price = 0
-                
                 sale_details['price'] = selling_price
                 sale_details['item'] = canonical_name
                 if 'unit_price' in sale_details: del sale_details['unit_price']
                 if 'service_name' in sale_details: del sale_details['service_name']
-
                 try:
                     quantity_sold = int(sale_details.get('quantity', 1))
                     if quantity_sold <= 0: return f"Sale failed for '{canonical_name}': Invalid quantity ({quantity_sold})."
                 except (ValueError, TypeError):
                     return f"Sale failed for '{canonical_name}': Invalid quantity format."
-
                 item_doc_ref = db.collection("users").document(user_phone).collection("inventory_and_services").document(canonical_name)
                 item_snapshot = item_doc_ref.get(transaction=transaction)
-                
                 item_cost = 0
                 if item_snapshot.exists:
                     inv_data = item_snapshot.to_dict()
                     inv_details = inv_data.get('details', {})
                     item_cost = inv_details.get('price') or inv_details.get('unit_price') or 0
-                    
                     if inv_data.get('type') == 'good':
                         stock_key = 'quantity'
                         current_stock = int(inv_details.get(stock_key, 0))
                         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)})
                 elif item_type == 'good':
                     return f"Sale failed for '{canonical_name}': Item not found in inventory. Please add it first."
@@ -402,7 +411,6 @@ def create_sale(user_phone: str, transaction_data: List[Dict]) -> tuple[bool, st
                         'last_updated': datetime.now(timezone.utc).isoformat()
                     }
                     transaction.set(item_doc_ref, service_record)
-
                 sale_doc_ref = sales_ref.document()
                 sale_record = {
                     'details': {**sale_details, 'cost': item_cost},
@@ -410,10 +418,8 @@ def create_sale(user_phone: str, transaction_data: List[Dict]) -> tuple[bool, st
                     'status': 'completed',
                     'transaction_id': sale_doc_ref.id
                 }
-                
                 transaction.set(sale_doc_ref, sale_record)
                 return f"Sale successful for {quantity_sold} x '{canonical_name}' at {sale_details.get('currency','')}{selling_price} each."
-
             transaction_feedback = process_one_sale(db.transaction(), details)
             feedback_messages.append(transaction_feedback)
             if "successful" in transaction_feedback:
@@ -421,7 +427,6 @@ def create_sale(user_phone: str, transaction_data: List[Dict]) -> tuple[bool, st
         except Exception as e:
             logger.error(f"Transactional sale failed for '{item_name}': {e}", exc_info=True)
             feedback_messages.append(f"Sale failed for '{item_name}': An unexpected error occurred.")
-            
     return any_success, "\n".join(feedback_messages)
 
 
@@ -430,15 +435,12 @@ def create_expense(user_phone: str, transaction_data: List[Dict]) -> tuple[bool,
     expenses_ref = db.collection("users").document(user_phone).collection("expenses")
     success_count = 0
     feedback_messages = []
-
     for transaction in transaction_data:
         details = transaction.get('details', {})
         expense_desc = details.get('description', details.get('category', 'Unnamed Expense'))
-        
         if 'amount' not in details:
             feedback_messages.append(f"Skipped expense '{expense_desc}': Missing amount.")
             continue
-        
         doc_ref = expenses_ref.document()
         expense_record = {
             'details': details, 'timestamp': datetime.now(timezone.utc).isoformat(),
@@ -447,10 +449,8 @@ def create_expense(user_phone: str, transaction_data: List[Dict]) -> tuple[bool,
         batch.set(doc_ref, expense_record)
         feedback_messages.append(f"Recorded expense: '{expense_desc}' for {details.get('currency','')}{details.get('amount')}.")
         success_count += 1
-
     if success_count == 0:
         return False, "\n".join(feedback_messages) if feedback_messages else "No valid expense transactions to create."
-
     try:
         batch.commit()
         return True, "\n".join(feedback_messages)
@@ -464,15 +464,12 @@ def create_asset(user_phone: str, transaction_data: List[Dict]) -> tuple[bool, s
     assets_ref = db.collection("users").document(user_phone).collection("assets")
     success_count = 0
     feedback_messages = []
-
     for transaction in transaction_data:
         details = transaction.get('details', {})
         asset_name = details.get('name', 'Unnamed Asset')
-
         if 'value' not in details:
             feedback_messages.append(f"Skipped asset '{asset_name}': Missing value.")
             continue
-
         doc_ref = assets_ref.document()
         asset_record = {
             'details': details, 'timestamp': datetime.now(timezone.utc).isoformat(),
@@ -481,10 +478,8 @@ def create_asset(user_phone: str, transaction_data: List[Dict]) -> tuple[bool, s
         batch.set(doc_ref, asset_record)
         feedback_messages.append(f"Recorded asset: '{asset_name}' with value {details.get('currency','')}{details.get('value')}.")
         success_count += 1
-
     if success_count == 0:
         return False, "\n".join(feedback_messages) if feedback_messages else "No valid asset transactions to create."
-
     try:
         batch.commit()
         return True, "\n".join(feedback_messages)
@@ -498,15 +493,12 @@ def create_liability(user_phone: str, transaction_data: List[Dict]) -> tuple[boo
     liabilities_ref = db.collection("users").document(user_phone).collection("liabilities")
     success_count = 0
     feedback_messages = []
-
     for transaction in transaction_data:
         details = transaction.get('details', {})
         creditor = details.get('creditor', 'Unnamed Creditor')
-
         if 'amount' not in details or not details.get('creditor'):
             feedback_messages.append(f"Skipped liability '{creditor}': Missing amount or creditor.")
             continue
-        
         doc_ref = liabilities_ref.document()
         liability_record = {
             'details': details, 
@@ -518,10 +510,8 @@ def create_liability(user_phone: str, transaction_data: List[Dict]) -> tuple[boo
         batch.set(doc_ref, liability_record)
         feedback_messages.append(f"Recorded liability to '{creditor}' for {details.get('currency','')}{details.get('amount')}.")
         success_count += 1
-
     if success_count == 0:
         return False, "\n".join(feedback_messages) if feedback_messages else "No valid liability transactions to create."
-
     try:
         batch.commit()
         return True, "\n".join(feedback_messages)
@@ -531,34 +521,24 @@ def create_liability(user_phone: str, transaction_data: List[Dict]) -> tuple[boo
 
 
 def _validate_dataframe(df: pd.DataFrame) -> pd.DataFrame:
-    """
-    Proactively cleans and validates a DataFrame to ensure data integrity for PandasAI.
-    """
+    """Proactively cleans and validates a DataFrame to ensure data integrity."""
     if df.empty:
         return df
-
     for col in ['timestamp', 'created_at', 'last_updated', 'acquisition_date', 'due_date']:
         if col in df.columns:
             df[col] = pd.to_datetime(df[col], errors='coerce', utc=True)
-
     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)
-
     for col in df.select_dtypes(include=['object']).columns:
         df[col] = df[col].fillna('Unknown')
-    
     return df
 
 
 def _fetch_all_collections_as_dfs(user_phone: str) -> List[Tuple[str, pd.DataFrame]]:
-    """
-    Fetches all user data, splits/validates DataFrames, and engineers features.
-    Returns a list of (name, dataframe) tuples.
-    """
+    """Fetches all user data, splits/validates DataFrames, and engineers features."""
     all_dfs_with_names = []
-
     inv_serv_docs = db.collection("users").document(user_phone).collection('inventory_and_services').stream()
     inventory_data, services_data = [], []
     for doc in inv_serv_docs:
@@ -569,118 +549,89 @@ def _fetch_all_collections_as_dfs(user_phone: str) -> List[Tuple[str, pd.DataFra
             services_data.append(flat_data)
         else:
             inventory_data.append(flat_data)
-
     if inventory_data:
         inventory_df = pd.DataFrame(inventory_data)
         all_dfs_with_names.append(("inventory", _validate_dataframe(inventory_df)))
-
     if services_data:
         services_df = pd.DataFrame(services_data)
         all_dfs_with_names.append(("services", _validate_dataframe(services_df)))
-
     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 = [doc.to_dict() for doc in docs]
-        
         if data:
             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)
             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']
-            
+            if df_name == 'sales' and 'price' in validated_df.columns and 'quantity' in validated_df.columns:
+                validated_df['sale_total'] = validated_df['price'] * validated_df['quantity']
             all_dfs_with_names.append((df_name, validated_df))
-            
     return all_dfs_with_names
 
 
 def read_datalake(user_phone: str, query: str) -> str:
     """
-    Handles queries with contextual pruning, temporal awareness, and robust error handling.
-    Dynamically generates reports or simple answers based on the query.
+    Handles data queries using a three-tiered hybrid system.
+    1. Programmatically calculates KPIs for reliability.
+    2. Uses a base LLM to synthesize KPIs into reports for high-level questions.
+    3. Uses an augmented PandasAI to answer specific questions, providing KPIs as context.
     """
-    # SURGICALLY MODIFIED: This function is enhanced to handle reporting.
     try:
         all_dfs_with_names = _fetch_all_collections_as_dfs(user_phone)
         if not all_dfs_with_names:
             return "You have no data recorded yet. Please add some transactions first."
 
-        # --- Dynamic Prompt Engineering Logic ---
-        report_keywords = ["report", "performance", "summary", "breakdown", "how did"]
-        is_report_request = any(keyword in query.lower() for keyword in report_keywords)
-        
-        today_str = datetime.now(timezone.utc).strftime('%Y-%m-%d')
-        
-        reporting_prompt_template = """
-        As a business data analyst, generate a report for the user's request: '{query}'.
-        Today's date is {today_str}. The data is in the provided DataFrames.
-
-        Structure your response in the following format.
-        **IMPORTANT: Use ONLY WhatsApp-compatible Markdown for formatting (*bold*, _italic_, ~strikethrough~, ```monospace```).**
+        # Tier 1: Programmatic KPI Calculation
+        engine = ReportEngine(all_dfs_with_names)
+        kpi_context = engine.calculate_all_kpis().format_kpis_for_llm()
 
-        *Report on:* {query}
-
-        *Headline:*
-        A one-sentence summary of the key finding.
+        # Routing Logic
+        report_keywords = ["report", "performance", "summary", "how did i do", "how are things"]
+        is_report_request = any(keyword in query.lower() for keyword in report_keywords)
 
-        *Key Metrics:*
-        - A bulleted list of the most important statistics.
+        if is_report_request:
+            # Tier 2: LLM as an "Insight Generator" (bypassing PandasAI)
+            logger.info(f"Handling '{query}' with the Insight Generator.")
+            synthesis_prompt = f"""
+            You are a business assistant. Based on the following real-time business KPIs, synthesize a concise, friendly report for the user.
+            The user's request was: '{query}'.
+            Format your entire response using WhatsApp-compatible markdown (*bold*, _italic_).
 
-        *Insight:*
-        - A brief, data-driven observation in plain English.
+            Here are the KPIs:
+            {kpi_context}
 
-        --- DETAILED BREAKDOWN BELOW ---
-        [This is where you will generate the code to produce the detailed dataframe]
-        """
+            Generate a summary based on these numbers.
+            """
+            response = llm.invoke(synthesis_prompt)
+            return response.content
 
-        if is_report_request:
-            contextual_query = reporting_prompt_template.format(query=query, today_str=today_str)
         else:
-            schema_description = "You have been provided with these Pandas DataFrames:\n"
-            for name, df in all_dfs_with_names:
-                if name == 'sales':
-                    schema_description += f"*   *{name}*: Contains sales records. **Crucially, the 'timestamp' column is timezone-aware (UTC). All date comparisons MUST also be timezone-aware.** For example, to create a date for comparison, use `pd.Timestamp('2025-07-23', tz='UTC')`.\n"
-                else:
-                    schema_description += f"*   *{name}*: Contains columns like {', '.join(df.columns.to_list())}.\n"
-            schema_description += "\nUse the 'sales' DataFrame for all questions about sales revenue and performance. Use the 'sale_total' column for calculating total sales."
+            # Tier 3: LLM as a "Code Assistant" (Augmented PandasAI)
+            logger.info(f"Handling '{query}' with Augmented PandasAI.")
+            augmented_prompt = f"""
+            For your context, here are the user's overall business KPIs:
+            {kpi_context}
+
+            Based on this context, please write Python code to answer the following specific user query: '{query}'.
+            Your code must end by declaring a `result` dictionary with "type" and "value" keys.
+            If the answer is a table, return a dataframe. My system will automatically convert it to an image.
+            """
             
-            contextual_query = (
-                f"{schema_description}\n\n"
-                f"For context, today's date is {today_str}. "
-                f"Based on this, please answer the following user query: '{query}'"
-            )
-
-        datalake_dfs = [df for _, df in all_dfs_with_names]
-
-        lake = SmartDatalake(datalake_dfs, config={
-            "llm": llm, "response_parser": FlaskResponse,
-            "save_charts_path": user_defined_path, "enable_cache": False,
-            "conversational": True
-        })
-        
-        logger.info(f"Contextual query for PandasAI: {contextual_query}")
-
-        try:
-            response = lake.chat(contextual_query)
-            # The response is now either a string (narrative/simple answer) or an image path
-            return str(response)
-        except NoCodeFoundError:
-            logger.warning(f"PandasAI failed on first attempt for query: '{query}'. Retrying.")
-            # Use a simpler prompt for retry, focusing on just getting the data
-            simplified_query = f"The previous attempt failed. Try a simpler way. Just return the data for this query: '{query}'"
-            response = lake.chat(simplified_query)
+            datalake_dfs = [df for _, df in all_dfs_with_names]
+            lake = SmartDatalake(datalake_dfs, config={
+                "llm": llm, "response_parser": FlaskResponse,
+                "save_charts_path": user_defined_path, "enable_cache": False
+            })
+            
+            response = lake.chat(augmented_prompt)
             return str(response)
 
     except NoCodeFoundError:
-        logger.error(f"PandasAI failed on second attempt for query: '{query}'")
+        logger.error(f"PandasAI failed to generate code for query: '{query}'")
         return "I'm sorry, I couldn't figure out how to answer that question with your data. Please try rephrasing it."
     except Exception as e:
         logger.error(f"Data query failed for user {user_phone}, query '{query}': {e}", exc_info=True)
@@ -689,22 +640,18 @@ def read_datalake(user_phone: str, query: str) -> str:
 
 def _find_document_by_details(user_phone: str, collection_name: str, details: Dict) -> Optional[Any]:
     col_ref = db.collection("users").document(user_phone).collection(collection_name)
-    
     if 'transaction_id' in details and details['transaction_id']:
         doc = col_ref.document(details['transaction_id']).get()
         if doc.exists: return {"id": doc.id, "data": doc.to_dict()}
-
     if collection_name in ['inventory_and_services', 'sales'] and ('item' in details or 'service_name' in details):
         item_name = details.get('item') or details.get('service_name')
         canonical_info = _get_canonical_info(user_phone, item_name)
         if canonical_info['doc']:
             doc = canonical_info['doc']
             return {"id": doc.id, "data": doc.to_dict()}
-
     query = col_ref
     key_map = {'expenses': 'description', 'assets': 'name', 'liabilities': 'creditor'}
     search_key = key_map.get(collection_name)
-    
     filters_applied = False
     if search_key and search_key in details:
         query = query.where(f'details.{search_key}', '==', details[search_key]); filters_applied = True
@@ -712,12 +659,9 @@ def _find_document_by_details(user_phone: str, collection_name: str, details: Di
         query = query.where('details.amount', '==', details['amount']); filters_applied = True
     if 'value' in details:
         query = query.where('details.value', '==', details['value']); filters_applied = True
-
     if not filters_applied: return None
-
     docs = query.limit(2).stream()
     found_docs = [{"id": doc.id, "data": doc.to_dict()} for doc in docs]
-
     if len(found_docs) == 1: return found_docs[0]
     elif len(found_docs) > 1: return "multiple_matches"
     else: return None
@@ -734,11 +678,9 @@ def update_transaction(user_phone: str, transaction_data: List[Dict]) -> tuple[b
             'service_offering': 'inventory_and_services', 'expense': 'expenses', 'asset': 'assets', 'liability': 'liabilities'
         }
         collection_name = collection_map.get(trans_type)
-
         if not collection_name:
             feedback.append(f"Update skipped: Unknown type '{trans_type}'.")
             continue
-
         target_doc = _find_document_by_details(user_phone, collection_name, details)
         if target_doc == "multiple_matches":
             feedback.append(f"Update for {trans_type} failed: Multiple records match. Please be more specific.")
@@ -746,16 +688,13 @@ def update_transaction(user_phone: str, transaction_data: List[Dict]) -> tuple[b
         if not target_doc:
             feedback.append(f"Update for {trans_type} failed: No record found matching your description.")
             continue
-        
         doc_id = target_doc["id"]
         doc_ref = db.collection("users").document(user_phone).collection(collection_name).document(doc_id)
-        
         try:
             updates = {f"details.{k}": v for k, v in details.items() if k != 'transaction_id'}
             if not updates:
                 feedback.append(f"Update for {trans_type} (ID: {doc_id}) skipped: No new data provided.")
                 continue
-
             updates['last_updated'] = datetime.now(timezone.utc).isoformat()
             doc_ref.update(updates)
             feedback.append(f"Successfully updated {trans_type} record (ID: {doc_id}).")
@@ -763,7 +702,6 @@ def update_transaction(user_phone: str, transaction_data: List[Dict]) -> tuple[b
         except Exception as e:
             logger.error(f"Update failed for doc '{doc_id}': {e}", exc_info=True)
             feedback.append(f"Update for {trans_type} (ID: {doc_id}) failed with an error.")
-
     return any_success, "\n".join(feedback)
 
 
@@ -778,11 +716,9 @@ def delete_transaction(user_phone: str, transaction_data: List[Dict]) -> tuple[b
             'service_offering': 'inventory_and_services', 'expense': 'expenses', 'asset': 'assets', 'liability': 'liabilities'
         }
         collection_name = collection_map.get(trans_type)
-
         if not collection_name:
             feedback.append(f"Delete skipped: Unknown type '{trans_type}'.")
             continue
-
         target_doc = _find_document_by_details(user_phone, collection_name, details)
         if target_doc == "multiple_matches":
             feedback.append(f"Delete for {trans_type} failed: Multiple records match.")
@@ -790,7 +726,6 @@ def delete_transaction(user_phone: str, transaction_data: List[Dict]) -> tuple[b
         if not target_doc:
             feedback.append(f"Delete for {trans_type} failed: No record found.")
             continue
-            
         doc_id = target_doc["id"]
         try:
             db.collection("users").document(user_phone).collection(collection_name).document(doc_id).delete()
@@ -799,7 +734,6 @@ def delete_transaction(user_phone: str, transaction_data: List[Dict]) -> tuple[b
         except Exception as e:
             logger.error(f"Delete failed for doc '{doc_id}': {e}", exc_info=True)
             feedback.append(f"Delete for {trans_type} (ID: {doc_id}) failed with an error.")
-
     return any_success, "\n".join(feedback)
 
 
@@ -813,38 +747,33 @@ def persist_temporary_transaction(transactions: List[Dict], mobile: str) -> bool
         logger.error(f"Failed to persist temporary transaction for user {mobile}: {e}", exc_info=True)
         return False
 
+
 CURRENCY_SYMBOL_REGEX = re.compile(r"^\s*[\$\£\€\¥\₹R]")
 
 def format_transaction_response(transactions: Union[List[Dict], Dict, None]) -> str:
     if not transactions: return "No transaction data to display."
     if isinstance(transactions, dict): transactions = [transactions]
-
     output_lines = []
     for idx, trans in enumerate(transactions):
         if not isinstance(trans, dict): continue
-
         details = trans.get('details', trans)
         trans_type = trans.get('transaction_type', 'Item').replace("_", " ").title()
         title = f"{trans_type}"
         if len(transactions) > 1: output_lines.append(f"--- {title} {idx + 1} ---")
         else: output_lines.append(f"--- {title} ---")
-
         key_order = [
             'transaction_id', 'item', 'service_name', 'name', 'creditor', 'category',
             'quantity', 'units_available', 'hours', 'price', 'rate', 'amount', 'cost', 'value',
             'customer', 'vendor', 'client', 'date', 'acquisition_date', 'due_date', 'description', 'type'
         ]
-        
         displayed_keys = set()
         if 'transaction_id' in trans:
             output_lines.append(f"• Transaction ID: {trans['transaction_id']}")
             displayed_keys.add('transaction_id')
-
         for key in key_order:
             if key in details and key not in displayed_keys:
                 output_lines.append(f"• {key.replace('_', ' ').title()}: {details[key]}")
                 displayed_keys.add(key)
-
         for key, value in details.items():
             if key not in displayed_keys and key != 'currency':
                 output_lines.append(f"• {key.replace('_', ' ').title()}: {value}")
@@ -868,7 +797,6 @@ def process_intent(parsed_trans_data: List[Dict], mobile: str) -> str:
     """
     if not parsed_trans_data:
         return "I couldn't understand the transaction details. Could you please try again?"
-
     grouped_transactions = {}
     for trans in parsed_trans_data:
         intent = trans.get('intent', 'unknown').lower()
@@ -876,7 +804,6 @@ def process_intent(parsed_trans_data: List[Dict], mobile: str) -> str:
         key = (intent, trans_type)
         if key not in grouped_transactions: grouped_transactions[key] = []
         grouped_transactions[key].append(trans)
-
     final_feedback = []
     for (intent, trans_type), transactions in grouped_transactions.items():
         logger.info(f"Processing group: {intent} - {trans_type} for user {mobile}")
@@ -917,6 +844,5 @@ def process_intent(parsed_trans_data: List[Dict], mobile: str) -> str:
         except Exception as e:
             logger.error(f"Error processing group ({intent}, {trans_type}) for user {mobile}: {e}", exc_info=True)
             final_feedback.append(f"An unexpected error occurred while processing your {trans_type} {intent} request.")
-
     if not final_feedback: return "No actions were processed from your request."
     return "\n\n".join(final_feedback).strip()