app.py

import base64
import io
import re
import time
import uuid
import warnings

import matplotlib
import numpy as np
import pandas as pd
import seaborn as sns
from flask import Flask, jsonify, render_template, request, send_file
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import MinMaxScaler, RobustScaler, StandardScaler

matplotlib.use("Agg")
import matplotlib.pyplot as plt  # noqa: E402


app = Flask(__name__)
sns.set_theme(style="whitegrid")

DOWNLOAD_TTL_SECONDS = 30 * 60
MAX_CACHED_DOWNLOADS = 20
DOWNLOAD_CACHE = {}
SUPPORTED_INPUT_EXTENSIONS = {".csv", ".tsv", ".txt", ".json", ".xlsx", ".xls"}
SUPPORTED_OUTPUT_FORMATS = {"csv", "json", "xlsx"}


def _figure_to_base64(fig) -> str:
    buffer = io.BytesIO()
    fig.savefig(buffer, format="png", dpi=120, bbox_inches="tight")
    plt.close(fig)
    buffer.seek(0)
    return base64.b64encode(buffer.read()).decode("utf-8")


def _prune_download_cache() -> None:
    now = time.time()
    expired_tokens = [
        token
        for token, payload in DOWNLOAD_CACHE.items()
        if now - payload["created_at"] > DOWNLOAD_TTL_SECONDS
    ]
    for token in expired_tokens:
        DOWNLOAD_CACHE.pop(token, None)

    overflow = len(DOWNLOAD_CACHE) - MAX_CACHED_DOWNLOADS
    if overflow > 0:
        oldest_tokens = sorted(
            DOWNLOAD_CACHE.items(),
            key=lambda item: item[1]["created_at"],
        )[:overflow]
        for token, _ in oldest_tokens:
            DOWNLOAD_CACHE.pop(token, None)


def _file_extension(filename: str) -> str:
    if not filename or "." not in str(filename):
        return ""
    return "." + str(filename).rsplit(".", 1)[1].lower()


def _build_download_filename(source_filename: str, output_extension: str) -> str:
    stem = re.sub(r"[^A-Za-z0-9_-]+", "_", str(source_filename).rsplit(".", 1)[0]).strip("_")
    safe_stem = stem or "processed_data"
    return f"{safe_stem}_processed.{output_extension}"


def _cache_download(payload_bytes: bytes, filename: str, mimetype: str) -> str:
    _prune_download_cache()
    token = uuid.uuid4().hex
    DOWNLOAD_CACHE[token] = {
        "payload_bytes": payload_bytes,
        "mimetype": mimetype,
        "filename": filename,
        "created_at": time.time(),
    }
    return token


def _safe_float(value, fallback: float) -> float:
    try:
        return float(value)
    except (TypeError, ValueError):
        return fallback


def _safe_int(value, fallback: int) -> int:
    try:
        return int(value)
    except (TypeError, ValueError):
        return fallback


def _is_true(value) -> bool:
    return str(value).lower() == "true"


def _numeric_columns(df: pd.DataFrame) -> list:
    return df.select_dtypes(include=[np.number]).columns.tolist()


def _categorical_columns(df: pd.DataFrame) -> list:
    return df.select_dtypes(include=["object", "category", "string", "bool"]).columns.tolist()


def _build_scaler(method: str):
    if method == "standard":
        return StandardScaler(), "standard"
    if method == "robust":
        return RobustScaler(), "robust"
    return MinMaxScaler(), "minmax"


def _normalize_column_names(df: pd.DataFrame, enabled: bool, transform_log: list) -> pd.DataFrame:
    if not enabled:
        return df

    seen = {}
    renamed = {}
    for original in df.columns:
        normalized = re.sub(r"[^A-Za-z0-9]+", "_", str(original).strip().lower()).strip("_")
        normalized = normalized or "column"
        if normalized in seen:
            seen[normalized] += 1
            normalized = f"{normalized}_{seen[normalized]}"
        else:
            seen[normalized] = 0
        renamed[original] = normalized

    if list(df.columns) != list(renamed.values()):
        df = df.rename(columns=renamed)
        transform_log.append("Normalized column names to lowercase snake_case.")
    else:
        transform_log.append("Column names were already normalized.")
    return df


