Spaces:
Build error
Build error
| 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() | |