def _present_count(df, column):
    if column not in df.columns:
        return 0

    return df[column].notna().sum()


def _yes_count(df, column):
    if column not in df.columns:
        return 0

    return (
        df[column]
        .fillna("")
        .astype(str)
        .str.strip()
        .str.lower()
        .eq("oui")
        .sum()
    )


def _complete_rows_count(df, columns):
    available_columns = [column for column in columns if column in df.columns]

    if not available_columns:
        return 0

    return df[available_columns].notna().all(axis=1).sum()


def _os_completion_rates(df):
    if "OS_Mobile" not in df.columns or "Contrat_Signe" not in df.columns:
        return {}

    valid_os = df["OS_Mobile"].isin(("Android", "iOS"))
    df = df[valid_os]

    if df.empty:
        return {}

    rates = {}
    signed_contracts = (
        _contract_mask(df)
    )

    for os_name, group in df.groupby("OS_Mobile"):
        if not os_name or group.empty:
            continue

        completed = signed_contracts.loc[group.index].sum()
        rates[str(os_name).lower()] = round((completed / len(group)) * 100, 2)

    return rates


def _contract_mask(df):
    mask = False

    if "Contrat_Signe" in df.columns:
        mask = (
            df["Contrat_Signe"]
            .fillna("")
            .astype(str)
            .str.strip()
            .str.lower()
            .eq("oui")
        )

    if "Statut_Lead" in df.columns:
        status_mask = (
            df["Statut_Lead"]
            .fillna("")
            .astype(str)
            .str.strip()
            .str.lower()
            .eq("souscrit")
        )
        mask = mask | status_mask

    return mask


def _contract_count(df):
    if "Contrat_Signe" not in df.columns and "Statut_Lead" not in df.columns:
        return 0

    return _contract_mask(df).sum()


def _os_avg_completion_days(df):
    required_columns = {"OS_Mobile", "Date_Contact", "Date_Souscription"}

    if not required_columns.issubset(df.columns):
        return {}

    df = df[df["OS_Mobile"].isin(("Android", "iOS"))]
    signed_df = df[_contract_mask(df)].copy()

    if signed_df.empty:
        return {}

    signed_df["completion_days"] = (
        signed_df["Date_Souscription"] - signed_df["Date_Contact"]
    ).dt.days
    signed_df = signed_df[signed_df["completion_days"].notna()]

    if signed_df.empty:
        return {}

    return {
        str(os_name).lower(): round(group["completion_days"].mean(), 1)
        for os_name, group in signed_df.groupby("OS_Mobile")
        if os_name
    }


def calculate_kpis(df):
    total_leads = len(df)
    telephone_count = _present_count(df, "Telephone")
    email_count = _present_count(df, "Email")
    biometric_count = _yes_count(df, "Verification_Biometrique")
    personal_info_count = _complete_rows_count(
        df,
        ("Nom", "Prenom", "Ville", "Offre_Souhaitee"),
    )
    app_installs_count = _present_count(df, "Date_Installation_App")
    contracts_count = _contract_count(df)

    return {
        "total_leads": int(total_leads),
        "telephone_count": int(telephone_count),
        "email_count": int(email_count),
        "biometric_count": int(biometric_count),
        "personal_info_count": int(personal_info_count),
        "app_installs_count": int(app_installs_count),
        "contracts_count": int(contracts_count),
        "os_completion_rates": _os_completion_rates(df),
        "os_avg_completion_days": _os_avg_completion_days(df),
    }