Python自动化清洗脚本
#!/usr/bin/env python3
-- coding: utf-8 --
"""
订单数据清洗与质量评估脚本
功能：
- 读取CSV，进行数据质量评估（缺失、重复、格式、异常）
- 执行清洗：缺失填充/标记、格式统一（日期/文本/数值/币种）、字段错位修复、重复合并、金额核对与修正、单位换算
- 分组异常检测（IQR法）并标记
- 输出清洗后的CSV、质量报告JSON和处理日志
- 严格遵循数据治理要求：不删除关键业务信息，所有修正可追溯（保留原字段副本与issue_flags）
使用方式：
python clean_orders.py --input input_orders.csv --output-dir output --base-currency CNY --rate-USD 7.0
若未提供 --input，将自动写入本题示例数据到 ./input_orders.csv 并读取。
"""
import os
import sys
import json
import math
import argparse
import logging
import datetime as dt
from typing import Dict, Any, List, Tuple
import pandas as pd
import numpy as np
EXPECTED_COLUMNS = [
"order_id","order_date","customer_id","region","item","qty",
"unit_price","currency","amount","discount","status","note"
]
可配置参数：汇率、基础币种
DEFAULT_BASE_CCY = "CNY"
DEFAULT_EXCHANGE_RATES = {
"USD": 7.0, # 可按需通过命令行覆盖
"CNY": 1.0,
"RMB": 1.0
}
地区与状态标准化映射
REGION_MAP = {
"华东": "华东",
"hua dong": "华东",
"hua dong": "华东",
"hua_dong": "华东",
"huadong": "华东",
"华東": "华东",
"华北": "华北",
"华南": "华南",
"华中": "华中",
"西南": "西南",
"西北": "西北",
"东北": "东北"
}
STATUS_MAP = {
"已支付": "已支付",
"paid": "已支付",
"paid ": "已支付",
"paid.": "已支付",
"paid,": "已支付",
"paid/": "已支付",
"paid\": "已支付",
"paid;": "已支付",
"paid:": "已支付",
"paid!": "已支付",
"paid?": "已支付",
"PAID": "已支付",
"Paid": "已支付",
"已取消": "已取消",
"取消": "已取消",
"未支付": "未支付",
"待支付": "未支付",
"pending": "未支付"
}
def setup_logging(output_dir:str):
os.makedirs(output_dir, exist_ok=True)
log_path = os.path.join(output_dir, "cleaning_log.txt")
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
handlers=[
logging.FileHandler(log_path, encoding="utf-8"),
logging.StreamHandler(sys.stdout)
]
)
logging.info("日志初始化完成 -> %s", log_path)
return log_path
def write_sample_if_missing(path:str):
if os.path.exists(path):
return
sample = """order_id,order_date,customer_id,region,item,qty,unit_price,currency,amount,discount,status,note
1001,2024-01-05,CU001,华东,咖啡豆A,2,58,CNY,116,0.05,已支付,首单优惠
1001,2024-01-05,CU001,华东,咖啡豆A,2,58,CNY,116,0.05,已支付,重复提交
1002,1/6/2024,CU002,Hua Dong,咖啡豆A,,58,CNY,,0,已支付,缺失数量与金额
1003,2024/01/07,CU003,华北,奶茶粉B,1000,12,CNY,12000,0.1,已支付,数量异常
1004,2024-01-08,CU004,华东,巧克力C,3,$9.5,USD,28.5,0,PAID,英文状态与币种
1005,2024-01-09,CU005,华南,曲奇D,5,15,CNY,60,0,已支付,金额不匹配(应=75)
1006,2024-13-01,CU006,华中,蛋糕E,1,50,CNY,50,0,已支付,日期错误
1007,2024-01-10,CU007,华东,酸奶F,2,???,CNY,??,0,已支付,文本乱码
1008,2024-01-11,CU008,华东,生鲜组合,1,30,CNY,30,0,已支付,正常记录
1009,2024-01-11,CU009,华东,生鲜,组合,1,30,CNY,30,0,已支付,字段错位
1010,2024-01-12,CU010,Ã¥—æ‘,牛奶G,2,5,CNY,10,0,已支付,编码问题
1011,2024-01-12,CU011,华東,面包H,3,7,CNY,21,0,已支付,地区拼写变体
"""
with open(path, "w", encoding="utf-8") as f:
f.write(sample)
def read_csv(input_path:str)->pd.DataFrame:
df = pd.read_csv(input_path, dtype=str, keep_default_na=False, na_values=["", "NA", "N/A", "null", "Null", "NULL"])
# 去除列名空格与隐藏字符
df.columns = [c.strip() for c in df.columns]
# 校验列
missing_cols = [c for c in EXPECTED_COLUMNS if c not in df.columns]
if missing_cols:
raise ValueError(f"输入缺少必要列: {missing_cols}")
# 仅保留所需列的顺序
df = df[EXPECTED_COLUMNS].copy()
return df
def safe_strip(x):
if isinstance(x, str):
return x.strip()
return x
def maybe_fix_encoding(s: str) -> str:
# 尝试修复常见mojibake：latin1误解码的UTF-8
if not isinstance(s, str) or s == "":
return s
try:
repaired = s.encode("latin1").decode("utf-8")
# 简单启发：若中文字符数量增加则采用
def zh_count(txt):
return sum(1 for ch in txt if '\u4e00' <= ch <= '\u9fff')
if zh_count(repaired) > zh_count(s):
return repaired
except Exception:
pass
return s
def normalize_region(s: str) -> Tuple[str,bool]:
if not isinstance(s, str):
return s, False
original = s
s1 = maybe_fix_encoding(s).strip()
key = s1.replace(" ", "_").lower()
normalized = REGION_MAP.get(key, REGION_MAP.get(s1, None))
changed = False
if normalized:
changed = (normalized != original)
return normalized, changed
# 若包含已知中文区域词根则保留原值
for cand in ["华东","华南","华北","华中","西南","西北","东北"]:
if cand in s1:
return cand, (cand != original)
# 无法识别则保留原值
return s1, (s1 != original)
def normalize_status(s: str) -> Tuple[str,bool]:
if not isinstance(s, str):
return s, False
original = s
key = s.strip().lower()
normalized = STATUS_MAP.get(key, STATUS_MAP.get(s.strip(), s.strip()))
return normalized, (normalized != original)
def parse_discount(x) -> Tuple[float, bool]:
"""
返回 (discount_in_0_1, changed?)
"""
if x is None or (isinstance(x, float) and np.isnan(x)) or x == "":
return (0.0, True)
s = str(x).strip()
original = s
changed = False
if s.endswith("%"):
try:
v = float(s[:-1])
return (round(v/100.0, 4), True)
except:
return (0.0, True)
# 纯数值
try:
v = float(s)
# 若>1，按百分比处理
if v > 1.0:
return (round(v/100.0, 4), True)
return (round(v, 4), False)
except:
return (0.0, True)
def clean_number(s: Any) -> Tuple[float, bool]:
"""
清理数值（去符号、去逗号、去未知值），返回(float或NaN, changed)
"""
if s is None:
return (np.nan, False)
if isinstance(s, (int, float)) and not isinstance(s, bool):
return (float(s), False)
txt = str(s).strip()
original = txt
# 处理未知/乱码
if txt in ["???","??","--","-","N/A","NA",""]:
return (np.nan, True)
# 去货币符号和逗号
txt = txt.replace(",", "").replace("¥","").replace("￥","").replace("$","")
changed = (txt != original)
try:
return (float(txt), changed)
except:
return (np.nan, True)
def is_number(s: Any) -> bool:
try:
float(str(s))
return True
except:
return False
def fix_field_misalignment(row: pd.Series) -> Tuple[pd.Series, bool]:
"""
针对 qty 非数值而后续字段呈现右移的情况进行修正。
规则：
- 若 qty 非数值 且 unit_price 为数值 且 currency 非标准币种（或为数值），
则认为 item 与 qty 误拆分，执行左移一位并合并 item。
"""
changed = False
qty = row["qty"]
unit_price = row["unit_price"]
currency = row["currency"]
qty_is_num = is_number(qty)
unit_price_is_num = is_number(unit_price)
# 若currency看起来不像币种（例如为数字），或长度>4非字母，视为错位信号
currency_suspect = (is_number(currency) or (isinstance(currency, str) and (len(currency) > 4 and not currency.isalpha())))
if (not qty_is_num) and unit_price_is_num and currency_suspect:
# 左移一位：item = item + qty； 其余字段整体左移
order = ["item","qty","unit_price","currency","amount","discount","status","note"]
vals = [row[k] for k in order]
# 合并 item 与 qty
new_item = str(vals[0]) + str(vals[1])
shifted = [new_item] + vals[2:] + ["字段错位-自动修复"] # 左移并填补note
for k, v in zip(order, shifted):
row[k] = v
changed = True
return row, changed
def parse_date_str(s: Any) -> Tuple[str, bool, str]:
"""
解析日期，统一为 YYYY-MM-DD。
返回: (标准化日期或空, 是否更改, 问题说明)
策略：
- 优先尝试格式：%Y-%m-%d, %Y/%m/%d, %m/%d/%Y
- 若失败，尝试 %Y-%d-%m（处理'2024-13-01' -> 2024-01-13）
- 对歧义 '1/6/2024' 按业务上下文（样本多为1月）假定为MM/DD/YYYY => 2024-01-06
"""
if not isinstance(s, str) or s.strip() == "":
return "", False, "empty"
original = s
s = s.strip()
fmts = ["%Y-%m-%d","%Y/%m/%d","%m/%d/%Y"]
for f in fmts:
try:
d = dt.datetime.strptime(s, f).date()
return d.isoformat(), (s != d.isoformat()), ""
except:
pass
# 尝试 %Y-%d-%m
try:
d = dt.datetime.strptime(s, "%Y-%d-%m").date()
# 记录更正
return d.isoformat(), True, "swapped_day_month"
except:
pass
return "", False, "invalid"
def currency_normalize(row: pd.Series, base_ccy: str, rates: Dict[str,float]) -> Tuple[pd.Series, bool]:
"""
将不同币种换算为基础币种，保留原值 original_* 字段，确保可追溯。
"""
changed = False
# 原始值备份（若无则写入）
if "original_currency" not in row or pd.isna(row.get("original_currency", np.nan)):
row["original_currency"] = row["currency"]
if "original_unit_price" not in row or pd.isna(row.get("original_unit_price", np.nan)):
row["original_unit_price"] = row["unit_price"]
if "original_amount" not in row or pd.isna(row.get("original_amount", np.nan)):
row["original_amount"] = row["amount"]
ccy = str(row["currency"]).strip() if isinstance(row["currency"], str) else row["currency"]
ccy_upper = str(ccy).upper()
# 同义词归一
if ccy_upper in ["RMB", "CNY", "￥", "¥"]:
ccy_upper = "CNY"
if ccy_upper not in rates:
# 未知币种，保持不变，仅记录
return row, changed
if ccy_upper != base_ccy:
rate = rates[ccy_upper]
if is_number(row["unit_price"]):
row["unit_price"] = round(float(row["unit_price"]) * rate, 4)
changed = True
if is_number(row["amount"]):
row["amount"] = round(float(row["amount"]) * rate, 4)
changed = True
row["currency"] = base_ccy
if changed:
row["issue_flags"] = append_flag(row.get("issue_flags",""), "currency_converted")
return row, changed
def append_flag(flags: str, new_flag: str) -> str:
s = (flags or "").strip()
if s == "":
return new_flag
parts = set([p.strip() for p in s.split(",") if p.strip()])
parts.add(new_flag)
return ",".join(sorted(parts))
def compute_expected_amount(qty, unit_price, discount):
if any([pd.isna(qty), pd.isna(unit_price), pd.isna(discount)]):
return np.nan
try:
return round(float(qty) * float(unit_price) * (1.0 - float(discount)), 2)
except:
return np.nan
def pre_quality_scan(df: pd.DataFrame) -> Dict[str, Any]:
m = {}
m["total_rows_raw"] = len(df)
m["duplicate_order_id_rows"] = int((df.duplicated(subset=["order_id"])).sum())
m["missing_qty_rows"] = int((df["qty"].astype(str).replace({"": np.nan}).isna() | ~df["qty"].astype(str).apply(is_number)).sum())
m["missing_unit_price_rows"] = int((df["unit_price"].astype(str).replace({"": np.nan}).isna() | ~df["unit_price"].astype(str).apply(lambda x: is_number(str(x).replace("$","").replace(",","")))).sum())
m["missing_amount_rows"] = int((df["amount"].astype(str).replace({"": np.nan}).isna() | ~df["amount"].astype(str).apply(lambda x: is_number(str(x).replace("$","").replace(",","")))).sum())
# 非标准日期（尝试解析失败计数）
invalid_dates = 0
for s in df["order_date"].astype(str).tolist():
std, chg, note = parse_date_str(s)
if note == "invalid":
invalid_dates += 1
m["invalid_date_rows"] = invalid_dates
# 非标准地区
nonstd_region = 0
for s in df["region"].astype(str).tolist():
norm, changed = normalize_region(s)
if changed:
nonstd_region += 1
m["region_normalization_needed_rows"] = nonstd_region
# 非标准状态
nonstd_status = 0
for s in df["status"].astype(str).tolist():
norm, changed = normalize_status(s)
if changed:
nonstd_status += 1
m["status_normalization_needed_rows"] = nonstd_status
# 可能错位
misaligned = 0
for _, row in df.iterrows():
_, changed = fix_field_misalignment(row.copy())
if changed:
misaligned += 1
m["field_misalignment_rows"] = misaligned
# 非基础币种
non_base_ccy = int((~df["currency"].str.upper().isin(["CNY","RMB"])).sum())
m["non_base_currency_rows"] = non_base_ccy
return m
def post_quality_scan(df: pd.DataFrame) -> Dict[str, Any]:
m = {}
m["total_rows_cleaned"] = len(df)
m["duplicate_order_id_rows"] = int((df.duplicated(subset=["order_id"])).sum())
m["missing_qty_rows"] = int(df["qty"].isna().sum())
m["missing_unit_price_rows"] = int(df["unit_price"].isna().sum())
m["missing_amount_rows"] = int(df["amount"].isna().sum())
m["invalid_date_rows"] = int(df["order_date"].isna().sum())
m["non_base_currency_rows"] = int((df["currency"].str.upper() != "CNY").sum())
# 统计标记
for flag in ["currency_converted","amount_corrected","field_misalignment_fixed","encoding_fixed","region_normalized","status_normalized","date_corrected","duplicate_merged","outlier_qty","outlier_unit_price"]:
m[f"flag_{flag}"] = int(df["issue_flags"].fillna("").str.contains(flag).sum())
return m
def merge_duplicates(df: pd.DataFrame, log: logging.Logger) -> Tuple[pd.DataFrame, int]:
"""
基于 order_id 合并重复：
- 数值字段优先非空且校验通过
- 文本字段合并备注
- 保留 earliest order_date（或修正后的标准日期）并统一
"""
if df["order_id"].duplicated().sum() == 0:
return df, 0
cols = df.columns.tolist()
groups = []
removed = 0
for oid, g in df.groupby("order_id", sort=False):
if len(g) == 1:
groups.append(g.iloc[0])
continue
removed += len(g) - 1
# 选取首条为基准
base = g.iloc[0].copy()
# 合并 note
all_notes = [str(x) for x in g["note"].fillna("").tolist() if str(x).strip()!=""]
base["note"] = ";".join(dict.fromkeys(all_notes)) if all_notes else base.get("note","")
# 状态取规范化后一致值，若不一致优先“已支付”
statuses = g["status"].fillna("").tolist()
if "已支付" in statuses:
base["status"] = "已支付"
else:
base["status"] = statuses[0] if statuses else base["status"]
# 对数值字段：qty, unit_price, amount, discount 选非空优先，并在后续统一校验
for c in ["qty","unit_price","amount","discount"]:
vals = [v for v in g[c].tolist() if not pd.isna(v)]
if len(vals)>0:
# 若存在多个不同值，记录冲突
if len(set([str(v) for v in vals]))>1:
base["issue_flags"] = append_flag(base.get("issue_flags",""), f"duplicate_conflict_{c}")
base[c] = vals[0]
# 其他文本字段以首条为主
base["issue_flags"] = append_flag(base.get("issue_flags",""), "duplicate_merged")
groups.append(base)
df2 = pd.DataFrame(groups, columns=cols)
return df2, removed
def detect_outliers_iqr(series: pd.Series) -> pd.Series:
"""
IQR法标记异常：低于 Q1-1.5IQR 或 高于 Q3+1.5IQR
返回布尔序列
"""
s = series.dropna().astype(float)
if len(s) < 4:
return pd.Series([False]len(series), index=series.index)
q1, q3 = s.quantile(0.25), s.quantile(0.75)
iqr = q3 - q1
low, high = q1 - 1.5iqr, q3 + 1.5*iqr
out = (series.astype(float) < low) | (series.astype(float) > high)
out = out.fillna(False)
return out
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--input", type=str, default="input_orders.csv", help="输入CSV路径")
parser.add_argument("--output-dir", type=str, default="output", help="输出目录")
parser.add_argument("--base-currency", type=str, default=DEFAULT_BASE_CCY, help="基础币种（默认CNY）")
parser.add_argument("--rate-USD", type=float, default=DEFAULT_EXCHANGE_RATES["USD"], help="USD兑基础币种的汇率")
args = parser.parse_args()
os.makedirs(args.output_dir, exist_ok=True)
log_path = setup_logging(args.output_dir)
write_sample_if_missing(args.input)
# 读取
df_raw = read_csv(args.input)
# 保存原始快照
raw_snapshot = os.path.join(args.output_dir, "raw_snapshot.csv")
df_raw.to_csv(raw_snapshot, index=False, encoding="utf-8-sig")
logging.info("原始数据快照保存: %s", raw_snapshot)
# 预评估
pre_metrics = pre_quality_scan(df_raw)
logging.info("清洗前质量评估: %s", json.dumps(pre_metrics, ensure_ascii=False))
df = df_raw.copy()
# 基础清理：去空格
for c in df.columns:
df[c] = df[c].apply(safe_strip)
# 字段错位修复
misaligned_fixed = 0
new_rows = []
for _, row in df.iterrows():
row, changed = fix_field_misalignment(row)
if changed:
row["issue_flags"] = append_flag(row.get("issue_flags",""), "field_misalignment_fixed")
misaligned_fixed += 1
new_rows.append(row)
df = pd.DataFrame(new_rows, columns=df.columns)
# 文本规范化与编码修复
enc_fixed = 0
region_normed = 0
status_normed = 0
std_dates = 0
invalid_dates = 0
std_dates_list = []
region_list = []
status_list = []
for i, row in df.iterrows():
# 日期
std, changed, note = parse_date_str(row["order_date"])
if std != "":
if changed or (row["order_date"] != std):
row["issue_flags"] = append_flag(row.get("issue_flags",""), "date_corrected")
std_dates += 1
row["order_date"] = std
else:
if note == "invalid":
invalid_dates += 1
row["issue_flags"] = append_flag(row.get("issue_flags",""), "date_invalid")
row["order_date"] = np.nan
else:
row["order_date"] = np.nan
# 地区
before_region = row["region"]
rnorm, rchg = normalize_region(before_region)
if rchg:
region_normed += 1
row["issue_flags"] = append_flag(row.get("issue_flags",""), "region_normalized")
if maybe_fix_encoding(before_region) != before_region:
enc_fixed += 1
row["issue_flags"] = append_flag(row.get("issue_flags",""), "encoding_fixed")
row["region"] = rnorm
# 状态
snorm, schg = normalize_status(row["status"])
if schg:
status_normed += 1
row["issue_flags"] = append_flag(row.get("issue_flags",""), "status_normalized")
row["status"] = snorm
std_dates_list.append(row["order_date"])
region_list.append(row["region"])
status_list.append(row["status"])
df.loc[i] = row
# 数值标准化：qty, unit_price, amount, discount
qty_list = []
up_list = []
amt_list = []
disc_list = []
for i, row in df.iterrows():
# qty
q_raw = row["qty"]
qv_changed = False
if q_raw is None or str(q_raw).strip()=="":
q = np.nan
qv_changed = True
else:
# 去掉非数字
if is_number(q_raw):
q = float(q_raw)
else:
q = np.nan
qv_changed = True
# unit_price
up, chg_up = clean_number(row["unit_price"])
# amount
amt, chg_amt = clean_number(row["amount"])
# discount
disc, chg_disc = parse_discount(row["discount"])
if qv_changed or chg_up or chg_amt or chg_disc:
row["issue_flags"] = append_flag(row.get("issue_flags",""), "numeric_cleaned")
qty_list.append(q if (pd.isna(q) or q.is_integer()) else q) # 可能是浮点，但后续转int
up_list.append(up)
amt_list.append(amt)
disc_list.append(disc)
df.loc[i] = row
df["qty"] = qty_list
df["unit_price"] = up_list
df["amount"] = amt_list
df["discount"] = disc_list
# 货币转换到基础币种
rates = dict(DEFAULT_EXCHANGE_RATES)
if args.base_currency.upper() not in rates:
rates[args.base_currency.upper()] = 1.0
# 覆盖USD汇率
rates["USD"] = float(args.rate_USD) if hasattr(args, "rate_USD") else DEFAULT_EXCHANGE_RATES["USD"]
conv_changed = 0
for i, row in df.iterrows():
row, changed = currency_normalize(row, args.base_currency.upper(), rates)
if changed:
conv_changed += 1
df.loc[i] = row
# 金额核对与修正：amount = qty * unit_price * (1 - discount)
amount_corrected = 0
for i, row in df.iterrows():
exp_amt = compute_expected_amount(row["qty"], row["unit_price"], row["discount"])
if not pd.isna(exp_amt):
if pd.isna(row["amount"]) or (abs(float(row["amount"]) - exp_amt) > 0.01):
df.at[i, "amount"] = exp_amt
df.at[i, "issue_flags"] = append_flag(row.get("issue_flags",""), "amount_corrected")
amount_corrected += 1
# 数据类型微调
# qty尽可能为整数
def to_int_or_nan(v):
if pd.isna(v):
return np.nan
try:
if float(v).is_integer():
return int(float(v))
return int(round(float(v)))
except:
return np.nan
df["qty"] = df["qty"].apply(to_int_or_nan)
# 单价保留两位
df["unit_price"] = df["unit_price"].apply(lambda x: round(float(x), 4) if (not pd.isna(x)) else x)
df["amount"] = df["amount"].apply(lambda x: round(float(x), 2) if (not pd.isna(x)) else x)
# 合并重复订单
df, removed_dup = merge_duplicates(df, logging)
logging.info("重复记录合并完成，移除行数: %s", removed_dup)
# 分组异常检测（按 item 分组）
df["outlier_qty"] = False
df["outlier_unit_price"] = False
for item, g in df.groupby("item"):
idx = g.index
if g["qty"].notna().sum() >= 4:
out_q = detect_outliers_iqr(g["qty"].astype(float))
df.loc[idx, "outlier_qty"] = out_q
if g["unit_price"].notna().sum() >= 4:
out_p = detect_outliers_iqr(g["unit_price"].astype(float))
df.loc[idx, "outlier_unit_price"] = out_p
# 将异常标记写入 issue_flags，不进行值修改以保护业务信息
df.loc[df["outlier_qty"]==True, "issue_flags"] = df.loc[df["outlier_qty"]==True, "issue_flags"].apply(lambda x: append_flag(x, "outlier_qty"))
df.loc[df["outlier_unit_price"]==True, "issue_flags"] = df.loc[df["outlier_unit_price"]==True, "issue_flags"].apply(lambda x: append_flag(x, "outlier_unit_price"))
# 事后质量评估
post_metrics = post_quality_scan(df)
# 输出文件
cleaned_path = os.path.join(args.output_dir, "cleaned_orders.csv")
report_path = os.path.join(args.output_dir, "quality_report.json")
# 输出顺序与追溯字段
output_cols = [
"order_id","order_date","customer_id","region","item",
"qty","unit_price","currency","amount","discount",
"status","note",
"original_currency","original_unit_price","original_amount",
"issue_flags","outlier_qty","outlier_unit_price"
]
for c in output_cols:
if c not in df.columns:
df[c] = np.nan
df = df[output_cols]
df.to_csv(cleaned_path, index=False, encoding="utf-8-sig")
full_report = {
"run_time": dt.datetime.now().isoformat(),
"base_currency": args.base_currency.upper(),
"exchange_rates_used": rates,
"paths": {
"log": log_path,
"raw_snapshot": raw_snapshot,
"cleaned_csv": cleaned_path
},
"metrics_before": pre_metrics,
"metrics_after": post_metrics,
"notes": {
"field_misalignment_fixed": "检测到qty非数值且后续字段错位时，将item与qty合并并整体左移一位",
"date_correction": "按多格式解析并在必要时采用YYYY-DD-MM作为兜底修正(示例: 2024-13-01 -> 2024-01-13)",
"currency_conversion": "将USD等换算为基础币种CNY，保留original_*追溯；汇率来自脚本参数",
"amount_validation": "使用 amount = qty * unit_price * (1 - discount) 校验并修正",
"duplicates_merge": "按order_id合并重复并合并备注，保留非空且一致性更高的值",
"outlier_detection": "按item分组使用IQR法标记极端值但不自动修改"
}
}
with open(report_path, "w", encoding="utf-8") as f:
json.dump(full_report, f, ensure_ascii=False, indent=2)
logging.info("清洗完成。清洗文件: %s, 质量报告: %s", cleaned_path, report_path)
print("\n=== 输出路径 ===")
print(f"清洗后CSV: {cleaned_path}")
print(f"质量报告JSON: {report_path}")
print(f"处理日志: {log_path}")
if name == "main":
main()
清洗效果评估
优化与持续监控建议
- 数据入口治理
- 上游系统强校验：在录入层面强制字段类型（qty为整数、单价/金额为数值、日期为YYYY-MM-DD）。
- 下拉字典统一：地区、状态、币种采用受控词表，避免自由文本（例如“Hua Dong”“华東”等变体）。
- 文件编码统一为UTF-8，并在ETL入口做编码探测与强制转码。
- 金额与汇率
- 汇率来源改为权威API或主数据系统，并按生效日期保留汇率版本与换算时间戳。
- 价格/金额校验前置到订单创建流程，异常即时阻断或提示。
- 异常监控
- 基于历史分布的自适应阈值，对qty、单价、折扣的异常进行持续监控并告警。
- 按item/region维度输出日常质量看板（缺失率、异常率、修正率、重复率）。
- 追溯与审计
- 保留original_*与issue_flags字段，坚持“不丢信息、可回溯”原则。
- 对“自动修正”类规则（金额修正、错位修复、日期纠偏）建立变更审计表与抽样复核流程。
- 性能与自动化
- 调度：将本脚本纳入定时任务（如Airflow/Cron），对新到文件自动执行。
- 测试：增加单元测试与样例快照测试，防止规则变更导致回归问题。
说明
- 脚本默认基础币种为CNY，USD汇率默认7.0，请在运行时按需通过参数覆盖，汇率取值应由业务或财务确认。
- 所有修正均通过issue_flags标记，并保留original_*字段，确保透明、可追溯。
- 对无法可靠推导的数据（如“???”单价、金额双缺失）不盲目填充，避免损害数据完整性。
