import sys, subprocess, os, io, traceback import pandas as pd import numpy as np import plotly.express as px import gradio as gr from scipy.stats import skew from sklearn.preprocessing import LabelEncoder, MinMaxScaler, StandardScaler # ----------------------------- # 1. HELPER FUNCTIONS (File Reading & Merge) # ----------------------------- def read_file(file_obj): if file_obj is None: return None if isinstance(file_obj, (tuple,list)): return io.BytesIO(file_obj[1]) if isinstance(file_obj, str) and os.path.exists(file_obj): return open(file_obj,"rb") return file_obj def smart_merge_files(file_objs): if not file_objs: return None dfs = [] if not isinstance(file_objs, list): file_objs = [file_objs] for f_obj in file_objs: f = read_file(f_obj) try: df = pd.read_csv(f) except: try: df = pd.read_excel(f) except: continue dfs.append(df) if not dfs: return None if len(dfs) == 1: return dfs[0] merged_df = dfs[0] for i in range(1, len(dfs)): current_df = dfs[i] common_cols = list(set(merged_df.columns) & set(current_df.columns)) if common_cols: join_key = common_cols[0] merged_df = pd.merge(merged_df, current_df, on=join_key, how='left') return merged_df # ----------------------------- # 2. THE ANALYST BRAIN (Auto-Dashboard Logic) # ----------------------------- def generate_smart_dashboard(df): if df is None or df.empty: return None, None, None, None num_cols = df.select_dtypes(include=np.number).columns.tolist() cat_cols = df.select_dtypes(include='object').columns.tolist() date_col = None is_month_name = False month_order = { 'jan':1, 'feb':2, 'mar':3, 'apr':4, 'may':5, 'jun':6, 'jul':7, 'aug':8, 'sep':9, 'oct':10, 'nov':11, 'dec':12, 'january':1, 'february':2, 'march':3, 'april':4, 'may':5, 'june':6, 'july':7, 'august':8, 'september':9, 'october':10, 'november':11, 'december':12 } for col in df.columns: if pd.api.types.is_datetime64_any_dtype(df[col]): date_col = col; break if df[col].dtype == 'object': sample_val = str(df[col].iloc[0]).lower().strip()[:3] if sample_val in ['jan', 'feb', 'mar', 'apr', 'may', 'jun', 'jul', 'aug', 'sep', 'oct', 'nov', 'dec']: date_col = col is_month_name = True break if 'date' in col.lower() and df[col].astype(str).str.match(r'\d{4}-\d{2}-\d{2}').any(): try: df[col] = pd.to_datetime(df[col], errors='coerce') date_col = col; break except: continue keywords = ['sale', 'profit', 'amount', 'price', 'revenue', 'cost', 'total', 'marks'] target_metric = None for col in num_cols: if any(k in col.lower() for k in keywords): target_metric = col; break if not target_metric and num_cols: target_metric = num_cols[0] # 🔥 Enhanced KPI Cards specifically designed for poster screenshots kpi_html = "
" display_cols = [c for c in num_cols if any(k in c.lower() for k in keywords)] if not display_cols: display_cols = num_cols[:3] for col in display_cols[:4]: val = df[col].sum() val_str = f"{val/1000000:.2f}M" if val > 1000000 else (f"{val/1000:.2f}K" if val > 1000 else f"{val:.2f}") kpi_html += f"""

Total {col}

{val_str}

