#!/usr/bin/env python3 """ 一键全量导入 + 物化视图刷新 用法: python3 run_all.py --month 2026-04-01 --db bill_query_test python3 run_all.py --month 2026-04-01 --db bill_query_test --skip-import # 仅刷新物化 python3 run_all.py --month 2026-04-01 --db bill_query_test --skip-views # 仅导入 python3 run_all.py --month 2026-04-01 --db bill_query_test --only bill_records # 仅导入某类 """ import argparse import os import subprocess import sys import time from config import DB_CONFIG, get_data_sources, PIPELINE_DIR def run_script(name, cmd, cwd=None): """运行子进程并实时输出""" print(f"\n{'='*60}") print(f" 执行: {name}") print(f" 命令: {' '.join(cmd)}") print(f"{'='*60}") t0 = time.time() result = subprocess.run(cmd, cwd=cwd or PIPELINE_DIR) elapsed = time.time() - t0 status = "成功" if result.returncode == 0 else f"失败(rc={result.returncode})" print(f" → {name}: {status} ({elapsed:.1f}s)") return result.returncode == 0 def main(): parser = argparse.ArgumentParser(description='全量数据导入与物化刷新流水线') parser.add_argument('--month', required=True, help='报告月份 (YYYY-MM-01)') parser.add_argument('--db', default='bill_query_test', help='目标数据库') parser.add_argument('--skip-import', action='store_true', help='跳过数据导入，仅刷新物化视图') parser.add_argument('--skip-views', action='store_true', help='跳过物化视图刷新') parser.add_argument('--only', default=None, help='仅导入指定数据集: bill_records, monthly, salary, store_location, derived') parser.add_argument('--diagnosis-date', default=None, help='诊断日期 (默认今天)') args = parser.parse_args() sources = get_data_sources(args.month) if not sources: print(f"错误: 无法找到月份 {args.month} 的数据源") sys.exit(1) print(f"\n{'#'*60}") print(f" 数据导入与物化刷新流水线") print(f" 月份: {args.month}") print(f" 数据库: {args.db}") print(f" 数据源根: {sources.get('bill_records', 'N/A')}") print(f"{'#'*60}") py = sys.executable results = [] # ============================================================ # Step 1: 数据导入 # ============================================================ if not args.skip_import: target = args.only # 1.1 薪资+考勤 if target in (None, 'salary'): cmd = [py, 'import_salary_attendance.py', '--month', args.month, '--db', args.db] sal_file = sources.get('salary') att_file = sources.get('attendance') if sal_file and os.path.isfile(sal_file): cmd.extend(['--salary-file', sal_file]) if att_file and os.path.isfile(att_file): cmd.extend(['--attendance-file', att_file]) ok = run_script("薪资+考勤导入", cmd) results.append(('薪资+考勤', ok)) # 1.2 账单查询原始数据（bill_records + bill_columns） if target in (None, 'bill_records'): bill_dir = sources.get('bill_records') if bill_dir and os.path.exists(bill_dir): ok = run_script("账单查询导入(bill_records)", [ py, 'import_bill_records.py', '--month', args.month, '--file', bill_dir, '--db', args.db, ]) results.append(('bill_records', ok)) else: print(f"\n 跳过 bill_records: 路径不存在 {bill_dir}") results.append(('bill_records', False)) # 1.3 月度业务数据（库存/销售/成本/费用/中央厨房/配送） if target in (None, 'monthly'): monthly_datasets = [ ('inventory', 'inventory', None), ('dish_sales', 'dish_sales', None), ('dish_cost', 'dish_cost', None), ('operating_expense', 'operating_expense', None), ('central_kitchen_finished', 'central_kitchen', 'finished'), ('central_kitchen_recipe', 'central_kitchen', 'recipe'), ('central_kitchen_material', 'central_kitchen', 'material'), ('central_kitchen_processing', 'central_kitchen', 'processing'), ('distribution', 'distribution', None), ('distribution_2', 'distribution', None), ] for src_key, dataset, ck_type in monthly_datasets: filepath = sources.get(src_key) if not filepath or not os.path.exists(filepath): print(f"\n 跳过 {dataset}/{ck_type or ''}: 文件不存在 {filepath}") results.append((f'{dataset}/{ck_type or src_key}', False)) continue cmd = [ py, 'import_monthly_data.py', '--dataset', dataset, '--month', args.month, '--file', filepath, '--db', args.db, ] if ck_type: cmd.extend(['--ck-type', ck_type]) label = f"月度数据-{dataset}" + (f"-{ck_type}" if ck_type else f"({os.path.basename(filepath)})") ok = run_script(label, cmd) results.append((label, ok)) # 1.4 门店位置与映射表 if target in (None, 'store_location'): store_file = sources.get('store_location') if store_file and os.path.exists(store_file): ok = run_script("门店位置与映射表导入", [ py, 'import_store_location.py', '--file', store_file, '--db', args.db, ]) results.append(('门店位置', ok)) else: print(f"\n 跳过门店位置: 文件不存在 {store_file}") results.append(('门店位置', False)) # 1.5 派生分析数据 if target in (None, 'derived'): diag_date = args.diagnosis_date or '' cmd = [ py, 'import_derived_data.py', '--month', args.month, '--db', args.db, ] if diag_date: cmd.extend(['--diagnosis-date', diag_date]) ok = run_script("派生分析数据生成", cmd) results.append(('派生数据', ok)) # ============================================================ # Step 2: 物化视图刷新 # ============================================================ if not args.skip_views: ok = run_script("物化视图刷新", [ py, 'refresh_materialized_views.py', '--db', args.db, ], cwd=PIPELINE_DIR) results.append(('物化视图', ok)) # ============================================================ # Step 3: 验证 # ============================================================ ok = run_script("数据一致性验证", [ py, 'verify.py', '--db', args.db, ], cwd=PIPELINE_DIR) results.append(('验证', ok)) # ============================================================ # 汇总 # ============================================================ print(f"\n{'#'*60}") print(f" 流水线执行汇总") print(f"{'#'*60}") all_ok = True for name, ok in results: status = "✓ 成功" if ok else "✗ 失败" print(f" {status} {name}") if not ok: all_ok = False if all_ok: print(f"\n 全部完成!") sys.exit(0) else: print(f"\n 部分失败，请检查上方日志") sys.exit(1) if __name__ == '__main__': main()