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)