---

# 6.1
def audit_history(
    clients: pd.DataFrame,
    campaigns: pd.DataFrame,
) -> dict[str, object]:
    campaign_clients = pd.Series(campaigns["client_id"].unique())
    return {
        "n_clients": int(len(clients)),
        "n_campaign_rows": int(len(campaigns)),
        "n_unique_campaign_clients": int(campaigns["client_id"].nunique()),
        "n_products": int(campaigns["product_id"].nunique()),
        "duplicate_client_ids": int(clients["client_id"].duplicated().sum()),
        "duplicate_client_product_rows": int(
            campaigns.duplicated(["client_id", "product_id"]).sum()
        ),
        "missing_campaign_clients": int(
            (~campaign_clients.isin(clients["client_id"])).sum()
        ),
    }


# 6.2
def make_client_folds(
    client_ids: pd.Series,
    n_folds: int,
    seed: int,
) -> pd.DataFrame:
    ids = pd.Series(client_ids).dropna().drop_duplicates()
    try:
        ids = ids.sort_values(kind="mergesort")
    except TypeError:  # смешанные типы id: сортируем по строковому представлению
        ids = ids.sort_values(kind="mergesort", key=lambda x: x.astype(str))
    ids = ids.to_numpy()
    n = len(ids)
    rng = np.random.RandomState(seed)
    perm = rng.permutation(n)
    fold = np.empty(n, dtype=np.int64)
    fold[perm] = np.arange(n) % int(n_folds)
    return pd.DataFrame({"client_id": ids, "fold": fold})


# 6.3
def summarize_campaigns(
    campaigns: pd.DataFrame,
) -> pd.DataFrame:
    columns = [
        "product_id",
        "n_rows",
        "n_offered",
        "n_control",
        "purchase_rate_offered",
        "purchase_rate_control",
    ]
    rows = []
    for product_id, group in campaigns.groupby("product_id", sort=True):
        offered = group[group["offered"] == 1]
        control = group[group["offered"] == 0]
        rows.append(
            {
                "product_id": product_id,
                "n_rows": int(len(group)),
                "n_offered": int(len(offered)),
                "n_control": int(len(control)),
                "purchase_rate_offered": float(offered["purchased"].mean())
                if len(offered)
                else float("nan"),
                "purchase_rate_control": float(control["purchased"].mean())
                if len(control)
                else float("nan"),
            }
        )
    out = pd.DataFrame(rows, columns=columns)
    return out.sort_values("product_id").reset_index(drop=True)


# 6.4
def build_product_table(
    clients: pd.DataFrame,
    campaigns: pd.DataFrame,
    product_id: int,
) -> pd.DataFrame:
    camp = campaigns[campaigns["product_id"] == product_id]
    camp = camp.drop_duplicates("client_id", keep="first")
    cl = clients.drop_duplicates("client_id", keep="first")
    # merge с левой таблицей clients сохраняет порядок clients
    try:
        out = cl.merge(camp, on="client_id", how="inner")
    except ValueError:  # разные dtype ключа: приводим обе стороны к строке
        cl = cl.assign(client_id=cl["client_id"].astype(str))
        camp = camp.assign(client_id=camp["client_id"].astype(str))
        out = cl.merge(camp, on="client_id", how="inner")
    return out.reset_index(drop=True)


# 6.5
def add_region_features(
    clients: pd.DataFrame,
    regions: pd.DataFrame,
) -> pd.DataFrame:
    out = clients.copy()
    reg = regions.drop_duplicates("region_code")
    for col in reg.columns:
        if col == "region_code":
            continue
        mapping = dict(zip(reg["region_code"], reg[col]))
        name = col if col not in out.columns else f"{col}_region"
        out[name] = out["region_code"].map(mapping)
    return out


# 6.6
def make_product_candidates(
    clients: pd.DataFrame,
    products: pd.DataFrame,
) -> pd.DataFrame:
    parts = []
    for row in products.itertuples(index=False):
        col = row.eligibility_column
        if col not in clients.columns:
            continue
        sel = clients.loc[clients[col] == 1, ["client_id"]].copy()
        sel["product_id"] = row.product_id
        parts.append(sel)
    if parts:
        out = pd.concat(parts, ignore_index=True)
    else:
        out = pd.DataFrame({"client_id": [], "product_id": []})
    out = out.drop_duplicates(["client_id", "product_id"])
    out = out.sort_values(["client_id", "product_id"]).reset_index(drop=True)
    return out[["client_id", "product_id"]]


