#!/usr/bin/env python3 """Seed tenant-scoped UAT T8 facts into the MDP staging layer. The script intentionally stops at ``mdp_stg_t8_*``. S5/S6/S7 module rebuilds must perform STG -> STD -> DWD -> KPI conversion afterwards. """ from __future__ import annotations import argparse import json from datetime import date, datetime, time, timedelta from typing import Any from apply_sql_file import connect TENANTS = ( ("A", 838257186181189, 10, 3, 8, 0.90), ("B", 838257212780613, 6, 2, 5, 0.75), ("DEMO", 838257237606469, 30, 5, 16, 0.83), ) SOURCE_SYSTEM = "UAT_GENERATOR" ZTID = "pbxfxp" TABLES = { "tz_head": "mdp_stg_t8_kc_tz_head", "tz_list": "mdp_stg_t8_kc_tz_list", "dd_head": "mdp_stg_t8_kc_dd_head", "dd_list": "mdp_stg_t8_kc_dd_list", "zj_list": "mdp_stg_t8_kc_zj_list", "pelist": "mdp_stg_t8_sys_pelist", "bg_head": "mdp_stg_t8_cj_bg_head_rep", "cllist": "mdp_stg_t8_kc_dd_list_cllist", } def previous_month_window(today: date) -> tuple[date, date]: month_start = today.replace(day=1) end = month_start - timedelta(days=1) return end.replace(day=1), end def dt(day: date, hour: int = 8) -> str: return datetime.combine(day, time(hour=hour)).isoformat(timespec="seconds") def stage_row( rows: list[tuple[str, int, str, str, dict[str, Any]]], table_key: str, tenant_id: int, row_id: str, raw: dict[str, Any], ) -> None: rows.append((TABLES[table_key], tenant_id, row_id, row_id, raw)) def build_rows(today: date) -> list[tuple[str, int, str, str, dict[str, Any]]]: month_start, month_end = previous_month_window(today) rows: list[tuple[str, int, str, str, dict[str, Any]]] = [] for code, tenant_id, order_count, warehouse_staff, production_staff, on_time_ratio in TENANTS: prefix = f"UAT-T8-{code}" id_base = tenant_id % 1_000_000_000 * 100_000 for index in range(warehouse_staff): rid = f"{prefix}-PE-WH-{index + 1:03d}" stage_row(rows, "pelist", tenant_id, rid, { "Id": id_base + 1_000 + index, "ztid": ZTID, "zzzt": "在职", "gw": "仓管", }) for index in range(production_staff): rid = f"{prefix}-PE-PROD-{index + 1:03d}" stage_row(rows, "pelist", tenant_id, rid, { "Id": id_base + 2_000 + index, "ztid": ZTID, "zzzt": "在职", "gw": "生产", }) for index in range(2): rid = f"{prefix}-PE-PLAN-{index + 1:03d}" stage_row(rows, "pelist", tenant_id, rid, { "Id": id_base + 3_000 + index, "ztid": ZTID, "zzzt": "在职", "gw": "计划", }) for index in range(order_count): seq = index + 1 product = f"{prefix}-FG-{seq % 5 + 1:02d}" material = f"{prefix}-MAT-{seq % 8 + 1:02d}" work_order = f"{prefix}-WO-{seq:03d}" task_no = f"{work_order}-TASK" sales_order = f"{prefix}-SO-{seq:03d}" sales_line = f"{sales_order}-L1" qty = 80 + (seq % 6) * 20 # Spread the facts through the previous month so Demo has a visible story. offset = min((seq * 2) % max(month_end.day - 8, 1), month_end.day - 9) receipt_day = month_start + timedelta(days=offset) issue_day = receipt_day + timedelta(days=2) start_day = issue_day + timedelta(days=1) plan_finish = start_day + timedelta(days=4) is_on_time = (seq / order_count) <= on_time_ratio finish_day = plan_finish - timedelta(days=1) if is_on_time else plan_finish + timedelta(days=2) fqc_day = finish_day ship_day = finish_day + timedelta(days=1) prod_head_id = id_base + 10_000 + seq prod_line_id = id_base + 20_000 + seq sales_head_id = id_base + 30_000 + seq sales_line_id = id_base + 40_000 + seq stage_row(rows, "dd_head", tenant_id, f"{prefix}-DDH-PROD-{seq:03d}", { "Id": prod_head_id, "ztid": ZTID, "lbs": "生产任务", "zf": 0, "shyn": 1, "noid": work_order, }) stage_row(rows, "dd_list", tenant_id, f"{prefix}-DDL-PROD-{seq:03d}", { "Id": prod_line_id, "idid": prod_head_id, "rwnoid": task_no, "code": product, "sl": qty, "slzx": qty if is_on_time else round(qty * 0.72, 2), "jhdate": plan_finish.isoformat(), "gdyn": 1 if is_on_time else 0, "gdtime": dt(finish_day, 17), "addtime": dt(start_day, 8), }) stage_row(rows, "dd_head", tenant_id, f"{prefix}-DDH-SALES-{seq:03d}", { "Id": sales_head_id, "ztid": ZTID, "lbs": "销售订单", "zf": 0, "shyn": 1, "noid": sales_order, }) stage_row(rows, "dd_list", tenant_id, f"{prefix}-DDL-SALES-{seq:03d}", { "Id": sales_line_id, "idid": sales_head_id, "rwnoid": sales_line, "code": product, "sl": qty, "slzx": qty, "jhdate": (ship_day + timedelta(days=1 if is_on_time else -1)).isoformat(), "gdyn": 1, "gdtime": dt(ship_day, 17), "addtime": dt(start_day, 8), }) stage_row(rows, "cllist", tenant_id, f"{prefix}-CL-{seq:03d}", { "Id": id_base + 50_000 + seq, "idid": prod_head_id, }) stage_row(rows, "bg_head", tenant_id, f"{prefix}-BG-{seq:03d}", { "Id": id_base + 60_000 + seq, "noid": work_order, "kgdate": dt(start_day, 8), "ztid": ZTID, }) stage_row(rows, "zj_list", tenant_id, f"{prefix}-FQC-{seq:03d}", { "Id": id_base + 70_000 + seq, "ztid": ZTID, "lyid": sales_line_id, "zjyn": 1, "shdate": dt(fqc_day, 9), }) quality_return_ratio = ( 0.003 if code == "A" else 0.008 if code == "B" else (0.002, 0.006, 0.012)[(seq - 1) % 3] ) return_day = min(ship_day + timedelta(days=2), month_end) transactions = ( ("采购入库", material, "", receipt_day, qty * 2), ("生产领料", material, work_order, issue_day, qty * 1.6), ("生产入库", product, task_no, finish_day, qty if is_on_time else qty * 0.72), ("销售出库", product, sales_line, ship_day, qty), ("销售退货", product, sales_line, return_day, qty * quality_return_ratio), ) for tx_index, (tx_type, item, linkage, tx_day, actual_qty) in enumerate(transactions, 1): tx_head_id = id_base + 100_000 + seq * 10 + tx_index tx_key = f"{prefix}-TZ-{seq:03d}-{tx_index}" stage_row(rows, "tz_head", tenant_id, f"{tx_key}-H", { "Id": tx_head_id, "ztid": ZTID, "lbs": tx_type, "lynoid": work_order if tx_type == "生产领料" else "", "hzyn": 0, "zfyn": 0, "shyn": 1, "shtime": dt(tx_day, 16), "date0": tx_day.isoformat(), }) stage_row(rows, "tz_list", tenant_id, f"{tx_key}-L", { "Id": id_base + 200_000 + seq * 10 + tx_index, "idid": tx_head_id, "code": item, "lynoid": linkage, "slzx": round(actual_qty, 2), "sl": qty, "gdyn": 1 if actual_qty >= qty else 0, "gdtime": dt(tx_day, 16), "rwnoid": task_no, "jhdate": plan_finish.isoformat(), "addtime": dt(tx_day, 8), }) return rows def apply(rows: list[tuple[str, int, str, str, dict[str, Any]]]) -> dict[str, int]: batch_id = f"UAT_T8_FACT_{datetime.now():%Y%m%d%H%M%S}" counts: dict[str, int] = {} conn = connect() try: conn.begin() with conn.cursor() as cursor: for table, tenant_id, row_id, biz_key, raw in rows: cursor.execute( f""" INSERT INTO {table} (tenant_id,source_system,source_table,source_row_id,source_biz_key, raw_data,sync_batch_id,sync_time,process_status,process_message) VALUES (%s,%s,%s,%s,%s,%s,%s,NOW(),'PENDING',NULL) ON DUPLICATE KEY UPDATE tenant_id=VALUES(tenant_id), source_row_id=VALUES(source_row_id), raw_data=VALUES(raw_data), sync_batch_id=VALUES(sync_batch_id), sync_time=NOW(), process_status='PENDING', process_message=NULL, update_time=NOW() """, ( tenant_id, SOURCE_SYSTEM, table, row_id, biz_key, json.dumps(raw, ensure_ascii=False), batch_id, ), ) counts[table] = counts.get(table, 0) + 1 conn.commit() except Exception: conn.rollback() raise finally: conn.close() return counts def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--apply", action="store_true", help="write staging rows") args = parser.parse_args() rows = build_rows(date.today()) summary: dict[str, Any] = { "sourceSystem": SOURCE_SYSTEM, "tenantCount": len(TENANTS), "rowCount": len(rows), "tables": {}, "mode": "apply" if args.apply else "dry-run", } for table, *_ in rows: summary["tables"][table] = summary["tables"].get(table, 0) + 1 if args.apply: summary["applied"] = apply(rows) print(json.dumps(summary, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())