Day 12 · AI 数据清洗：脏数据一键变干净

「AI Python 系列」第 03 栏 · Python 爬虫实战 全栏 15 篇 · 零成本跟完 🍃 作者：梅雅达编程笔记 首发：CSDN

摘要： 爬虫抓回来的数据到底有多脏？格式不统一、分类混乱、有错别字、字段缺失——手动清洗能把你逼疯。传统方法靠正则和pandas一条条处理，写规则的时间比爬数据还长。Day 12教你用AI做数据清洗：让大模型帮你标准化字段、自动分类、修正错误、补全缺失。一篇真实的脏数据清洗全流程，从乱糟糟到整整齐齐。

数据抓下来了，接下来面对的现实是——脏得没法直接用。

一、爬虫数据到底有多脏

从真实网站抓回来的数据，典型问题：

手动一条条改？几百条数据就让你怀疑人生。

二、传统清洗 vs AI 清洗

传统做法

import pandas as pd
import re

df = pd.read_csv("raw_jobs.csv")

# 1. 统一城市名
city_map = {"bj": "北京", "beijing": "北京", "上海市": "上海", "深圳市": "深圳"}
df["city"] = df["city"].str.strip().str.lower().map(city_map).fillna(df["city"])

# 2. 去掉多余空白
df["title"] = df["title"].str.strip()
df["company"] = df["company"].str.strip()

# 3. 处理薪资格式
def clean_salary(salary):
if pd.isna(salary):
return None
# 把 "25k-50k·15薪" 提取出 25 和 50
match = re.search(r'(\\d+)k?[\\-\\~](\\d+)k?', str(salary), re.I)
if match:
return int(match.group(1)), int(match.group(2))
return None

df[["salary_min", "salary_max"]] = df["salary"].apply(
lambda x: pd.Series(clean_salary(x))
)

# 4. 去重
df = df.drop_duplicates(subset=["title", "company"])

# 5. 删除缺失值
df = df.dropna(subset=["title", "company"])

看着不复杂？那是因为这只是 5 个字段。如果数据有 20 个字段，每个字段都有各自的清洗规则……写规则比抓数据还累。

AI 做法

prompt = f"""请清洗以下职位数据，做以下处理：
1. 统一城市名（如 "bj" → "北京"，"上海市" → "上海"）
2. 统一职位名称（如 "python工程师" → "Python开发工程师"）
3. 修正公司名错别字
4. 薪资统一为数字（单位：千元），如 "25k-50k" → min=25, max=50
5. 标记明显异常的数据（如薪资100k以上）

返回 JSON 数组。

原始数据：
{json.dumps(raw_data, ensure_ascii=False)}"""

一段 prompt，全部清洗规则都搞定。

三、实战：一份脏数据的全流程清洗

脏数据样本

raw_data = [
{"title": " Python开发 ", "company": "腾迅", "city": "shenzhen", "salary": "25k-50k"},
{"title": "python工程师", "company": "阿里巳", "city": "杭州市", "salary": "30k-60k·15薪"},
{"title": "后端-Python", "company": "字节的跳动", "city": "bj", "salary": "20k-40k"},
{"title": " Python ", "company": "美团点评", "city": "北京", "salary": "面议"},
{"title": "python dev", "company": "网异", "city": "beijing", "salary": "100k-200k"},
{"title": "", "company": "某公司", "city": "上海", "salary": "15k-25k"},
{"title": "爬虫工程师", "company": "", "city": "深圳", "salary": "18k-35k"},
]

问题一目了然：多余空格、错别字（腾迅→腾讯、阿里巳→阿里巴巴）、城市格式不统一、异常薪资、空字段。

AI 清洗代码

"""
Day 12 示例代码：AI 数据清洗
作者：梅雅达编程笔记
"""
import json
from openai import OpenAI

client = OpenAI(
api_key="YOUR_API_KEY",
base_url="https://open.bigmodel.cn/api/paas/v4/"
)

