Spaces:
Configuration error
Configuration error
File size: 7,260 Bytes
2567e7e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 | 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)
|