Upload order_tracking_agent.py

agent/order_tracking_agent.py

@@ -0,0 +1,149 @@
+import pandas as pd
+import os
+import logging
+import re
+
+# Configure logging
+logging.basicConfig(level=logging.DEBUG)
+logger = logging.getLogger(__name__)
+
+def extract_order_number(query):
+    """Extracts the order number from the user's query using regex."""
+    match = re.search(r'\b\d{3}-\d{7}-\d{7}\b', query)
+    if match:
+        return match.group(0)
+    return None
+
+def extract_tracking_number(query):
+    """Extracts the tracking number from the user's query using regex."""
+    match = re.search(r'\b[A-Z0-9]{10,}\b', query)
+    if match:
+        return match.group(0)
+    return None
+
+class OrderTrackingAgent:
+    def __init__(self, csv_path='./data/order_track.csv'):
+        self.csv_path = csv_path
+        self.order_data = None
+        self.pending_actions = {}  # To handle ongoing refund/replacement flows
+        self.load_data()
+
+    def load_data(self):
+        """Load the order tracking data from the CSV file."""
+        try:
+            if os.path.exists(self.csv_path):
+                self.order_data = pd.read_csv(self.csv_path)
+                logger.info("Order tracking data loaded successfully.")
+                logger.debug(f"Columns in CSV: {self.order_data.columns.tolist()}")
+            else:
+                raise FileNotFoundError(f"CSV file not found at {self.csv_path}")
+        except Exception as e:
+            logger.error(f"Error loading order tracking data: {e}")
+
+    def process_query(self, query, order_number=None, tracking_number=None):
+        """
+        Process order-related queries.
+
+        Args:
+            query (str): The user's query.
+            order_number (str, optional): The order number extracted from the query.
+            tracking_number (str, optional): The tracking number extracted from the query.
+        Returns:
+            str: The response to the user's query.
+        """
+        logger.info(f"Processing query: {query}")
+
+        # Extract identifiers if not explicitly provided
+        if not order_number and not tracking_number:
+            order_number = self.extract_order_number(query)
+            tracking_number = self.extract_tracking_number(query)
+
+        logger.debug(f"Extracted Order Number: {order_number}")
+        logger.debug(f"Extracted Tracking Number: {tracking_number}")
+
+        if "refund" in query.lower():
+            return self.handle_refund_flow(query, order_number)
+        elif "replace" in query.lower():
+            return self.handle_replacement_flow(query, order_number)
+
+        # Normal tracking queries
+        if order_number:
+            return self._find_order_by_column("Order number", order_number)
+        elif tracking_number:
+            return self._find_order_by_column("Tracking number", tracking_number)
+
+        return "Please provide a valid order number or tracking number to proceed."
+
+    def _find_order_by_column(self, column_name, value):
+        """Find order details based on a specific column."""
+        try:
+            result = self.order_data[self.order_data[column_name].astype(str).str.strip() == value.strip()]
+            if not result.empty:
+                order_info = result.to_dict(orient="records")[0]
+                response = f"**Order Details:**\n"
+                for key, val in order_info.items():
+                    response += f"**{key}:** {val}\n"
+                return response.strip()
+            else:
+                return f"No order found with {column_name.lower()}: {value}. Please check the number and try again."
+        except KeyError:
+            return f"Invalid column name: {column_name}. Unable to process the query."
+        except Exception as e:
+            logger.error(f"Error retrieving order data: {e}")
+            return "An error occurred while processing your request."
+
+    def handle_refund_flow(self, query, order_number):
+        """Handles refund flow step-by-step."""
+        if "order number" not in self.pending_actions:
+            if not order_number:
+                return "Sure! Please provide your order number to proceed with the refund."
+            self.pending_actions["order number"] = order_number
+            return "Order found! Confirm if you'd like to initiate the refund."
+
+        if "confirmed" not in self.pending_actions:
+            if "yes" in query.lower():
+                self.pending_actions["confirmed"] = True
+                return "Thank you! I have initiated your refund process. What time shall I arrange to pick up the item? Please provide a convenient date and time."
+
+        if "pickup time" not in self.pending_actions:
+            pickup_time = query.strip()
+            self.pending_actions["pickup time"] = pickup_time
+            return f"Got it! Your pickup is scheduled for {pickup_time}. Please ensure the parcel is securely packed. Once we receive it, the refund will be credited to your Amazon Pay balance."
+
+        # Clear the flow once complete
+        self.pending_actions.clear()
+        return "Refund process complete. Let me know if there's anything else I can help with!"
+
+    def handle_replacement_flow(self, query, order_number):
+        """Handles replacement flow step-by-step."""
+        if "order number" not in self.pending_actions:
+            if not order_number:
+                return "Sure! Please provide your order number to process the replacement."
+            self.pending_actions["order number"] = order_number
+            return "Order found! Confirm if you'd like to proceed with the replacement request."
+
+        if "confirmed" not in self.pending_actions:
+            if "yes" in query.lower():
+                self.pending_actions["confirmed"] = True
+                return "Your replacement process has been initiated. What time shall I arrange to pick up the item for replacement? Please provide a convenient date and time."
+
+        if "pickup time" not in self.pending_actions:
+            pickup_time = query.strip()
+            self.pending_actions["pickup time"] = pickup_time
+            return f"Got it! Your pickup is scheduled for {pickup_time}. Please ensure the item is in its original condition and securely packed. Once we receive it, your replacement will be processed."
+
+        # Clear the flow once complete
+        self.pending_actions.clear()
+        return "Replacement process complete. Let me know if there's anything else I can help with!"
+
+    def extract_order_number(self, query):
+        """Extracts the order number from the query."""
+        return extract_order_number(query)
+
+    def extract_tracking_number(self, query):
+        """Extracts the tracking number from the query."""
+        return extract_tracking_number(query)
+
+    def clear_context(self):
+        """Reset the context if required (placeholder for compatibility)."""
+        self.pending_actions.clear()