Compétence : nettoyage et analyse des données
Ce projet construit une compétence de nettoyage de données qui traite automatiquement les données CSV/Excel téléchargées par l'utilisateur, effectue le nettoyage, les statistiques et la génération de rapports.
Objectif : Maîtriser les pipelines de traitement de données pandas et la conception de compétences multi-étapes.
Aperçu des fonctions du projet
| Étape | Opération | Sortie |
|---|---|---|
| 1. Lecture des données | Prend en charge CSV, Excel | Nombre de lignes, noms de colonnes, types de données |
| 2. Contrôle qualité | Valeurs nulles, lignes dupliquées, détection des valeurs aberrantes | Rapport sur les problèmes de qualité |
| 3. Nettoyage des données | Dédoublonnage, remplissage des valeurs nulles, normalisation des noms de colonnes | CSV nettoyé |
| 4. Analyse statistique | Moyenne, médiane, distribution | Tableau récapitulatif statistique |
| 5. Générer le rapport | Résumer les résultats ci-dessus | Fichier de rapport Excel |
Structure du répertoire
data-cleaner/
├── SKILL.md
└── scripts/
├── requirements.txt
├── load_data.py
├── quality_check.py
├── clean_data.py
├── calc_stats.py
└── gen_report.py
Étape 1 : Charger les données
Exemple
# Chemin du fichier : scripts/load_data.py
import pandas as pd
import sys, json, os
def load(file_path: str) -> dict:
"""Charger un fichier CSV ou Excel, renvoyer des informations de base"""
if not os.path.exists(file_path):
return {"status": "error", "message": fLe fichier n'existe pas : {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": fFormat non pris en charge : {ext}, veuillez utiliser .csv ou .xlsx}
except UnicodeDecodeError:
# Essayer l'encodage GBK (courant dans les CSV exportés depuis Windows chinois)
df = pd.read_csv(file_path, encoding="gbk")
# Enregistrer df dans un fichier temporaire pour les étapes suivantes
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:
"""Charger un fichier CSV ou Excel, renvoyer des informations de base"""
if not os.path.exists(file_path):
return {"status": "error", "message": fLe fichier n'existe pas : {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": fFormat non pris en charge : {ext}, veuillez utiliser .csv ou .xlsx}
except UnicodeDecodeError:
# Essayer l'encodage GBK (courant dans les CSV exportés depuis Windows chinois)
df = pd.read_csv(file_path, encoding="gbk")
# Enregistrer df dans un fichier temporaire pour les étapes suivantes
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))
Étape 2 : Vérification de la qualité
Exemple
# Chemin du fichier : scripts/quality_check.py
import pandas as pd
import sys, json
def check(file_path: str) -> dict:
"""Rechercher les problèmes de qualité des données"""
df = pd.read_csv(file_path)
# Statistiques des valeurs nulles
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
]
# Lignes dupliquées
dup_count = int(df.duplicated().sum())
# Détection des valeurs aberrantes (méthode IQR pour les colonnes numériques)
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:
"""Rechercher les problèmes de qualité des données"""
df = pd.read_csv(file_path)
# Statistiques des valeurs nulles
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
]
# Lignes dupliquées
dup_count = int(df.duplicated().sum())
# Détection des valeurs aberrantes (méthode IQR pour les colonnes numériques)
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))
Étape 3 : Nettoyage des données
Exemple
# Chemin du fichier : 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:
"""Exécuter le processus de nettoyage standard"""
df = pd.read_csv(file_path)
stats = {}
# 1. Standardiser les noms de colonnes (supprimer les espaces, convertir en minuscules, remplacer les espaces par des traits de soulignement)
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. Supprimer les lignes entièrement dupliquées
before = len(df)
df.drop_duplicates(inplace=True)
stats["dup_removed"] = before - len(df)
# 3. Remplir les valeurs nulles des colonnes numériques avec 0, et celles des colonnes de chaînes de caractères avec une chaîne vide
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. Supprimer les espaces en début et en fin des chaînes
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:
"""Exécuter le processus de nettoyage standard"""
df = pd.read_csv(file_path)
stats = {}
# 1. Standardiser les noms de colonnes (supprimer les espaces, convertir en minuscules, remplacer les espaces par des traits de soulignement)
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. Supprimer les lignes entièrement dupliquées
before = len(df)
df.drop_duplicates(inplace=True)
stats["dup_removed"] = before - len(df)
# 3. Remplir les valeurs nulles des colonnes numériques avec 0, et celles des colonnes de chaînes de caractères avec une chaîne vide
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. Supprimer les espaces en début et en fin des chaînes
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 Contenu complet
--- 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 展示。Autres extensions