def _trim_text_columns(df: pd.DataFrame, enabled: bool, transform_log: list) -> pd.DataFrame:
    if not enabled:
        return df

    text_cols = df.select_dtypes(include=["object", "string"]).columns.tolist()
    if not text_cols:
        transform_log.append("Skipped text trimming because no text columns were found.")
        return df

    changed_cols = 0
    for col in text_cols:
        series = df[col]
        trimmed = series.where(series.isna(), series.astype(str).str.strip())
        if not trimmed.equals(series):
            changed_cols += 1
        df[col] = trimmed

    transform_log.append(f"Trimmed leading/trailing spaces in {changed_cols} text columns.")
    return df


def _auto_cast_numeric_columns(
    df: pd.DataFrame,
    enabled: bool,
    min_success_ratio: float,
    transform_log: list,
) -> tuple:
    if not enabled:
        return df, []

    object_cols = df.select_dtypes(include=["object", "string"]).columns.tolist()
    converted_cols = []
    for col in object_cols:
        numeric_series = pd.to_numeric(df[col], errors="coerce")
        non_null_count = int(df[col].notna().sum())
        if non_null_count == 0:
            continue
        success_ratio = float(numeric_series.notna().sum() / non_null_count)
        if success_ratio >= min_success_ratio:
            df[col] = numeric_series
            converted_cols.append(col)

    if converted_cols:
        transform_log.append(
            f"Auto-cast {len(converted_cols)} columns to numeric using ratio >= {min_success_ratio:.2f}."
        )
    else:
        transform_log.append("No text columns qualified for auto numeric casting.")
    return df, converted_cols


def _drop_high_missing_columns(
    df: pd.DataFrame,
    enabled: bool,
    threshold: float,
    transform_log: list,
) -> tuple:
    if not enabled:
        return df, []

    if df.empty:
        transform_log.append("Skipped high-missing column drop because dataset is empty after preprocessing.")
        return df, []

    missing_ratio = df.isna().mean()
    dropped_cols = [col for col, ratio in missing_ratio.items() if float(ratio) > threshold]
    if dropped_cols:
        df = df.drop(columns=dropped_cols)
        transform_log.append(
            f"Dropped {len(dropped_cols)} columns with missing ratio above {threshold:.2f}."
        )
    else:
        transform_log.append(
            f"No columns exceeded missing ratio threshold {threshold:.2f}."
        )
    return df, dropped_cols


def _apply_missing_strategy(df: pd.DataFrame, strategy: str, transform_log: list) -> pd.DataFrame:
    numeric_cols = _numeric_columns(df)

    if strategy == "drop":
        before_rows = len(df)
        df = df.dropna()
        transform_log.append(f"Dropped {before_rows - len(df)} rows with missing values.")
    elif strategy in {"mean", "median"}:
        if numeric_cols:
            imputer = SimpleImputer(strategy=strategy)
            df[numeric_cols] = imputer.fit_transform(df[numeric_cols])
            transform_log.append(f"Imputed missing numeric values using {strategy}.")
        else:
            transform_log.append("Skipped numeric imputation because no numeric columns were found.")
    elif strategy == "most_frequent":
        imputer = SimpleImputer(strategy="most_frequent")
        df[df.columns] = imputer.fit_transform(df[df.columns])
        transform_log.append("Imputed missing values with the most frequent value per column.")
    elif strategy == "zero":
        if numeric_cols:
            df[numeric_cols] = df[numeric_cols].fillna(0.0)
            transform_log.append("Filled missing numeric values with zero.")
        else:
            transform_log.append("Skipped zero fill because no numeric columns were found.")
    else:
        transform_log.append("Missing-value strategy set to no change.")

    return df


def _add_datetime_features(df: pd.DataFrame, enabled: bool, transform_log: list) -> tuple:
    if not enabled:
        return df, 0

    candidate_cols = df.select_dtypes(include=["object", "string"]).columns
    generated_features = 0

    for col in candidate_cols:
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", UserWarning)
            parsed = pd.to_datetime(df[col], errors="coerce")
        parse_ratio = float(parsed.notna().mean())
        if parse_ratio < 0.7:
            continue

        safe_name = str(col).strip().replace(" ", "_")
        df[f"{safe_name}_year"] = parsed.dt.year
        df[f"{safe_name}_month"] = parsed.dt.month
        df[f"{safe_name}_day"] = parsed.dt.day
        generated_features += 3

    if generated_features:
        transform_log.append(f"Generated {generated_features} datetime feature columns.")
    else:
        transform_log.append("Datetime feature generation was enabled, but no columns passed the parse threshold.")

    return df, generated_features