MODEL = "glm-4-flash"

def clean_data_with_ai(raw_data):
"""用 AI 清洗脏数据"""

prompt = f"""你是一个数据清洗专家。请清洗以下职位数据，执行以下操作：

1. **去空值**：如果 title 或 company 为空/null/纯空格，直接删除该条记录
2. **标准化城市**：统一为简短中文名（如 bj→北京、shenzhen→深圳、杭州市→杭州）
3. **修正公司名**：修正明显的错别字（如 腾迅→腾讯、阿里巳→阿里巴巴）
4. **标准化职位名**：相似的职位统一名称（如 "python工程师"、"Python开发"、"后端-Python"、"python dev" 统一为 "Python开发工程师"）
5. **处理薪资**：
– "25k-50k" → salary_min=25, salary_max=50
– "30k-60k·15薪" → salary_min=30, salary_max=60, pay_months=15
– "面议" → salary_min=null, salary_max=null
– 标记明显不合理的薪资（如 min > 100）
6. **去空格**：所有字符串字段 trim 首尾空格

返回清洗后的 JSON 数组，每条记录额外加一个 "clean_status" 字段：
– "ok"：正常
– "corrected"：有修正（在 "corrections" 字段说明改了什么）
– "abnormal"：有异常值

只返回 JSON。

原始数据：
{json.dumps(raw_data, ensure_ascii=False, indent=2)}"""

response = client.chat.completions.create(
model=MODEL,
messages=[{"role": "user", "content": prompt}],
temperature=0.1
)

content = response.choices[0].message.content.strip()

# 清理 markdown 标记
if content.startswith("```json"):
content = content[7:]
if content.startswith("```"):
content = content[3:]
if content.endswith("```"):
content = content[:–3]

return json.loads(content.strip())

# ============ 执行 ============
if __name__ == "__main__":
raw_data = [
{"title": " Python开发 ", "company": "腾迅", "city": "shenzhen", "salary": "25k-50k"},
{"title": "python工程师", "company": "阿里巳", "city": "杭州市", "salary": "30k-60k·15薪"},
{"title": "后端-Python", "company": "字节的跳动", "city": "bj", "salary": "20k-40k"},
{"title": " Python ", "company": "美团点评", "city": "北京", "salary": "面议"},
{"title": "python dev", "company": "网易", "city": "beijing", "salary": "100k-200k"},
{"title": "", "company": "某公司", "city": "上海", "salary": "15k-25k"},
{"title": "爬虫工程师", "company": "", "city": "深圳", "salary": "18k-35k"},
]

print(f"原始数据：{len(raw_data)} 条")
print("\\n原始数据预览：")
for item in raw_data:
print(f" {item}")

print("\\n正在用 AI 清洗…")
cleaned = clean_data_with_ai(raw_data)

print(f"\\n清洗后：{len(cleaned)} 条")
print("\\n清洗结果：")
print("-" * 80)

for item in cleaned:
status = item.get("clean_status", "unknown")
emoji = {"ok": "✅", "corrected": "🔧", "abnormal": "⚠️"}.get(status, "?")
print(f"\\n{emoji} [{status}] {item.get('title')} @ {item.get('company')}")
print(f" 城市：{item.get('city')} | 薪资：{item.get('salary_min')}–{item.get('salary_max')}k")
if item.get("corrections"):
print(f" 修正：{item['corrections']}")

# 统计
ok_count = sum(1 for x in cleaned if x.get("clean_status") == "ok")
corrected_count = sum(1 for x in cleaned if x.get("clean_status") == "corrected")
abnormal_count = sum(1 for x in cleaned if x.get("clean_status") == "abnormal")

print(f"\\n{'='*80}")
print(f"统计：正常 {ok_count} 条 | 已修正 {corrected_count} 条 | 异常 {abnormal_count} 条")

