Skill Data Cleaning and Analysis
This project builds a data cleaning Skill that automatically processes user-uploaded CSV/Excel data, performs cleaning, statistics, and report generation.
Goal: Master pandas data processing pipelines and the design of multi-step Skills.
Project Function Overview
| Step | Operation | Output |
|---|---|---|
| 1. Read Data | Supports CSV, Excel | Row count, column names, data types |
| 2. Quality Check | Null values, duplicate rows, outlier scanning | Quality issue report |
| 3. Data Cleaning | Deduplication, fill null values, standardize column names | Cleaned CSV |
| 4. Statistical Analysis | Mean, median, distribution | Statistical summary table |
| 5. Generate Report | Summarize the above results | Excel report file |
Directory Structure
data-cleaner/
├── SKILL.md
└── scripts/
├── requirements.txt
├── load_data.py
├── quality_check.py
├── clean_data.py
├── calc_stats.py
└── gen_report.py
Step 1: Load Data
Example
# File path: scripts/load_data.py
import pandas as pd
import sys, json, os
def load(file_path: str) -> dict:
"""Load CSV or Excel file, return basic information"""
if not os.path.exists(file_path):
return {"status": "error", "message": fFile does not exist: {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": fUnsupported format: {ext}, please use .csv or .xlsx}
except UnicodeDecodeError:
# Try GBK encoding (common in CSV exported from Chinese Windows)
df = pd.read_csv(file_path, encoding="gbk")
# Save df to a temporary file for subsequent steps
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))
import pandas as pd
import sys, json, os
def load(file_path: str) -> dict:
"""Load CSV or Excel file, return basic information"""
if not os.path.exists(file_path):
return {"status": "error", "message": fFile does not exist: {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": fUnsupported format: {ext}, please use .csv or .xlsx}
except UnicodeDecodeError:
# Try GBK encoding (common in CSV exported from Chinese Windows)
df = pd.read_csv(file_path, encoding="gbk")
# Save df to a temporary file for subsequent steps
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))
Step 2: Quality Check
Example
# File path: scripts/quality_check.py
import pandas as pd
import sys, json
def check(file_path: str) -> dict:
"""Scan for data quality issues"""
df = pd.read_csv(file_path)
# Null value statistics
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
]
# Duplicate rows
dup_count = int(df.duplicated().sum())
# Outlier detection (IQR method for numeric columns)
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))
import pandas as pd
import sys, json
def check(file_path: str) -> dict:
"""Scan for data quality issues"""
df = pd.read_csv(file_path)
# Null value statistics
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
]
# Duplicate rows
dup_count = int(df.duplicated().sum())
# Outlier detection (IQR method for numeric columns)
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))
Step 3: Data Cleaning
Example
# File path: 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:
"""Execute standard cleaning process"""
df = pd.read_csv(file_path)
stats = {}
# 1. Standardize column names (remove spaces, convert to lowercase, replace spaces with underscores)
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. Remove completely duplicate rows
before = len(df)
df.drop_duplicates(inplace=True)
stats["dup_removed"] = before - len(df)
# 3. Fill null values in numeric columns with 0, and string columns with empty string
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. Strip leading/trailing spaces from strings
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))
import pandas as pd
import sys, json, os
def clean(file_path: str, output_path: str = "/home/claude/cleaned_data.csv") -> dict:
"""Execute standard cleaning process"""
df = pd.read_csv(file_path)
stats = {}
# 1. Standardize column names (remove spaces, convert to lowercase, replace spaces with underscores)
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. Remove completely duplicate rows
before = len(df)
df.drop_duplicates(inplace=True)
stats["dup_removed"] = before - len(df)
# 3. Fill null values in numeric columns with 0, and string columns with empty string
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. Strip leading/trailing spaces from strings
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 Full Content
--- 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 展示。Other Extensions