"""
cleaning.py -- Data cleaning and feature engineering for hedonic pricing analysis.

Provides deterministic, auditable transformations that convert raw
Airbnb listing data into analysis-ready features.  Every function
logs what it changes so that a full cleaning report can be generated.
"""

from __future__ import annotations

import logging
import re
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple, Union

import numpy as np
import pandas as pd

from src.config import (
    DEFAULT_WINSORIZE,
    WINSORIZE_1_99,
    WINSORIZE_2_98,
    WINSORIZE_5_95,
    PRICE_FLOOR,
    PRICE_CEILING,
    QUEBEC_LAT_RANGE,
    QUEBEC_LON_RANGE,
    PROPERTY_TYPE_MAP,
    ROOM_TYPE_MAP,
    LOG_LEVEL,
    LOG_FORMAT,
    LOG_DATE_FORMAT,
)

# ---------------------------------------------------------------------------
# Logger
# ---------------------------------------------------------------------------

logging.basicConfig(format=LOG_FORMAT, datefmt=LOG_DATE_FORMAT, level=LOG_LEVEL)
logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Cleaning audit record
# ---------------------------------------------------------------------------


@dataclass
class CleaningRecord:
    """Accumulates counts and descriptions of every cleaning action."""

    actions: List[Dict[str, Any]] = field(default_factory=list)

    def log(
        self,
        step: str,
        column: str,
        description: str,
        rows_affected: int,
        total_rows: int,
    ) -> None:
        pct = 100.0 * rows_affected / total_rows if total_rows > 0 else 0.0
        entry = {
            "step": step,
            "column": column,
            "description": description,
            "rows_affected": rows_affected,
            "total_rows": total_rows,
            "pct_affected": round(pct, 2),
        }
        self.actions.append(entry)
        logger.info(
            "[%s] %s | %s rows (%.1f%%): %s",
            step, column, f"{rows_affected:,}", pct, description,
        )

    def to_dataframe(self) -> pd.DataFrame:
        return pd.DataFrame(self.actions)

    def summary_dict(self) -> Dict[str, Any]:
        return {
            "total_actions": len(self.actions),
            "total_rows_affected": sum(a["rows_affected"] for a in self.actions),
            "steps": [a["step"] for a in self.actions],
        }


# Module-level record -- reset by ``run_full_cleaning``
_record = CleaningRecord()


def _get_record() -> CleaningRecord:
    return _record


# ---------------------------------------------------------------------------
# 1. Price cleaning
# ---------------------------------------------------------------------------

_PRICE_STRIP_RE = re.compile(r"[^\d.]")