def _encode_categoricals(
    df: pd.DataFrame,
    enabled: bool,
    max_levels: int,
    transform_log: list,
) -> pd.DataFrame:
    if not enabled:
        return df

    categorical_cols = _categorical_columns(df)
    if not categorical_cols:
        transform_log.append("Skipped one-hot encoding because no categorical columns were found.")
        return df

    eligible_cols = []
    skipped_cols = []
    for col in categorical_cols:
        levels = int(df[col].nunique(dropna=True))
        if levels <= max_levels:
            eligible_cols.append(col)
        else:
            skipped_cols.append(f"{col}({levels})")

    if eligible_cols:
        df = pd.get_dummies(df, columns=eligible_cols, dtype=float)
        transform_log.append(
            f"One-hot encoded {len(eligible_cols)} categorical columns (max levels: {max_levels})."
        )
    else:
        transform_log.append(
            f"Skipped one-hot encoding because all categorical columns exceeded {max_levels} levels."
        )

    if skipped_cols:
        suffix = "..." if len(skipped_cols) > 5 else ""
        preview = ", ".join(skipped_cols[:5])
        transform_log.append(f"High-cardinality columns left unchanged: {preview}{suffix}")

    return df


def _drop_constant_columns(df: pd.DataFrame, enabled: bool, transform_log: list) -> pd.DataFrame:
    if not enabled:
        return df

    constant_cols = [col for col in df.columns if df[col].nunique(dropna=False) <= 1]
    if not constant_cols:
        transform_log.append("No constant columns were found.")
        return df

    df = df.drop(columns=constant_cols)
    transform_log.append(f"Dropped {len(constant_cols)} constant-value columns.")
    return df


def _clip_outliers(df: pd.DataFrame, enabled: bool, sigma: float, transform_log: list) -> pd.DataFrame:
    if not enabled:
        return df

    numeric_cols = _numeric_columns(df)
    clipped_cols = 0

    for col in numeric_cols:
        series = pd.to_numeric(df[col], errors="coerce")
        std = float(series.std())
        if std == 0 or np.isnan(std):
            continue

        mean = float(series.mean())
        lower = mean - (sigma * std)
        upper = mean + (sigma * std)
        clipped_series = series.clip(lower=lower, upper=upper)
        if not clipped_series.equals(series):
            clipped_cols += 1
        df[col] = clipped_series

    transform_log.append(f"Applied outlier clipping with sigma={sigma:.2f} on {clipped_cols} columns.")
    return df


def _scale_numeric_data(
    df: pd.DataFrame,
    enabled: bool,
    method: str,
    transform_log: list,
) -> pd.DataFrame:
    if not enabled:
        return df

    numeric_cols = _numeric_columns(df)
    if not numeric_cols:
        transform_log.append("Skipped scaling because no numeric columns were found.")
        return df

    scaler, normalized_method = _build_scaler(method)
    numeric_frame = df[numeric_cols].apply(pd.to_numeric, errors="coerce")
    nan_mask = numeric_frame.isna()
    fill_values = numeric_frame.mean(numeric_only=True).fillna(0.0)
    scaled_values = scaler.fit_transform(numeric_frame.fillna(fill_values))
    scaled_df = pd.DataFrame(scaled_values, columns=numeric_cols, index=df.index).mask(nan_mask)
    df[numeric_cols] = scaled_df

    transform_log.append(
        f"Scaled {len(numeric_cols)} numeric columns using {normalized_method} scaling."
    )
    return df


def _build_numeric_summary(df: pd.DataFrame) -> dict:
    numeric_cols = _numeric_columns(df)
    if not numeric_cols:
        return {}

    stats = df[numeric_cols].describe().round(4).fillna(0)
    return {
        col: {key: float(value) for key, value in stats[col].to_dict().items()}
        for col in stats.columns
    }


