Skill 数据清洗与分析

本项目构建一个数据清洗 Skill，自动处理用户上传的 CSV/Excel 数据，完成清洗、统计和报告生成。

目标：掌握 pandas 数据处理流水线与多步骤 Skill 的设计。

---

项目功能概览

步骤	操作	输出
1. 读取数据	支持 CSV、Excel	行数、列名、数据类型
2. 质量检测	空值、重复行、异常值扫描	质量问题报告
3. 数据清洗	去重、填充空值、标准化列名	清洗后的 CSV
4. 统计分析	均值、中位数、分布	统计摘要表格
5. 生成报告	汇总以上结果	Excel 报告文件

---

目录结构

data-cleaner/
├── SKILL.md
└── scripts/
    ├── requirements.txt
    ├── load_data.py
    ├── quality_check.py
    ├── clean_data.py
    ├── calc_stats.py
    └── gen_report.py

---

步骤一：加载数据

实例

# 文件路径：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))

---

SKILL.md 完整内容

---
name: data-cleaner
version: 1.0.0
description: >
  清洗并分析用户上传的 CSV/Excel 数据，包含质量检测、去重、
  空值处理、统计分析和报告生成。当用户需要数据清洗、数据预处理、
  统计分析、数据质量报告时触发。
---

# 数据清洗与分析助手

## 执行流程

### 第一步：加载数据
```bash
cd scripts/
python load_data.py <文件路径>
```
告知用户数据规模（行数、列数）和列名。

### 第二步：质量检测
```bash
python quality_check.py /home/claude/loaded_data.csv
```
展示质量报告：有哪些列有空值、有多少重复行、哪些列有异常值。
询问用户：是否按默认规则清洗？或需要自定义处理方式？

### 第三步：执行清洗
用户确认后执行：
```bash
python clean_data.py /home/claude/loaded_data.csv
```
告知清洗结果：删除了多少重复行、填充了多少空值。

### 第四步：输出清洗后的文件
将 /home/claude/cleaned_data.csv 复制到输出目录后调用 present_files 展示。

AI 思考中...

Skill 代码审查

多 Skill 协作工作流