def clean_price(
    df: pd.DataFrame,
    *,
    price_col: str = "price_numeric",
    raw_price_col: str = "price",
    floor: float = PRICE_FLOOR,
    ceiling: float = PRICE_CEILING,
    record: Optional[CleaningRecord] = None,
) -> pd.DataFrame:
    """Clean and standardise the price column.

    Handles both the pre-parsed ``price_numeric`` column in the cleaned
    sample and the raw ``$1,234.00``-style ``price`` string column in
    full Inside Airbnb data.

    Steps
    -----
    1. If *raw_price_col* exists and *price_col* does not, parse the
       raw column (strip ``$``, commas, whitespace) to float.
    2. Replace non-positive values with ``NaN``.
    3. Replace values outside ``[floor, ceiling]`` with ``NaN``.
    """
    rec = record or _get_record()
    n = len(df)
    df = df.copy()

    # Parse raw string price if needed
    if raw_price_col in df.columns and price_col not in df.columns:
        raw = df[raw_price_col].astype(str)
        parsed = raw.apply(lambda s: _PRICE_STRIP_RE.sub("", s))
        df[price_col] = pd.to_numeric(parsed, errors="coerce")
        failed = df[price_col].isnull().sum() - df[raw_price_col].isnull().sum()
        rec.log("clean_price", price_col, "Parsed raw price string to numeric", max(failed, 0), n)
    elif price_col not in df.columns:
        logger.warning("Neither '%s' nor '%s' found in DataFrame", price_col, raw_price_col)
        return df

    before_missing = int(df[price_col].isnull().sum())

    # Non-positive -> NaN
    mask_nonpos = df[price_col] <= 0
    n_nonpos = int(mask_nonpos.sum())
    if n_nonpos > 0:
        df.loc[mask_nonpos, price_col] = np.nan
        rec.log("clean_price", price_col, f"Set {n_nonpos} non-positive prices to NaN", n_nonpos, n)

    # Below floor -> NaN
    mask_low = df[price_col] < floor
    n_low = int(mask_low.sum())
    if n_low > 0:
        df.loc[mask_low, price_col] = np.nan
        rec.log("clean_price", price_col, f"Set {n_low} prices below floor (${floor}) to NaN", n_low, n)

    # Above ceiling -> NaN
    mask_high = df[price_col] > ceiling
    n_high = int(mask_high.sum())
    if n_high > 0:
        df.loc[mask_high, price_col] = np.nan
        rec.log("clean_price", price_col, f"Set {n_high} prices above ceiling (${ceiling:,}) to NaN", n_high, n)

    after_missing = int(df[price_col].isnull().sum())
    total_set_nan = after_missing - before_missing
    rec.log(
        "clean_price", price_col,
        f"Total new NaN from price cleaning: {total_set_nan} "
        f"({after_missing} total missing)",
        total_set_nan, n,
    )

    return df


def _winsorize_series(s: pd.Series, lower: float, upper: float) -> pd.Series:
    """Winsorize *s* at the given quantile bounds, ignoring NaN."""
    q_lo = s.quantile(lower)
    q_hi = s.quantile(upper)
    return s.clip(lower=q_lo, upper=q_hi)


def create_price_variants(
    df: pd.DataFrame,
    *,
    price_col: str = "price_numeric",
    record: Optional[CleaningRecord] = None,
) -> pd.DataFrame:
    """Create multiple price representations for modelling.

    Columns added
    -------------
    ``price_raw``
        Copy of the cleaned numeric price.
    ``price_winsor_1_99``
        Winsorized at 1st / 99th percentile.
    ``price_winsor_2_98``
        Winsorized at 2nd / 98th percentile.
    ``price_winsor_5_95``
        Winsorized at 5th / 95th percentile.
    ``log_price_raw``
        Natural log of raw price (NaN for non-positive).
    ``log_price_winsor``
        Natural log of the 1-99 winsorized price.
    """
    rec = record or _get_record()
    n = len(df)
    df = df.copy()

    if price_col not in df.columns:
        logger.warning("Price column '%s' not found; skipping variant creation", price_col)
        return df

    prices = df[price_col]

    # Raw copy
    df["price_raw"] = prices.copy()

    # Winsorized variants
    df["price_winsor_1_99"] = _winsorize_series(prices, *WINSORIZE_1_99)
    df["price_winsor_2_98"] = _winsorize_series(prices, *WINSORIZE_2_98)
    df["price_winsor_5_95"] = _winsorize_series(prices, *WINSORIZE_5_95)

    # Log transforms (only for positive values)
    df["log_price_raw"] = np.where(prices > 0, np.log(prices), np.nan)
    df["log_price_winsor"] = np.where(
        df["price_winsor_1_99"] > 0,
        np.log(df["price_winsor_1_99"]),
        np.nan,
    )

    variants_created = [
        "price_raw", "price_winsor_1_99", "price_winsor_2_98",
        "price_winsor_5_95", "log_price_raw", "log_price_winsor",
    ]
    rec.log(
        "create_price_variants", price_col,
        f"Created {len(variants_created)} price variants: {variants_created}",
        n, n,
    )

    # Log winsorization bounds for reproducibility
    valid_prices = prices.dropna()
    if len(valid_prices) > 0:
        for label, lo, hi in [
            ("1-99", *WINSORIZE_1_99),
            ("2-98", *WINSORIZE_2_98),
            ("5-95", *WINSORIZE_5_95),
        ]:
            q_lo = valid_prices.quantile(lo)
            q_hi = valid_prices.quantile(hi)
            logger.info(
                "Winsor %s bounds: $%.2f -- $%.2f", label, q_lo, q_hi
            )

    return df