def _build_charts(df: pd.DataFrame) -> dict:
    charts = {}
    numeric_cols = _numeric_columns(df)
    if not numeric_cols:
        return charts

    top_cols = numeric_cols[: min(6, len(numeric_cols))]
    plot_frame = df[top_cols].melt(var_name="column", value_name="value").dropna()

    if not plot_frame.empty:
        fig, ax = plt.subplots(figsize=(9, 4.6))
        sns.histplot(
            data=plot_frame,
            x="value",
            hue="column",
            bins=20,
            stat="count",
            common_norm=False,
            element="step",
            fill=True,
            alpha=0.25,
            ax=ax,
        )
        ax.set_title("Distribution of Numeric Columns")
        ax.set_xlabel("Value")
        ax.set_ylabel("Count")
        charts["histogram"] = _figure_to_base64(fig)

    corr = df[top_cols].corr(numeric_only=True).fillna(0)
    if not corr.empty:
        fig, ax = plt.subplots(figsize=(7, 5))
        sns.heatmap(
            corr,
            cmap="crest",
            linewidths=0.3,
            cbar=True,
            ax=ax,
        )
        ax.set_title("Correlation Matrix")
        charts["correlation"] = _figure_to_base64(fig)

    box_frame = df[top_cols].apply(pd.to_numeric, errors="coerce")
    if not box_frame.dropna(how="all").empty:
        fig, ax = plt.subplots(figsize=(9, 4.6))
        sns.boxplot(data=box_frame, orient="h", ax=ax)
        ax.set_title("Boxplot of Numeric Columns")
        ax.set_xlabel("Value")
        charts["boxplot"] = _figure_to_base64(fig)

    return charts


def _build_profile(df: pd.DataFrame) -> dict:
    numeric_cols = _numeric_columns(df)
    categorical_cols = _categorical_columns(df)
    return {
        "numeric_count": len(numeric_cols),
        "categorical_count": len(categorical_cols),
        "numeric_columns": numeric_cols[:25],
        "categorical_columns": categorical_cols[:25],
    }


def _ratio(part: int, whole: int) -> float:
    if whole <= 0:
        return 0.0
    return round(float(part / whole), 6)


def _numeric_missing_count(df: pd.DataFrame) -> int:
    numeric_cols = _numeric_columns(df)
    if not numeric_cols:
        return 0
    return int(df[numeric_cols].isna().sum().sum())


def _add_check(checks: list, key: str, status: str, message: str) -> None:
    checks.append({"key": key, "status": status, "message": message})


def _read_uploaded_dataframe(file_obj) -> tuple:
    extension = _file_extension(file_obj.filename)
    if extension not in SUPPORTED_INPUT_EXTENSIONS:
        raise ValueError("Unsupported file type.")

    file_obj.stream.seek(0)
    if extension == ".csv":
        return pd.read_csv(file_obj), extension
    if extension == ".tsv":
        return pd.read_csv(file_obj, sep="\t"), extension
    if extension == ".txt":
        return pd.read_csv(file_obj, sep=None, engine="python"), extension
    if extension == ".json":
        return pd.read_json(file_obj), extension

    # Excel formats (.xlsx/.xls). .xls may require extra engines depending on environment.
    return pd.read_excel(file_obj), extension


def _build_download_artifact(
    df: pd.DataFrame,
    source_filename: str,
    requested_output_format: str,
) -> tuple:
    output_format = str(requested_output_format or "csv").strip().lower()
    if output_format not in SUPPORTED_OUTPUT_FORMATS:
        output_format = "csv"

    if output_format == "json":
        payload_bytes = df.to_json(orient="records", force_ascii=False, indent=2).encode("utf-8")
        mimetype = "application/json"
        filename = _build_download_filename(source_filename, "json")
        return payload_bytes, mimetype, filename, output_format

    if output_format == "xlsx":
        buffer = io.BytesIO()
        with pd.ExcelWriter(buffer, engine="openpyxl") as writer:
            df.to_excel(writer, index=False, sheet_name="processed_data")
        payload_bytes = buffer.getvalue()
        mimetype = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
        filename = _build_download_filename(source_filename, "xlsx")
        return payload_bytes, mimetype, filename, output_format

    payload_bytes = df.to_csv(index=False).encode("utf-8")
    mimetype = "text/csv"
    filename = _build_download_filename(source_filename, "csv")
    return payload_bytes, mimetype, filename, "csv"


