Update utility.py

utility.py

@@ -124,7 +124,7 @@ except Exception as e:
 
 # --- START: VISION PROCESSING FUNCTIONS ---
 
-def _transpile_vision_json_to_query(vision_json: List[Dict], caption: Optional[str]) -> str:
+def _transpile_vision_json_to_query(vision_json: List[Dict]) -> str:
     """Converts the structured JSON list from the Vision AI into a natural language query."""
     if not vision_json:
         return "Error: Could not extract any transactions from the image."
@@ -156,10 +156,6 @@ def _transpile_vision_json_to_query(vision_json: List[Dict], caption: Optional[s
         query_parts.append(part)
 
     final_query = " and ".join(query_parts)
-    
-    # --- REMOVED --- The caption is now handled inside the vision prompt, not appended here.
-    # if caption:
-    #     final_query += f" {caption}"
         
     return final_query.strip()
 
@@ -171,7 +167,6 @@ def _analyze_image_with_vision(image_bytes: bytes, caption: Optional[str]) -> Li
     try:
         image_pil = Image.open(io.BytesIO(image_bytes))
         
-        # --- MODIFIED --- Added caption handling directly into the prompt.
         prompt = f"""
         You are a bookkeeping vision model. Analyze the image (receipt, invoice, handwritten note, *catalog/menu/price list*, product photo, shelf photo). Return ONLY a valid JSON array [] of transaction objects that our TEXT PIPELINE can consume directly.
 
@@ -315,8 +310,7 @@ def process_image_and_generate_query(image_bytes: bytes, caption: Optional[str])
         return "Error: I couldn't find any actionable transactions in the image."
         
     logger.info(f"Vision AI analysis complete. Result: {vision_json_list}")
-    # --- MODIFIED --- Caption is no longer passed here.
-    return _transpile_vision_json_to_query(vision_json_list, None)
+    return _transpile_vision_json_to_query(vision_json_list)
 
 # --- END: VISION PROCESSING FUNCTIONS ---
 
@@ -329,9 +323,7 @@ class ReportEngine:
         self.currency = self._get_user_currency()
 
     def _get_user_currency(self) -> str:
-        """
-        Determines the user's primary currency from their data.
-        """
+        """Determines the user's primary currency from their data."""
         for df_name in ['sales', 'expenses', 'assets', 'liabilities']:
             if df_name in self.dfs and 'currency' in self.dfs[df_name].columns:
                 mode = self.dfs[df_name]['currency'].mode()
@@ -342,36 +334,31 @@ class ReportEngine:
         return "$"
 
     def _get_time_filter(self, target_df: pd.DataFrame) -> Optional[pd.Series]:
-        """
-        REWRITTEN: A robust, ordered temporal filter to correctly interpret user requests.
-        """
+        """A robust, ordered temporal filter to correctly interpret user requests."""
         if target_df is None or 'timestamp' not in target_df.columns or target_df.empty:
             return None
 
         # The order of these checks is critical: from most specific to most general.
         
-        # 1. Check for "yesterday"
         if "yesterday" in self.query:
             yesterday = (self.now - timedelta(days=1)).date()
             start_of_yesterday = pd.Timestamp(yesterday, tz='UTC')
             end_of_yesterday = start_of_yesterday + timedelta(days=1)
             return (target_df['timestamp'] >= start_of_yesterday) & (target_df['timestamp'] < end_of_yesterday)
 
-        # 2. Check for "today"
         if "today" in self.query:
             today = self.now.date()
             start_of_today = pd.Timestamp(today, tz='UTC')
             end_of_today = start_of_today + timedelta(days=1)
             return (target_df['timestamp'] >= start_of_today) & (target_df['timestamp'] < end_of_today)
 
-        # 3. Check for "last month"
         if "last month" in self.query:
             first_day_current_month = self.now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
             last_day_last_month = first_day_current_month - timedelta(days=1)
             first_day_last_month = last_day_last_month.replace(day=1)
-            return (target_df['timestamp'] >= pd.Timestamp(first_day_last_month, tz='UTC')) & (target_df['timestamp'] < pd.Timestamp(first_day_current_month, tz='UTC'))
+            # --- FIX --- Removed redundant tz parameter to prevent ValueError
+            return (target_df['timestamp'] >= pd.Timestamp(first_day_last_month)) & (target_df['timestamp'] < pd.Timestamp(first_day_current_month))
 
-        # 4. Check for a specific month name (e.g., "in july")
         month_match = re.search(r"\b(january|february|march|april|may|june|july|august|september|october|november|december)\b", self.query)
         if month_match:
             month_name = month_match.group(1)
@@ -379,36 +366,30 @@ class ReportEngine:
             current_year = self.now.year
             return (target_df['timestamp'].dt.month == month_number) & (target_df['timestamp'].dt.year == current_year)
 
-        # 5. Check for "this month" or a standalone "month"
         if "this month" in self.query or "month" in self.query:
              return (target_df['timestamp'].dt.month == self.now.month) & (target_df['timestamp'].dt.year == self.now.year)
         
-        # 6. Check for "last week"
         if "last week" in self.query:
             start_of_this_week = self.now.date() - timedelta(days=self.now.weekday())
             start_of_last_week = start_of_this_week - timedelta(days=7)
             return (target_df['timestamp'].dt.date >= start_of_last_week) & (target_df['timestamp'].dt.date < start_of_this_week)
 
-        # 7. Check for "this week" or a standalone "week"
         if "this week" in self.query or "week" in self.query:
             start_of_week = self.now.date() - timedelta(days=self.now.weekday())
             return target_df['timestamp'].dt.date >= start_of_week
 
-        # 8. Check for a specific day of the week (e.g., "on monday")
         day_match = re.search(r"on (monday|tuesday|wednesday|thursday|friday|saturday|sunday)", self.query)
         if day_match and 'day_of_week' in target_df.columns:
             day_name = day_match.group(1).title()
             return target_df['day_of_week'] == day_name
 
-        # 9. Check for "year"
         if "year" in self.query:
             return target_df['timestamp'].dt.year == self.now.year
 
-        # Default: if no time filter is found, return all data
         return pd.Series(True, index=target_df.index)
 
     def generate_report(self) -> str:
-        """RESTORED: Generates the original, simple Sales or Expenses report."""
+        """Generates a simple Sales or Expenses report."""
         subject = "sales"
         if "expense" in self.query:
             subject = "expenses"
@@ -452,7 +433,7 @@ class ReportEngine:
         return json.dumps(self.results, indent=2)
 
     def generate_profit_report(self) -> str:
-        """NEW: Generates a comprehensive profitability report."""
+        """Generates a comprehensive profitability report."""
         sales_df = self.dfs.get('sales', pd.DataFrame())
         expenses_df = self.dfs.get('expenses', pd.DataFrame())
         
@@ -494,7 +475,7 @@ class ReportEngine:
         return json.dumps(self.results, indent=2)
         
     def generate_item_report(self, subject_item: str) -> str:
-        """NEW: Generates a performance report for a specific item."""
+        """Generates a performance report for a specific item."""
         sales_df = self.dfs.get('sales', pd.DataFrame())
         if sales_df.empty: return json.dumps({"error": f"No sales data found for '{subject_item}'."})
         
@@ -521,7 +502,7 @@ class ReportEngine:
         return json.dumps(self.results, indent=2)
 
     def generate_day_of_week_report(self) -> str:
-        """NEW: Generates a report analyzing sales by day of the week."""
+        """Generates a report analyzing sales by day of the week."""
         sales_df = self.dfs.get('sales', pd.DataFrame())
         if sales_df.empty or 'day_of_week' not in sales_df.columns: return json.dumps({"error": "No data available to analyze by day."})
         time_filter = self._get_time_filter(sales_df)
@@ -571,11 +552,8 @@ class ReportEngine:
         return json.dumps(self.results, indent=2)
 
     def generate_business_snapshot(self) -> Dict[str, Any]:
-        """
-        NEW: Creates a high-level summary of the entire business for contextual AI coaching.
-        """
+        """Creates a high-level summary of the entire business for contextual AI coaching."""
         snapshot = {}
-        # Financial KPIs
         sales_df = self.dfs.get('sales', pd.DataFrame())
         expenses_df = self.dfs.get('expenses', pd.DataFrame())
         total_revenue = sales_df['sale_total'].sum() if not sales_df.empty else 0
@@ -588,7 +566,6 @@ class ReportEngine:
             "Net Profit": f"{self.currency}{net_profit:.2f}"
         }
 
-        # Inventory Overview
         inventory_df = self.dfs.get('inventory', pd.DataFrame())
         if not inventory_df.empty and 'item' in inventory_df.columns and 'quantity' in inventory_df.columns:
             snapshot['inventory_overview'] = "\n".join(
@@ -597,7 +574,6 @@ class ReportEngine:
         else:
             snapshot['inventory_overview'] = "No inventory items recorded."
 
-        # Asset Register
         assets_df = self.dfs.get('assets', pd.DataFrame())
         if not assets_df.empty and 'name' in assets_df.columns and 'value' in assets_df.columns:
             snapshot['asset_register'] = "\n".join(
@@ -606,7 +582,6 @@ class ReportEngine:
         else:
             snapshot['asset_register'] = "No assets recorded."
 
-        # Liabilities Ledger
         liabilities_df = self.dfs.get('liabilities', pd.DataFrame())
         if not liabilities_df.empty and 'creditor' in liabilities_df.columns and 'amount' in liabilities_df.columns:
             snapshot['liabilities_ledger'] = "\n".join(
@@ -725,9 +700,7 @@ def add_timestamp(transaction: Dict) -> Dict:
     return transaction
 
 def _get_canonical_info(user_phone: str, item_name: str) -> Dict[str, Any]:
-    """
-    Finds the canonical version of an item using an "exact match first" hybrid approach.
-    """
+    """Finds the canonical version of an item using an "exact match first" hybrid approach."""
     inventory_ref = db.collection("users").document(user_phone).collection("inventory_and_services")
     name_lower = item_name.lower().strip()
     all_item_docs = list(inventory_ref.stream())
@@ -789,9 +762,7 @@ def create_or_update_inventory_or_service_offering(user_phone: str, transaction_
         return False, f"An error occurred during inventory update: {e}"
 
 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.
-    """
+    """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:
@@ -1012,21 +983,14 @@ def _fetch_all_collections_as_dfs(user_phone: str) -> List[Tuple[str, pd.DataFra
     return all_dfs_with_names
 
 def _get_relative_date_context() -> str:
-    """
-    Generates a string of pre-calculated dates to inject into the PandasAI prompt
-    for improved temporal awareness.
-    """
+    """Generates a string of pre-calculated dates for improved temporal awareness."""
     today = datetime.now(timezone.utc)
     
-    def fmt(d):
-        return d.strftime('%Y-%m-%d')
+    def fmt(d): return d.strftime('%Y-%m-%d')
 
     yesterday = today - timedelta(days=1)
     start_of_this_week = today - timedelta(days=today.weekday())
-    end_of_this_week = start_of_this_week + timedelta(days=6)
     start_of_last_week = start_of_this_week - timedelta(days=7)
-    end_of_last_week = start_of_last_week + timedelta(days=6)
-    
     last_monday = start_of_this_week - timedelta(days=7)
     
     context = [
@@ -1041,9 +1005,7 @@ def _get_relative_date_context() -> str:
     return "\n".join(context)
 
 def read_datalake(user_phone: str, query: str) -> str:
-    """
-    Implements the final Unified Strategy for robust, intelligent data analysis.
-    """
+    """Implements the final Unified Strategy for robust, intelligent data analysis."""
     def _to_text(resp) -> str:
         try:
             if resp is None: return ""
@@ -1065,13 +1027,11 @@ def read_datalake(user_phone: str, query: str) -> str:
 
         # --- REFACTORED ROUTING LOGIC ---
 
-        # --- Tier 0: Simple Direct Lookups (Unchanged) ---
+        # --- Tier 0: Simple Direct Lookups ---
         simple_lookup_map = {
             "inventory": ["stock", "inventory", "in stock", "what do i have"],
             "assets": ["asset", "assets", "my assets"],
             "liabilities": ["liabilities", "i owe", "creditor", "my debts"],
-            "sales": ["show my sales", "list sales"],
-            "expenses": ["show my expenses", "list expenses"]
         }
         for df_name, keywords in simple_lookup_map.items():
             if any(keyword in query_lower for keyword in keywords):
@@ -1081,7 +1041,7 @@ def read_datalake(user_phone: str, query: str) -> str:
                     return render_df_as_image(target_df_tuple[1])
                 return f"You don't have any {df_name} recorded yet."
 
-        # --- Tier 1: Specific, Pre-canned Reports ---
+        # --- Tier 1: Canned & Temporal Reports (NEW UNIFIED LOGIC) ---
         item_report_match = re.search(r"(?:sales report for|report on|performance of)\s+([\w\s]+?)(?:\s+(?:this|last|on|in|for|today|yesterday)|$)", query_lower)
         report_json = None
         
@@ -1096,15 +1056,20 @@ def read_datalake(user_phone: str, query: str) -> str:
         elif any(k in query_lower for k in ["best day", "busiest day", "sales by day"]):
             logger.info(f"Handling '{query}' with the Day of Week Report Path.")
             report_json = engine.generate_day_of_week_report()
+        # --- FIX --- This new, simple route handles all sales/expense queries directly.
+        elif "sales" in query_lower or "expense" in query_lower:
+            logger.info(f"Handling '{query}' with the General Sales/Expense Report Path.")
+            report_json = engine.generate_report()
         
         if report_json:
             report_data = json.loads(report_json)
             if "error" in report_data: return report_data["error"]
             synthesis_prompt = f"""
             Directly synthesize a professional business report from the following JSON data. Omit conversational introductions or summaries. Present only the data-driven report, formatted for WhatsApp (*bold*, _italic_, emojis).
-
+            For sales reports, if helpful, provide a creative and actionable "Insight" section at the end based on the best/worst selling items.
+            
             **IMPORTANT INSTRUCTIONS:**
-            - If `report_subject` is "Profitability", present a clear financial summary: start with Revenue, subtract COGS for Gross Profit, then subtract Expenses for Net Profit. Also mention other KPIs.
+            - If `report_subject` is "Profitability", present a clear financial summary.
             - If `report_subject` is "Item Report", state the item name and present its performance KPIs.
             - If `report_subject` is "Day of Week Analysis", state the best day and list daily sales.
 
@@ -1114,29 +1079,8 @@ def read_datalake(user_phone: str, query: str) -> str:
             response = llm.invoke(synthesis_prompt)
             return _to_text(response)
 
-        # --- Tier 1.5: General Temporal Reports ---
-        subjects = ["sales", "expenses"]
-        # --- MODIFIED --- Expanded temporals list for better routing
-        temporals = [
-            "today", "yesterday", "week", "month", "year", "monday", "tuesday", 
-            "wednesday", "thursday", "friday", "saturday", "sunday", "january", 
-            "february", "march", "april", "may", "june", "july", "august", 
-            "september", "october", "november", "december"
-        ]
-        if any(sub in query_lower for sub in subjects) and any(temp in query_lower for temp in temporals):
-            logger.info(f"Handling '{query}' with the General Temporal Report Path.")
-            report_json = engine.generate_report()
-            report_data = json.loads(report_json)
-            if "error" in report_data: return report_data["error"]
-            synthesis_prompt = f"""Synthesize a professional business report from the following JSON data. Omit conversational introductions or summaries. For sales reports, you MUST provide a creative and actionable "Insight" section at the end based on the best/worst selling items. Present only the data-driven report and the insight, formatted for WhatsApp (*bold*, _italic_, emojis).
-            Data: {report_json}"""
-            response = llm.invoke(synthesis_prompt)
-            return _to_text(response)
-
-        # --- Tier 2: Predictive & Generic Summary Fallback ---
+        # --- Tier 2: Predictive Queries ---
         predictive_keywords = ["expect", "forecast", "predict"]
-        historical_report_keywords = ["sales report", "expense report", "performance summary", "how did i do", "overview"]
-        
         if any(keyword in query_lower for keyword in predictive_keywords):
             logger.info(f"Handling '{query}' with the Forecasting Path.")
             forecast_json = engine.generate_forecast_data()
@@ -1145,16 +1089,6 @@ def read_datalake(user_phone: str, query: str) -> str:
             synthesis_prompt = f"Synthesize a sales forecast from the following JSON data. Omit conversational introductions or summaries. Present only the forecast. Data: {forecast_json}"
             response = llm.invoke(synthesis_prompt)
             return _to_text(response)
-
-        elif any(keyword in query_lower for keyword in historical_report_keywords):
-            logger.info(f"Handling '{query}' with the General Reporting Path (Sales/Expense).")
-            report_json = engine.generate_report()
-            report_data = json.loads(report_json)
-            if "error" in report_data: return report_data["error"]
-            synthesis_prompt = f"""Synthesize a professional business report from the following JSON data. Omit conversational introductions or summaries. For sales reports, you MUST provide a creative and actionable "Insight" section at the end based on the best/worst selling items. Present only the data-driven report and the insight, formatted for WhatsApp (*bold*, _italic_, emojis).
-            Data: {report_json}"""
-            response = llm.invoke(synthesis_prompt)
-            return _to_text(response)
         
         # --- Tier 3: Business Coach & Help Layer ---
         help_keywords = ['help', 'tutorial', 'guide', 'how do you work', 'what can you do', 'how can', 'how would']
@@ -1165,16 +1099,15 @@ def read_datalake(user_phone: str, query: str) -> str:
             snapshot_str = json.dumps(snapshot, indent=2)
 
             synthesis_prompt = f"""
-            You are Qx, a friendly and insightful business coach and financial expert. The user is asking a general question. Only perform any calculations when necessary. Your task is to provide a clear, helpful, and strategic answer based on their question, using your general knowledge and the business snapshot provided below for context.
+            You are Qx, a friendly and insightful business coach and financial expert. Your task is to provide a clear, helpful, and strategic answer based on the user's question, using your general business knowledge combined with the business snapshot provided below for context.
 
             **IMPORTANT RULES:**
-            1.  **Use the Context:** Use the Business Snapshot as your internal knowledge to make your advice relevant and personalized. For example, if inventory is high for an item, you might suggest a promotion. If profit is low, you might suggest cost-cutting measures.
-            2.  **You can state the numbers, info or metrics from the data where helpful. Synthesize them into your advice. generate insight from the data if needed.
-            3.  **Stay in Character:** Act as a coach. Be encouraging and provide actionable advice.
-            4.  **Handle 'Help' Queries:** If asked about your capabilities, explain that you can record transactions (sales, expenses, etc.) via text or images, generate detailed reports (profit, sales by item), answer questions about their data, and provide business advice.
-            5.  **Format for WhatsApp:** Use *bold*, _italic_, and emojis to make your response clear and engaging.
+            1.  **Synthesize, Don't Just Report:** Use the Business Snapshot to make your advice relevant and personalized. For example, if inventory is high for an item, you might suggest a promotion. If profit is low, you might suggest cost-cutting measures.
+            2.  **Act as a Coach:** Be encouraging and provide actionable advice.
+            3.  **Handle 'Help' Queries:** If asked about your capabilities, explain that you can record transactions (sales, expenses, etc.) via text or images, generate detailed reports (profit, sales by item), answer questions about their data, and provide business advice.
+            4.  **Format for WhatsApp:** Use *bold*, _italic_, and emojis to make your response clear and engaging.
 
-            **BUSINESS SNAPSHOT (INTERNAL CONTEXT ONLY):**
+            **Business Snapshot for Context:**
             {snapshot_str}
 
             **User's Question:**
@@ -1231,16 +1164,15 @@ def read_datalake(user_phone: str, query: str) -> str:
                 snapshot_str = json.dumps(snapshot, indent=2)
 
                 synthesis_prompt = f"""
-                You are Qx, a friendly and insightful business coach and financial expert. The user is asking a general question. Only perform calculations when necessary. Your task is to provide a clear, helpful, and strategic answer based on their question, using your general knowledge and the business snapshot provided below for context.
+                You are Qx, a friendly and insightful business coach and financial expert. Your task is to provide a clear, helpful, and strategic answer based on the user's question, using your general business knowledge combined with the business snapshot provided below for context.
 
                 **IMPORTANT RULES:**
-                1.  **Use the Context:** Use the Business Snapshot as your internal knowledge to make your advice relevant and personalized. For example, if inventory is high for an item, you might suggest a promotion. If profit is low, you might suggest cost-cutting measures.
-                2.  **You can state the numbers, info or metrics from the data where helpful. Synthesize them into your advice. generate insight from data if needed.
-                3.  **Stay in Character:** Act as a coach. Be encouraging and provide actionable advice.
-                4.  **Handle 'Help' Queries:** If asked about your capabilities, explain that you can record transactions (sales, expenses, etc.) via text or images, generate detailed reports (profit, sales by item), answer questions about their data, and provide business advice.
-                5.  **Format for WhatsApp:** Use *bold*, _italic_, and emojis to make your response clear and engaging.
+                1.  **Synthesize, Don't Just Report:** Use the Business Snapshot to make your advice relevant and personalized. For example, if inventory is high for an item, you might suggest a promotion. If profit is low, you might suggest cost-cutting measures.
+                2.  **Act as a Coach:** Be encouraging and provide actionable advice.
+                3.  **Handle 'Help' Queries:** If asked about your capabilities, explain that you can record transactions (sales, expenses, etc.) via text or images, generate detailed reports (profit, sales by item), answer questions about their data, and provide business advice.
+                4.  **Format for WhatsApp:** Use *bold*, _italic_, and emojis to make your response clear and engaging.
 
-                **BUSINESS SNAPSHOT (INTERNAL CONTEXT ONLY):**
+                **Business Snapshot for Context:**
                 {snapshot_str}
 
                 **User's Question:**
@@ -1396,9 +1328,7 @@ def fetch_transaction(user_phone: str, identifier: str, collection: str = "inven
         return None
 
 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.
-    """
+    """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?"
     grouped_transactions = {}