# ---------------------------------------------------------------------------
# 2. Rating cleaning
# ---------------------------------------------------------------------------


def clean_rating(
    df: pd.DataFrame,
    *,
    rating_col: str = "rating",
    raw_rating_col: str = "review_scores_rating",
    record: Optional[CleaningRecord] = None,
) -> pd.DataFrame:
    """Parse and validate rating values.

    Handles both pre-parsed ``rating`` (0-5 scale) and the Inside
    Airbnb ``review_scores_rating`` column (may be 0-100 scale in older
    extracts or 0-5 in newer ones).
    """
    rec = record or _get_record()
    n = len(df)
    df = df.copy()

    # If raw_rating_col exists but rating_col does not, create it
    if raw_rating_col in df.columns and rating_col not in df.columns:
        df[rating_col] = pd.to_numeric(df[raw_rating_col], errors="coerce")
        rec.log("clean_rating", rating_col, "Parsed raw rating column to numeric", n, n)

    if rating_col not in df.columns:
        logger.info("No rating column found; skipping rating cleaning")
        return df

    # Convert old 0-100 scale to 0-5
    if df[rating_col].max() > 5.5:
        mask_100 = df[rating_col] > 5.5
        n_converted = int(mask_100.sum())
        df.loc[mask_100, rating_col] = df.loc[mask_100, rating_col] / 20.0
        rec.log(
            "clean_rating", rating_col,
            f"Converted {n_converted} ratings from 0-100 scale to 0-5",
            n_converted, n,
        )

    # Clip to valid range
    before_nan = int(df[rating_col].isnull().sum())
    mask_invalid = (df[rating_col] < 0) | (df[rating_col] > 5)
    n_invalid = int(mask_invalid.sum())
    if n_invalid > 0:
        df.loc[mask_invalid, rating_col] = np.nan
        rec.log(
            "clean_rating", rating_col,
            f"Set {n_invalid} out-of-range ratings to NaN",
            n_invalid, n,
        )

    # Create rating category if not present
    if "rating_category" not in df.columns:
        df["rating_category"] = pd.cut(
            df[rating_col],
            bins=[-np.inf, 4.0, 4.5, 4.7, 4.9, 5.01],
            labels=["Average (<4.0)", "Good (4.0-4.5)", "Very Good (4.5-4.7)",
                     "Excellent (4.7-4.9)", "Exceptional (4.9-5.0)"],
            right=False,
        ).astype("object")
        df.loc[df[rating_col].isnull(), "rating_category"] = "No rating"
        rec.log("clean_rating", "rating_category", "Created rating_category from rating", n, n)

    after_nan = int(df[rating_col].isnull().sum())
    rec.log(
        "clean_rating", rating_col,
        f"Rating cleaning complete. Missing: {after_nan} ({100*after_nan/n:.1f}%)",
        after_nan - before_nan, n,
    )

    return df


# ---------------------------------------------------------------------------
# 3. Text field cleaning
# ---------------------------------------------------------------------------

_WHITESPACE_RE = re.compile(r"\s+")
_HTML_TAG_RE = re.compile(r"<[^>]+>")
_URL_RE = re.compile(r"https?://\S+")


