#WorkBuddy# 18 个分表 2.7 万行，我把"多表合并"跑成一条带 6 类校验的流水线

import glob, re, unicodedata
import pandas as pd
from datetime import datetime

HEADER_MAP = {"订单编号":"order_id","订单号":"order_id","order_id":"order_id",
              "下单日期":"order_date","日期":"order_date","date":"order_date",
              "商品编码":"sku","SKU":"sku","货号":"sku",
              "数量":"qty","件数":"qty","qty":"qty",
              "单价":"price","price":"price",
              "金额":"amount","总价":"amount","amount":"amount","total":"amount",
              "客户名称":"customer","客户":"customer",
              "订单状态":"status","状态":"status"}
DATE_FMTS = ["%Y-%m-%d","%Y/%m/%d","%Y%m%d","%d-%m-%Y"]

def norm(s):
    s = unicodedata.normalize("NFKC", str(s)).strip().lower()
    return HEADER_MAP.get(s, s)

frames = []
for f in glob.glob("./raw/*.xlsx"):
    df = pd.read_excel(f, dtype=object)
    df.columns = [norm(c) for c in df.columns]
    df["__src"] = f
    frames.append(df)
m = pd.concat(frames, ignore_index=True, sort=False)

def num(v):
    s = str(v).replace("¥","").replace(",","").strip()
    try: return round(float(s), 2)
    except Exception: return None

def pdate(v):
    for f in DATE_FMTS:
        try: return datetime.strptime(str(v).strip(), f)
        except Exception: pass
    return None

m["amount"] = m["amount"].map(num)
m["price"]  = m["price"].map(num)
m["qty"]    = m["qty"].map(num)
m["order_date"] = m["order_date"].map(pdate)

# 规则1：完全重复行
dup = m.duplicated(keep="first")
clean = m[~dup].copy()
# 规则2：金额勾稽 qty*price == amount
clean["calc"] = (clean["qty"]*clean["price"]).round(2)
bad_amt = clean[(clean["amount"]-clean["calc"]).abs() > 0.01]
# 规则3：必填缺失
miss = clean[["order_id","sku","qty","price","amount"]].isna().any(axis=1)
# 规则4：非法值（负数量 / 坏日期）
inv = (clean["qty"] <= 0) | clean["order_date"].isna()
# 规则5：跨表冲突（同单号多来源且金额不一致）
g = clean.groupby("order_id").agg(n_src=("__src","nunique"), n_amt=("amount","nunique"))
conf = g[(g["n_src"]>1) & (g["n_amt"]>1)]

clean[~(miss | inv | clean["order_id"].isin(conf.index))].to_excel("merged_clean.xlsx")