# 6.7
def check_eligibility(
    clients: pd.DataFrame,
    offers: pd.DataFrame,
) -> pd.DataFrame:
    out = offers[["client_id", "offer_id"]].copy()
    cl = clients.drop_duplicates("client_id", keep="first").set_index("client_id")
    elig_cols = {j: f"eligible_p{j}" for j in PRODUCT_IDS}
    known = set(cl.index)
    result = np.zeros(len(out), dtype=bool)
    for i, (cid, oid) in enumerate(zip(out["client_id"], out["offer_id"])):
        if cid not in known:
            continue
        try:
            o = int(oid)
        except (TypeError, ValueError):
            continue
        if o != oid:
            continue
        if o == 0:
            result[i] = True
        elif o in elig_cols:
            col = elig_cols[o]
            result[i] = bool(col in cl.columns and cl.at[cid, col] == 1)
    out["is_eligible"] = result.astype(bool)
    return out


# 6.8
def make_submission(
    clients: pd.DataFrame,
    offers: pd.DataFrame,
) -> pd.DataFrame:
    if offers["client_id"].duplicated().any():
        raise ValueError("duplicate client_id in offers")
    mapping = dict(zip(offers["client_id"], offers["offer_id"]))
    out = pd.DataFrame({"client_id": clients["client_id"].to_numpy()})
    out["offer_id"] = out["client_id"].map(mapping).fillna(0).astype(int)
    return out


# 6.9
def contact_statistics(
    submission: pd.DataFrame,
) -> dict[str, object]:
    offer = submission["offer_id"]
    n_clients = int(len(submission))
    n_contacts = int(offer.isin(PRODUCT_IDS).sum())
    out = {
        "n_clients": n_clients,
        "n_no_contact": int((offer == 0).sum()),
        "n_contacts": n_contacts,
        "contact_rate": float(n_contacts / n_clients) if n_clients else 0.0,
    }
    for j in PRODUCT_IDS:
        out[f"n_offer_{j}"] = int((offer == j).sum())
    return out


# 6.10
def validate_submission_frame(
    submission: pd.DataFrame,
    clients: pd.DataFrame,
    contact_fraction: float = 0.20,
) -> dict[str, object]:
    sub_ids = submission["client_id"]
    client_ids = clients["client_id"].drop_duplicates()
    n_rows = int(len(submission))
    n_unique_clients = int(sub_ids.nunique())
    missing_clients = int((~client_ids.isin(sub_ids)).sum())
    extra_clients = int((~sub_ids.drop_duplicates().isin(client_ids)).sum())
    duplicate_clients = int(sub_ids.duplicated().sum())
    invalid_offer_ids = int((~submission["offer_id"].isin(VALID_OFFER_IDS)).sum())

    product_mask = submission["offer_id"].isin(PRODUCT_IDS).to_numpy()
    known_mask = sub_ids.isin(client_ids).to_numpy()
    eligible = check_eligibility(clients, submission)["is_eligible"].to_numpy()
    ineligible_offers = int((known_mask & product_mask & ~eligible).sum())

    n_contacts = int(product_mask.sum())
    # 1e-9 компенсирует погрешность float (0.29 * 100 = 28.999999999999996)
    max_contacts = int(math.floor(contact_fraction * len(clients) + 1e-9))
    contact_limit_exceeded = bool(n_contacts > max_contacts)

    is_valid = bool(
        missing_clients == 0
        and extra_clients == 0
        and duplicate_clients == 0
        and invalid_offer_ids == 0
        and ineligible_offers == 0
        and not contact_limit_exceeded
        and n_rows == len(client_ids)
        and n_unique_clients == len(client_ids)
    )
    return {
        "is_valid": is_valid,
        "n_rows": n_rows,
        "n_unique_clients": n_unique_clients,
        "missing_clients": missing_clients,
        "extra_clients": extra_clients,
        "duplicate_clients": duplicate_clients,
        "invalid_offer_ids": invalid_offer_ids,
        "ineligible_offers": ineligible_offers,
        "n_contacts": n_contacts,
        "max_contacts": max_contacts,
        "contact_limit_exceeded": contact_limit_exceeded,
    }