#!/usr/bin/env python3 """计算 (店, 部门, 班次) 三维交叉营收,并按比例校正到权威 部门业绩 合计。 输入: 月度账务目录(包含订单明细 xlsx + 菜品库 xlsx + 营收分析 xlsx) 输出: 标准输出打印交叉矩阵 + 12 名员工的 (部门业绩 / 班次业绩 / 部门×班次业绩) 三元组 用法: python3 compute_cross.py <月度账务目录> # 例: python3 compute_cross.py ~/Downloads/大梦5月账务处理 依赖: openpyxl """ import glob import sys from collections import defaultdict try: import openpyxl except ImportError: sys.exit("缺少依赖: python3 -m pip install openpyxl") # ============================================================ # 部门归口规则(来自 分析方法.md,已根据实际菜品库一级分类核对) # ============================================================ def categorize(primary, secondary, store): p = (primary or "").lower().strip() s = (secondary or "").strip() if store == "西湖": if p in ["小吃", "主食", "brunch", "零食"]: return "厨房" if p in ["咖啡", "甜品", "茶饮tea"]: return "咖啡" if p == "软饮": return "咖啡" if s in ["可尔必思", "海盐荔枝"] else "调酒" if p in ["精酿", "精酿 老菜单"]: return "精酿" if p in ["鸡尾酒", "纯饮"]: return "调酒" elif store == "滨江": if p in ["肉肉肉", "小吃", "主食", "brunch"]: return "厨房" if p in ["咖啡", "甜品点心", "茶饮tea"]: return "咖啡" if p == "无咖无醇": return "调酒" if s == "无醇鸡尾酒" else "咖啡" if p in ["精酿", "瓶罐精酿", "精酿 老菜单(已废弃)"]: return "精酿" if p in ["鸡尾酒", "纯饮酒"]: return "调酒" return None # ============================================================ # 计算每店 (餐段, 部门) 营收 # ============================================================ def load_store(base_dir, store_cn): """计算指定店 (餐段, 部门) → 顾客实付 (POS 菜品收入) 矩阵""" print(f"\n===== {store_cn}店 =====", file=sys.stderr) # 1) 订单 → 餐段 order_files = glob.glob(f"{base_dir}/大梦可能实验室({store_cn}店)_全渠道订单明细_*.xlsx") if not order_files: sys.exit(f"找不到 {store_cn}店 全渠道订单明细 xlsx") wb1 = openpyxl.load_workbook(order_files[0], data_only=True) ws1 = wb1.active order_shift = {} header_row = None for i, r in enumerate(ws1.iter_rows(values_only=True), start=1): if r and r[0] == "营业日期": header_row = i cols = list(r) col_segment = cols.index("餐段") col_order = cols.index("订单号") continue if header_row and i > header_row and r[col_order]: order_shift[str(r[col_order])] = str(r[col_segment]) if r[col_segment] else None print(f" Loaded {len(order_shift)} 订单", file=sys.stderr) # 2) 菜品名 → 一级/二级分类 menu_files = glob.glob(f"{base_dir}/大梦_可能实验室_{store_cn}店_菜品库_*.xlsx") if not menu_files: sys.exit(f"找不到 {store_cn}店 菜品库 xlsx") wb2 = openpyxl.load_workbook(menu_files[0], data_only=True) ws2 = wb2["菜品"] name_to_cat = {} header_row2 = None for i, r in enumerate(ws2.iter_rows(values_only=True), start=1): if r and r[0] == "菜品编码(SPUID)": header_row2 = i cols = list(r) ci_name = cols.index("菜品名称") ci_cat = cols.index("基础分类") continue if header_row2 and i > header_row2 and r[ci_name] and r[ci_cat]: parts = str(r[ci_cat]).split("/") primary = parts[0] secondary = parts[1] if len(parts) > 1 else "" name_to_cat[str(r[ci_name]).strip()] = (primary, secondary) print(f" Loaded {len(name_to_cat)} 菜品", file=sys.stderr) # 3) 菜品明细 → 聚合 detail_files = glob.glob(f"{base_dir}/大梦_可能实验室_{store_cn}店__店内订单明细*.xlsx") if not detail_files: sys.exit(f"找不到 {store_cn}店 店内订单明细 xlsx") wb3 = openpyxl.load_workbook(detail_files[0], data_only=True) ws3 = wb3["菜品明细"] bucket = defaultdict(float) # (shift, dept) -> revenue unmatched_count = 0 header_row3 = None for i, r in enumerate(ws3.iter_rows(values_only=True), start=1): if r and r[0] == "订单编号": header_row3 = i cols = list(r) ci_order = cols.index("订单编号") ci_revenue = cols.index("菜品收入(元)") ci_name = cols.index("菜品名称") continue if header_row3 and i > header_row3: order = str(r[ci_order]) if r[ci_order] else None name = str(r[ci_name]).strip() if r[ci_name] else None revenue = r[ci_revenue] if not order or revenue is None or not name: continue shift = order_shift.get(order) cat = name_to_cat.get(name) if not shift or not cat: unmatched_count += 1 continue dept = categorize(cat[0], cat[1], store_cn) if not dept: unmatched_count += 1 continue bucket[(shift, dept)] += float(revenue) print(f" unmatched: {unmatched_count}", file=sys.stderr) return bucket # ============================================================ # 读取权威 部门业绩 / 班次业绩(从月度营收分析 xlsx) # ============================================================ def read_official(base_dir): rev_files = glob.glob(f"{base_dir}/大梦可能实验室_*月营收分析_西湖店vs滨江店.xlsx") if not rev_files: sys.exit("找不到月度营收分析 xlsx") wb = openpyxl.load_workbook(rev_files[0], data_only=True) dept = {"西湖店": {}, "滨江店": {}} ws = wb["部门收入"] in_section = False for r in ws.iter_rows(values_only=True): if r and r[0] and "部门 |" in str(r[0]) or (r and r[0] == "部门"): in_section = True continue if in_section and r and r[0]: name = str(r[0]).strip() if name in ["厨房", "咖啡", "精酿", "调酒"]: # cols: 部门, 滨江_POS, 滨江_团购, 滨江_合计, 西湖_POS, 西湖_团购, 西湖_合计 dept["滨江店"][name] = float(r[3]) if r[3] else 0 dept["西湖店"][name] = float(r[6]) if r[6] else 0 else: if "部门收入小计" in name: break shift = {"西湖店": {}, "滨江店": {}} ws = wb["班次营收"] for r in ws.iter_rows(values_only=True): if r and r[0] in ["西湖店", "滨江店"] and r[1] in ["白班", "晚班"]: # cols: 门店, 餐段, 订单数, 订单金额, 顾客实付 shift[str(r[0])][str(r[1])] = float(r[4]) if r[4] else 0 return dept, shift # ============================================================ # 12 员工 (店, 部门, 班次) 配置(行号约定) # ============================================================ EMPLOYEES = [ (14, "蔡逸丰", "西湖店", "精酿", "晚班"), (15, "何简", "西湖店", "厨房", "白班"), (16, "宋群喜", "西湖店", "咖啡", "白班"), (17, "胡舒", "西湖店", "调酒", "晚班"), (18, "郭思儒", "西湖店", "咖啡", "白班"), (19, "秦天", "西湖店", "厨房", "晚班"), (20, "李想", "滨江店", "调酒", "晚班"), (21, "王瑛胤", "滨江店", "咖啡", "白班"), (22, "刘润祥", "滨江店", "厨房", "晚班"), # 不算班次 (23, "朱秋风", "滨江店", "精酿", "晚班"), (24, "叶磊", "滨江店", "厨房", "白班"), # 不算班次 (25, "尹志艳", "滨江店", "厨房", "中班"), # 不算班次/中班无班次业绩 ] # 滨江厨房团队:只算 部门业绩,班次 + 交叉 都为 0 BINJIANG_KITCHEN = {"刘润祥", "叶磊", "尹志艳"} def main(): if len(sys.argv) < 2: sys.exit("用法: python3 compute_cross.py <月度账务目录>") base = sys.argv[1].rstrip("/") # 计算 raw cross-tab raw_xihu = load_store(base, "西湖") raw_binjiang = load_store(base, "滨江") raw = {"西湖店": raw_xihu, "滨江店": raw_binjiang} # 读权威 部门 / 班次 dept_official, shift_official = read_official(base) # 按部门比例校正:scale factor = 权威总 / raw 部门小计 scaled = {} for store in ["西湖店", "滨江店"]: for d in ["厨房", "咖啡", "精酿", "调酒"]: raw_dept_sum = sum(raw[store].get((sh, d), 0) for sh in ["白班", "晚班"]) if raw_dept_sum > 0 and d in dept_official[store]: factor = dept_official[store][d] / raw_dept_sum for sh in ["白班", "晚班"]: scaled[(store, d, sh)] = round(raw[store].get((sh, d), 0) * factor) # 打印交叉表 print("\n=== 校正后 部门×班次(用作 V2 col 21 部门x班次业绩)===") for store in ["西湖店", "滨江店"]: print(f"\n{store}:") print(f" {'部门':<6}{'白班':>10}{'晚班':>10}") for d in ["厨房", "咖啡", "精酿", "调酒"]: wb = scaled.get((store, d, "白班"), 0) nb = scaled.get((store, d, "晚班"), 0) print(f" {d:<6}{wb:>10}{nb:>10}") # 输出 12 员工三元组 print("\n=== 12 员工 (部门业绩 / 班次业绩 / 部门×班次业绩) ===") print(f"{'行':>3} {'姓名':<6} {'店':<5} {'部门':<5} {'班次':<5} {'部门业绩':>10} {'班次业绩':>10} {'部门×班次':>10}") for row, name, store, dept, shift in EMPLOYEES: is_dept = dept in dept_official[store] dr = dept_official[store].get(dept, 0) sr = shift_official[store].get(shift, 0) cr = scaled.get((store, dept, shift), 0) # 特殊规则:滨江厨房团队不算班次 if name in BINJIANG_KITCHEN: sr = 0 cr = 0 # 中班无班次业绩 if shift == "中班": sr = 0 cr = 0 # 前厅、空部门 if not is_dept: dr = 0 cr = 0 print(f"{row:>3} {name:<6} {store:<5} {dept:<5} {shift:<5} {dr:>10.2f} {sr:>10.2f} {cr:>10}") if __name__ == "__main__": main()