def clean_text_fields(
    df: pd.DataFrame,
    *,
    text_cols: Optional[List[str]] = None,
    record: Optional[CleaningRecord] = None,
) -> pd.DataFrame:
    """Clean free-text columns (name, description, neighborhood_overview, etc.).

    Operations: strip HTML tags, normalise whitespace, strip leading/
    trailing whitespace, replace empty strings with NaN.
    """
    rec = record or _get_record()
    n = len(df)
    df = df.copy()

    if text_cols is None:
        text_cols = [
            c for c in ("name", "description", "neighborhood_overview",
                         "host_about", "transit", "access", "interaction",
                         "house_rules", "notes")
            if c in df.columns
        ]

    for col in text_cols:
        if col not in df.columns:
            continue

        original_missing = int(df[col].isnull().sum())
        s = df[col].astype(str)

        # Remove HTML
        s = s.str.replace(_HTML_TAG_RE, " ", regex=True)
        # Remove URLs
        s = s.str.replace(_URL_RE, " ", regex=True)
        # Collapse whitespace
        s = s.str.replace(_WHITESPACE_RE, " ", regex=True)
        # Strip
        s = s.str.strip()
        # Empty / 'nan' -> actual NaN
        s = s.replace({"": np.nan, "nan": np.nan, "None": np.nan, "N/A": np.nan})

        df[col] = s
        new_missing = int(df[col].isnull().sum())
        cleaned_count = new_missing - original_missing

        rec.log(
            "clean_text", col,
            f"Stripped HTML/URLs/whitespace; {cleaned_count} new NaN",
            max(cleaned_count, 0), n,
        )

    return df


# ---------------------------------------------------------------------------
# 4. Host field cleaning
# ---------------------------------------------------------------------------

_PCT_RE = re.compile(r"[\d.]+")


def _parse_pct_to_float(s: pd.Series) -> pd.Series:
    """Convert percentage strings like '87%' or '0.87' to float [0, 1]."""
    out = pd.to_numeric(s.astype(str).str.replace("%", "", regex=False), errors="coerce")
    # If values are > 1, assume they are 0-100 scale
    mask_pct = out > 1.0
    out = out.where(~mask_pct, out / 100.0)
    return out


def clean_host_fields(
    df: pd.DataFrame,
    *,
    record: Optional[CleaningRecord] = None,
) -> pd.DataFrame:
    """Parse host_response_rate, host_acceptance_rate to float [0, 1].

    Also converts host_is_superhost / host_identity_verified from
    't'/'f' strings to boolean if needed.
    """
    rec = record or _get_record()
    n = len(df)
    df = df.copy()

    # Response / acceptance rates
    for col in ("host_response_rate", "host_acceptance_rate"):
        if col not in df.columns:
            continue
        before_valid = int(df[col].notna().sum())
        df[col] = _parse_pct_to_float(df[col])
        # Clip to [0, 1]
        df[col] = df[col].clip(0.0, 1.0)
        after_valid = int(df[col].notna().sum())
        rec.log(
            "clean_host", col,
            f"Parsed to float [0,1]. Valid: {before_valid} -> {after_valid}",
            abs(after_valid - before_valid), n,
        )

    # Boolean-like host columns ('t'/'f' -> True/False)
    bool_cols = [
        "host_is_superhost", "host_identity_verified",
        "host_has_profile_pic",
    ]
    for col in bool_cols:
        if col not in df.columns:
            continue
        if df[col].dtype == "bool":
            continue
        mapping = {"t": True, "f": False, "True": True, "False": False}
        df[col] = df[col].map(mapping)
        rec.log("clean_host", col, "Converted t/f strings to boolean", n, n)

    # is_superhost: unify from host_is_superhost if our column is missing
    if "is_superhost" not in df.columns and "host_is_superhost" in df.columns:
        df["is_superhost"] = df["host_is_superhost"].fillna(False).astype(bool)
        rec.log("clean_host", "is_superhost", "Created from host_is_superhost", n, n)

    # host_listings_count -> numeric
    if "host_listings_count" in df.columns:
        df["host_listings_count"] = pd.to_numeric(
            df["host_listings_count"], errors="coerce"
        )

    # host_total_listings_count -> numeric
    if "host_total_listings_count" in df.columns:
        df["host_total_listings_count"] = pd.to_numeric(
            df["host_total_listings_count"], errors="coerce"
        )

    return df


# ---------------------------------------------------------------------------
# 5. Coordinate cleaning
# ---------------------------------------------------------------------------


