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

# -*- coding: utf-8 -*-
"""Extract UPDATE SET column hints and key business assignments from procs."""
from __future__ import annotations

import json
import re
from pathlib import Path

import pyodbc

OUT = Path(__file__).with_name("_tmp_field_level_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"
)

PROCS = [
    # WMS path
    "pr_WMS_SavePrepareMaterial",
    "pr_WMS_SaveWorkOrdMaterialIssue",
    "pr_WMS_SaveWorkOrdMaterialIssueByQty",
    "pr_WMS_SaveWorkOrdMaterialConfirm",
    "pr_WMS_BPM_SaveCancelPrepareMaterial",
    "pr_WMS_CreateBarCodeCard",
    "pr_SFM_InventoryTransactionProcessing2",
    # MES path
    "pr_SFM_SaveWorkOrdChooseByOpData",
    "pr_SFM_SaveWorkOrdChooseByRWData",
    "pr_SFM_SaveEmployeeWorkStatus",
    "pr_SFM_LinePlanedStart",
    "pr_SFM_LinePlanedEnd",
    "pr_SFM_RejectsTransactionProcessingByOp",
    "pr_SFM_InventoryMesTransactionProcessingByOp",
    "pr_SFM_InventoryMesCartonTransactionProcessingByOp",
    "pr_SFM_EndWorkOrdByOp",
    "pr_SFM_AccPeriodSequenceDet",
    "pr_MES_UpdateWorkOrder",
    "pr_MES_CloseWorkOrders",
]

# UPDATE dbo.Table SET a=..., b=...
UPD_BLOCK = re.compile(
    r"(?is)\bUPDATE\s+(?:\[?dbo\]?\.)?\[?([A-Za-z_][A-Za-z0-9_]*)\]?\s+SET\s+(.*?)(?:\bWHERE\b|\bFROM\b|;|$)"
)
# INSERT INTO Table (cols)
INS_COLS = re.compile(
    r"(?is)\bINSERT\s+(?:INTO\s+)?(?:\[?dbo\]?\.)?\[?([A-Za-z_][A-Za-z0-9_]*)\]?\s*\(([^)]{1,800})\)"
)
EXEC_RE = re.compile(
    r"(?i)\b(?:EXEC|EXECUTE)\s+(?:\[?dbo\]?\.)?\[?([A-Za-z_][A-Za-z0-9_]*)\]?"
)
# assignment-like mentions Table.Col =
ASSIGN = re.compile(
    r"(?i)\b([A-Za-z_][A-Za-z0-9_]*)\.([A-Za-z_][A-Za-z0-9_]*)\s*="
)

CORE = {
    "WorkOrdMaster",
    "WorkOrdRouting",
    "WorkOrdDetail",
    "PeriodSequenceDet",
    "NbrMaster",
    "NbrDetail",
    "MissedPrint",
    "InvMaster",
    "InvTransHist",
    "LocationDetail",
    "LocationMaster",
    "LineMaster",
    "LineStatusDet",
    "LineRunRestDet",
    "EmployeeStatus",
    "EmpWorkHist",
    "EmpWorkWmsHist",
    "WorkOrdIssued",
    "ReleaseDet",
    "SingleBarCode",
    "MobileTask",
    "PurOrdDetail",
    "PurOrdRctDetail",
    "ItemMaster",
}


def split_set_cols(set_clause: str) -> list[str]:
    cols = []
    # rough split by comma at top level
    depth = 0
    buf = []
    for ch in set_clause:
        if ch == "(":
            depth += 1
        elif ch == ")":
            depth = max(0, depth - 1)
        if ch == "," and depth == 0:
            part = "".join(buf).strip()
            if part:
                cols.append(part)
            buf = []
        else:
            buf.append(ch)
    part = "".join(buf).strip()
    if part:
        cols.append(part)
    names = []
    for p in cols:
        m = re.match(r"\[?([A-Za-z_][A-Za-z0-9_]*)\]?\s*=", p.strip())
        if m:
            names.append(m.group(1))
    return names


