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.

Default window is the **current calendar month** (unique ``YYYYMM`` row ids) so
an August run writes August facts without touching July keys, and a September
run can seed September the same way.
"""

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 current_month_window(today: date) -> tuple[date, date]:
    start = today.replace(day=1)
    if start.month == 12:
        nxt = start.replace(year=start.year + 1, month=1, day=1)
    else:
        nxt = start.replace(month=start.month + 1, day=1)
    return start, nxt - timedelta(days=1)


def clamp_day(day: date, month_end: date) -> date:
    return day if day <= month_end else month_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 seed_staff(
    rows: list[tuple[str, int, str, str, dict[str, Any]]],
    prefix: str,
    tenant_id: int,
    id_base: int,
    warehouse_staff: int,
    production_staff: int,
) -> None:
    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": "计划",
        })


def seed_month(
    rows: list[tuple[str, int, str, str, dict[str, Any]]],
    *,
    code: str,
    prefix: str,
    tenant_id: int,
    id_base: int,
    order_count: int,
    on_time_ratio: float,
    month_start: date,
    month_end: date,
) -> None:
    month_tag = month_start.strftime("%Y%m")
    month_shift = ((month_start.year * 12 + month_start.month) % 100) * 500_000
    span = max(month_end.day - 8, 1)

    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-{month_tag}-{seq:03d}"
        task_no = f"{work_order}-TASK"
        sales_order = f"{prefix}-SO-{month_tag}-{seq:03d}"
        sales_line = f"{sales_order}-L1"
        qty = 80 + (seq % 6) * 20

        offset = min((seq * 2) % span, max(month_end.day - 9, 0))
        receipt_day = clamp_day(month_start + timedelta(days=offset), month_end)
        issue_day = clamp_day(receipt_day + timedelta(days=2), month_end)
        start_day = clamp_day(issue_day + timedelta(days=1), month_end)
        plan_finish = clamp_day(start_day + timedelta(days=4), month_end)
        is_on_time = (seq / order_count) <= on_time_ratio
        finish_day = clamp_day(
            plan_finish - timedelta(days=1) if is_on_time else plan_finish + timedelta(days=2),
            month_end,
        )
        fqc_day = finish_day
        ship_day = clamp_day(finish_day + timedelta(days=1), month_end)
        return_day = clamp_day(ship_day + timedelta(days=2), month_end)

        prod_head_id = id_base + month_shift + 10_000 + seq
        prod_line_id = id_base + month_shift + 20_000 + seq
        sales_head_id = id_base + month_shift + 30_000 + seq
        sales_line_id = id_base + month_shift + 40_000 + seq

        stage_row(rows, "dd_head", tenant_id, f"{prefix}-DDH-PROD-{month_tag}-{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-{month_tag}-{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-{month_tag}-{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-{month_tag}-{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-{month_tag}-{seq:03d}", {
            "Id": id_base + month_shift + 50_000 + seq,
            "idid": prod_head_id,
        })
        stage_row(rows, "bg_head", tenant_id, f"{prefix}-BG-{month_tag}-{seq:03d}", {
            "Id": id_base + month_shift + 60_000 + seq,
            "noid": work_order,
            "kgdate": dt(start_day, 8),
            "ztid": ZTID,
        })
        stage_row(rows, "zj_list", tenant_id, f"{prefix}-FQC-{month_tag}-{seq:03d}", {
            "Id": id_base + month_shift + 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]
        )
        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 + month_shift + 100_000 + seq * 10 + tx_index
            tx_key = f"{prefix}-TZ-{month_tag}-{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 + month_shift + 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),
            })


def build_rows(today: date, tenant_codes: set[str] | None = None) -> list[tuple[str, int, str, str, dict[str, Any]]]:
    month_start, month_end = current_month_window(today)
    rows: list[tuple[str, int, str, str, dict[str, Any]]] = []
    selected = [
        item for item in TENANTS
        if not tenant_codes or item[0] in tenant_codes
    ]

    for code, tenant_id, order_count, warehouse_staff, production_staff, on_time_ratio in selected:
        prefix = f"UAT-T8-{code}"
        id_base = tenant_id % 1_000_000_000 * 100_000
        seed_staff(rows, prefix, tenant_id, id_base, warehouse_staff, production_staff)
        seed_month(
            rows,
            code=code,
            prefix=prefix,
            tenant_id=tenant_id,
            id_base=id_base,
            order_count=order_count,
            on_time_ratio=on_time_ratio,
            month_start=month_start,
            month_end=month_end,
        )
    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")
    parser.add_argument("--as-of", help="YYYY-MM-DD; default today. Controls which calendar month is seeded.")
    parser.add_argument("--tenants", default="A", help="comma-separated tenant codes: A,B,DEMO")
    args = parser.parse_args()
    as_of = date.fromisoformat(args.as_of) if args.as_of else date.today()
    tenant_codes = {x.strip().upper() for x in args.tenants.split(",") if x.strip()}
    rows = build_rows(as_of, tenant_codes)
    month_start, month_end = current_month_window(as_of)
    summary: dict[str, Any] = {
        "sourceSystem": SOURCE_SYSTEM,
        "asOf": as_of.isoformat(),
        "window": f"{month_start.isoformat()}~{month_end.isoformat()}",
        "tenantCodes": sorted(tenant_codes),
        "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())