def clean_coordinates(
    df: pd.DataFrame,
    *,
    lat_col: str = "lat",
    lon_col: str = "long",
    lat_range: Tuple[float, float] = QUEBEC_LAT_RANGE,
    lon_range: Tuple[float, float] = QUEBEC_LON_RANGE,
    record: Optional[CleaningRecord] = None,
) -> pd.DataFrame:
    """Validate and clean geographic coordinates for Quebec listings.

    Also handles the Inside Airbnb column names ``latitude`` /
    ``longitude`` by renaming them to ``lat`` / ``long``.
    """
    rec = record or _get_record()
    n = len(df)
    df = df.copy()

    # Rename full names to short aliases if needed
    rename_map: Dict[str, str] = {}
    if "latitude" in df.columns and lat_col not in df.columns:
        rename_map["latitude"] = lat_col
    if "longitude" in df.columns and lon_col not in df.columns:
        rename_map["longitude"] = lon_col
    if rename_map:
        df.rename(columns=rename_map, inplace=True)
        rec.log(
            "clean_coords", "rename",
            f"Renamed columns: {rename_map}", len(rename_map), n,
        )

    if lat_col not in df.columns or lon_col not in df.columns:
        logger.warning("Coordinate columns not found; skipping coordinate cleaning")
        return df

    # Ensure numeric
    df[lat_col] = pd.to_numeric(df[lat_col], errors="coerce")
    df[lon_col] = pd.to_numeric(df[lon_col], errors="coerce")

    # Flag out-of-range
    lat_min, lat_max = lat_range
    lon_min, lon_max = lon_range

    mask_bad_lat = (df[lat_col] < lat_min) | (df[lat_col] > lat_max)
    mask_bad_lon = (df[lon_col] < lon_min) | (df[lon_col] > lon_max)
    mask_bad = mask_bad_lat | mask_bad_lon
    n_bad = int(mask_bad.sum())

    if n_bad > 0:
        df.loc[mask_bad, lat_col] = np.nan
        df.loc[mask_bad, lon_col] = np.nan
        rec.log(
            "clean_coords", f"{lat_col}/{lon_col}",
            f"Set {n_bad} out-of-bounds coordinates to NaN "
            f"(lat: [{lat_min}, {lat_max}], lon: [{lon_min}, {lon_max}])",
            n_bad, n,
        )

    # Flag missing coordinates
    mask_missing = df[lat_col].isnull() | df[lon_col].isnull()
    n_missing = int(mask_missing.sum())
    rec.log(
        "clean_coords", f"{lat_col}/{lon_col}",
        f"Total missing coordinates after cleaning: {n_missing}",
        n_missing, n,
    )

    return df


# ---------------------------------------------------------------------------
# 6. Property type standardisation
# ---------------------------------------------------------------------------


def clean_property_type(
    df: pd.DataFrame,
    *,
    col: str = "property_type",
    mapping: Optional[Dict[str, str]] = None,
    record: Optional[CleaningRecord] = None,
) -> pd.DataFrame:
    """Standardise property types to a canonical set.

    Uses ``PROPERTY_TYPE_MAP`` from config by default.  Unmapped values
    are assigned ``"Other"``.
    """
    rec = record or _get_record()
    n = len(df)
    df = df.copy()

    if col not in df.columns:
        logger.info("Column '%s' not found; skipping property type cleaning", col)
        return df

    mapping = mapping or PROPERTY_TYPE_MAP
    original_unique = df[col].nunique()

    # Map known types; assign "Other" to unknowns
    df[col] = df[col].map(mapping).fillna("Other")

    new_unique = df[col].nunique()
    mapped_to_other = int((df[col] == "Other").sum())

    rec.log(
        "clean_property_type", col,
        f"Standardised {original_unique} types -> {new_unique} types. "
        f"{mapped_to_other} mapped to 'Other'",
        n, n,
    )

    return df


# ---------------------------------------------------------------------------
# 7. Room type standardisation
# ---------------------------------------------------------------------------


