| import pandas as pd |
| import numpy as np |
| import re |
|
|
| def get_suggestions(df: pd.DataFrame, col: str, c_type: str) -> list: |
| if col not in df.columns: |
| return [] |
|
|
| s = df[col].dropna() |
| if len(s) == 0: |
| return [] |
|
|
| suggestions = [] |
|
|
| if c_type in ["text", "categorical", "id"]: |
| |
| suggestions.append({ |
| "id": "uppercase", |
| "name": "Convert to UPPERCASE", |
| "confidence": 95, |
| "preview": s.astype(str).str.upper().head(5).tolist() |
| }) |
| suggestions.append({ |
| "id": "lowercase", |
| "name": "Convert to lowercase", |
| "confidence": 95, |
| "preview": s.astype(str).str.lower().head(5).tolist() |
| }) |
| suggestions.append({ |
| "id": "titlecase", |
| "name": "Convert to Title Case", |
| "confidence": 90, |
| "preview": s.astype(str).str.title().head(5).tolist() |
| }) |
| suggestions.append({ |
| "id": "extract_numbers", |
| "name": "Extract numbers only", |
| "confidence": 85 if s.astype(str).str.contains(r'\d').any() else 20, |
| "preview": s.astype(str).apply(lambda x: re.sub(r'\D', '', x) if pd.notnull(x) else x).head(5).tolist() |
| }) |
| suggestions.append({ |
| "id": "remove_special", |
| "name": "Remove special characters", |
| "confidence": 80, |
| "preview": s.astype(str).apply(lambda x: re.sub(r'[^a-zA-Z0-9\s]', '', x) if pd.notnull(x) else x).head(5).tolist() |
| }) |
| suggestions.append({ |
| "id": "first_word", |
| "name": "Extract first word", |
| "confidence": 75 if s.astype(str).str.contains(r'\s').mean() > 0.5 else 30, |
| "preview": s.astype(str).apply(lambda x: x.split()[0] if pd.notnull(x) and str(x).strip() else x).head(5).tolist() |
| }) |
| suggestions.append({ |
| "id": "last_word", |
| "name": "Extract last word", |
| "confidence": 70 if s.astype(str).str.contains(r'\s').mean() > 0.5 else 20, |
| "preview": s.astype(str).apply(lambda x: x.split()[-1] if pd.notnull(x) and str(x).strip() else x).head(5).tolist() |
| }) |
| suggestions.append({ |
| "id": "remove_whitespace", |
| "name": "Remove extra whitespace", |
| "confidence": 60, |
| "preview": s.astype(str).apply(lambda x: re.sub(r'\s+', ' ', str(x)).strip() if pd.notnull(x) else x).head(5).tolist() |
| }) |
| suggestions.append({ |
| "id": "split_delimiter", |
| "name": "Split by delimiter (take first)", |
| "confidence": 50, |
| "preview": s.astype(str).apply(lambda x: re.split(r'[,;|/]', str(x))[0].strip() if pd.notnull(x) and re.search(r'[,;|/]', str(x)) else x).head(5).tolist() |
| }) |
| suggestions.append({ |
| "id": "word_length", |
| "name": "Count word length", |
| "confidence": 40, |
| "preview": s.astype(str).apply(lambda x: len(str(x)) if pd.notnull(x) else x).head(5).tolist() |
| }) |
|
|
| elif c_type == "numeric": |
| s_num = pd.to_numeric(s, errors='coerce').dropna() |
| if len(s_num) > 0: |
| min_val = s_num.min() |
| max_val = s_num.max() |
| range_val = max_val - min_val if max_val != min_val else 1 |
| mean_val = s_num.mean() |
| std_val = s_num.std() if s_num.std() > 0 else 1 |
|
|
| suggestions.append({ |
| "id": "normalize", |
| "name": "Normalize (0-1 min-max scaling)", |
| "confidence": 90, |
| "preview": ((s_num - min_val) / range_val).round(4).head(5).tolist() |
| }) |
| suggestions.append({ |
| "id": "standardize", |
| "name": "Standardize (z-score)", |
| "confidence": 85, |
| "preview": ((s_num - mean_val) / std_val).round(4).head(5).tolist() |
| }) |
| if (s_num > -1).all(): |
| suggestions.append({ |
| "id": "log1p", |
| "name": "Log transform (log1p)", |
| "confidence": 80 if s_num.skew() > 1 else 40, |
| "preview": np.log1p(s_num).round(4).head(5).tolist() |
| }) |
| suggestions.append({ |
| "id": "integer_part", |
| "name": "Extract integer part", |
| "confidence": 95 if (s_num % 1 != 0).any() else 10, |
| "preview": np.floor(s_num).astype(int).head(5).tolist() |
| }) |
| suggestions.append({ |
| "id": "bin_5", |
| "name": "Bin into 5 equal buckets", |
| "confidence": 75, |
| "preview": pd.cut(s_num, bins=5, labels=False, duplicates='drop').head(5).tolist() |
| }) |
| suggestions.append({ |
| "id": "round_2", |
| "name": "Round to 2 decimal places", |
| "confidence": 70 if (s_num % 1 != 0).any() else 5, |
| "preview": s_num.round(2).head(5).tolist() |
| }) |
|
|
| elif c_type == "email": |
| suggestions.append({ |
| "id": "extract_domain", |
| "name": "Extract domain", |
| "confidence": 95, |
| "preview": s.astype(str).apply(lambda x: x.split('@')[-1] if pd.notnull(x) and '@' in str(x) else x).head(5).tolist() |
| }) |
| suggestions.append({ |
| "id": "extract_username", |
| "name": "Extract username", |
| "confidence": 90, |
| "preview": s.astype(str).apply(lambda x: x.split('@')[0] if pd.notnull(x) and '@' in str(x) else x).head(5).tolist() |
| }) |
| suggestions.append({ |
| "id": "mask_email", |
| "name": "Mask email", |
| "confidence": 85, |
| "preview": s.astype(str).apply(lambda x: str(x)[:2] + "***@" + str(x).split('@')[-1] if pd.notnull(x) and '@' in str(x) and len(str(x).split('@')[0]) >= 2 else x).head(5).tolist() |
| }) |
|
|
| elif c_type == "phone": |
| suggestions.append({ |
| "id": "mask_phone", |
| "name": "Mask last 4 digits", |
| "confidence": 90, |
| "preview": s.astype(str).apply(lambda x: re.sub(r'\d{4}$', '****', str(x)) if pd.notnull(x) else x).head(5).tolist() |
| }) |
| suggestions.append({ |
| "id": "extract_country_code", |
| "name": "Extract country code", |
| "confidence": 85 if s.astype(str).str.startswith('+').mean() > 0.5 else 30, |
| "preview": s.astype(str).apply(lambda x: x.split('-')[0] if pd.notnull(x) and '-' in str(x) and str(x).startswith('+') else x).head(5).tolist() |
| }) |
|
|
| elif c_type == "datetime": |
| try: |
| s_dt = pd.to_datetime(s, errors='coerce').dropna() |
| if len(s_dt) > 0: |
| suggestions.append({ |
| "id": "extract_year", |
| "name": "Extract year", |
| "confidence": 95, |
| "preview": s_dt.dt.year.head(5).tolist() |
| }) |
| suggestions.append({ |
| "id": "extract_month", |
| "name": "Extract month", |
| "confidence": 90, |
| "preview": s_dt.dt.month.head(5).tolist() |
| }) |
| suggestions.append({ |
| "id": "extract_dayofweek", |
| "name": "Extract day of week", |
| "confidence": 85, |
| "preview": s_dt.dt.day_name().head(5).tolist() |
| }) |
| suggestions.append({ |
| "id": "extract_hour", |
| "name": "Extract hour", |
| "confidence": 80 if s_dt.dt.hour.max() > 0 else 10, |
| "preview": s_dt.dt.hour.head(5).tolist() |
| }) |
| suggestions.append({ |
| "id": "days_since", |
| "name": "Calculate days since today", |
| "confidence": 75, |
| "preview": (pd.Timestamp.today() - s_dt).dt.days.head(5).tolist() |
| }) |
| suggestions.append({ |
| "id": "format_ymd", |
| "name": "Format as YYYY-MM-DD", |
| "confidence": 80, |
| "preview": s_dt.dt.strftime('%Y-%m-%d').head(5).tolist() |
| }) |
| except: |
| pass |
|
|
| |
| suggestions.sort(key=lambda x: x["confidence"], reverse=True) |
| return suggestions[:3] |
|
|
| def apply_transformation(df: pd.DataFrame, col: str, transform_id: str) -> dict: |
| new_col = f"{col}_transformed" |
| s = df[col] |
|
|
| |
| success_count = 0 |
| fail_count = 0 |
|
|
| def safe_apply(val, func): |
| nonlocal success_count, fail_count |
| if pd.isna(val): |
| return np.nan |
| try: |
| res = func(val) |
| success_count += 1 |
| return res |
| except Exception: |
| fail_count += 1 |
| return np.nan |
|
|
| if transform_id == "uppercase": |
| df[new_col] = s.apply(lambda x: safe_apply(x, lambda v: str(v).upper())) |
| elif transform_id == "lowercase": |
| df[new_col] = s.apply(lambda x: safe_apply(x, lambda v: str(v).lower())) |
| elif transform_id == "titlecase": |
| df[new_col] = s.apply(lambda x: safe_apply(x, lambda v: str(v).title())) |
| elif transform_id == "extract_numbers": |
| df[new_col] = s.apply(lambda x: safe_apply(x, lambda v: re.sub(r'\D', '', str(v)))) |
| elif transform_id == "remove_special": |
| df[new_col] = s.apply(lambda x: safe_apply(x, lambda v: re.sub(r'[^a-zA-Z0-9\s]', '', str(v)))) |
| elif transform_id == "first_word": |
| df[new_col] = s.apply(lambda x: safe_apply(x, lambda v: str(v).split()[0] if str(v).strip() else "")) |
| elif transform_id == "last_word": |
| df[new_col] = s.apply(lambda x: safe_apply(x, lambda v: str(v).split()[-1] if str(v).strip() else "")) |
| elif transform_id == "remove_whitespace": |
| df[new_col] = s.apply(lambda x: safe_apply(x, lambda v: re.sub(r'\s+', ' ', str(v)).strip())) |
| elif transform_id == "split_delimiter": |
| df[new_col] = s.apply(lambda x: safe_apply(x, lambda v: re.split(r'[,;|/]', str(v))[0].strip() if re.search(r'[,;|/]', str(v)) else v)) |
| elif transform_id == "word_length": |
| df[new_col] = s.apply(lambda x: safe_apply(x, lambda v: len(str(v)))) |
|
|
| |
| elif transform_id in ["normalize", "standardize", "log1p", "integer_part", "bin_5", "round_2"]: |
| s_num = pd.to_numeric(s, errors='coerce') |
| if transform_id == "normalize": |
| min_val = s_num.min() |
| range_val = s_num.max() - min_val if s_num.max() != min_val else 1 |
| df[new_col] = (s_num - min_val) / range_val |
| elif transform_id == "standardize": |
| mean_val = s_num.mean() |
| std_val = s_num.std() if s_num.std() > 0 else 1 |
| df[new_col] = (s_num - mean_val) / std_val |
| elif transform_id == "log1p": |
| df[new_col] = s_num.apply(lambda x: safe_apply(x, lambda v: np.log1p(max(0, v)) if v > -1 else np.nan)) |
| elif transform_id == "integer_part": |
| df[new_col] = s_num.apply(lambda x: safe_apply(x, lambda v: int(np.floor(v)))) |
| elif transform_id == "bin_5": |
| df[new_col] = pd.cut(s_num, bins=5, labels=False, duplicates='drop') |
| elif transform_id == "round_2": |
| df[new_col] = s_num.round(2) |
|
|
| success_count = s_num.notna().sum() |
| fail_count = len(s) - success_count - s.isna().sum() |
|
|
| |
| elif transform_id == "extract_domain": |
| df[new_col] = s.apply(lambda x: safe_apply(x, lambda v: str(v).split('@')[-1] if '@' in str(v) else np.nan)) |
| elif transform_id == "extract_username": |
| df[new_col] = s.apply(lambda x: safe_apply(x, lambda v: str(v).split('@')[0] if '@' in str(v) else np.nan)) |
| elif transform_id == "mask_email": |
| df[new_col] = s.apply(lambda x: safe_apply(x, lambda v: str(v)[:2] + "***@" + str(v).split('@')[-1] if '@' in str(v) and len(str(v).split('@')[0]) >= 2 else np.nan)) |
|
|
| |
| elif transform_id == "mask_phone": |
| df[new_col] = s.apply(lambda x: safe_apply(x, lambda v: re.sub(r'\d{4}$', '****', str(v)))) |
| elif transform_id == "extract_country_code": |
| df[new_col] = s.apply(lambda x: safe_apply(x, lambda v: str(v).split('-')[0] if '-' in str(v) and str(v).startswith('+') else np.nan)) |
|
|
| |
| elif transform_id in ["extract_year", "extract_month", "extract_dayofweek", "format_ymd", "extract_hour", "days_since"]: |
| s_dt = pd.to_datetime(s, errors='coerce') |
| if transform_id == "extract_year": |
| df[new_col] = s_dt.dt.year |
| elif transform_id == "extract_month": |
| df[new_col] = s_dt.dt.month |
| elif transform_id == "extract_dayofweek": |
| df[new_col] = s_dt.dt.day_name() |
| elif transform_id == "extract_hour": |
| df[new_col] = s_dt.dt.hour |
| elif transform_id == "days_since": |
| df[new_col] = (pd.Timestamp.today() - s_dt).dt.days |
| elif transform_id == "format_ymd": |
| df[new_col] = s_dt.dt.strftime('%Y-%m-%d') |
|
|
| success_count = s_dt.notna().sum() |
| fail_count = len(s) - success_count - s.isna().sum() |
| else: |
| raise ValueError(f"Unknown transformation: {transform_id}") |
|
|
| return { |
| "df": df, |
| "new_column": new_col, |
| "success_count": int(success_count), |
| "fail_count": int(fail_count) |
| } |
|
|
