- 错误识别 🧭
- 行动描述
- 列出触发点与可能原因：分隔符混用导致列数不一致；多种日期格式与时区导致解析失败；user_id被推断为整数导致前导零丢失；amount含币种与欧式小数；文件内混杂UTF-8与GBK编码。
- 适用代码片段
- 预期结果
- 明确五类错误来源：分隔符、日期、多编码、ID类型、金额格式。
- 调试步骤 🔍
- 行动描述
- 依序执行：编码嗅探与统一；分隔符自动检测与统一；惰性读取与坏行隔离；多格式日期解析与本地化；数值规范化与币种剥离；ID强制字符串；重复检测与去重。
- 适用代码片段
- 预期结果
- 生成统一UTF-8、统一逗号分隔的CSV；隔离坏行到 quarantine.csv；产出规范化DataFrame。
- 解决方案实施 🛠️
- 行动描述
- 使用 pandas + python csv + dateutil。包含容错读、隔离坏行、字段规范化。请先设置配置项。
- 适用代码片段
- 可执行脚本（保存为 clean_events.py 或在Notebook运行）：
import os, io, csv, re, random
from decimal import Decimal, InvalidOperation
import numpy as np
import pandas as pd
from dateutil import parser, tz
# ===== 配置 =====
CONFIG = {
"expected_cols": 7, # 头部列数；可留None自动取header
"local_tz": "Asia/Shanghai", # 本地时区
"user_id_col": "user_id", # 如果列名不同，请修改
"timestamp_col": None, # None时自动检测
"amount_col": None, # None时自动检测
"dedupe_subset": None # 例如 ["user_id", "timestamp_utc"]；None则全列去重
}
# ===== 1) 编码嗅探与统一（逐行解码，UTF-8优先，失败退GBK；仍失败->隔离） =====
def unify_encoding(src_path, utf8_path, quarantine_path):
total, good, bad = 0, 0, 0
with open(src_path, "rb") as f_in, \
open(utf8_path, "w", encoding="utf-8", newline="") as f_out, \
open(quarantine_path, "w", encoding="utf-8", newline="") as f_bad:
for raw in f_in:
total += 1
for enc in ("utf-8-sig", "gbk"):
try:
line = raw.decode(enc)
# 去除BOM
line = line.replace("\ufeff", "")
f_out.write(line)
good += 1
break
except UnicodeDecodeError:
continue
else:
# 两种编码都失败，隔离
try:
# 尝试二进制转义保留信息
safe = raw.decode("utf-8", errors="replace")
except Exception:
safe = str(raw)
f_bad.write(safe)
bad += 1
return {"total_lines": total, "encoded_ok": good, "encoded_bad": bad}
# ===== 2) 分隔符统一与坏行隔离（逗号/分号自动切换；统一写成逗号） =====
def parse_with_delim(line, delim):
buf = io.StringIO(line)
reader = csv.reader(buf, delimiter=delim, quotechar='"', escapechar="\\")
return next(reader)
def unify_delimiters(utf8_path, unified_path, quarantine_path, expected_cols=None):
with open(utf8_path, "r", encoding="utf-8", newline="") as f_in, \
open(unified_path, "w", encoding="utf-8", newline="") as f_out, \
open(quarantine_path, "a", encoding="utf-8", newline="") as f_bad: # 追加写入
writer = csv.writer(f_out, delimiter=",", quotechar='"', escapechar="\\")
# 读header，确定主分隔符和列数
header_line = f_in.readline()
if not header_line:
return {"rows_ok": 0, "rows_bad": 0, "expected_cols": 0}
try:
h_comma = parse_with_delim(header_line, ",")
except Exception:
h_comma = []
try:
h_semi = parse_with_delim(header_line, ";")
except Exception:
h_semi = []
header = h_comma if len(h_comma) >= len(h_semi) else h_semi
main_delim = "," if header is h_comma else ";"
exp_cols = expected_cols or len(header)
writer.writerow(header)
rows_ok, rows_bad = 0, 0
# 逐行处理
for line in f_in:
row = None
for delim in (main_delim, ";" if main_delim == "," else ","):
try:
fields = parse_with_delim(line, delim)
if len(fields) == exp_cols:
row = fields
break
except Exception:
continue
if row is None:
f_bad.write(line)
rows_bad += 1
continue
writer.writerow(row)
rows_ok += 1
return {"rows_ok": rows_ok, "rows_bad": rows_bad, "expected_cols": exp_cols}
# ===== 3) 自动检测候选列 =====
def detect_datetime_column(df):
candidates = [c for c in df.columns if re.search(r"(time|date)", c, flags=re.I)]
if not candidates:
candidates = list(df.select_dtypes(include=["object", "string"]).columns)
best_col, best_ratio = None, 0.0
for c in candidates:
s = df[c].dropna().astype(str).head(200)
ok = 0
for v in s:
try:
parser.parse(v, dayfirst=True, fuzzy=True)
ok += 1
except Exception:
pass
ratio = ok / max(1, len(s))
if ratio > best_ratio:
best_ratio, best_col = ratio, c
return best_col if best_ratio >= 0.6 else None
def detect_amount_column(df):
name_hits = [c for c in df.columns if re.search(r"(amount|price|total)", c, flags=re.I)]
if name_hits:
return name_hits[0]
# 回退：找含货币符号或小数的大概率金额列
for c in df.select_dtypes(include=["object", "string"]).columns:
s = df[c].dropna().astype(str).head(50)
hit = sum(bool(re.search(r"[$€£¥]|[A-Z]{3}|^\d+[.,]\d{2}$", v)) for v in s)
if hit / max(1, len(s)) >= 0.4:
return c
return None
# ===== 4) 日期解析（统一UTC，输出为带时区的datetime64[ns, UTC]） =====
LOCAL_TZ_CACHE = {}
def get_local_tz(name):
if name not in LOCAL_TZ_CACHE:
LOCAL_TZ_CACHE[name] = tz.gettz(name)
return LOCAL_TZ_CACHE[name]
def parse_datetime_utc(val, local_tz_name):
if pd.isna(val):
return pd.NaT
text = str(val)
try:
dt = parser.parse(text, dayfirst=True, fuzzy=True)
except Exception:
return pd.NaT
# 赋默认本地时区，再转UTC
local_zone = get_local_tz(local_tz_name)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=local_zone)
return pd.Timestamp(dt).tz_convert("UTC")
# ===== 5) 金额规范化与币种剥离 =====
SYM_TO_CODE = {"$": "USD", "€": "EUR", "£": "GBP", "¥": "CNY"}
EURO_PATTERN = re.compile(r"^\d{1,3}(\.\d{3})*,\d{2}$")
def normalize_amount_cell(val):
if pd.isna(val):
return (np.nan, None)
s = str(val).strip()
# 抽取币种
m_code = re.search(r"(?i)\b([A-Z]{3})\b", s)
currency = m_code.group(1).upper() if m_code else None
m_sym = re.search(r"[$€£¥]", s)
if currency is None and m_sym:
currency = SYM_TO_CODE.get(m_sym.group(0))
# 去除币种与符号
s = re.sub(r"(?i)\b[A-Z]{3}\b", "", s)
s = re.sub(r"[$€£¥]", "", s).strip()
# 欧式小数
if EURO_PATTERN.fullmatch(s):
num_str = s.replace(".", "").replace(",", ".")
else:
num_str = s.replace(",", "").replace(" ", "")
num_str = re.sub(r"[^\d\.\-]", "", num_str)
if num_str == "":
return (np.nan, currency)
try:
dec = Decimal(num_str)
except InvalidOperation:
return (np.nan, currency)
return (dec, currency)
# ===== 6) 主清洗流程 =====
def clean_events_csv(source_csv, output_clean_csv, quarantine_csv, config=CONFIG):
tmp_utf8 = source_csv + ".utf8.tmp"
tmp_unified = source_csv + ".unified.tmp"
stats_enc = unify_encoding(source_csv, tmp_utf8, quarantine_csv)
stats_sep = unify_delimiters(tmp_utf8, tmp_unified, quarantine_csv, expected_cols=config.get("expected_cols"))
# pandas惰性读取可用chunksize，但这里先一次读入（文件大时可改为chunksize处理）
df = pd.read_csv(
tmp_unified,
engine="python",
dtype={config["user_id_col"]: "string"},
keep_default_na=False,
na_values=["N/A", "-", ""],
)
# user_id 强制字符串 & 去空白
if config["user_id_col"] in df.columns:
df[config["user_id_col"]] = df[config["user_id_col"]].astype("string").str.strip()
# 自动检测列名
ts_col = config.get("timestamp_col") or detect_datetime_column(df)
amt_col = config.get("amount_col") or detect_amount_column(df)
# 日期解析 -> UTC
parse_ok = 0
if ts_col:
df["timestamp_utc"] = df[ts_col].map(lambda v: parse_datetime_utc(v, config["local_tz"]))
parse_ok = df["timestamp_utc"].notna().sum()
else:
df["timestamp_utc"] = pd.NaT
# 金额规范化与币种剥离
amt_ok = 0
if amt_col:
pairs = df[amt_col].map(normalize_amount_cell)
df["amount_dec"] = pairs.map(lambda x: x[0])
df["currency"] = pairs.map(lambda x: x[1])
# 转float避免后续计算问题；保留原Decimal列以检查
df["amount"] = df["amount_dec"].astype("float64")
amt_ok = df["amount"].notna().sum()
# 去重策略
before = len(df)
if config.get("dedupe_subset"):
df_clean = df.drop_duplicates(subset=config["dedupe_subset"], keep="first")
else:
df_clean = df.drop_duplicates(keep="first")
after = len(df_clean)
# 输出
df_clean.to_csv(output_clean_csv, index=False)
# 清理临时文件
for p in (tmp_utf8, tmp_unified):
try:
os.remove(p)
except Exception:
pass
metrics = {
"total_lines": stats_enc["total_lines"],
"encoded_ok": stats_enc["encoded_ok"],
"encoded_bad": stats_enc["encoded_bad"],
"rows_ok": stats_sep["rows_ok"],
"rows_bad": stats_sep["rows_bad"],
"expected_cols": stats_sep["expected_cols"],
"parse_time_ok": parse_ok,
"parse_amount_ok": amt_ok,
"rows_before": before,
"rows_after": after,
}
return df_clean, metrics
# ===== 7) 验证与打分 =====
def validate(df, metrics, config=CONFIG, sample_n=20, min_year=2000, max_year=2035, max_abs_amount=1e9):
report = {}
# 读入成功率
read_rate = metrics["rows_ok"] / max(1, metrics["rows_ok"] + metrics["rows_bad"])
report["read_rate"] = read_rate
# 日期解析成功率
if "timestamp_utc" in df.columns:
time_rate = df["timestamp_utc"].notna().mean()
report["time_parse_rate"] = time_rate
# 日期范围约束
dt_min = pd.Timestamp(f"{min_year}-01-01", tz="UTC")
dt_max = pd.Timestamp(f"{max_year}-12-31", tz="UTC")
in_range = df["timestamp_utc"].dropna().between(dt_min, dt_max).mean() if df["timestamp_utc"].notna().any() else np.nan
report["time_in_range_rate"] = in_range
# 金额约束
if "amount" in df.columns:
amt_valid = df["amount"].dropna().apply(lambda x: abs(x) <= max_abs_amount).mean() if df["amount"].notna().any() else np.nan
report["amount_valid_rate"] = amt_valid
# 重复率统计
dupe_dropped = metrics["rows_before"] - metrics["rows_after"]
dupe_rate = dupe_dropped / max(1, metrics["rows_before"])
report["dedupe_drop_rate"] = dupe_rate
# 随机抽样检查
sample = df.sample(min(sample_n, len(df)), random_state=42) if len(df) else df
sample_ok = {
"timestamp_non_null": int(sample["timestamp_utc"].notna().sum()) if "timestamp_utc" in df.columns else 0,
"amount_non_null": int(sample["amount"].notna().sum()) if "amount" in df.columns else 0,
}
report["sample_ok"] = sample_ok
# 简单断言（不抛错，只记录）
report["assertions"] = {
"no_int_user_id": df[CONFIG["user_id_col"]].dtype.name == "string",
}
# 交互式打分
# 成功读入率、日期解析成功率、去重消除率各占权重
score = 0
score += int(round(read_rate * 4)) # 0-4分
score += int(round(report.get("time_parse_rate", 0) * 3)) # 0-3分
score += int(round(dupe_rate * 3)) # 0-3分
report["score"] = score
return report
if __name__ == "__main__":
# 示例调用：修改路径与配置后运行
src = "user_events.csv"
out = "user_events_clean.csv"
quarantine = "quarantine.csv"
CONFIG.update({
"expected_cols": 7,
"local_tz": "Asia/Shanghai",
"user_id_col": "user_id",
"timestamp_col": None, # 自动检测
"amount_col": None, # 自动检测
"dedupe_subset": None # 或例如 ["user_id", "timestamp_utc"]
})
df_clean, metrics = clean_events_csv(src, out, quarantine, CONFIG)
report = validate(df_clean, metrics, CONFIG)
print("Metrics:", metrics)
print("Report:", report)
- 预期结果
- 读入阶段不再触发 ParserError；坏行写入 quarantine.csv。
- 日期统一为 UTC 带时区列 timestamp_utc，兼容格式：2024/11/02 14:23、02-11-2024 2:23 PM、2024-11-02T14:23:01Z。
- user_id 保持字符串类型，前导零不丢失。
- amount 产出数值列 amount 与币种列 currency，兼容“1.234,56”“USD 23.50”。
- 输出 user_events_clean.csv 可复现。
- 测试与验证 ✅
- 行动描述
- 运行内置 validate，执行随机抽样校验、日期范围与金额约束断言、重复率统计；打印评分。
- 适用代码片段
- 已在脚本的 validate 中实现；也可在Notebook中手动执行：
df_clean, metrics = clean_events_csv("user_events.csv", "user_events_clean.csv", "quarantine.csv", CONFIG)
report = validate(df_clean, metrics, CONFIG)
print(f"成功读入率: {report['read_rate']:.2%}")
print(f"日期解析成功率: {report.get('time_parse_rate', 0):.2%}")
print(f"日期范围命中率: {report.get('time_in_range_rate', np.nan)}")
print(f"金额有效率: {report.get('amount_valid_rate', np.nan)}")
print(f"去重消除率: {report['dedupe_drop_rate']:.2%}")
print(f"得分: {report['score']}/10")
加分建议 🎯
- 如果文件很大：把 pandas 读入改为 chunksize=100_000，按块应用解析与规范化，再拼接或直接写出。
- 如果日期列确定：设置 CONFIG["timestamp_col"] 避免自动检测误判。
- 若币种需保留映射或做汇率换算：保留 currency 列，另行处理。
互动打分与积分规则 🕹️
- 成功读入率≥95%：4分；85–95%：3分；70–85%：2分；<70%：1分
- 日期解析成功率≥90%：3分；75–90%：2分；50–75%：1分；<50%：0分
- 去重消除率≥20%：3分；10–20%：2分；1–10%：1分；<1%：0分
任务完成！获得10分。
当前等级：初级调试员
下一等级：代码修复者（距离20分）