def clean_room_type(
    df: pd.DataFrame,
    *,
    col: str = "room_type",
    mapping: Optional[Dict[str, str]] = None,
    record: Optional[CleaningRecord] = None,
) -> pd.DataFrame:
    """Standardise room types to the canonical four-level set.

    Maps case-insensitive variants to:
    ``Entire home/apt``, ``Private room``, ``Shared room``, ``Hotel room``.
    """
    rec = record or _get_record()
    n = len(df)
    df = df.copy()

    if col not in df.columns:
        logger.info("Column '%s' not found; skipping room type cleaning", col)
        return df

    mapping = mapping or ROOM_TYPE_MAP
    original_unique = df[col].nunique()

    # Try direct map first, then lower-case map
    mapped = df[col].map(mapping)
    still_null = mapped.isnull()
    if still_null.any():
        lower_mapped = df.loc[still_null, col].str.lower().str.strip().map(mapping)
        mapped = mapped.fillna(lower_mapped)

    # Anything still unmapped -> try substring matching
    still_null = mapped.isnull()
    if still_null.any():
        for idx in df.loc[still_null].index:
            raw = str(df.at[idx, col]).lower()
            if "entire" in raw:
                mapped.at[idx] = "Entire home/apt"
            elif "private" in raw:
                mapped.at[idx] = "Private room"
            elif "shared" in raw:
                mapped.at[idx] = "Shared room"
            elif "hotel" in raw:
                mapped.at[idx] = "Hotel room"
            else:
                mapped.at[idx] = "Entire home/apt"  # default

    df[col] = mapped
    new_unique = df[col].nunique()

    rec.log(
        "clean_room_type", col,
        f"Standardised {original_unique} room types -> {new_unique}",
        n, n,
    )

    return df


# ---------------------------------------------------------------------------
# 8. Additional numeric cleaning helpers
# ---------------------------------------------------------------------------


def _ensure_numeric(
    df: pd.DataFrame,
    cols: List[str],
    record: Optional[CleaningRecord] = None,
) -> pd.DataFrame:
    """Coerce columns to numeric, logging conversion failures."""
    rec = record or _get_record()
    n = len(df)
    for col in cols:
        if col not in df.columns:
            continue
        before_na = int(df[col].isnull().sum())
        df[col] = pd.to_numeric(df[col], errors="coerce")
        after_na = int(df[col].isnull().sum())
        new_na = after_na - before_na
        if new_na > 0:
            rec.log(
                "ensure_numeric", col,
                f"Coerced to numeric; {new_na} new NaN from unparseable values",
                new_na, n,
            )
    return df


def _clean_boolean(
    df: pd.DataFrame,
    cols: List[str],
    record: Optional[CleaningRecord] = None,
) -> pd.DataFrame:
    """Convert t/f/True/False/1/0 columns to proper bool dtype."""
    rec = record or _get_record()
    n = len(df)
    truthy = {"t", "true", "1", "yes", "True"}
    for col in cols:
        if col not in df.columns:
            continue
        if df[col].dtype == "bool":
            continue
        df[col] = df[col].astype(str).str.strip().str.lower().isin(truthy)
        rec.log("clean_boolean", col, "Converted to boolean dtype", n, n)
    return df


# ---------------------------------------------------------------------------
# 9. Inside Airbnb amenities parsing
# ---------------------------------------------------------------------------


