# -*- 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()