数据精洗工坊 (datapurifier) — 完整 Skill 文档

版本信息

• 版本号: 1.0.0
• 开发者: yznocz
• 许可证: MIT
• 兼容平台: SkillHub / ClawHub / 任意 AI Agent 平台

---

一、功能全景（30 项有序功能，形成合力）

Layer 1 · 基础清洗 (Foundation) — 解决"数据能不能用"

#1 缺失值智能填充

• 数值列 → 均值/中位数/众数（自动选择最合理方式）
• 文本列 → 众数/"未知"
• 时间序列 → 前向填充 / 插值

#2 精确 + 模糊去重

• 完全重复行删除
• 近似重复检测（编辑距离 < 阈值）
• 自定义保留策略（first / last / custom）

#3 数据类型自动推断与强制转换

• 智能识别：int / float / str / datetime / bool
• 解析科学计数法、千分位分隔符
• 转换失败 → 标记而非崩溃

#4 异常值检测与处理

• IQR 法（Q1 - 1.5IQR, Q3 + 1.5IQR）
• Z-score 法（阈值可配，默认 3σ）
• 处理策略：删除 / 截断 / 标记

#5 字符串深度标准化

• 去首尾空格、合并多余空白
• 统一大小写策略（lower/upper/title）
• 清除不可见字符（\u200b, \ufeff, \n, \r, \t）

#6 日期时间格式统一

• 自动解析多格式 → ISO 8601
• 时区 → UTC+8 北京时间
• 相对日期解析（今天/昨天/N天前）

#7 数值范围校验

• 合法区间 [min, max] 配置
• 超范围 → 截断 / 标记 / 删除
• 百分比 ↔ 小数 自动转换

#8 单位统一换算

• 长度 / 重量 / 金额 / 时间 四大类
• 自动识别原单位 → 统一目标单位

---

Layer 2 · 结构治理 (Structure) — 解决"数据是否规范"

#9 列名规范化 (snake_case)

• 转小写 + 下划线
• 去除特殊字符
• 自动去重追加序号

#10 空行/空列/损坏行过滤

• 删除全空维度
• 跳过解码错误行
• 可配置"标记而非删除"

#11 智能列分割与合并

• 自动检测分隔符
• 姓名 ↔ 姓+名 拆分
• 多列合并（自定义连接符）

#12 分类变量编码

• Label / One-Hot / Ordinal 三种编码
• 自动检测分类列（unique ratio < 阈值）

#13 正则批量提取与替换

• 手机号 / 邮箱 / URL / 身份证 预设模式
• 自定义正则支持
• 模板替换（反向引用）

#14 跨列关联一致性校验

• 省-市-区三级联动
• 年龄 ↔ 出生日期
• 金额 = 单价 × 数量

#15 脏数据模式自动识别

• 混合类型检测
• 编码混乱检测（UTF-8/GBK）
• 输出"脏数据体检报告"

#16 条件格式规则引擎

• / < / == / contains / not_contains

• AND / OR / NOT 逻辑组合
• 输出满足条件的行索引

---

Layer 3 · 质量增强 (Enhancement) — 解决"数据是否可信"

#17 敏感信息自动脱敏

• 手机号 138****8000
• 身份证 440305********1234
• 邮箱 z****@example.com
• 姓名 张* / 地址 广东省****

#18 数据质量评分（六维）

• 完整性 / 唯一性 / 有效性 / 一致性 / 时效性 / 准确性
• 加权综合得分（0-100）
• 可配置权重

#19 智能拼写纠错

• 编辑距离 + 词频
• 自定义词典注入
• 修正 / 建议列表 两种模式

#20 文本去重（语义级）

• 编辑距离近似匹配
• 可配置相似度阈值（默认 0.85）
• 输出去重映射表

#21 参照完整性修复

• 外键引用检查
• 孤儿记录标记
• 缺失关联自动填充

#22 数据版本对比（增量更新）

• 新旧集对比 → added / modified / deleted
• 主键匹配
• 支持输出 diff 补丁

#23 文本辅助标注

• 关键词规则匹配
• 自定义标签体系
• 输出标签列 + 置信度

#24 数据聚合预计算

• GROUP BY 多字段
• SUM / COUNT / AVG / MAX / MIN / STD
• 透视表 / 交叉表生成

---

Layer 4 · 编排交付 (Orchestration) — 解决"流程是否可复用"

#25 清洗流水线 DAG 编排

• 操作编排为有向无环图
• 支持条件分支 (if-then-else)
• 自动检测循环依赖

#26 数据采样与分层

• 随机采样（可设 seed）
• 分层抽样（保持原始比例）
• 时间序列截断

#27 导出格式自适应

• CSV / TSV / JSON / XLSX
• 中文编码自动选 UTF-8-BOM
• 大数据集自动分片

#28 清洗报告生成

• JSON 清洗日志
• Markdown 可读摘要
• 影响行数/列数统计

#29 流水线配置序列化

• JSON / YAML 导出
• 一键复用：加载即执行
• 版本可追溯

