TwinklData/Community_Collections_App

1from __future__ import annotations  # harmless on 3.11+, useful on 3.7‑3.102import re3import string4from typing import Sequence, Dict, Tuple, Optional5import pandas as pd6 7 8# ========= HELPER FUNCTIONS ========9 10def _max_or_eps(values, eps: float = 1e-9) -> float:11    """Avoid divide‑by‑zero during normalisation."""12    return max(values) or eps13 14 15def _normalise(value: float, max_value: float) -> float:16    return value / max_value if max_value else 0.017 18# =================== FREEFORM COL =====================   19 20def detect_freeform_col(21    df: pd.DataFrame,22    *,23    length_weight: float = 0.4,24    punct_weight: float = 0.3,25    unique_weight: float = 0.3,26    low_uniqueness_penalty: float = 0.4,27    name_boosts: dict[str, float] | None = None,28    min_score: float = 0.50,29    return_scores: bool = False,30) -> str | None | Tuple[str | None, Dict[str, float]]:31    """32    Guess which *object* column contains free‑text answers or comments.33 34    A good free‑text column tends to be longish, rich in punctuation,35    and fairly unique row‑to‑row.36 37    name_boosts38        e.g. ``{"additional_comment": 3.1, "usage_reason": 0.5}``39        Multiplicative factors applied if the token appears in the header.40    """41    name_boosts = name_boosts or {}42    obj_cols = df.select_dtypes(include=["object"]).columns43 44    # quick exit45    if not obj_cols.size:46        return (None, {}) if return_scores else None47 48    # pre‑compute raw metrics49    raw: Dict[str, dict[str, float]] = {}50    for col in obj_cols:51        ser = df[col].dropna().astype(str)52        if ser.empty:53            continue54        raw[col] = {55            "avg_len": ser.str.len().mean(),56            "avg_punct": ser.apply(lambda s: sum(c in string.punctuation for c in s)).mean(),57            "unique_ratio": ser.nunique() / len(ser),58        }59 60    if not raw:61        return (None, {}) if return_scores else None62 63    # normalisers64    max_len = _max_or_eps([m["avg_len"] for m in raw.values()])65    max_punc = _max_or_eps([m["avg_punct"] for m in raw.values()])66 67    # composite scores68    scores: Dict[str, float] = {}69    for col, m in raw.items():70        score = (71            length_weight * _normalise(m["avg_len"], max_len)72            + punct_weight * _normalise(m["avg_punct"], max_punc)73            + unique_weight * m["unique_ratio"]74        )75 76        # header boosts / penalties77        for token, factor in name_boosts.items():78            if token in col.lower():79                score *= factor80 81        # penalise low uniqueness82        if m["unique_ratio"] < low_uniqueness_penalty:83            score *= 0.584 85        scores[col] = score86 87    best_col, best_score = max(scores.items(), key=lambda kv: kv[1])88    passed = best_score >= min_score89 90    if return_scores:91        return (best_col if passed else None, scores)92    return best_col if passed else None93 94 95# ================= ID COLUMN =================96 97def detect_id_col(df: pd.DataFrame) -> str | None:98    n_rows = len(df)99 100    # 1) Name‐based detection101    name_pattern = re.compile(r'\b(id|identifier|key)\b', re.IGNORECASE)102    for col in df.columns:103        if name_pattern.search(col):104            return col105 106    # 2) Uniqueness detection: columns where every row is unique107    unique_cols = [108        col for col in df.columns109        if df[col].nunique(dropna=False) == n_rows110    ]111    if not unique_cols:112        return None113 114    # 3) Prioritise int cols over object cols when both are unique115    non_unnamed = [c for c in unique_cols if not c.startswith("Unnamed")]116    candidates = non_unnamed or unique_cols117 118    # 4) Prefer integer dtypes among candidates119    for col in candidates:120        if pd.api.types.is_integer_dtype(df[col]):121            return col122 123    # Fallback: return the first candidate124    return candidates[0]125 126 127# ============== SCHOOL TYPE COLUMN =============128 129def detect_school_type_col(130    df: pd.DataFrame,131    *,132    uniqueness_weight: float = 0.3,133    content_match_weight: float = 0.4, # <-- New weight for content134    length_weight: float = 0.2,135    punct_weight: float = 0.1,136    name_boosts: dict[str, float] | None = None,137    value_keywords: set[str] | None = None, # <-- New parameter for keywords138    min_score: float = 0.40,139    high_uniqueness_penalty: float = 0.95,140    return_scores: bool = False,141) -> str | None | Tuple[str | None, Dict[str, float]]:142    """143    Analyzes a DataFrame to find the column that most likely represents a 'school type'.144 145    The function operates on heuristics based on common characteristics of a school-type col:146    1.  **Content Match**: A significant portion of values match known school types (the strongest signal).147    2.  **Low Uniqueness**: Values are often repeated (e.g., 'Primary', 'All-through').148    3.  **Short Text**: Entries are typically brief.149    4.  **Minimal Punctuation**: Values are clean strings, not sentences.150    5.  **Header Keywords**: The column name itself is a strong indicator (e.g., 'School Type').151    """152    # More robust default name boosts153    if name_boosts is None:154        name_boosts = {'school': 3.0, 'type': 2.0}155 156    # Default set of keywords to search for within the column's values157    if value_keywords is None:158        value_keywords = {159            'nursery', 'primary', 'secondary', 'infant', 'junior',160            'college', 'academy', 'independent', 'special', 'pru',161            'all-through', 'middle', 'state', 'educator', 'home'162        }163 164    obj_cols = df.select_dtypes(include=["object"]).columns165    if not obj_cols.size:166        return (None, {}) if return_scores else None167 168    # Pre-compute raw metrics for each object column169    raw_metrics: Dict[str, dict[str, float]] = {}170    for col in obj_cols:171        ser = df[col].dropna().astype(str)172        if ser.empty:173            continue174 175        # --- New Content Match Calculation ---176        unique_values = ser.unique()177        content_match_score = 0.0178        if len(unique_values) > 0:179            match_count = 0180            for val in unique_values:181                # Check if any keyword is a substring of the lowercase value182                if any(keyword in val.lower() for keyword in value_keywords):183                    match_count += 1184            content_match_score = match_count / len(unique_values)185        # --- End of New Calculation ---186 187        raw_metrics[col] = {188            "avg_len": ser.str.len().mean(),189            "avg_punct": ser.apply(lambda s: sum(c in string.punctuation for c in s)).mean(),190            "unique_ratio": ser.nunique() / len(ser) if len(ser) > 0 else 0.0,191            "content_match": content_match_score # Store the new score192        }193 194    if not raw_metrics:195        return (None, {}) if return_scores else None196 197    # Get max values for normalization198    max_len = _max_or_eps([m["avg_len"] for m in raw_metrics.values()])199    max_punc = _max_or_eps([m["avg_punct"] for m in raw_metrics.values()])200 201    # Calculate a final score for each column202    scores: Dict[str, float] = {}203    for col, metrics in raw_metrics.items():204        len_score = 1 - _normalise(metrics["avg_len"], max_len)205        punc_score = 1 - _normalise(metrics["avg_punct"], max_punc)206        uniq_score = 1 - metrics["unique_ratio"]207 208        # --- Updated Final Scoring Formula ---209        score = (210            content_match_weight * metrics["content_match"] # Use the new score directly211            + uniqueness_weight * uniq_score212            + length_weight * len_score213            + punct_weight * punc_score214        )215 216        # Apply boosts for matching header keywords217        for token, factor in name_boosts.items():218            if token in col.lower().strip():219                score *= factor220 221        # Apply penalty for columns that are almost entirely unique222        if metrics["unique_ratio"] > high_uniqueness_penalty:223            score *= 0.1  # Heavy penalty224 225        scores[col] = score226 227    if not scores:228         return (None, {}) if return_scores else None229 230    best_col, best_score = max(scores.items(), key=lambda item: item[1])231    passed = best_score >= min_score232 233    if return_scores:234        return (best_col if passed else None, scores)235    return best_col if passed else None236# =========== USAGE ============237 238def main():239 240    df = pd.read_csv('data/raw/new-application-format-data.csv')241    df.columns = df.columns.str.strip()242 243    print("--- Testing Column Detection Functions ---")244 245    id_col = detect_id_col(df)246    freeform_col, freeform_scores = detect_freeform_col(df, return_scores=True)247    school_type_col, school_type_scores = detect_school_type_col(df, return_scores=True)248 249    print(f"\nDetected ID Column: '{id_col}'")250    print(f"Detected Free-Form Column: '{freeform_col}'")251    print(f"Detected School Type Column: '{school_type_col}'")252    print()253    print("\n--- Free-form Column Scores (Higher is better) ---")254    if freeform_scores:255        sorted_scores = sorted(freeform_scores.items(), key=lambda item: item[1], reverse=True)256        for col, score in sorted_scores:257            print(f"  - {col:<25}: {score:.4f}")258    else:259        print("No object columns found to score for freeform col...")260 261 262    print("\n--- School Type Column Scores (Higher is better) ---")263    if school_type_scores:264        sorted_scores = sorted(school_type_scores.items(), key=lambda item: item[1], reverse=True)265        for col, score in sorted_scores:266            print(f"  - {col:<25}: {score:.4f}")267    else:268        print("No object columns found to score for career.")269 270if __name__ == '__main__':271    main()272