def _build_accuracy_report(
    original: pd.DataFrame,
    processed: pd.DataFrame,
    options_used: dict,
    runtime_meta: dict,
) -> dict:
    checks = []

    duplicate_count = int(processed.duplicated().sum())
    if options_used.get("drop_duplicates"):
        status = "pass" if duplicate_count == 0 else "fail"
        _add_check(
            checks,
            "duplicates_removed",
            status,
            f"Duplicate rows after processing: {duplicate_count}.",
        )

    missing_strategy = options_used.get("missing_strategy")
    if missing_strategy in {"mean", "median", "zero"}:
        before_missing = _numeric_missing_count(original)
        after_missing = _numeric_missing_count(processed)
        status = "pass" if after_missing <= before_missing else "fail"
        _add_check(
            checks,
            "numeric_missing_reduction",
            status,
            f"Numeric missing values changed from {before_missing} to {after_missing}.",
        )

    if options_used.get("scale_data") and options_used.get("scale_method") == "minmax":
        tolerance = 1e-6
        violations = []
        for col in _numeric_columns(processed):
            series = processed[col].dropna()
            if series.empty:
                continue
            col_min = float(series.min())
            col_max = float(series.max())
            if col_min < -tolerance or col_max > (1.0 + tolerance):
                violations.append(col)
        status = "pass" if not violations else "fail"
        detail = (
            "All min-max scaled columns are within [0, 1]."
            if not violations
            else f"Columns outside expected min-max range: {', '.join(violations[:5])}."
        )
        _add_check(checks, "minmax_range", status, detail)

    if options_used.get("drop_constant_columns"):
        constant_cols = [col for col in processed.columns if processed[col].nunique(dropna=False) <= 1]
        status = "pass" if not constant_cols else "fail"
        detail = (
            "No constant columns remain."
            if not constant_cols
            else f"Constant columns still present: {', '.join(constant_cols[:5])}."
        )
        _add_check(checks, "constant_columns_removed", status, detail)

    if options_used.get("parse_datetime"):
        generated = int(runtime_meta.get("generated_datetime_features", 0))
        status = "pass" if generated > 0 else "warn"
        _add_check(
            checks,
            "datetime_features",
            status,
            f"Datetime features generated: {generated}.",
        )

    if options_used.get("auto_cast_numeric"):
        converted_cols = runtime_meta.get("auto_casted_columns", [])
        status = "pass" if converted_cols else "warn"
        _add_check(
            checks,
            "auto_numeric_cast",
            status,
            (
                f"Auto-cast numeric columns: {', '.join(converted_cols[:6])}."
                if converted_cols
                else "No columns met auto-cast threshold."
            ),
        )

    if options_used.get("drop_high_missing_cols"):
        dropped = runtime_meta.get("dropped_high_missing_columns", [])
        status = "pass" if dropped else "warn"
        _add_check(
            checks,
            "high_missing_column_drop",
            status,
            (
                f"Dropped high-missing columns: {', '.join(dropped[:6])}."
                if dropped
                else "No columns exceeded high-missing threshold."
            ),
        )

    numeric_values = processed.select_dtypes(include=[np.number]).to_numpy()
    inf_count = int(np.isinf(numeric_values).sum()) if numeric_values.size else 0
    inf_status = "pass" if inf_count == 0 else "fail"
    _add_check(
        checks,
        "finite_numeric_values",
        inf_status,
        f"Infinite numeric values after processing: {inf_count}.",
    )

    pass_fail_checks = [item for item in checks if item["status"] in {"pass", "fail"}]
    pass_count = len([item for item in pass_fail_checks if item["status"] == "pass"])
    fail_count = len([item for item in pass_fail_checks if item["status"] == "fail"])
    warn_count = len([item for item in checks if item["status"] == "warn"])
    score = round((100.0 * pass_count / len(pass_fail_checks)), 2) if pass_fail_checks else 100.0

    return {
        "score": score,
        "pass_count": pass_count,
        "fail_count": fail_count,
        "warn_count": warn_count,
        "checks": checks,
    }