#30 一键自检与合规验证

• 验证输入输出 Schema
• 检查权限声明完整性
• 生成上架自检清单

---

二、输入规范 (Input Schema)

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "type": "object",
  "properties": {
    "input": {
      "type": "object",
      "properties": {
        "data": {
          "oneOf": [
            {"type": "array", "items": {"type": "object"}},
            {"type": "string", "description": "CSV/TSV/JSON 字符串"}
          ]
        },
        "format": {
          "type": "string",
          "enum": ["json", "csv", "tsv"],
          "default": "json"
        }
      },
      "required": ["data"]
    },
    "config": {
      "type": "object",
      "properties": {
        "pipeline": {
          "type": "array",
          "items": {
            "type": "object",
            "properties": {
              "step": {"type": "string"},
              "params": {"type": "object"}
            },
            "required": ["step"]
          }
        },
        "strategy": {"type": "object"},
        "output_format": {
          "type": "string",
          "enum": ["json", "csv", "tsv", "xlsx"]
        }
      }
    }
  },
  "required": ["input"]
}

三、输出规范 (Output Schema)

{
  "type": "object",
  "properties": {
    "output": {
      "type": "object",
      "properties": {
        "cleaned_data": {
          "type": "array",
          "items": {"type": "object"}
        },
        "quality_report": {
          "type": "object",
          "properties": {
            "score": {"type": "number", "minimum": 0, "maximum": 100},
            "dimensions": {
              "type": "object",
              "properties": {
                "completeness": {"type": "number"},
                "uniqueness": {"type": "number"},
                "validity": {"type": "number"},
                "consistency": {"type": "number"},
                "timeliness": {"type": "number"},
                "accuracy": {"type": "number"}
              }
            }
          }
        },
        "pipeline_log": {
          "type": "array",
          "items": {
            "type": "object",
            "properties": {
              "step": {"type": "string"},
              "status": {"type": "string", "enum": ["success", "warning", "skipped"]},
              "rows_affected": {"type": "number"},
              "columns_modified": {"type": "array", "items": {"type": "string"}},
              "message": {"type": "string"}
            }
          }
        },
        "metadata": {
          "type": "object",
          "properties": {
            "rows_before": {"type": "number"},
            "rows_after": {"type": "number"},
            "columns_total": {"type": "number"},
            "execution_time_ms": {"type": "number"},
            "version": {"type": "string"}
          }
        }
      }
    }
  }
}

四、标准 JSON 结构（OpenAI / 插件 / Skill 规范）

{
  "name": "datapurifier",
  "display_name": "数据精洗工坊",
  "version": "1.0.0",
  "author": "yznocz",
  "description": "数据精洗工坊 (datapurifier) — 一站式数据清洗 Skill，融合 30 项有序功能，从原始脏数据到分析就绪，一条流水线全搞定。零外部依赖，纯标准库实现。",
  "category": {
    "primary": "数据分析",
    "secondary": "数据清洗"
  },
  "tags": ["数据清洗", "数据预处理", "质量管理", "脱敏", "流水线", "零依赖", "标准化"],
  "permissions": {
    "required": [],
    "optional": []
  },
  "external_apis": [],
  "runtime": {
    "language": "python",
    "version": ">=3.8",
    "dependencies": []
  },
  "icon": "icon.png",
  "entry": {
    "skill_md": "SKILL.md",
    "handler": "handler.py"
  },
  "input_schema": {
    "type": "object",
    "properties": {
      "input": {"type": "object", "required": ["data"]},
      "config": {"type": "object"}
    }
  },
  "output_schema": {
    "type": "object",
    "properties": {
      "output": {"type": "object"}
    }
  },
  "license": "MIT",
  "repository": "",
  "keywords": ["data-cleaning", "data-quality", "etl", "preprocessing"]
}

五、上架文案（提交到平台时用）

短描述（一句话）

数据精洗工坊 — 30 项功能一站式数据清洗，零依赖、标准化、可复用。

长描述

🧹 数据精洗工坊 (datapurifier) v1.0.0

让脏数据在 30 秒内变得干净整洁。

【什么时候用？】
✅ 拿到一份乱七八糟的 CSV/Excel，需要快速清洗
✅ 数据有缺失值、重复行、格式混乱，手动处理太慢
✅ 需要对敏感信息（手机号、身份证）自动脱敏
✅ 想建立可复用的数据清洗流水线
✅ 需要数据质量评分报告

【核心能力】
📋 30 项有序功能，4 层架构：
  Layer 1 基础清洗 — 缺失填充/去重/类型转换/异常检测/字符串标准化/日期统一/范围校验/单位换算
  Layer 2 结构治理 — 列名规范/空行过滤/列分割合并/编码/正则/一致性校验/脏数据识别/条件规则
  Layer 3 质量增强 — 脱敏/六维评分/拼写纠错/语义去重/参照完整性/版本对比/标注/聚合
  Layer 4 编排交付 — DAG流水线/采样/导出/报告/序列化/自检