# 保存
with open("cleaned_jobs.json", "w", encoding="utf-8") as f:
json.dump(cleaned, f, ensure_ascii=False, indent=2)

print(f"已保存到 cleaned_jobs.json")

预期输出

原始数据：7 条

正在用 AI 清洗…

清洗后：5 条

清洗结果：
——————————————————————————–

🔧 [corrected] Python开发工程师 @ 腾讯
城市：深圳 | 薪资：25-50k
修正：职位名标准化、公司名修正（腾迅→腾讯）、城市标准化

🔧 [corrected] Python开发工程师 @ 阿里巴巴
城市：杭州 | 薪资：30-60k
修正：职位名标准化、公司名修正（阿里巳→阿里巴巴）、城市标准化

🔧 [corrected] Python开发工程师 @ 字节跳动
城市：北京 | 薪资：20-40k
修正：职位名标准化、公司名修正、城市标准化

🔧 [corrected] Python开发工程师 @ 美团点评
城市：北京 | 薪资：null-nullk（面议）
修正：职位名标准化、空格清理

⚠️ [abnormal] Python开发工程师 @ 网易
城市：北京 | 薪资：100-200k
修正：职位名标准化、公司名修正（网易→网易）、城市标准化
异常：薪资超过100k，可能不准确

统计：正常 0 条 | 已修正 4 条 | 异常 1 条
已保存到 cleaned_jobs.json

注意：空 title 和空 company的记录被自动删除了，7 条变成了 5 条。

四、批量清洗策略

数据量大（几千、几万条）的时候，不能一次性全丢给 AI。

分批处理

def batch_clean(data, batch_size=20):
"""分批清洗数据"""
all_cleaned = []

for i in range(0, len(data), batch_size):
batch = data[i:i + batch_size]
print(f"正在清洗第 {i+1}–{i+len(batch)} 条…")

cleaned = clean_data_with_ai(batch)
all_cleaned.extend(cleaned)

# 控制频率
import time
time.sleep(1)

return all_cleaned

先粗筛再精洗

def smart_clean(data):
"""智能清洗：先用规则快速处理，再用 AI 精修"""
import pandas as pd

df = pd.DataFrame(data)

# 第一步：规则快速处理
# 去空格
for col in ["title", "company", "city"]:
if col in df.columns:
df[col] = df[col].str.strip()

# 删除明显无效数据
df = df[df["title"].notna() & (df["title"] != "")]
df = df[df["company"].notna() & (df["company"] != "")]

# 第二步：AI 精修（只处理需要语义判断的部分）
need_ai_clean = df.to_dict("records")
cleaned = batch_clean(need_ai_clean, batch_size=20)

return cleaned

先做不花钱的清洗（去空格、删空值），再用 AI 做需要语义理解的部分（修正错别字、统一名称），省钱。

五、常见清洗场景

1. 地址标准化

prompt = """将以下地址统一为"省-市-区"格式。
地址列表：
["北京市海淀区中关村", "上海浦东", "深圳南山科技园", "杭州余杭区文一西路"]

返回 JSON 数组，格式：[{{"original": "原文", "province": "省", "city": "市", "district": "区"}}]"""

2. 文本分类

prompt = """将以下职位分类到预定义类别中。
类别：["后端开发", "前端开发", "数据分析", "产品经理", "运维", "测试", "其他"]

职位列表：
["Python后端", "React前端", "数据分析师", "PM", "DevOps", "QA工程师", "挖掘机司机"]

返回 JSON：[{{"title": "职位", "category": "类别", "confidence": 0.95}}]"""

3. 信息补全

prompt = """根据公司名称，补全以下信息（如果不确定就填 null）：
[
{{"company": "字节跳动"}},
{{"company": "蚂蚁集团"}},
{{"company": "某不知名小公司"}}
]

返回 JSON：[{{"company": "公司名", "industry": "行业", "size": "规模", "founded_year": 年份}}]"""

4. 异常检测

