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

#!/usr/bin/env python3
"""Create tenant-scoped UAT sales shipment business facts.

Facts are written to the normal ASN business tables. A subsequent S1 rebuild
must pull them through mdp_stg_ship_trans -> mdp_std_ship_trans -> DWD -> KPI.
"""

from __future__ import annotations

import argparse
import json
from collections import defaultdict
from datetime import datetime, timedelta
from decimal import Decimal
from typing import Any

from apply_sql_file import connect


TENANTS = {
    838257186181189: "A",
    838257212780613: "B",
    838257237606469: "DEMO",
}
PREFIX = "UAT-SHIP-"


def shipment_ratio(code: str, order_no: str, index: int) -> Decimal:
    if code == "A":
        scenario = next((x for x in ("A01", "A02", "A03", "A04", "A05") if x in order_no), "")
        return {
            "A01": Decimal("1.00"),
            "A02": Decimal("0.60"),
            "A03": Decimal("0.20"),
            "A04": Decimal("0.80"),
            "A05": Decimal("1.00"),
        }.get(scenario, Decimal("0.75"))
    if code == "B":
        return Decimal("0.65")
    return (Decimal("1.00"), Decimal("0.82"), Decimal("0.55"), Decimal("0.25"))[index % 4]


def load_order_lines(cursor: Any) -> dict[int, list[dict[str, Any]]]:
    tenant_ids = ",".join(str(x) for x in TENANTS)
    cursor.execute(
        f"""
        SELECT tenant_id,order_no,order_line,item_code,item_name,order_qty
        FROM mdp_std_so
        WHERE tenant_id IN ({tenant_ids})
          AND IFNULL(order_no,'')<>''
          AND IFNULL(item_code,'')<>''
          AND IFNULL(order_qty,0)>0
        ORDER BY tenant_id,order_no,CAST(order_line AS UNSIGNED),item_code
        """
    )
    result: dict[int, list[dict[str, Any]]] = defaultdict(list)
    for row in cursor.fetchall():
        result[int(row["tenant_id"])].append(row)
    return result


def apply() -> dict[str, Any]:
    conn = connect()
    summary: dict[str, Any] = {"masters": 0, "details": 0, "tenants": {}}
    try:
        conn.begin()
        with conn.cursor() as cursor:
            lines_by_tenant = load_order_lines(cursor)
            tenant_ids = tuple(TENANTS)
            placeholders = ",".join(["%s"] * len(tenant_ids))
            cursor.execute(
                f"DELETE FROM ASNBOLShipperDetail WHERE tenant_id IN ({placeholders}) AND Id LIKE %s",
                (*tenant_ids, f"{PREFIX}%"),
            )
            cursor.execute(
                f"DELETE FROM ASNBOLShipperMaster WHERE tenant_id IN ({placeholders}) AND Id LIKE %s",
                (*tenant_ids, f"{PREFIX}%"),
            )

            now = datetime.now().replace(microsecond=0)
            for tenant_id, code in TENANTS.items():
                grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
                for row in lines_by_tenant.get(tenant_id, []):
                    grouped[str(row["order_no"])].append(row)

                tenant_summary = {"orders": 0, "lines": 0}
                for order_index, (order_no, order_lines) in enumerate(grouped.items()):
                    shipper_id = f"{PREFIX}{code}-{order_index + 1:03d}"
                    if code == "DEMO":
                        ship_date = now - timedelta(days=(order_index * 2) % 90)
                    else:
                        ship_date = now - timedelta(days=order_index % 10)
                    ratio = shipment_ratio(code, order_no, order_index)
                    total_qty = sum(Decimal(str(row["order_qty"])) for row in order_lines)
                    real_total = (total_qty * ratio).quantize(Decimal("0.0001"))

                    cursor.execute(
                        """
                        INSERT INTO ASNBOLShipperMaster
                          (Id,OrdNbr,QtyToShip,QtyShipped,ShipDate,Status,IsConfirm,
                           SoldTo,Remark,CreateTime,UpdateTime,IsActive,tenant_id)
                        VALUES
                          (%s,%s,%s,%s,%s,'SHIPPED',1,'UAT-CUSTOMER',
                           'UAT_GENERATOR：销售发运事实',NOW(),NOW(),1,%s)
                        """,
                        (shipper_id, order_no, total_qty, real_total, ship_date, tenant_id),
                    )
                    master_rec_id = cursor.lastrowid
                    summary["masters"] += 1
                    tenant_summary["orders"] += 1

                    for line_index, row in enumerate(order_lines, 1):
                        qty = Decimal(str(row["order_qty"]))
                        real_qty = (qty * ratio).quantize(Decimal("0.0001"))
                        cursor.execute(
                            """
                            INSERT INTO ASNBOLShipperDetail
                              (ASNBOLShipperRecID,Id,Line,OrdNbr,OrdLine,ContainerItem,
                               Descr,QtyToShip,PickingQty,RealQty,QtyShipped,ShipDate,
                               Status,IsConfirm,Remark,CreateTime,UpdateTime,IsActive,tenant_id)
                            VALUES
                              (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,
                               'SHIPPED',1,'UAT_GENERATOR：销售发运明细',NOW(),NOW(),1,%s)
                            """,
                            (
                                master_rec_id,
                                f"{shipper_id}-L{line_index:03d}",
                                line_index,
                                order_no,
                                int(row["order_line"] or line_index),
                                row["item_code"],
                                row["item_name"] or row["item_code"],
                                qty,
                                real_qty,
                                real_qty,
                                real_qty,
                                ship_date,
                                tenant_id,
                            ),
                        )
                        summary["details"] += 1
                        tenant_summary["lines"] += 1
                summary["tenants"][code] = tenant_summary
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()
    return summary


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--apply", action="store_true")
    args = parser.parse_args()
    if not args.apply:
        print(json.dumps({"mode": "dry-run", "tenants": TENANTS, "prefix": PREFIX}, ensure_ascii=False, indent=2))
        return 0
    result = apply()
    result["mode"] = "apply"
    print(json.dumps(result, ensure_ascii=False, indent=2, default=str))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())