AiDOPWarehouse/doc/plan/旧DOP-MES-WMS/tools/_tmp_parse_wms_proc_io.py

# -*- coding: utf-8 -*-
"""Parse pr_WMS_/qms_WMS_/key SFM procs for table READ/WRITE against dopdemorq."""
from __future__ import annotations

import json
import re
from collections import Counter
from pathlib import Path

import pyodbc

OUT = Path(__file__).with_name("_tmp_old_wms_proc_io.json")

CONN = (
    "DRIVER={ODBC Driver 17 for SQL Server};"
    "SERVER=123.60.180.165,1433;DATABASE=dopdemorq;"
    "UID=dopsa;PWD=5h3n9)uN;TrustServerCertificate=yes"
)

INTEREST = {
    "WorkOrdMaster",
    "WorkOrdRouting",
    "WorkOrdDetail",
    "srm_polist_ds",
    "PurOrdMaster",
    "PurOrdDetail",
    "PurOrdDetailBatch",
    "PeriodSequenceDet",
    "NbrMaster",
    "NbrDetail",
    "MissedPrint",
    "ASNBOLShipperMaster",
    "ASNBOLShipperDetail",
    "InvMaster",
    "InvTransHist",
    "LocationDetail",
    "LocationMaster",
    "PurOrdRctMaster",
    "PurOrdRctDetail",
    "SalesOrdMaster",
    "SalesOrdDetail",
    "ItemMaster",
    "EmpWorkHist",
    "EmpWorkWmsHist",
    "WMSInvTransHist",
    "qms_qcp_inspbill",
    "InvTransHistDay",
    "ProductStructureMaster",
    "RoutingOpDetail",
}

SKIP = {
    "SET",
    "SELECT",
    "VALUES",
    "WHERE",
    "AND",
    "OR",
    "CASE",
    "WHEN",
    "THEN",
    "ELSE",
    "END",
    "BEGIN",
    "DECLARE",
    "IF",
    "WHILE",
    "RETURN",
    "EXEC",
    "EXECUTE",
    "PRINT",
    "WITH",
    "AS",
    "ON",
    "INNER",
    "LEFT",
    "RIGHT",
    "OUTER",
    "CROSS",
    "APPLY",
    "TOP",
    "DISTINCT",
    "GROUP",
    "ORDER",
    "BY",
    "HAVING",
    "UNION",
    "ALL",
    "EXISTS",
    "NOT",
    "NULL",
    "INTO",
    "FROM",
    "JOIN",
    "UPDATE",
    "DELETE",
    "INSERT",
    "TABLE",
    "TEMP",
    "OUTPUT",
    "INSERTED",
    "DELETED",
}

UPD_RE = re.compile(
    r"(?i)\bUPDATE\s+(?:\[?dbo\]?\.)?\[?([A-Za-z_][A-Za-z0-9_]*)\]?\s+SET"
)
INS_RE = re.compile(
    r"(?i)\bINSERT\s+(?:INTO\s+)?(?:\[?dbo\]?\.)?\[?([A-Za-z_][A-Za-z0-9_]*)\]?"
)
DEL_RE = re.compile(
    r"(?i)\bDELETE\s+(?:FROM\s+)?(?:\[?dbo\]?\.)?\[?([A-Za-z_][A-Za-z0-9_]*)\]?"
)
FROM_RE = re.compile(
    r"(?i)\b(?:FROM|JOIN)\s+(?:\[?dbo\]?\.)?\[?([A-Za-z_][A-Za-z0-9_]*)\]?"
)


def keep(t: str) -> bool:
    if t in INTEREST:
        return True
    if t.startswith(("qms_", "scm_", "srm_", "mes_")):
        return True
    if t[:1].isupper() and len(t) > 3:
        return True
    return False


def parse(text: str) -> tuple[set[str], set[str]]:
    writes: set[str] = set()
    reads: set[str] = set()
    for m in UPD_RE.finditer(text):
        t = m.group(1)
        if t.upper() not in SKIP and not t.startswith("#"):
            writes.add(t)
    for m in INS_RE.finditer(text):
        t = m.group(1)
        if t.upper() not in SKIP and not t.startswith("#"):
            writes.add(t)
    for m in DEL_RE.finditer(text):
        t = m.group(1)
        if t.upper() not in SKIP and not t.startswith("#"):
            writes.add(t)
    for m in FROM_RE.finditer(text):
        t = m.group(1)
        if t.upper() not in SKIP and not t.startswith("#"):
            reads.add(t)
    return reads, writes