@app.route("/")
def home():
    return render_template("index.html")


@app.route("/about")
def about():
    return render_template("about.html")


@app.route("/developer-guide")
def developer_guide():
    return render_template("developer-guide.html")


@app.get("/api/download/<token>")
def download_processed_csv(token: str):
    _prune_download_cache()
    payload = DOWNLOAD_CACHE.get(token)
    if not payload:
        return jsonify({"error": "Download link expired. Reprocess the dataset and try again."}), 404

    return send_file(
        io.BytesIO(payload["payload_bytes"]),
        mimetype=payload["mimetype"],
        as_attachment=True,
        download_name=payload["filename"],
        max_age=0,
    )


@app.post("/api/process")
def process_data():
    process_start = time.perf_counter()
    if "file" not in request.files:
        return jsonify({"error": "Please upload a data file."}), 400

    file = request.files["file"]
    if not file:
        return jsonify({"error": "Please upload a data file."}), 400

    try:
        df, input_extension = _read_uploaded_dataframe(file)
    except ValueError:
        supported_text = ", ".join(sorted(SUPPORTED_INPUT_EXTENSIONS))
        return jsonify({"error": f"Unsupported file type. Supported: {supported_text}"}), 400
    except Exception:
        return jsonify({"error": "Could not parse file. Check format and structure."}), 400

    if df.empty:
        return jsonify({"error": "Uploaded file is empty."}), 400

    original = df.copy()
    transform_log = []
    transform_log.append(f"Loaded input file type: {input_extension[1:].upper()}.")
    runtime_meta = {
        "generated_datetime_features": 0,
        "auto_casted_columns": [],
        "dropped_high_missing_columns": [],
    }

    normalize_columns = _is_true(request.form.get("normalize_columns"))
    trim_text = _is_true(request.form.get("trim_text"))
    auto_cast_numeric = _is_true(request.form.get("auto_cast_numeric"))
    auto_cast_threshold = _safe_float(request.form.get("auto_cast_threshold"), 0.9)
    auto_cast_threshold = min(max(auto_cast_threshold, 0.5), 1.0)
    drop_high_missing_cols = _is_true(request.form.get("drop_high_missing_cols"))
    missing_col_threshold = _safe_float(request.form.get("missing_col_threshold"), 0.7)
    missing_col_threshold = min(max(missing_col_threshold, 0.05), 0.99)
    preview_rows = _safe_int(request.form.get("preview_rows"), 15)
    preview_rows = min(max(preview_rows, 5), 100)
    requested_output_format = str(request.form.get("output_format", "csv")).strip().lower()

    df = _normalize_column_names(df, normalize_columns, transform_log)
    df = _trim_text_columns(df, trim_text, transform_log)
    df, converted_cols = _auto_cast_numeric_columns(
        df,
        auto_cast_numeric,
        auto_cast_threshold,
        transform_log,
    )
    runtime_meta["auto_casted_columns"] = converted_cols
    df, dropped_cols = _drop_high_missing_columns(
        df,
        drop_high_missing_cols,
        missing_col_threshold,
        transform_log,
    )
    runtime_meta["dropped_high_missing_columns"] = dropped_cols

    if _is_true(request.form.get("drop_duplicates")):
        before_rows = len(df)
        df = df.drop_duplicates()
        transform_log.append(f"Removed {before_rows - len(df)} duplicate rows.")
    else:
        transform_log.append("Duplicate-row removal skipped.")

    missing_strategy = str(request.form.get("missing_strategy", "none")).strip().lower()
    df = _apply_missing_strategy(df, missing_strategy, transform_log)

    parse_datetime = _is_true(request.form.get("parse_datetime"))
    df, generated_datetime_features = _add_datetime_features(df, parse_datetime, transform_log)
    runtime_meta["generated_datetime_features"] = generated_datetime_features

    encode_categoricals = _is_true(request.form.get("encode_categoricals"))
    encoding_max_levels = max(2, _safe_int(request.form.get("encoding_max_levels"), 15))
    df = _encode_categoricals(df, encode_categoricals, encoding_max_levels, transform_log)

    drop_constant_columns = _is_true(request.form.get("drop_constant_columns"))
    df = _drop_constant_columns(df, drop_constant_columns, transform_log)

    clip_enabled = _is_true(request.form.get("clip_outliers"))
    clip_limit = max(0.5, _safe_float(request.form.get("outlier_sigma"), 3.0))
    df = _clip_outliers(df, clip_enabled, clip_limit, transform_log)

    scale_enabled = _is_true(request.form.get("scale_data"))
    scale_method = str(request.form.get("scale_method", "minmax")).strip().lower()
    df = _scale_numeric_data(df, scale_enabled, scale_method, transform_log)

    overview = {
        "rows_before": int(original.shape[0]),
        "rows_after": int(df.shape[0]),
        "cols": int(df.shape[1]),
        "missing_before": int(original.isna().sum().sum()),
        "missing_after": int(df.isna().sum().sum()),
        "duplicates_before": int(original.duplicated().sum()),
        "duplicates_after": int(df.duplicated().sum()),
        "numeric_cols_before": len(_numeric_columns(original)),
        "numeric_cols_after": len(_numeric_columns(df)),
        "categorical_cols_before": len(_categorical_columns(original)),
        "categorical_cols_after": len(_categorical_columns(df)),
    }

    summary = _build_numeric_summary(df)
    generate_charts = _is_true(request.form.get("generate_charts"))
    charts = _build_charts(df) if generate_charts else {}
    if not generate_charts:
        transform_log.append("Chart generation skipped for faster processing.")
    profile = _build_profile(df)
    quality_summary = {
        "missing_ratio_before": _ratio(
            int(original.isna().sum().sum()),
            int(original.shape[0] * original.shape[1]),
        ),
        "missing_ratio_after": _ratio(
            int(df.isna().sum().sum()),
            int(df.shape[0] * df.shape[1]),
        ),
        "duplicate_ratio_before": _ratio(
            int(original.duplicated().sum()),
            int(original.shape[0]),
        ),
        "duplicate_ratio_after": _ratio(
            int(df.duplicated().sum()),
            int(df.shape[0]),
        ),
    }

    options_used = {
        "missing_strategy": missing_strategy,
        "drop_duplicates": _is_true(request.form.get("drop_duplicates")),
        "parse_datetime": parse_datetime,
        "encode_categoricals": encode_categoricals,
        "encoding_max_levels": encoding_max_levels,
        "drop_constant_columns": drop_constant_columns,
        "clip_outliers": clip_enabled,
        "outlier_sigma": clip_limit,
        "scale_data": scale_enabled,
        "scale_method": scale_method,
        "generate_charts": generate_charts,
        "normalize_columns": normalize_columns,
        "trim_text": trim_text,
        "auto_cast_numeric": auto_cast_numeric,
        "auto_cast_threshold": auto_cast_threshold,
        "drop_high_missing_cols": drop_high_missing_cols,
        "missing_col_threshold": missing_col_threshold,
        "preview_rows": preview_rows,
    }

    preview_df = df.head(preview_rows).replace([np.inf, -np.inf], np.nan).fillna("")
    download_bytes, download_mimetype, download_filename, output_format = _build_download_artifact(
        df,
        file.filename,
        requested_output_format,
    )
    options_used["output_format"] = output_format
    accuracy_report = _build_accuracy_report(
        original=original,
        processed=df,
        options_used=options_used,
        runtime_meta=runtime_meta,
    )
    download_token = _cache_download(download_bytes, download_filename, download_mimetype)
    overview["processing_ms"] = int((time.perf_counter() - process_start) * 1000)

    response = {
        "overview": overview,
        "columns": df.columns.tolist(),
        "preview": preview_df.to_dict(orient="records"),
        "summary": summary,
        "charts": charts,
        "profile": profile,
        "quality_summary": quality_summary,
        "accuracy_report": accuracy_report,
        "transform_log": transform_log,
        "options_used": options_used,
        "input_file_type": input_extension[1:],
        "download_filename": download_filename,
        "download_url": f"/api/download/{download_token}",
        "download_expires_in_seconds": DOWNLOAD_TTL_SECONDS,
        "supported_output_formats": sorted(SUPPORTED_OUTPUT_FORMATS),
    }
    return jsonify(response)


if __name__ == "__main__":
    app.run(debug=True, host="0.0.0.0", port=5000)