def parse_amenities(
    df: pd.DataFrame,
    *,
    col: str = "amenities",
    record: Optional[CleaningRecord] = None,
) -> pd.DataFrame:
    """Parse the JSON-like amenities string into count and flag columns.

    Inside Airbnb stores amenities as: ``'["Wifi", "Kitchen", ...]'``.
    We extract the count and create binary flags for key amenities.
    """
    rec = record or _get_record()
    n = len(df)
    df = df.copy()

    if col not in df.columns:
        logger.info("Column '%s' not found; skipping amenity parsing", col)
        return df

    def _count_amenities(val: Any) -> int:
        if pd.isnull(val):
            return 0
        s = str(val)
        # Count quoted items
        items = re.findall(r'"([^"]+)"', s)
        return len(items) if items else 0

    def _has_amenity(val: Any, keyword: str) -> bool:
        if pd.isnull(val):
            return False
        return keyword.lower() in str(val).lower()

    if "amenities_count" not in df.columns:
        df["amenities_count"] = df[col].apply(_count_amenities)
        rec.log("parse_amenities", "amenities_count", "Counted amenities from raw string", n, n)

    # Key amenity flags
    key_amenities = {
        "has_wifi": "wifi",
        "has_kitchen": "kitchen",
        "has_parking": "parking",
        "has_pool": "pool",
        "has_ac": "air conditioning",
        "has_heating": "heating",
        "has_washer": "washer",
        "has_dryer": "dryer",
        "has_tv": "tv",
        "has_hot_tub": "hot tub",
        "has_fireplace": "fireplace",
        "has_elevator": "elevator",
        "has_gym": "gym",
    }

    created_flags: List[str] = []
    for flag_col, keyword in key_amenities.items():
        if flag_col not in df.columns:
            df[flag_col] = df[col].apply(lambda v, kw=keyword: _has_amenity(v, kw))
            created_flags.append(flag_col)

    if created_flags:
        rec.log(
            "parse_amenities", "amenity_flags",
            f"Created {len(created_flags)} binary amenity flags",
            n, n,
        )

    return df


# ---------------------------------------------------------------------------
# 10. Master cleaning pipeline
# ---------------------------------------------------------------------------


def run_full_cleaning(
    df: pd.DataFrame,
    *,
    price_floor: float = PRICE_FLOOR,
    price_ceiling: float = PRICE_CEILING,
    create_variants: bool = True,
    parse_amenities_col: bool = True,
    verbose: bool = True,
) -> pd.DataFrame:
    """Execute the full cleaning pipeline in the canonical order.

    Parameters
    ----------
    df : pd.DataFrame
        Raw or semi-cleaned listings data.
    price_floor / price_ceiling : float
        Plausible nightly price bounds (CAD).
    create_variants : bool
        Whether to create winsorized / log price columns.
    parse_amenities_col : bool
        Whether to parse the raw amenities string (if present).
    verbose : bool
        Extra logging.

    Returns
    -------
    pd.DataFrame
        Cleaned DataFrame with all new columns appended.
    """
    global _record
    _record = CleaningRecord()
    rec = _record

    n_original = len(df)
    logger.info("=" * 70)
    logger.info("STARTING FULL CLEANING PIPELINE  |  %s rows  |  %s cols", f"{n_original:,}", len(df.columns))
    logger.info("=" * 70)

    # 1. Text fields
    df = clean_text_fields(df, record=rec)

    # 2. Host fields (before price so is_superhost is available)
    df = clean_host_fields(df, record=rec)

    # 3. Prices
    df = clean_price(df, floor=price_floor, ceiling=price_ceiling, record=rec)
    if create_variants:
        df = create_price_variants(df, record=rec)

    # 4. Ratings
    df = clean_rating(df, record=rec)

    # 5. Coordinates
    df = clean_coordinates(df, record=rec)

    # 6. Property type
    df = clean_property_type(df, record=rec)

    # 7. Room type
    df = clean_room_type(df, record=rec)

    # 8. Amenities (raw string parsing)
    if parse_amenities_col:
        df = parse_amenities(df, record=rec)

    # 9. Ensure key numeric columns
    numeric_cols = [
        "bedrooms", "bathrooms", "guests_count", "accommodates",
        "beds", "amenities_count", "num_images", "num_reviews",
        "quality_score", "number_of_reviews",
        "minimum_nights", "maximum_nights",
        "availability_30", "availability_60", "availability_90", "availability_365",
        "reviews_per_month", "calculated_host_listings_count",
    ]
    df = _ensure_numeric(df, [c for c in numeric_cols if c in df.columns], record=rec)

    # 10. Ensure boolean columns
    bool_cols = ["is_superhost", "is_guest_favorite", "pets_allowed",
                 "instant_bookable"]
    df = _clean_boolean(df, [c for c in bool_cols if c in df.columns], record=rec)

    # Summary
    n_final = len(df)
    n_cols_final = len(df.columns)
    logger.info("=" * 70)
    logger.info(
        "CLEANING COMPLETE  |  %s rows  |  %s cols  |  %s actions logged",
        f"{n_final:,}", n_cols_final, len(rec.actions),
    )
    logger.info("=" * 70)

    return df


