Skill 数据清洗与分析

# 文件路径：scripts/load_data.py
import pandas as pd
import sys, json, os

def load(file_path: str) -> dict:
    """加载 CSV 或 Excel 文件，返回基本信息"""
    if not os.path.exists(file_path):
        return {"status": "error", "message": f"文件不存在：{file_path}"}

    ext = os.path.splitext(file_path)[1].lower()
    try:
        if ext == ".csv":
            df = pd.read_csv(file_path, encoding="utf-8")
        elif ext in (".xlsx", ".xls"):
            df = pd.read_excel(file_path)
        else:
            return {"status": "error",
                    "message": f"不支持的格式：{ext}，请使用 .csv 或 .xlsx"}
    except UnicodeDecodeError:
        # 尝试 GBK 编码（常见于中文 Windows 导出的 CSV）
        df = pd.read_csv(file_path, encoding="gbk")

    # 将 df 保存到临时文件供后续步骤使用
    tmp_path = "/home/claude/loaded_data.csv"
    df.to_csv(tmp_path, index=False, encoding="utf-8")

    return {
        "status":   "success",
        "rows":     len(df),
        "cols":     len(df.columns),
        "columns":  list(df.columns),
        "dtypes":   df.dtypes.astype(str).to_dict(),
        "tmp_path": tmp_path
    }

if __name__ == "__main__":
    print(json.dumps(load(sys.argv[1] if len(sys.argv) > 1 else ""),
                     ensure_ascii=False, indent=2))

# 文件路径：scripts/quality_check.py
import pandas as pd
import sys, json

def check(file_path: str) -> dict:
    """扫描数据质量问题"""
    df = pd.read_csv(file_path)

    # 空值统计
    null_counts = df.isnull().sum()
    null_issues = [
        {"column": col, "null_count": int(cnt),
         "null_pct": round(cnt / len(df) * 100, 1)}
        for col, cnt in null_counts.items() if cnt > 0
    ]

    # 重复行
    dup_count = int(df.duplicated().sum())

    # 异常值检测（数值列的 IQR 方法）
    outlier_issues = []
    for col in df.select_dtypes(include="number").columns:
        q1, q3 = df[col].quantile(0.25), df[col].quantile(0.75)
        iqr    = q3 - q1
        lower, upper = q1 - 1.5 * iqr, q3 + 1.5 * iqr
        outliers = df[(df[col] < lower) | (df[col] > upper)]
        if len(outliers) > 0:
            outlier_issues.append({
                "column":  col,
                "count":   len(outliers),
                "range":   f"[{lower:.2f}, {upper:.2f}]"
            })

    issues_count = len(null_issues) + (1 if dup_count > 0 else 0) + len(outlier_issues)
    return {
        "status":          "success",
        "total_issues":    issues_count,
        "duplicate_rows":  dup_count,
        "null_columns":    null_issues,
        "outlier_columns": outlier_issues
    }

if __name__ == "__main__":
    print(json.dumps(check(sys.argv[1] if len(sys.argv) > 1 else ""),
                     ensure_ascii=False, indent=2))

# 文件路径：scripts/clean_data.py
import pandas as pd
import sys, json, os

def clean(file_path: str, output_path: str = "/home/claude/cleaned_data.csv") -> dict:
    """执行标准清洗流程"""
    df = pd.read_csv(file_path)
    stats = {}

    # 1. 标准化列名（去除空格，转小写，空格换下划线）
    original_cols  = list(df.columns)
    df.columns     = [c.strip().lower().replace(" ", "_") for c in df.columns]
    stats["cols_renamed"] = sum(a != b for a, b in zip(original_cols, df.columns))

    # 2. 删除完全重复的行
    before = len(df)
    df.drop_duplicates(inplace=True)
    stats["dup_removed"] = before - len(df)

    # 3. 数值列空值填 0，字符串列填空字符串
    null_before = int(df.isnull().sum().sum())
    for col in df.columns:
        if df[col].dtype in ("int64", "float64"):
            df[col].fillna(0, inplace=True)
        else:
            df[col].fillna("", inplace=True)
    stats["nulls_filled"] = null_before - int(df.isnull().sum().sum())

    # 4. 去除字符串首尾空格
    for col in df.select_dtypes(include="object").columns:
        df[col] = df[col].str.strip()

    df.to_csv(output_path, index=False, encoding="utf-8")
    return {
        "status":         "success",
        "output":         output_path,
        "rows_after":     len(df),
        "cols_renamed":   stats["cols_renamed"],
        "dup_removed":    stats["dup_removed"],
        "nulls_filled":   stats["nulls_filled"]
    }

if __name__ == "__main__":
    print(json.dumps(clean(sys.argv[1] if len(sys.argv) > 1 else ""),
                     ensure_ascii=False, indent=2))