【为什么选它？】
🔒 零外部 API 依赖 — 纯 Python 标准库
📦 输入输出完全标准化 — 统一 JSON Schema
🚫 无敏感权限 — 不需要文件/网络/环境变量
♻️ 可复用可扩展 — 流水线配置可保存复用
📊 全链路可观测 — 每步操作都有详细日志

开发者: yznocz | 许可证: MIT | 兼容: SkillHub / ClawHub / 任意平台

---

六、handler.py — 核心实现（零依赖，纯标准库）

完整实现见同目录 handler.py，此处仅展示接口签名。

"""
datapurifier · 数据精洗工坊
核心处理器 — 纯标准库实现，零第三方依赖
"""

import json
import re
import statistics
import datetime
import hashlib
from typing import Any, Optional


class DataPurifier:
    """数据精洗工坊主类 — 30 项功能统一入口"""

    VERSION = "1.0.0"

    def __init__(self, config: Optional[dict] = None):
        self.config = config or {}
        self.log = []

    # ===== Layer 1: 基础清洗 =====
    def fill_missing(self, data, method="auto") -> list: ...
    def dedup(self, data, mode="exact", threshold=0.85) -> list: ...
    def infer_types(self, data) -> list: ...
    def detect_outliers(self, data, method="iqr") -> dict: ...
    def normalize_strings(self, data, columns=None) -> list: ...
    def unify_datetime(self, data, columns=None, tz="UTC+8") -> list: ...
    def validate_ranges(self, data, rules=None) -> list: ...
    def convert_units(self, data, rules=None) -> list: ...

    # ===== Layer 2: 结构治理 =====
    def normalize_columns(self, data) -> list: ...
    def filter_empty(self, data) -> list: ...
    def split_merge_columns(self, data, rules) -> list: ...
    def encode_categorical(self, data, method="label") -> list: ...
    def regex_extract_replace(self, data, patterns) -> list: ...
    def check_consistency(self, data, rules) -> dict: ...
    def detect_dirty_patterns(self, data) -> dict: ...
    def conditional_rules(self, data, rules) -> list: ...

    # ===== Layer 3: 质量增强 =====
    def mask_sensitive(self, data, columns=None) -> list: ...
    def quality_score(self, data) -> dict: ...
    def spell_correct(self, data, columns=None, custom_dict=None) -> list: ...
    def text_dedup(self, data, column, threshold=0.85) -> list: ...
    def referential_integrity(self, data, references) -> dict: ...
    def diff_versions(self, old_data, new_data, key) -> dict: ...
    def text_annotate(self, data, rules) -> list: ...
    def aggregate(self, data, group_by, agg_funcs) -> list: ...

    # ===== Layer 4: 编排交付 =====
    def run_pipeline(self, data, pipeline_config) -> dict: ...
    def sample_data(self, data, strategy, **kwargs) -> list: ...
    def export(self, data, format="json", **kwargs) -> str: ...
    def generate_report(self) -> dict: ...
    def serialize_config(self, path=None) -> str: ...
    def self_check(self) -> dict: ...

---

七、流水线使用示例

{
  "input": {
    "data": [
      {"name": "张三 ", "age": "25", "phone": "13800138000", "city": "深圳", "province": "广东"},
      {"name": "张三 ", "age": "25", "phone": "13800138000", "city": "深圳", "province": "广东"},
      {"name": "李四", "age": "", "phone": "13900139000", "city": "广州", "province": "广东"}
    ],
    "format": "json"
  },
  "config": {
    "pipeline": [
      {"step": "fill_missing", "params": {"method": "mode"}},
      {"step": "dedup", "params": {"mode": "exact", "keep": "first"}},
      {"step": "normalize_strings", "params": {}},
      {"step": "infer_types", "params": {}},
      {"step": "mask_sensitive", "params": {"columns": ["phone"]}},
      {"step": "quality_score", "params": {}},
      {"step": "export", "params": {"format": "json"}}
    ]
  }
}

预期输出

{
  "output": {
    "cleaned_data": [
      {"name": "张三", "age": 25, "phone": "138****8000", "city": "深圳", "province": "广东"},
      {"name": "李四", "age": 25, "phone": "139****9000", "city": "广州", "province": "广东"}
    ],
    "quality_report": {
      "score": 95,
      "dimensions": {
        "completeness": 100,
        "uniqueness": 100,
        "validity": 100,
        "consistency": 100,
        "timeliness": 90,
        "accuracy": 95
      }
    },
    "pipeline_log": [
      {"step": "fill_missing", "status": "success", "rows_affected": 1, "message": "Filled 1 missing value in 'age'"},
      {"step": "dedup", "status": "success", "rows_affected": 1, "message": "Removed 1 duplicate row"},
      {"step": "mask_sensitive", "status": "success", "rows_affected": 2, "message": "Masked 2 phone numbers"}
    ],
    "metadata": {
      "rows_before": 3,
      "rows_after": 2,
      "columns_total": 5,
      "execution_time_ms": 12,
      "version": "1.0.0"
    }
  }
}