AiDOPWarehouse/doc/plan/sql/uat-data/run_uat_t8_fact_seed.py

#!/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())