prompt = """检查以下数据中的异常值：
[
{{"title": "实习生", "salary": "50k-80k"}},
{{"title": "CEO", "salary": "5k-8k"}},
{{"title": "Python开发", "salary": "25k-50k"}}
]

返回 JSON：[{{"index": 0, "is_abnormal": true, "reason": "实习生薪资过高"}}]"""

六、清洗质量的自我检查

AI 清洗完不代表就对了，要做检查：

def verify_cleaning(cleaned_data):
"""检查清洗结果是否合理"""
issues = []

for i, item in enumerate(cleaned_data):
# 检查必填字段
if not item.get("title"):
issues.append(f"第 {i+1} 条：缺少 title")

# 检查薪资合理性
salary_min = item.get("salary_min")
salary_max = item.get("salary_max")

if salary_min and salary_max:
if salary_min > salary_max:
issues.append(f"第 {i+1} 条：最低薪资 > 最高薪资")
if salary_max > 200:
issues.append(f"第 {i+1} 条：薪资异常高 {salary_max}k")

# 检查城市
valid_cities = ["北京", "上海", "广州", "深圳", "杭州", "成都", "南京", "武汉", "西安", "苏州"]
if item.get("city") and item["city"] not in valid_cities:
issues.append(f"第 {i+1} 条：城市可能不正确 '{item['city']}'")

if issues:
print("发现以下问题：")
for issue in issues:
print(f" ⚠️ {issue}")
else:
print("✅ 清洗结果检查通过")

return issues

📊 本篇成本透明栏

GLM-4.7-Flash 免费额度足够清洗几千条数据。

练手改造题

改造 1（基础）： 用爬虫从任意网站抓一批数据（不用多，20 条就够），然后写 AI 清洗代码把数据整理干净。

改造 2（进阶）： 写一个"清洗管道"：先从 CSV 读取原始数据 → 用规则做基础清洗（去空格、删空值）→ 用 AI 做语义清洗（标准化、纠错）→ 输出干净的 CSV。整个流程一键运行。

下期预告

Day 13 · AI 辅助解析：复杂表格和嵌套结构

有些页面的结构复杂到 XPath 写不出来——嵌套表格、混合排版、图文混排。Day 13 教你把页面截图丢给视觉模型 GLM-4.6V-Flash，直接让它"看图说话"输出结构化数据。

📚 资源与工具

• 智谱 BigModel（GLM-4.7-Flash 免费）： https://open.bigmodel.cn/
• pandas 数据清洗文档： https://pandas.pydata.org/docs/user_guide/text.html
• 本专栏配套代码： CSDN 下载区（每篇更新）
• 梅雅达编程笔记： 专注 Python + AI 实战教程，提供数据采集与自动化定制服务

往期回顾

Day 01 · 爬虫能干啥不能干啥

Day 02 · 环境搭建与第一个爬虫

Day 03 · 列表页抓取：批量拿数据

Day 04 · 详情页 + 翻页：从一条到一百条

Day 05 · Ajax 请求拦截：抓看不到的数据

Day 06 · 浏览器自动化：DrissionPage 实战

Day 07 · Cookie 与 Session：处理登录状态

Day 08 · 反爬对策：Headers + 限速 + 代理

Day 09 · 存成文件：CSV / Excel / JSON

Day 10 · 存进数据库：SQLite 从零上手

Day 11 · 用 AI 替代正则：智能提取结构化数据

专栏推荐

• 「AI Python 系列」第 03 栏 · Python 爬虫实战（连载中）
• 「AI Python 系列」第 01 栏 · AI+自动化办公（连载中）
• 「AI Python 系列」第 02 栏 · AI+数据可视化（连载中）
• 「AI Python 系列」第05栏 · AI Agent实战（连载中）

资源领取

• 关注作者获取本栏完整代码和数据集

原创声明：本文为梅雅达编程笔记原创作品，未经允许不得转载。