# ---------------------------------------------------------------------------
# 11. Cleaning report
# ---------------------------------------------------------------------------


def generate_cleaning_report(
    df_before: pd.DataFrame,
    df_after: pd.DataFrame,
    *,
    record: Optional[CleaningRecord] = None,
) -> Dict[str, Any]:
    """Generate a comprehensive report comparing pre- and post-cleaning data.

    Parameters
    ----------
    df_before : pd.DataFrame
        The DataFrame *before* cleaning.
    df_after : pd.DataFrame
        The DataFrame *after* cleaning.
    record : CleaningRecord, optional
        The cleaning audit trail; defaults to the module-level record.

    Returns
    -------
    dict
        A structured report suitable for logging or serialisation.
    """
    rec = record or _get_record()

    report: Dict[str, Any] = {
        "rows_before": len(df_before),
        "rows_after": len(df_after),
        "rows_dropped": len(df_before) - len(df_after),
        "cols_before": len(df_before.columns),
        "cols_after": len(df_after.columns),
        "cols_added": sorted(set(df_after.columns) - set(df_before.columns)),
        "cols_removed": sorted(set(df_before.columns) - set(df_after.columns)),
    }

    # Missing value comparison
    missing_before: Dict[str, int] = {}
    missing_after: Dict[str, int] = {}
    for col in df_before.columns:
        missing_before[col] = int(df_before[col].isnull().sum())
    for col in df_after.columns:
        missing_after[col] = int(df_after[col].isnull().sum())
    report["missing_before"] = missing_before
    report["missing_after"] = missing_after

    # Price comparison
    if "price_numeric" in df_before.columns and "price_numeric" in df_after.columns:
        p_before = df_before["price_numeric"].dropna()
        p_after = df_after["price_numeric"].dropna()
        report["price_comparison"] = {
            "valid_before": len(p_before),
            "valid_after": len(p_after),
            "mean_before": float(p_before.mean()) if len(p_before) > 0 else None,
            "mean_after": float(p_after.mean()) if len(p_after) > 0 else None,
            "median_before": float(p_before.median()) if len(p_before) > 0 else None,
            "median_after": float(p_after.median()) if len(p_after) > 0 else None,
            "std_before": float(p_before.std()) if len(p_before) > 0 else None,
            "std_after": float(p_after.std()) if len(p_after) > 0 else None,
            "min_before": float(p_before.min()) if len(p_before) > 0 else None,
            "min_after": float(p_after.min()) if len(p_after) > 0 else None,
            "max_before": float(p_before.max()) if len(p_before) > 0 else None,
            "max_after": float(p_after.max()) if len(p_after) > 0 else None,
        }

    # Cleaning actions log
    report["actions"] = rec.actions
    report["total_actions"] = len(rec.actions)

    # Log the report highlights
    logger.info("-" * 70)
    logger.info("CLEANING REPORT")
    logger.info("-" * 70)
    logger.info("Rows: %s -> %s (dropped %s)", f"{report['rows_before']:,}",
                f"{report['rows_after']:,}", report["rows_dropped"])
    logger.info("Columns: %s -> %s", report["cols_before"], report["cols_after"])
    if report["cols_added"]:
        logger.info("Columns added: %s", report["cols_added"])
    if report["cols_removed"]:
        logger.info("Columns removed: %s", report["cols_removed"])
    logger.info("Total cleaning actions: %s", report["total_actions"])
    logger.info("-" * 70)

    return report