""" kpi_html += "
" fig1, fig2, fig3 = None, None, None valid_cat_cols = [c for c in cat_cols if c != date_col] # Updated chart templates for a cleaner, more professional look chart_template = "plotly_white" if valid_cat_cols and target_metric: cat_1 = valid_cat_cols[0] df_g1 = df.groupby(cat_1)[target_metric].sum().reset_index().sort_values(target_metric, ascending=False).head(10) fig1 = px.bar(df_g1, x=cat_1, y=target_metric, color=target_metric, color_continuous_scale="Viridis", title=f"📊 Analysis by {cat_1}", template=chart_template) fig1.update_layout(margin=dict(l=20, r=20, t=50, b=20), font=dict(family="Inter, sans-serif")) if date_col and target_metric: trend_df = df.groupby(date_col)[target_metric].sum().reset_index() if is_month_name: trend_df['month_num'] = trend_df[date_col].apply(lambda x: month_order.get(str(x).lower().strip()[:3], 99)) trend_df = trend_df.sort_values('month_num') else: trend_df = trend_df.sort_values(date_col) fig2 = px.line(trend_df, x=date_col, y=target_metric, title=f"📈 Monthly Trend ({date_col})", markers=True, template=chart_template) fig2.update_traces(line=dict(width=3, color="#4f46e5"), marker=dict(size=8, color="#0ea5e9")) fig2.update_layout(margin=dict(l=20, r=20, t=50, b=20), font=dict(family="Inter, sans-serif")) elif len(valid_cat_cols) > 0 and target_metric: cat_2 = valid_cat_cols[0] if len(valid_cat_cols)==1 else valid_cat_cols[1] fig2 = px.bar(df, x=cat_2, y=target_metric, title=f"Analysis by {cat_2}", template=chart_template) fig2.update_layout(margin=dict(l=20, r=20, t=50, b=20)) if len(num_cols) >= 2: fig3 = px.scatter(df, x=num_cols[0], y=num_cols[1], color=valid_cat_cols[0] if valid_cat_cols else None, title=f"Correlation: {num_cols[0]} vs {num_cols[1]}", template=chart_template) elif valid_cat_cols: fig3 = px.pie(df, names=valid_cat_cols[0], values=target_metric, title=f"Distribution by {valid_cat_cols[0]}", hole=0.4, template=chart_template) else: fig3 = px.histogram(df, x=target_metric, title="Distribution", template=chart_template) if fig3: fig3.update_layout(margin=dict(l=20, r=20, t=50, b=20), font=dict(family="Inter, sans-serif")) return kpi_html, fig1, fig2, fig3 # ----------------------------- # 3. REPORTING & CLEANING UTILS # ----------------------------- def data_scan_report(df): report = {} report['shape'] = df.shape report['missing'] = df.isna().sum().to_dict() report['duplicates'] = df.duplicated().sum() report['numerical_cols'] = df.select_dtypes(include=np.number).columns.tolist() report['categorical_cols'] = df.select_dtypes(include='object').columns.tolist() try: report['skew'] = {col: skew(df[col].dropna()) for col in report['numerical_cols']} except: report['skew'] = {} report['outliers'] = {} for col in report['numerical_cols']: Q1 = df[col].quantile(0.25); Q3 = df[col].quantile(0.75); IQR = Q3 - Q1 report['outliers'][col] = df[(df[col] < Q1 - 1.5*IQR) | (df[col] > Q3 + 1.5*IQR)].shape[0] return report def suggest_methods(df, report): missing_num = {col: ('mean' if abs(report['skew'].get(col,0)) < 0.5 else 'median') for col in report['numerical_cols'] if report['missing'][col]>0} missing_cat = {col: 'mode' for col in report['categorical_cols'] if report['missing'][col]>0} encode = {col: ('OneHot' if df[col].nunique() <= 5 else 'Label') for col in report['categorical_cols'] if df[col].nunique()>1} outliers = {col: True for col in report['numerical_cols'] if report['outliers'].get(col,0) > 0} num_ranges = {col: df[col].max() - df[col].min() for col in report['numerical_cols'] if not df[col].empty} needs_norm = any(rng > 10 for rng in num_ranges.values()) return missing_num, missing_cat, encode, needs_norm, outliers def apply_preprocessing(df, miss_num, miss_cat, enc, norm_method, rem_out): df_proc = df.copy() for c, m in miss_num.items(): if c in df_proc: df_proc[c].fillna(df_proc[c].mean() if m=='mean' else df_proc[c].median(), inplace=True) for c, m in miss_cat.items(): if c in df_proc: df_proc[c].fillna(df_proc[c].mode()[0], inplace=True) for c, m in enc.items(): if c in df_proc: if m=='OneHot': df_proc = pd.get_dummies(df_proc, columns=[c], prefix=c) else: df_proc[c] = LabelEncoder().fit_transform(df_proc[c].astype(str)) for c, rem in rem_out.items(): if rem and c in df_proc: Q1=df_proc[c].quantile(0.25); Q3=df_proc[c].quantile(0.75); IQR=Q3-Q1 df_proc = df_proc[~((df_proc[c] < Q1-1.5*IQR)|(df_proc[c] > Q3+1.5*IQR))] if norm_method in ['MinMax','Standard']: cols = df_proc.select_dtypes(include=np.number).columns scaler = MinMaxScaler() if norm_method=='MinMax' else StandardScaler() df_proc[cols] = scaler.fit_transform(df_proc[cols]) return df_proc def dict_to_df(d): return pd.DataFrame(list(d.items()), columns=['Column','Method']) def df_to_dict(df, val_type=str): if df.empty: return {} s = pd.Series(df.iloc[:,1].values, index=df.iloc[:,0]) if val_type==bool: s = s.map(lambda x: str(x).lower()=='true') return s.to_dict() # ----------------------------- # 4. CALLBACKS # ----------------------------- def on_file_upload(file_objs): try: df = smart_merge_files(file_objs) if df is None: return ("**Dataset shape:** (0, 0)", "**Duplicates:** 0", pd.DataFrame(), pd.DataFrame(), pd.DataFrame(), "", pd.DataFrame(), pd.DataFrame(), pd.DataFrame(), "None", pd.DataFrame()) report = data_scan_report(df) m_num, m_cat, enc, needs_norm, out = suggest_methods(df, report) shape_str = f"**Dataset shape:** {report['shape']}" dupes_str = f"**Duplicates:** {report['duplicates']}" types_str = f"**Columns types:**\nNumerical: {report['numerical_cols']}\nCategorical: {report['categorical_cols']}" df_missing = pd.DataFrame(list(report['missing'].items()), columns=['Column', 'Missing']) df_skew = pd.DataFrame(list(report['skew'].items()), columns=['Column', 'Skew']) df_outliers_summary = pd.DataFrame(list(report['outliers'].items()), columns=['Column', 'Outlier Count']) norm_val = "MinMax" if needs_norm else "None" return (shape_str, dupes_str, df_missing, df_skew, df_outliers_summary, types_str, dict_to_df(m_num), dict_to_df(m_cat), dict_to_df(enc), norm_val, dict_to_df(out)) except Exception as e: return (f"Error: {str(e)}", "", pd.DataFrame(), pd.DataFrame(), pd.DataFrame(), "", pd.DataFrame(), pd.DataFrame(), pd.DataFrame(), "None", pd.DataFrame()) def on_clean_and_dashboard(file_objs, mn, mc, en, nrm, ro): try: df = smart_merge_files(file_objs) df_clean = apply_preprocessing(df, df_to_dict(mn), df_to_dict(mc), df_to_dict(en), nrm, df_to_dict(ro, bool)) kpi, fig1, fig2, fig3 = generate_smart_dashboard(df_clean) df_clean.to_csv("cleaned_data.csv", index=False) return f"✅ Data successfully cleaned! ({len(df_clean)} rows remaining)", df_clean.head(10).to_html(classes='table'), "cleaned_data.csv", kpi, fig1, fig2, fig3 except Exception as e: return f"Error: {e}", None, None, None, None, None, None def on_dashboard_only(file_objs): try: df = smart_merge_files(file_objs) kpi, fig1, fig2, fig3 = generate_smart_dashboard(df) return f"⚠️ Analyzing raw data directly ({len(df)} rows)", df.head(10).to_html(classes='table'), None, kpi, fig1, fig2, fig3 except Exception as e: return f"Error: {e}", None, None, None, None, None, None # ----------------------------- # 5. UI LAYOUT & PREMIUM STYLING # ----------------------------- custom_css = """ /* Font Import for modern look */ @import url('https://fonts.googleapis.com/css2?family=Inter:wght@400;500;600;700;800&display=swap'); body, .gradio-container { font-family: 'Inter', sans-serif !important; background-color: #f8fafc; } /* 🔥 TITLE */ .main-title { text-align: center; font-size: 52px; font-weight: 900; background: linear-gradient(90deg, #6366f1, #ec4899, #22c55e); -webkit-background-clip: text; -webkit-text-fill-color: transparent; margin-bottom: 5px; } /* Sleek Subtitle */ .main-subtitle { text-align: center; font-size: 18px; color: #64748b; margin-bottom: 35px; font-weight: 500; letter-spacing: 0.5px; } /* Section Headers */ .section-header { font-size: 20px; font-weight: 700; margin-bottom: 15px; color: #0f172a; border-bottom: 2px solid #e2e8f0; padding-bottom: 10px; display: flex; align-items: center; gap: 8px; } /* Tab Styling override for cleaner UI */ .tabs { border-radius: 12px !important; overflow: hidden; box-shadow: 0 4px 6px -1px rgba(0, 0, 0, 0.05) !important; } """ with gr.Blocks( theme=gr.themes.Soft( primary_hue="indigo", secondary_hue="blue", neutral_hue="slate", font=[gr.themes.GoogleFont("Inter"), "sans-serif"] ), css=custom_css ) as demo: gr.Markdown("
🤖 AI-Based Data Preprocessing and Analytics System
") gr.Markdown("
Automated Data Cleaning • Smart Visualization Dashboard • Actionable Insights
") with gr.Tabs(): # 🔷 TAB 1: UPLOAD & PREPROCESSING with gr.Tab("⚙️ Upload & Setup Configuration"): file_in = gr.File(label="📂 Upload Dataset (CSV/Excel)", file_count="multiple") with gr.Row(): # --- LEFT PANEL: Dataset Summary --- with gr.Column(scale=1): gr.Markdown("
📊 Dataset Summary
") with gr.Group(): txt_shape = gr.Markdown("**Dataset shape:** (waiting...)") txt_dupes = gr.Markdown("**Duplicates:** 0") txt_types = gr.Markdown("**Columns types:**\nNumerical: []\nCategorical: []") gr.Markdown("
**Missing Values Snapshot:**") df_missing_summary = gr.Dataframe(headers=["Column", "Missing Count"], interactive=False) gr.Markdown("**Skewness (Numerical):**") df_skewness = gr.Dataframe(headers=["Column", "Skew Score"], interactive=False) gr.Markdown("**Outliers Detected (IQR):**") df_outliers_summary = gr.Dataframe(headers=["Column", "Outlier Count"], interactive=False) # --- RIGHT PANEL: Preprocessing Configuration --- with gr.Column(scale=1): gr.Markdown("
🛠️ Automated Pipeline Configuration
") mn = gr.Dataframe(headers=["Column", "Method"], label="Impute Numerical Variables", interactive=True) mc = gr.Dataframe(headers=["Column", "Method"], label="Impute Categorical Variables", interactive=True) en = gr.Dataframe(headers=["Column", "Method"], label="Feature Encoding", interactive=True) with gr.Row(): nrm = gr.Dropdown(choices=["MinMax", "Standard", "None"], label="Scale/Normalize Features", value="MinMax") ro = gr.Dataframe(headers=["Column", "Method"], label="Outlier Management", interactive=True) gr.Markdown("
") with gr.Row(): btn_clean = gr.Button("🚀 Execute Pipeline & Generate Dashboard", variant="primary", size="lg") with gr.Row(): btn_dash = gr.Button("📊 Skip Cleaning (Analyze Raw Data)", variant="secondary") status_msg = gr.Markdown("**System Status:** 🟢 Ready for upload.") # 🔷 TAB 2: DASHBOARD with gr.Tab("📈 Analytical Dashboard"): kpi_out = gr.HTML() with gr.Row(): plot1 = gr.Plot(label="Primary Category Breakdown") plot2 = gr.Plot(label="Time-Series / Secondary Trend") with gr.Row(): plot3 = gr.Plot(label="Correlation Matrix / Distribution") # 🔷 TAB 3: DATA PREVIEW with gr.Tab("📄 Cleaned Data Explorer"): gr.Markdown("### Previewing Top 10 Records") clean_head = gr.HTML(label="Data Table") gr.Markdown("
") dl = gr.File(label="💾 Export Cleaned Dataset for ML Models") # ----------------------------- # 🔗 EVENT CONNECTIONS # ----------------------------- file_in.upload( on_file_upload, inputs=file_in, outputs=[ txt_shape, txt_dupes, df_missing_summary, df_skewness, df_outliers_summary, txt_types, mn, mc, en, nrm, ro ] ) btn_clean.click( on_clean_and_dashboard, inputs=[file_in, mn, mc, en, nrm, ro], outputs=[status_msg, clean_head, dl, kpi_out, plot1, plot2, plot3] ) btn_dash.click( on_dashboard_only, inputs=[file_in], outputs=[status_msg, clean_head, dl, kpi_out, plot1, plot2, plot3] ) if __name__ == "__main__": demo.launch()