import flet as ft from flet import canvas import requests from pages.clients_page import ClientsPage from pages.transporters_page import TransportersPage from pages.destinations_page import DestinationsPage from pages.orders_page import OrdersPage from pages.report_page import ReportPage from pages.profile_page import ProfilePage from pages.subscription_page import Subscription from datetime import datetime from pages.admin_page import Admin from config import API_BASE_URL from email.utils import parsedate_to_datetime class DashboardPage: def __init__(self, page: ft.Page): self.page = page self.placeholder = ft.Container( expand=True, padding=ft.padding.only( top=30, left=20, right=20, bottom=20 ), ) self.all_clients = self.get_all_clients() self.all_orders = self.get_all_orders() self.all_transporters = self.get_all_transporters() self.orders = [] self.profit = self.get_profit() self.logo = ft.Image( src=f"images/{self.page.client_storage.get('logo_filename') or 'truck_logo_black.png'}", width=60, height=60, fit=ft.ImageFit.CONTAIN ) self.subscription_status = ft.Text( "", color=ft.Colors.RED ) self.user = self.get_user() # --- helper to parse created_at values coming from API in multiple formats --- def _parse_dt(self, s): if not s: return None if isinstance(s, datetime): return s try: # First try ISO 8601 (with or without microseconds) return datetime.fromisoformat(s.replace("Z", "+00:00")) except Exception: pass try: # Then try RFC 2822 / RFC 1123 (e.g., 'Mon, 18 Aug 2025 19:22:19 GMT') return parsedate_to_datetime(s) except Exception: pass try: # Fallback: common ISO without microseconds return datetime.strptime(s, "%Y-%m-%dT%H:%M:%S") except Exception: return None def get_user(self): token = self.page.client_storage.get("token") if not token: self.message.value = "Unauthorized: No token" return response = requests.get(f"{API_BASE_URL}/profile/", headers={"Authorization": f"Bearer {token}"}) #print(response.text) return response.json() if response.status_code == 200 else None def on_admin_btn_click(self, e): admin = Admin(self.page, self) self.placeholder.content = admin.build() self.placeholder.update() def get_all_transporters(self): try: token = self.page.client_storage.get("token") headers = {"Authorization": f"Bearer {token}"} response = requests.get(f"{API_BASE_URL}/transporters/", headers=headers) return response.json() if response.status_code == 200 else [] except Exception as e: print("Error loading transporters:", e) def get_all_orders(self): try: token = self.page.client_storage.get("token") headers = {"Authorization": f"Bearer {token}"} response = requests.get(f"{API_BASE_URL}/orders/list", headers=headers) return response.json() if response.status_code == 200 else [] except Exception as e: print("Error loading orders:", e) def get_all_clients(self): try: token = self.page.client_storage.get("token") headers = {"Authorization": f"Bearer {token}"} response = requests.get(f"{API_BASE_URL}/clients/", headers=headers) return response.json() if response.status_code == 200 else [] except Exception as e: print("Error loading clients:", e) def get_profit(self): date_today = datetime.now() month_first_day = date_today.replace(day=1, hour=0, minute=0, second=0, microsecond=0) #print(month_first_day) if len(self.all_orders)>0: for order in self.all_orders: if order.get('status') != 'active': continue dt = self._parse_dt(order.get('created_at')) if dt and dt.date() >= month_first_day.date(): self.orders.append(order) all_profit = 0 for order in self.orders: profit = round(order['received_price'] - order['paid_price'], 2) all_profit += profit return all_profit else: return 0 def _build_dashboard_content(self): self.all_clients = self.get_all_clients() self.all_orders = self.get_all_orders() self.all_transporters = self.get_all_transporters() self.orders = [] self.profit = self.get_profit() summary_cards = ft.Row( controls=[ #self._summary_card("Clients", len(self.all_clients), ft.Icons.PEOPLE), #self._summary_card("Transporters", len(self.all_transporters), ft.Icons.LOCAL_SHIPPING), self._summary_card("Monthly Orders", len(self.orders), ft.Icons.RECEIPT_LONG), self._summary_card("Monthly Profit", self.profit, ft.Icons.SHOW_CHART), ], spacing=20, wrap=True ) # Prepare data for BarChart using flet_charts from collections import defaultdict daily_orders = defaultdict(int) for order in self.orders: try: dt = self._parse_dt(order.get('created_at')) if dt: daily_orders[dt.day] += 1 except Exception as e: print("Invalid date:", order['created_at']) max_value = max(daily_orders.values(), default=1) bar_groups = [] for day in sorted(daily_orders.keys()): value = daily_orders[day] bar_groups.append( ft.BarChartGroup( x=day, bar_rods=[ ft.BarChartRod( from_y=0, to_y=value, width=20, color=ft.Colors.BLUE ) ] ) ) chart_1 = ft.Container( expand=True, bgcolor=ft.Colors.BLUE_50, padding=10, border_radius=20, content=ft.Column([ ft.Text("Monthly Orders", size=16, weight=ft.FontWeight.BOLD), ft.BarChart( bar_groups=bar_groups, border=ft.border.all(1, ft.Colors.GREY_300), left_axis=ft.ChartAxis( labels_size=40, labels=[ ft.ChartAxisLabel( value=i, label=ft.Text(str(i), size=10) ) for i in range(0, max_value + 1, max(1, max_value // 5)) ], ), bottom_axis=ft.ChartAxis( labels=[ ft.ChartAxisLabel( value=day, label=ft.Text(str(day), size=10) ) for day in sorted(daily_orders.keys()) ], ), tooltip_bgcolor=ft.Colors.with_opacity(0.8, ft.Colors.BLUE), expand=True, animate=True ) ]) ) # Calculate daily profit from collections import defaultdict daily_profit = defaultdict(float) for order in self.orders: try: dt = self._parse_dt(order.get('created_at')) if dt: profit = order["received_price"] - order["paid_price"] daily_profit[dt.day] += profit except Exception as e: print("Invalid date in profit chart:", order['created_at']) max_profit = max(daily_profit.values(), default=1) profit_bar_groups = [] for day in sorted(daily_profit.keys()): value = daily_profit[day] profit_bar_groups.append( ft.BarChartGroup( x=day, bar_rods=[ ft.BarChartRod( from_y=0, to_y=value, width=20, color=ft.Colors.GREEN ) ] ) ) chart_2 = ft.Container( expand=True, bgcolor=ft.Colors.BLUE_50, padding=10, border_radius=20, content=ft.Column([ ft.Text("Daily Profit", size=16, weight=ft.FontWeight.BOLD), ft.BarChart( bar_groups=profit_bar_groups, border=ft.border.all(1, ft.Colors.GREY_300), left_axis=ft.ChartAxis( labels_size=40, labels=[ ft.ChartAxisLabel( value=i, label=ft.Text(str(i), size=10) ) for i in range(0, int(max_profit)+1, max(1, int(max_profit)//5)) ], ), bottom_axis=ft.ChartAxis( labels=[ ft.ChartAxisLabel( value=day, label=ft.Text(str(day), size=10) ) for day in sorted(daily_profit.keys()) ], ), tooltip_bgcolor=ft.Colors.with_opacity(0.8, ft.Colors.GREEN), expand=True, animate=True ) ]) ) charts = ft.Row( controls=[chart_1, chart_2], expand=True, spacing=20 ) if self.user: if self.get_subscription(): if self.get_subscription()['end_date'] < datetime.today()-3: self.subscription_status.value = "Your subscription will expire soon. Please Renew!" self.subscription_status.update() elif self.get_subscription()['end_date'] > datetime.today(): self.subscription_status.value = "Your subscription has expired!" self.subscription_status.update() return ft.Column( [ ft.Row( [ summary_cards, self.subscription_status ], alignment=ft.MainAxisAlignment.SPACE_BETWEEN ), charts ], spacing=20, expand=True ) else: return ft.Column( controls=[ summary_cards, charts ], spacing=20, expand=True ) else: return ft.Column( controls=[ summary_cards, charts ], spacing=20, expand=True ) def _summary_card(self, title, value, icon): return ft.Container( content=ft.Column( controls=[ ft.Icon(name=icon, size=30), ft.Text(value, weight=ft.FontWeight.BOLD, size=18), ft.Text(title, size=14) ], spacing=5, alignment=ft.MainAxisAlignment.CENTER ), width=150, height=100, padding=10, alignment=ft.alignment.center, border_radius=10, bgcolor=ft.Colors.BLUE_100 ) def _navigate(self, index): self.placeholder.content.clean() if index == 0: self.placeholder.content = self._build_dashboard_content() elif index == 1: client = ClientsPage(self.page, self) self.placeholder.content = client.build() elif index == 2: transporters = TransportersPage(self.page, self) self.placeholder.content = transporters.build() elif index == 3: destinations = DestinationsPage(self.page, self) self.placeholder.content = destinations.build() elif index == 4: orders = OrdersPage(self.page, self) self.placeholder.content = orders.build() elif index == 5: reports = ReportPage(self.page, self) self.placeholder.content = reports.build() elif index == 6: profile = ProfilePage(self.page, self) self.placeholder.content = profile.build() elif index == 7: subscription = Subscription(self.page, self) self.placeholder.content = subscription.build() self.placeholder.update() def got_to_profile(self, e): self.placeholder.content.clean() profile = ProfilePage(self.page, self) self.placeholder.content = profile.build() self.placeholder.update() def _navigate_or_logout(self, index): if index == 8: # Index of the "Logout" destination self._logout() else: self._navigate(index) def _logout(self): self.page.client_storage.remove("token") self.page.session.clear() self.page.go("/auth") def get_subscription(self): try: user_id = self.page.session.get("user_id") token = self.page.client_storage.get("token") headers = {"Authorization": f"Bearer {token}"} response = requests.get(f"{API_BASE_URL}/admin/subscriptions/{user_id}", headers=headers) return response.json() if response.status_code == 200 else None except Exception as e: print("Error loading clients:", e) def get_current_subscription_plan(self): try: token = self.page.client_storage.get("token") headers = {"Authorization": f"Bearer {token}"} response = requests.get(f"{API_BASE_URL}/subscription/", headers=headers) #print(response.text) return response.json()[-1] if response.status_code == 200 else None except Exception as e: print("Error loading subscription:", e) def build(self): self.status = { 'active':'Active', 'cancelled':'Cancelled', 'expired':'Expired', 'less_than_5_days':'Less than 5 days' } self.subscription = self.get_current_subscription_plan() if self.subscription: self.subscription_status_bottom = ft.Text( f'{self.status[self.subscription['status']] if self.subscription else "None"}', size=12, weight=ft.FontWeight.BOLD, color=ft.Colors.GREEN if self.subscription['status'] != 'expired' else ft.Colors.RED ) else: self.subscription_status_bottom = ft.Text( f'{self.status[self.subscription['status']] if self.subscription else "None"}', size=12, weight=ft.FontWeight.BOLD, color=ft.Colors.RED ) nav_rail = ft.NavigationRail( selected_index=0, label_type=ft.NavigationRailLabelType.ALL, on_change=lambda e: self._navigate_or_logout(e.control.selected_index), destinations=[ ft.NavigationRailDestination( icon=ft.Icons.DASHBOARD, label="Dashboard" ), ft.NavigationRailDestination( icon=ft.Icons.PEOPLE, label="Clients" ), ft.NavigationRailDestination( icon=ft.Icons.LOCAL_SHIPPING, label="Transporters" ), ft.NavigationRailDestination( icon=ft.Icons.LOCATION_ON, label="Address" ), ft.NavigationRailDestination( icon=ft.Icons.RECEIPT_LONG, label="Orders" ), ft.NavigationRailDestination( icon=ft.Icons.ASSESSMENT, label="Reports" ), ft.NavigationRailDestination( icon=ft.Icons.MANAGE_ACCOUNTS, label="Profile" ), ft.NavigationRailDestination( icon=ft.Icons.AUTORENEW, label="Subscription" ), ft.NavigationRailDestination( icon=ft.Icons.LOGOUT, label="Logout" ) ], leading=ft.Container( content=ft.Column( controls=[self.logo], alignment=ft.MainAxisAlignment.CENTER, horizontal_alignment=ft.CrossAxisAlignment.CENTER, spacing=5, ), padding=10 ), #bgcolor = ft.Colors.BLUE, trailing = ft.Column( [ ft.Text( f'\nSubsctiption:', size=12, ), self.subscription_status_bottom ], horizontal_alignment=ft.CrossAxisAlignment.CENTER ) ) self.placeholder.content = self._build_dashboard_content() return ft.Row( controls=[ nav_rail, ft.VerticalDivider(), self.placeholder ], expand=True )