Spaces:
Configuration error
Configuration error
| import os | |
| from datetime import datetime | |
| from typing import List, Dict, Any, Optional | |
| import openpyxl | |
| import pandas as pd | |
| from app.config import EXCEL_PATH, SNAPSHOT_DATETIME | |
| from app.core.security import UserContext | |
| class DataStore: | |
| def __init__(self, excel_path: str = str(EXCEL_PATH)): | |
| self.excel_path = excel_path | |
| self.snapshot_datetime = SNAPSHOT_DATETIME | |
| self.accounts: List[Dict[str, Any]] = [] | |
| self.orders: List[Dict[str, Any]] = [] | |
| self.tickets: List[Dict[str, Any]] = [] | |
| self.readme_info: Dict[str, Any] = {} | |
| self.load_data() | |
| def load_data(self): | |
| """Loads all sheets from Excel workbook into memory structures.""" | |
| if not os.path.exists(self.excel_path): | |
| raise FileNotFoundError(f"Data file not found at {self.excel_path}") | |
| wb = openpyxl.load_workbook(self.excel_path, data_only=True) | |
| # 1. README | |
| if "README" in wb.sheetnames: | |
| sheet = wb["README"] | |
| for row in sheet.iter_rows(values_only=True): | |
| if row and len(row) >= 2 and row[0]: | |
| self.readme_info[str(row[0])] = str(row[1]) | |
| # 2. Accounts | |
| if "accounts" in wb.sheetnames: | |
| df_acc = pd.read_excel(self.excel_path, sheet_name="accounts").where(pd.notnull, None) | |
| self.accounts = df_acc.to_dict(orient="records") | |
| for acc in self.accounts: | |
| acc["premium_support"] = bool(acc.get("premium_support", False)) | |
| # 3. Orders | |
| if "orders" in wb.sheetnames: | |
| df_ord = pd.read_excel(self.excel_path, sheet_name="orders").where(pd.notnull, None) | |
| self.orders = df_ord.to_dict(orient="records") | |
| for ord_item in self.orders: | |
| # Convert timestamps to string/datetime | |
| for col in ["booked_at", "pickup_window_start", "pickup_window_end", "pickup_actual_at", "cancellation_requested_at"]: | |
| if pd.notna(ord_item.get(col)): | |
| ord_item[col] = str(ord_item[col]) | |
| else: | |
| ord_item[col] = None | |
| ord_item["carrier_fault"] = bool(ord_item.get("carrier_fault", False)) | |
| ord_item["customer_fault"] = bool(ord_item.get("customer_fault", False)) | |
| # 4. Tickets | |
| if "tickets" in wb.sheetnames: | |
| df_tkt = pd.read_excel(self.excel_path, sheet_name="tickets").where(pd.notnull, None) | |
| self.tickets = df_tkt.to_dict(orient="records") | |
| for tkt in self.tickets: | |
| for col in ["created_at", "last_customer_message_at"]: | |
| if pd.notna(tkt.get(col)): | |
| tkt[col] = str(tkt[col]) | |
| else: | |
| tkt[col] = None | |
| # --- Query Methods with Access Control --- | |
| def get_account(self, account_id: str, user_context: UserContext) -> Optional[Dict[str, Any]]: | |
| """Retrieve account by ID, checking access control.""" | |
| if not user_context.can_access_account(account_id): | |
| return None | |
| for acc in self.accounts: | |
| if acc["account_id"] == account_id: | |
| return acc | |
| return None | |
| def get_accounts(self, user_context: UserContext) -> List[Dict[str, Any]]: | |
| """Retrieve all accounts visible to user_context.""" | |
| if user_context.is_internal: | |
| return self.accounts | |
| return [acc for acc in self.accounts if acc["account_id"] == user_context.account_id] | |
| def get_order(self, order_id: str, user_context: UserContext) -> Optional[Dict[str, Any]]: | |
| """Retrieve order by ID, checking access control.""" | |
| for ord_item in self.orders: | |
| if ord_item["order_id"] == order_id: | |
| if not user_context.can_access_account(ord_item["account_id"]): | |
| return None | |
| return ord_item | |
| return None | |
| def get_orders(self, user_context: UserContext, account_id: Optional[str] = None) -> List[Dict[str, Any]]: | |
| """Retrieve orders visible to user_context, optionally filtered by account_id.""" | |
| results = [] | |
| for ord_item in self.orders: | |
| if account_id and ord_item["account_id"] != account_id: | |
| continue | |
| if user_context.can_access_account(ord_item["account_id"]): | |
| results.append(ord_item) | |
| return results | |
| def get_ticket(self, ticket_id: str, user_context: UserContext) -> Optional[Dict[str, Any]]: | |
| """Retrieve ticket by ID, checking access control.""" | |
| for tkt in self.tickets: | |
| if tkt["ticket_id"] == ticket_id: | |
| if not user_context.can_access_account(tkt["account_id"]): | |
| return None | |
| return tkt | |
| return None | |
| def get_tickets(self, user_context: UserContext, account_id: Optional[str] = None, status: Optional[str] = None) -> List[Dict[str, Any]]: | |
| """Retrieve tickets visible to user_context.""" | |
| results = [] | |
| for tkt in self.tickets: | |
| if account_id and tkt["account_id"] != account_id: | |
| continue | |
| if status and tkt["status"].lower() != status.lower(): | |
| continue | |
| if user_context.can_access_account(tkt["account_id"]): | |
| results.append(tkt) | |
| return results | |
| # --- Calculations --- | |
| def calculate_order_delay_hours(self, order_id: str) -> Optional[float]: | |
| """Calculates late pickup hours relative to pickup_window_end or snapshot time.""" | |
| # Find order | |
| order = None | |
| for o in self.orders: | |
| if o["order_id"] == order_id: | |
| order = o | |
| break | |
| if not order: | |
| return None | |
| window_end_str = order.get("pickup_window_end") | |
| if not window_end_str: | |
| return 0.0 | |
| window_end_dt = datetime.strptime(window_end_str, "%Y-%m-%d %H:%M") | |
| actual_str = order.get("pickup_actual_at") | |
| if actual_str: | |
| compare_dt = datetime.strptime(actual_str, "%Y-%m-%d %H:%M") | |
| else: | |
| # Not yet picked up -> calculate delay relative to current snapshot timestamp | |
| compare_dt = self.snapshot_datetime | |
| if compare_dt > window_end_dt: | |
| diff_hours = (compare_dt - window_end_dt).total_seconds() / 3600.0 | |
| return round(diff_hours, 2) | |
| return 0.0 | |
| def calculate_cancellation_elapsed_minutes(self, order_id: str) -> Optional[float]: | |
| """Calculates elapsed minutes between booked_at and cancellation_requested_at (or snapshot).""" | |
| order = None | |
| for o in self.orders: | |
| if o["order_id"] == order_id: | |
| order = o | |
| break | |
| if not order or not order.get("booked_at"): | |
| return None | |
| booked_dt = datetime.strptime(order["booked_at"], "%Y-%m-%d %H:%M") | |
| cancel_str = order.get("cancellation_requested_at") | |
| if cancel_str: | |
| cancel_dt = datetime.strptime(cancel_str, "%Y-%m-%d %H:%M") | |
| else: | |
| cancel_dt = self.snapshot_datetime | |
| elapsed_mins = (cancel_dt - booked_dt).total_seconds() / 60.0 | |
| return round(elapsed_mins, 1) | |