Spaces:
Sleeping
Sleeping
| import pandas as pd | |
| import gspread | |
| import gradio as gr | |
| from oauth2client.service_account import ServiceAccountCredentials | |
| from datetime import datetime | |
| from math import radians, cos, sin, asin, sqrt | |
| # ------------------ AUTH ------------------ | |
| VALID_USERS = { | |
| "[email protected]": "Pass.123", | |
| "[email protected]": "Pass.123", | |
| "[email protected]": "Pass.123" | |
| } | |
| # ------------------ GOOGLE SHEET SETUP ------------------ | |
| scope = ["https://spreadsheets.google.com/feeds", "https://www.googleapis.com/auth/drive"] | |
| creds = ServiceAccountCredentials.from_json_keyfile_name("credentials.json", scope) | |
| client = gspread.authorize(creds) | |
| sheet_file = client.open("userAccess") | |
| # ------------------ HELPERS ------------------ | |
| def load_tab(sheet_name): | |
| try: | |
| df = pd.DataFrame(sheet_file.worksheet(sheet_name).get_all_records()) | |
| return df | |
| except: | |
| return pd.DataFrame([["⚠️ Could not load sheet."]], columns=["Error"]) | |
| def haversine(coord1, coord2): | |
| try: | |
| lon1, lat1 = map(radians, map(float, coord1.split(',')[::-1])) | |
| lon2, lat2 = map(radians, map(float, coord2.split(',')[::-1])) | |
| dlon = lon2 - lon1 | |
| dlat = lat2 - lat1 | |
| a = sin(dlat/2)**2 + cos(lat1) * cos(lat2) * sin(dlon/2)**2 | |
| c = 2 * asin(sqrt(a)) | |
| return round(6371 * c, 2) | |
| except: | |
| return 0.0 | |
| # ------------------ FIELD SALES ------------------ | |
| def load_field_sales(): | |
| df = load_tab("Field Sales") | |
| if df.empty: | |
| return pd.DataFrame(columns=["Date", "Rep", "Order Value", "Order Received", "Location", "DateStr", "KM Travelled"]) | |
| df['Date'] = pd.to_datetime(df.get("Date", datetime.today()), errors='coerce') | |
| df = df.dropna(subset=["Date"]) | |
| df['DateStr'] = df['Date'].dt.date.astype(str) | |
| df["Order Value"] = pd.to_numeric(df.get("Order Value", 0), errors="coerce").fillna(0) | |
| df["KM Travelled"] = 0.0 | |
| for rep in df["Rep"].unique(): | |
| rep_df = df[df["Rep"] == rep].sort_values(by="Date") | |
| prev_coord = None | |
| for idx, row in rep_df.iterrows(): | |
| curr_coord = row.get("Location", "") | |
| if prev_coord: | |
| df.at[idx, "KM Travelled"] = haversine(prev_coord, curr_coord) | |
| prev_coord = curr_coord | |
| return df | |
| def get_field_summary(): | |
| df = load_field_sales() | |
| if df.empty: | |
| return pd.DataFrame([["No data available"]], columns=["Message"]) | |
| summary = df.groupby("Rep").agg({ | |
| "Order Value": "sum", | |
| "Order Received": lambda x: (x == "Yes").sum(), | |
| "KM Travelled": "sum" | |
| }).reset_index().rename(columns={ | |
| "Order Value": "Total Order Value", | |
| "Order Received": "Orders Received" | |
| }) | |
| return summary | |
| # ------------------ TELESALES ------------------ | |
| def get_telesales_summary(): | |
| df = load_tab("TeleSales") | |
| if df.empty or "Rep" not in df.columns: | |
| return pd.DataFrame([["No data available"]], columns=["Message"]) | |
| df["Date"] = pd.to_datetime(df.get("Date", datetime.today()), errors='coerce') | |
| df["DateStr"] = df["Date"].dt.date.astype(str) | |
| df["Order Value"] = pd.to_numeric(df.get("Order Value", 0), errors="coerce").fillna(0) | |
| summary = df.groupby("Rep").agg({ | |
| "Order Value": "sum", | |
| "Order Received": lambda x: (x == "Yes").sum() | |
| }).reset_index().rename(columns={ | |
| "Order Value": "Total Order Value", | |
| "Order Received": "Orders Received" | |
| }) | |
| return summary | |
| # ------------------ COMBINED ORDERS ------------------ | |
| def get_combined_orders(): | |
| fs = get_field_summary() | |
| ts = get_telesales_summary() | |
| fs["Source"] = "Field Sales" | |
| ts["Source"] = "TeleSales" | |
| combined = pd.concat([fs, ts], ignore_index=True) | |
| return combined[["Rep", "Orders Received", "Total Order Value", "Source"]].sort_values(by="Total Order Value", ascending=False) | |
| # ------------------ OEM VISITS ------------------ | |
| def get_oem_summary(): | |
| df = load_tab("OEM Visit") | |
| if df.empty or "Rep" not in df.columns: | |
| return pd.DataFrame([["No data available"]], columns=["Message"]) | |
| df["Date"] = pd.to_datetime(df.get("Date", datetime.today()), errors='coerce') | |
| df["DateStr"] = df["Date"].dt.date.astype(str) | |
| return df.groupby(["Rep", "DateStr"]).size().reset_index(name="OEM Visits") | |
| # ------------------ OTHER TABS ------------------ | |
| def get_requests(): | |
| df = load_tab("Customer Requests") | |
| return df if not df.empty else pd.DataFrame([["No requests yet."]], columns=["Message"]) | |
| def get_listings(): | |
| df = load_tab("CustomerListings") | |
| return df if not df.empty else pd.DataFrame([["No listings found."]], columns=["Message"]) | |
| def get_users(): | |
| df = load_tab("Users") | |
| return df if not df.empty else pd.DataFrame([["No users configured."]], columns=["Message"]) | |
| def get_escalations(): | |
| df = load_field_sales() | |
| col = "Customer Type & Status" | |
| if col in df.columns: | |
| flagged = df[df[col].str.contains("Second", na=False)] | |
| return flagged if not flagged.empty else pd.DataFrame([["No second-hand dealerships flagged."]], columns=["Message"]) | |
| else: | |
| return pd.DataFrame([["⚠️ Column 'Customer Type & Status' not found."]], columns=["Message"]) | |
| # ------------------ GRADIO APP ------------------ | |
| with gr.Blocks() as app: | |
| with gr.Row(): | |
| with gr.Column(visible=True) as login_ui: | |
| gr.Markdown("## 🔐 Login Required") | |
| email = gr.Textbox(label="Email") | |
| password = gr.Textbox(label="Password", type="password") | |
| login_btn = gr.Button("Login") | |
| login_msg = gr.Markdown("") | |
| with gr.Column(visible=False) as main_ui: | |
| gr.Markdown("## 🗂️ CarMat Dashboard") | |
| # --- Summary Tab --- | |
| with gr.Tab("📊 Summary"): | |
| summary_table = gr.Dataframe(label="Combined Orders", value=get_combined_orders) | |
| # --- Field Sales Tab --- | |
| with gr.Tab("🛣️ Field Sales"): | |
| fs_table = gr.Dataframe(label="Field Sales Summary", value=get_field_summary) | |
| fs_raw = gr.Dataframe(label="Raw Field Sales", value=load_field_sales) | |
| # --- Telesales Tab --- | |
| with gr.Tab("📞 TeleSales"): | |
| ts_table = gr.Dataframe(label="TeleSales Summary", value=get_telesales_summary) | |
| # --- Orders Tab --- | |
| with gr.Tab("📦 Orders"): | |
| order_table = gr.Dataframe(label="All Orders Combined", value=get_combined_orders) | |
| # --- Escalations --- | |
| with gr.Tab("🚨 Escalations"): | |
| esc_table = gr.Dataframe(value=get_escalations, label="Second-hand Dealerships") | |
| esc_btn = gr.Button("🔄 Refresh") | |
| esc_btn.click(fn=get_escalations, outputs=esc_table) | |
| # --- OEM Visits --- | |
| with gr.Tab("🏭 OEM Visits"): | |
| oem_table = gr.Dataframe(value=get_oem_summary, label="OEM Visit Summary") | |
| oem_refresh = gr.Button("🔄 Refresh OEM") | |
| oem_refresh.click(fn=get_oem_summary, outputs=oem_table) | |
| # --- Requests --- | |
| with gr.Tab("📬 Customer Requests"): | |
| req_table = gr.Dataframe(value=get_requests, label="Customer Requests", interactive=False) | |
| req_refresh = gr.Button("🔄 Refresh Requests") | |
| req_refresh.click(fn=get_requests, outputs=req_table) | |
| # --- Dealership Listings --- | |
| with gr.Tab("📋 Dealership Directory"): | |
| listings_table = gr.Dataframe(value=get_listings, label="Customer Listings") | |
| listings_refresh = gr.Button("🔄 Refresh Listings") | |
| listings_refresh.click(fn=get_listings, outputs=listings_table) | |
| # --- Users --- | |
| with gr.Tab("👤 Users"): | |
| users_table = gr.Dataframe(value=get_users, label="Users") | |
| users_refresh = gr.Button("🔄 Refresh Users") | |
| users_refresh.click(fn=get_users, outputs=users_table) | |
| def do_login(user, pw): | |
| if VALID_USERS.get(user) == pw: | |
| return gr.update(visible=False), gr.update(visible=True), "" | |
| else: | |
| return gr.update(visible=True), gr.update(visible=False), "❌ Invalid login." | |
| login_btn.click(fn=do_login, inputs=[email, password], outputs=[login_ui, main_ui, login_msg]) | |
| app.launch() | |