def main() -> None:
    conn = pyodbc.connect(CONN, timeout=90)
    cur = conn.cursor()
    result = {}
    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:
            # fuzzy search
            cur.execute(
                "SELECT name FROM sys.procedures WHERE name LIKE ?",
                f"%{name.replace('pr_WMS_', '').replace('pr_SFM_', '')}%",
            )
            alts = [r[0] for r in cur.fetchall()[:8]]
            result[name] = {"empty": True, "alts": alts}
            continue
        updates = {}
        for m in UPD_BLOCK.finditer(text):
            tbl, clause = m.group(1), m.group(2)
            if tbl not in CORE and not tbl[:1].isupper():
                continue
            cols = split_set_cols(clause)
            updates.setdefault(tbl, [])
            for c in cols:
                if c not in updates[tbl]:
                    updates[tbl].append(c)
        inserts = {}
        for m in INS_COLS.finditer(text):
            tbl, cols = m.group(1), m.group(2)
            clist = [c.strip().strip("[]") for c in cols.split(",") if c.strip()]
            inserts.setdefault(tbl, [])
            for c in clist:
                if re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", c) and c not in inserts[tbl]:
                    inserts[tbl].append(c)
        # filter inserts/updates to interesting
        updates = {k: v for k, v in updates.items() if k in CORE or k.startswith(("qms_", "mes_"))}
        inserts = {k: v for k, v in inserts.items() if k in CORE or k.startswith(("qms_", "mes_"))}
        nested = sorted(
            {
                n
                for n in EXEC_RE.findall(text)
                if n.lower().startswith(("pr_", "qms_"))
            }
        )
        # status / qty literals near key tables
        snippets = []
        for pat in [
            r"(?is)UPDATE\s+(?:\[?dbo\]?\.)?\[?WorkOrdMaster\]?\s+SET.{0,400}",
            r"(?is)UPDATE\s+(?:\[?dbo\]?\.)?\[?WorkOrdRouting\]?\s+SET.{0,400}",
            r"(?is)UPDATE\s+(?:\[?dbo\]?\.)?\[?PeriodSequenceDet\]?\s+SET.{0,400}",
            r"(?is)UPDATE\s+(?:\[?dbo\]?\.)?\[?NbrDetail\]?\s+SET.{0,400}",
            r"(?is)UPDATE\s+(?:\[?dbo\]?\.)?\[?NbrMaster\]?\s+SET.{0,400}",
            r"(?is)UPDATE\s+(?:\[?dbo\]?\.)?\[?MissedPrint\]?\s+SET.{0,400}",
            r"(?is)UPDATE\s+(?:\[?dbo\]?\.)?\[?LineStatusDet\]?\s+SET.{0,400}",
            r"(?is)UPDATE\s+(?:\[?dbo\]?\.)?\[?EmployeeStatus\]?\s+SET.{0,400}",
            r"(?is)UPDATE\s+(?:\[?dbo\]?\.)?\[?WorkOrdDetail\]?\s+SET.{0,400}",
        ]:
            for m in re.finditer(pat, text):
                s = re.sub(r"\s+", " ", m.group(0))[:350]
                snippets.append(s)
                if len(snippets) >= 12:
                    break
            if len(snippets) >= 12:
                break
        result[name] = {
            "empty": False,
            "def_len": len(text),
            "updates": updates,
            "inserts": {k: v[:40] for k, v in inserts.items()},
            "nested": nested[:25],
            "update_snippets": snippets[:12],
        }
    OUT.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
    print("WROTE", OUT)
    for name, v in result.items():
        if v.get("empty"):
            print(name, "EMPTY", v.get("alts"))
            continue
        print("==", name, "len", v["def_len"])
        for t, cols in v["updates"].items():
            print("  UPD", t, ":", ", ".join(cols[:25]))
        for t, cols in list(v["inserts"].items())[:6]:
            print("  INS", t, ":", ", ".join(cols[:15]))
        if v["nested"]:
            print("  NEST", ", ".join(v["nested"][:8]))
    conn.close()


if __name__ == "__main__":
    main()