# -*- coding: utf-8 -*- """ 门店销售表合并工具 用途：把各门店交上来的表（CSV 或 Excel 都行，列名还不统一）合并成一张干净的汇总表， 顺手把重复行、异常值、缺字段挑出来，最后出一个能直接看的汇总 Excel。 用法：python merge_reports.py --input ./input --output ./输出/汇总.xlsx """ import argparse import csv import os import re from datetime import datetime from openpyxl import Workbook, load_workbook from openpyxl.chart import BarChart, Reference from openpyxl.styles import Alignment, Border, Font, PatternFill, Side # 列名别名表：左边是标准名，右边是各门店可能写出来的写法 ALIAS = { "月份": ["月份", "月", "日期", "date", "month"], "门店": ["门店", "店", "店铺", "store", "shop"], "品类": ["品类", "分类", "类目", "category"], "销售额": ["销售额", "金额", "销售金额", "amount", "sales"], "订单数": ["订单数", "单量", "订单量", "orders", "qty"], } FIELDS = ["月份", "门店", "品类", "销售额", "订单数"] HEAD = PatternFill("solid", fgColor="1F3864") WHITEB = Font(bold=True, color="FFFFFF") THIN = Side(style="thin", color="AAAAAA") BOX = Border(left=THIN, right=THIN, top=THIN, bottom=THIN) def norm_month(v): """把 2026-04、2026/4、2026年4月 这类写法统一成 2026-04""" s = str(v).strip() m = re.match(r"(\d{4})\D+(\d{1,2})", s) if not m: return s return "%s-%02d" % (m.group(1), int(m.group(2))) def read_any(path): """CSV 和 xlsx 都能读，统一返回 [dict]""" rows = [] if path.lower().endswith((".xlsx", ".xlsm")): wb = load_workbook(path, data_only=True) ws = wb.active data = list(ws.iter_rows(values_only=True)) else: with open(path, encoding="utf-8-sig", newline="") as f: data = [tuple(r) for r in csv.reader(f)] if not data: return rows, [] head = [str(c).strip() if c is not None else "" for c in data[0]] # 建立 标准名 -> 实际列号 的映射 mapping = {} for std, names in ALIAS.items(): for i, h in enumerate(head): if h and h.lower() in [n.lower() for n in names]: mapping[std] = i break missing = [f for f in FIELDS if f not in mapping] for raw in data[1:]: if not any(x not in (None, "") for x in raw): continue item = {"__src": os.path.basename(path)} for f in FIELDS: if f in mapping and mapping[f] < len(raw): item[f] = raw[mapping[f]] else: item[f] = "" rows.append(item) return rows, missing def clean(rows, issues): out = [] for r in rows: r["月份"] = norm_month(r["月份"]) for k in ("门店", "品类"): r[k] = str(r[k]).strip() if r[k] is not None else "" try: r["销售额"] = float(str(r["销售额"]).replace(",", "").replace("¥", "").strip() or 0) except ValueError: issues.append(["销售额不是数字", r["__src"], r["月份"], str(r["销售额"])]) r["销售额"] = 0.0 try: r["订单数"] = int(float(str(r["订单数"]).strip() or 0)) except ValueError: issues.append(["订单数不是数字", r["__src"], r["月份"], str(r["订单数"])]) r["订单数"] = 0 if r["销售额"] < 0: issues.append(["销售额为负数", r["__src"], r["月份"], r["销售额"]]) if not r["品类"]: issues.append(["品类为空", r["__src"], r["月份"], ""]) out.append(r) seen, deduped = set(), [] for r in out: key = (r["月份"], r["门店"], r["品类"]) if key in seen: issues.append(["重复行已删除", r["__src"], r["月份"], "%s %s" % (r["门店"], r["品类"])]) continue seen.add(key) deduped.append(r) return deduped def summarize(rows): """按月份和门店两个维度汇总""" by_month, by_store = {}, {} for r in rows: m = by_month.setdefault(r["月份"], [0.0, 0]) m[0] += r["销售额"] m[1] += r["订单数"] s = by_store.setdefault(r["门店"], [0.0, 0]) s[0] += r["销售额"] s[1] += r["订单数"] return by_month, by_store def write_xlsx(path, rows, issues, by_month, by_store): os.makedirs(os.path.dirname(path), exist_ok=True) wb = Workbook() ws = wb.active ws.title = "汇总明细" ws.append(FIELDS + ["来源文件"]) for c in ws[1]: c.fill, c.font, c.border = HEAD, WHITEB, BOX for r in sorted(rows, key=lambda x: (x["月份"], x["门店"], x["品类"])): ws.append([r[f] for f in FIELDS] + [r["__src"]]) for col, w in zip("ABCDEF", (12, 12, 10, 12, 10, 26)): ws.column_dimensions[col].width = w ws.freeze_panes = "A2" ws2 = wb.create_sheet("月度汇总") ws2.append(["月份", "销售额", "订单数", "客单价"]) for c in ws2[1]: c.fill, c.font, c.border = HEAD, WHITEB, BOX for m in sorted(by_month): v, o = by_month[m] ws2.append([m, round(v, 2), o, round(v / o, 2) if o else 0]) ws2.column_dimensions["A"].width = 12 for col in "BCD": ws2.column_dimensions[col].width = 12 if ws2.max_row > 1: ch = BarChart() ch.title = "月度销售额" ch.height, ch.width = 8, 16 ch.add_data(Reference(ws2, min_col=2, min_row=1, max_row=ws2.max_row), titles_from_data=True) ch.set_categories(Reference(ws2, min_col=1, min_row=2, max_row=ws2.max_row)) ws2.add_chart(ch, "F2") ws3 = wb.create_sheet("门店汇总") ws3.append(["门店", "销售额", "订单数", "客单价"]) for c in ws3[1]: c.fill, c.font, c.border = HEAD, WHITEB, BOX for s in sorted(by_store, key=lambda k: -by_store[k][0]): v, o = by_store[s] ws3.append([s, round(v, 2), o, round(v / o, 2) if o else 0]) for col, w in zip("ABCD", (14, 12, 10, 12)): ws3.column_dimensions[col].width = w ws4 = wb.create_sheet("数据质量") ws4.append(["问题类型", "来源文件", "月份", "原值"]) for c in ws4[1]: c.fill, c.font, c.border = HEAD, WHITEB, BOX for row in issues: ws4.append(row) for col, w in zip("ABCD", (18, 26, 12, 20)): ws4.column_dimensions[col].width = w if ws4.max_row == 1: ws4.append(["无问题", "", "", ""]) wb.save(path) def main(): ap = argparse.ArgumentParser() ap.add_argument("--input", default="./input") ap.add_argument("--output", default="./输出/汇总.xlsx") args = ap.parse_args() files = sorted(f for f in os.listdir(args.input) if f.lower().endswith((".csv", ".xlsx", ".xlsm"))) if not files: print("输入目录里没有找到 CSV 或 Excel 文件") return all_rows, issues, missing_report = [], [], [] for f in files: rows, missing = read_any(os.path.join(args.input, f)) if missing: missing_report.append((f, missing)) all_rows.extend(rows) print("读入 %-28s %3d 行" % (f, len(rows))) cleaned = clean(all_rows, issues) by_month, by_store = summarize(cleaned) write_xlsx(args.output, cleaned, issues, by_month, by_store) total = sum(r["销售额"] for r in cleaned) orders = sum(r["订单数"] for r in cleaned) print("-" * 46) print("合并后 %d 行，总销售额 %.2f，总订单 %d" % (len(cleaned), total, orders)) print("区间 %s 至 %s" % (min(by_month), max(by_month))) print("发现问题 %d 条" % len(issues)) for t, src, m, v in issues: print(" - %s（%s %s %s）" % (t, src, m, v)) if missing_report: for f, miss in missing_report: print(" ! %s 缺少列：%s" % (f, "、".join(miss))) print("已写出：" + os.path.abspath(args.output)) if __name__ == "__main__": main()