| import pandas as pd
|
| import os
|
| import logging
|
| import re
|
|
|
|
|
| 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 = {}
|
| 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}")
|
|
|
|
|
| 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)
|
|
|
|
|
| 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."
|
|
|
|
|
| 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."
|
|
|
|
|
| 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()
|
|
|