def main() -> None:
    conn = pyodbc.connect(CONN, timeout=60)
    cur = conn.cursor()
    cur.execute(
        """
        SELECT name FROM sys.procedures
        WHERE name LIKE 'pr_WMS_Save%' OR name LIKE 'pr_WMS_BPM_%'
           OR name LIKE 'pr_WMS_Direct%' OR name LIKE 'pr_WMS_Cancel%'
           OR name LIKE 'pr_WMS_Update%' OR name LIKE 'pr_WMS_Auto%'
           OR name LIKE 'pr_WMS_Unplanned%'
           OR name LIKE 'qms_WMS_%'
           OR name IN (
             'pr_SFM_InventoryMesTransactionProcessingByOp',
             'pr_SFM_InventoryMesWorkOrdIsses',
             'pr_SFM_EndWorkOrdByOp',
             'pr_SFM_InventoryTransactionProcessing2',
             'pr_SFM_InventoryMesCartonTransactionProcessingByOp',
             'pr_MES_UpdateWorkOrder',
             'pr_MES_CloseWorkOrders',
             'pr_WMS_SaveWorkOrdMaterialConfirm',
             'pr_WMS_SaveWorkOrdMaterialIssue',
             'pr_WMS_SaveWorkOrdMaterialIssueByQty',
             'pr_WMS_SaveTWorkOrdMaterialReturn',
             'pr_WMS_SaveShipperConfim',
             'pr_WMS_SaveProductStorage',
             'pr_WMS_SaveFQCcheck',
             'pr_WMS_SaveFQCResult',
             'pr_WMS_SaveIQCResult',
             'pr_WMS_SaveIQCResultByRctNbr'
           )
        ORDER BY name
        """
    )
    procs = [r[0] for r in cur.fetchall()]
    results: dict = {}
    for name in procs:
        cur.execute("SELECT OBJECT_DEFINITION(OBJECT_ID(?))", name)
        row = cur.fetchone()
        text = row[0] if row and row[0] else ""
        if not text:
            results[name] = {
                "reads": [],
                "writes": [],
                "reads_core": [],
                "writes_core": [],
                "empty": True,
            }
            continue
        reads, writes = parse(text)
        results[name] = {
            "reads": sorted(t for t in reads if keep(t)),
            "writes": sorted(t for t in writes if keep(t)),
            "reads_core": sorted(t for t in reads if t in INTEREST),
            "writes_core": sorted(t for t in writes if t in INTEREST),
            "empty": False,
            "def_len": len(text),
        }

    OUT.write_text(json.dumps(results, ensure_ascii=False, indent=2), encoding="utf-8")
    print("PROC_COUNT", len(procs))
    print("WROTE", OUT)

    wc, rc = Counter(), Counter()
    for v in results.values():
        for t in v.get("writes_core", []):
            wc[t] += 1
        for t in v.get("reads_core", []):
            rc[t] += 1
    print("WRITE_CORE_FREQ")
    for t, c in wc.most_common():
        print(f"  {t}: {c}")
    print("READ_CORE_FREQ")
    for t, c in rc.most_common(25):
        print(f"  {t}: {c}")

    # domain summary for user tables
    domain_map = {
        "采购收货/IQC": [
            "pr_WMS_BPM_RctPurOrdAndPurOrdByBarcode",
            "pr_WMS_SavePurOrdRctByReconfirmQty",
            "pr_WMS_CancelPurOrdRct",
            "pr_WMS_SaveIQCResult",
            "pr_WMS_SaveIQCResultByRctNbr",
            "qms_WMS_SaveIQCResult",
            "pr_WMS_UpdatePurOrdRctTransactionProcessing2",
        ],
        "领料/备料/退料": [
            "pr_WMS_SavePrepareMaterial",
            "pr_WMS_SaveWorkOrdMaterialIssue",
            "pr_WMS_SaveWorkOrdMaterialIssueByQty",
            "pr_WMS_SaveWorkOrdMaterialConfirm",
            "pr_WMS_BPM_SaveCancelPrepareMaterial",
            "pr_WMS_SaveTWorkOrdMaterialReturn",
            "pr_SFM_InventoryMesWorkOrdIsses",
        ],
        "报工/工序完工": [
            "pr_SFM_InventoryMesTransactionProcessingByOp",
            "pr_SFM_EndWorkOrdByOp",
            "pr_MES_UpdateWorkOrder",
            "pr_MES_CloseWorkOrders",
            "pr_WMS_SaveProcessingByOtherOp",
            "pr_WMS_SaveMoveOpConfirm",
        ],
        "成品入库/上架": [
            "pr_WMS_SaveScanStorage",
            "pr_WMS_SaveProductStorage",
            "pr_WMS_SaveProductStorageByBox",
            "pr_WMS_BPM_SaveScanStorageCancel",
            "pr_WMS_BPM_SaveInvUpShelf",
            "pr_WMS_DirectInBound",
            "pr_WMS_SavePlanInBound",
        ],
        "发货/备货/ASN": [
            "pr_WMS_SavePrepareGoods",
            "pr_WMS_SavePrepareGoodsCancel",
            "pr_WMS_SaveShipperConfim",
            "pr_WMS_AutoCreateShipper",
            "pr_WMS_UpdateSalesOrdShippTransactionProcessing",
            "pr_WMS_SaveCancelSalesOrd",
            "pr_WMS_DirectOutBound",
            "pr_WMS_SavePlanOutBound",
        ],
        "FQC/检验": [
            "pr_WMS_SaveFQCcheck",
            "pr_WMS_SaveFQCResult",
            "pr_WMS_SaveInspectionConfirm",
            "pr_WMS_SaveInspectionKanbanOp",
            "pr_WMS_SaveInspectionScrap",
        ],
        "库存移动/盘点/标签": [
            "pr_WMS_SaveInvMove",
            "pr_WMS_SaveInvMoveByNbr",
            "pr_WMS_SaveCycleCounting",
            "pr_WMS_SaveInventoryCheck",
            "pr_WMS_SaveBarCodeFreezing",
            "pr_WMS_SaveMergeBarcode",
            "pr_WMS_SaveBreakUpBarcodes",
            "pr_SFM_InventoryTransactionProcessing2",
        ],
    }
    print("DOMAIN_CORE_TOUCH")
    for dom, plist in domain_map.items():
        tw, tr = set(), set()
        missing = []
        for p in plist:
            v = results.get(p)
            if not v:
                missing.append(p)
                continue
            tw.update(v.get("writes_core", []))
            tr.update(v.get("reads_core", []))
        print(f"## {dom}")
        print("  writes_core:", ", ".join(sorted(tw)) or "(none parsed)")
        print("  reads_core:", ", ".join(sorted(tr)) or "(none parsed)")
        if missing:
            print("  missing_defs:", ", ".join(missing))
    conn.close()


if __name__ == "__main__":
    main()