#!/usr/bin/env python3 """Build and execute tenant-scoped UAT migration batches. The source SQL files contain extracted, anonymized S1-S4 chains. This runner assigns target-specific IDs and business keys, persists old/new mappings, and keeps credentials outside the repository by reading the local App.json. """ from __future__ import annotations import argparse import json import re from dataclasses import dataclass from datetime import datetime from pathlib import Path from typing import Iterable import pymysql from pymysql.constants import CLIENT ROOT = Path(__file__).resolve().parents[4] SQL_ROOT = ROOT / "doc" / "plan" / "sql" DATABASE_JSON = ROOT / "server" / "Admin.NET.Application" / "Configuration" / "Database.json" SOURCE_TENANT = 797403760988229 @dataclass(frozen=True) class Target: code: str tenant_id: int factory_id: int operator_id: int source_file: str source_batch: str target_batch: str id_base: int key_prefix: str TARGETS = { "A": Target( "A", 838257186181189, 838257186320453, 838257187360837, "S1S4_UAT_P0_execute_write.sql", "S1S4_UAT_20260604_RQ_V1", "UAT_WP1_A_20260817_V1", 9206081700000001, "UATA", ), "B": Target( "B", 838257212780613, 838257212858437, 838257213620293, "S1S4_UAT_P0_execute_write.sql", "S1S4_UAT_20260604_RQ_V1", "UAT_WP1_B_20260817_V1", 9206081800000001, "UATB", ), "DEMO": Target( "DEMO", 838257237606469, 838257237676101, 838257238302789, "S1S4_UAT_FULL_execute_write.sql", "S1S4_UAT_FULL_20260605_V1", "UAT_WP1_DEMO_20260817_V1", 9206081900000001, "DEMO", ), } def walk_strings(value: object) -> Iterable[str]: if isinstance(value, str): yield value elif isinstance(value, dict): for child in value.values(): yield from walk_strings(child) elif isinstance(value, list): for child in value: yield from walk_strings(child) def load_connection() -> pymysql.Connection: raw = DATABASE_JSON.read_text(encoding="utf-8-sig") candidates = [ s for s in re.findall(r'(?m)^\s*"ConnectionString"\s*:\s*"([^"]+)"', raw) if "Database=aidopdev" in s ] if not candidates: raise RuntimeError("Database.json 中未找到启用的 aidopdev 连接串") parts = {} for pair in candidates[0].split(";"): if "=" in pair: key, val = pair.split("=", 1) parts[key.strip().lower()] = val.strip() return pymysql.connect( host=parts.get("server", "127.0.0.1"), port=int(parts.get("port", "3306")), user=parts.get("uid") or parts.get("user id"), password=parts.get("pwd") or parts.get("password"), database=parts.get("database", "aidopdev"), charset="utf8mb4", autocommit=True, client_flag=CLIENT.MULTI_STATEMENTS, connect_timeout=20, read_timeout=600, write_timeout=600, ) def ordered_tokens(sql: str, pattern: str) -> list[str]: return list(dict.fromkeys(re.findall(pattern, sql))) def build_key_map(sql: str, target: Target) -> dict[str, tuple[str, str]]: mapping: dict[str, tuple[str, str]] = {} orders = ordered_tokens(sql, r"'((?:MPO|SO)\d{10,})'") works = ordered_tokens(sql, r"'(M\d{9,12})'") prs = ordered_tokens(sql, r"'(PR-UAT-[^']+)'") pos = ordered_tokens(sql, r"'(PO-UAT-[^']+)'") ships = ordered_tokens(sql, r"'(SH-UAT-[^']+)'") for idx, old in enumerate(orders, 1): scenario = f"A{idx:02d}" if target.code == "A" else f"CP{idx:02d}" new = ( f"UATA-{scenario}-SO" if target.code == "A" else f"UATB-{scenario}-SO" if target.code == "B" else f"DEMO-SO-{idx:03d}" ) mapping[old] = ("SALES_ORDER", new) for idx, old in enumerate(works, 1): new = f"{target.key_prefix}-WO-{idx:03d}" mapping[old] = ("WORK_ORDER", new) for kind, tokens, label in ( ("PURCHASE_REQUEST", prs, "PR"), ("PURCHASE_ORDER", pos, "PO"), ("SUPPLIER_SHIPMENT", ships, "SH"), ): for idx, old in enumerate(tokens, 1): mapping[old] = (kind, f"{target.key_prefix}-{label}-{idx:03d}") return mapping def patch_tenant_columns(sql: str) -> str: sql = sql.replace( "need_number, morder_production_number, IsDeleted,\n" " create_by_name, create_time, update_by_name, update_time,\n" " tenant_id, factory_id\n" ")\n" "SELECT target_mo_id, morder_no, @now, 'Released', product_code, product_code,\n" " need_number, need_number, 0,", "need_number, morder_production_number, IsDeleted, urgent,\n" " create_by_name, create_time, update_by_name, update_time,\n" " tenant_id, factory_id\n" ")\n" "SELECT target_mo_id, morder_no, @now, 'Released', product_code, product_code,\n" " need_number, need_number, 0, 0,", ) sql = sql.replace( "INSERT INTO scm_shd (id, po_billno, shddh, sh_purchase_num, " "estimated_delivery_date, tjrxm, tjrq, state)", "INSERT INTO scm_shd (id, po_billno, shddh, sh_purchase_num, " "estimated_delivery_date, tjrxm, tjrq, state, tenant_id)", ) sql = sql.replace( "DATE_FORMAT(@now,'%Y-%m-%d'), 0\nFROM tmp_s4_uat_ship", "DATE_FORMAT(@now,'%Y-%m-%d'), 0, @tenant_id\nFROM tmp_s4_uat_ship", ) sql = sql.replace( "INSERT INTO scm_shdzb (id, glid, sh_material_code, " "sh_delivery_quantity, po_bill, po_billline)", "INSERT INTO scm_shdzb (id, glid, sh_material_code, " "sh_delivery_quantity, po_bill, po_billline, tenant_id)", ) sql = sql.replace( "pl.item_num, pl.qty, pl.pur_ord, CAST(pl.line_no AS CHAR)\n" "FROM tmp_s3_uat_po_line", "pl.item_num, pl.qty, pl.pur_ord, CAST(pl.line_no AS CHAR), @tenant_id\n" "FROM tmp_s3_uat_po_line", ) sql += """ -- Keep both work-order representations aligned with the linked sales-order item. UPDATE mes_morder m JOIN mes_moentry me ON me.tenant_id=m.tenant_id AND me.moentry_mono=m.morder_no JOIN crm_seorderentry e ON e.tenant_id=me.tenant_id AND e.Id=me.soentry_id SET m.product_code=e.item_number, m.product_name=COALESCE(NULLIF(e.item_name,''),e.item_number) WHERE m.tenant_id=@tenant_id; UPDATE WorkOrdMaster w JOIN mes_moentry me ON me.tenant_id=w.tenant_id AND me.moentry_mono=w.WorkOrd JOIN crm_seorderentry e ON e.tenant_id=me.tenant_id AND e.Id=me.soentry_id SET w.ItemNum=e.item_number WHERE w.tenant_id=@tenant_id; UPDATE PurOrdDetail d JOIN PurOrdMaster h ON h.tenant_id=d.tenant_id AND h.PurOrd=d.PurOrd SET d.PurOrdRecID=h.RecID WHERE d.tenant_id=@tenant_id AND d.PurOrdRecID=0; """ return sql def split_statements(sql: str) -> list[str]: """Split these migration files without breaking on comment semicolons.""" body = "\n".join( line for line in sql.splitlines() if not line.lstrip().startswith("--") ) return [statement.strip() for statement in body.split(";") if statement.strip()] def adapt_sql(target: Target) -> tuple[str, dict[str, tuple[str, str]]]: source = (SQL_ROOT / target.source_file).read_text(encoding="utf-8-sig") mapping = build_key_map(source, target) sql = source sql = re.sub(r"SET @tenant_id\s*:=\s*\d+;", f"SET @tenant_id := {target.tenant_id};", sql) sql = re.sub( r"SET @factory_id\s*:=\s*\d+;", f"SET @factory_id := {target.factory_id};", sql, ) sql = re.sub(r"SET @operator_id\s*:=\s*\d+;", f"SET @operator_id := {target.operator_id};", sql) sql = re.sub( r"SET @operator_name\s*:=\s*'[^']*';", f"SET @operator_name := 'UAT-{target.code}-Migration';", sql, ) sql = re.sub(r"SET @batch_no\s*:=\s*'[^']+';", f"SET @batch_no := '{target.target_batch}';", sql) sql = re.sub(r"SET @id_base\s*:=\s*\d+;", f"SET @id_base := {target.id_base};", sql) # FULL uses literal IDs in its temporary mapping rows. if target.source_file.startswith("S1S4_UAT_FULL"): sql = sql.replace("91060005", str(target.id_base)[:8]) # The historical template was slimmed; use an extant anonymized source PO. sql = sql.replace( "WHERE PurOrd='PW202605100002' AND tenant_id=@tenant_id", f"WHERE PurOrd='DO20260711001' AND tenant_id={SOURCE_TENANT}", ) for old, (_, new) in sorted(mapping.items(), key=lambda item: len(item[0]), reverse=True): sql = sql.replace(f"'{old}'", f"'{new}'") sql = patch_tenant_columns(sql) return sql, mapping def ensure_audit_tables(cur: pymysql.cursors.Cursor) -> None: cur.execute( """ CREATE TABLE IF NOT EXISTS uat_data_migration_batch ( batch_code varchar(100) NOT NULL PRIMARY KEY, target_tenant_id bigint NOT NULL, target_set varchar(16) NOT NULL, source_file varchar(255) NOT NULL, status varchar(20) NOT NULL, started_at datetime(3) NOT NULL, finished_at datetime(3) NULL, summary_json json NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """ ) cur.execute( """ CREATE TABLE IF NOT EXISTS uat_data_migration_map ( batch_code varchar(100) NOT NULL, object_type varchar(40) NOT NULL, source_key varchar(200) NOT NULL, target_key varchar(200) NOT NULL, target_tenant_id bigint NOT NULL, created_at datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), PRIMARY KEY(batch_code, object_type, source_key), KEY ix_uat_map_target(target_tenant_id, object_type, target_key) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """ ) def execute_target(target: Target, dry_run: bool) -> dict[str, object]: sql, mapping = adapt_sql(target) generated = Path(__file__).with_name(f"{target.target_batch}.sql") generated.write_text(sql, encoding="utf-8") if dry_run: return {"target": target.code, "generated": str(generated), "mapping_count": len(mapping)} conn = load_connection() try: with conn.cursor() as cur: ensure_audit_tables(cur) cur.execute( "SELECT status FROM uat_data_migration_batch WHERE batch_code=%s", (target.target_batch,), ) existing = cur.fetchone() if existing and existing[0] == "SUCCESS": raise RuntimeError(f"{target.target_batch} 已成功执行，拒绝重复导入") cur.execute( """ INSERT INTO uat_data_migration_batch (batch_code,target_tenant_id,target_set,source_file,status,started_at) VALUES(%s,%s,%s,%s,'RUNNING',NOW(3)) ON DUPLICATE KEY UPDATE status='RUNNING',started_at=NOW(3),finished_at=NULL """, (target.target_batch, target.tenant_id, target.code, target.source_file), ) try: for index, statement in enumerate(split_statements(sql), 1): try: cur.execute(statement) except Exception as exc: cur.execute("ROLLBACK") excerpt = " ".join(statement.split())[:240] raise RuntimeError( f"SQL #{index} 执行失败: {excerpt}: {exc}" ) from exc cur.executemany( """ INSERT INTO uat_data_migration_map (batch_code,object_type,source_key,target_key,target_tenant_id) VALUES(%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE target_key=VALUES(target_key) """, [ (target.target_batch, kind, old, new, target.tenant_id) for old, (kind, new) in mapping.items() ], ) summary = verify(cur, target) cur.execute( """ UPDATE uat_data_migration_batch SET status='SUCCESS',finished_at=NOW(3),summary_json=%s WHERE batch_code=%s """, (json.dumps(summary, ensure_ascii=False), target.target_batch), ) return summary except Exception as exc: cur.execute( """ UPDATE uat_data_migration_batch SET status='FAILED',finished_at=NOW(3),summary_json=%s WHERE batch_code=%s """, (json.dumps({"error": str(exc)}, ensure_ascii=False), target.target_batch), ) raise finally: conn.close() def verify(cur: pymysql.cursors.Cursor, target: Target) -> dict[str, object]: queries = { "orders": ( "SELECT COUNT(*) FROM crm_seorder WHERE tenant_id=%s AND bill_from LIKE %s", (target.tenant_id, f"%{target.target_batch}%"), ), "order_lines": ( """ SELECT COUNT(*) FROM crm_seorderentry e JOIN crm_seorder h ON h.Id=e.seorder_id AND h.tenant_id=e.tenant_id WHERE h.tenant_id=%s AND h.bill_from LIKE %s """, (target.tenant_id, f"%{target.target_batch}%"), ), "work_orders": ( "SELECT COUNT(*) FROM mes_morder WHERE tenant_id=%s AND morder_no LIKE %s", (target.tenant_id, f"{target.key_prefix}-WO-%"), ), "purchase_requests": ( "SELECT COUNT(*) FROM srm_pr_main WHERE tenant_id=%s AND pr_billno LIKE %s", (target.tenant_id, f"{target.key_prefix}-PR-%"), ), "purchase_orders": ( "SELECT COUNT(*) FROM PurOrdMaster WHERE tenant_id=%s AND PurOrd LIKE %s", (target.tenant_id, f"{target.key_prefix}-PO-%"), ), "supplier_shipments": ( "SELECT COUNT(*) FROM scm_shd WHERE tenant_id=%s AND shddh LIKE %s", (target.tenant_id, f"{target.key_prefix}-SH-%"), ), } result: dict[str, object] = {"target": target.code, "tenant_id": target.tenant_id} for key, (query, params) in queries.items(): cur.execute(query, params) result[key] = int(cur.fetchone()[0]) return result def table_columns(cur: pymysql.cursors.Cursor, table: str) -> tuple[list[str], str]: cur.execute(f"SHOW COLUMNS FROM `{table}`") rows = cur.fetchall() writable = [row[0] for row in rows if "auto_increment" not in (row[5] or "")] primary = next(row[0] for row in rows if row[3] == "PRI") return writable, primary def ensure_master_rows( cur: pymysql.cursors.Cursor, target: Target, table: str, key_column: str, keys: list[str], fallback: bool = False, ) -> int: columns, primary = table_columns(cur, table) inserted = 0 for key in keys: cur.execute( f"SELECT `{primary}` FROM `{table}` WHERE tenant_id=%s AND `{key_column}`=%s LIMIT 1", (target.tenant_id, key), ) if cur.fetchone(): continue cur.execute( f""" SELECT {",".join(f"`{column}`" for column in columns)} FROM `{table}` WHERE tenant_id IN (%s,%s) AND `{key_column}`=%s ORDER BY FIELD(tenant_id,%s,%s) LIMIT 1 """, (SOURCE_TENANT, 824585161322565, key, SOURCE_TENANT, 824585161322565), ) row = cur.fetchone() if row is None and fallback: cur.execute( f""" SELECT {",".join(f"`{column}`" for column in columns)} FROM `{table}` WHERE tenant_id=%s ORDER BY `{primary}` LIMIT 1 """, (SOURCE_TENANT,), ) row = cur.fetchone() if row is None: raise RuntimeError(f"{table}.{key_column}={key} 无可用脱敏源记录") values = dict(zip(columns, row)) values[key_column] = key for column in ("tenant_id", "TenantId"): if column in values: values[column] = target.tenant_id for column in ( "company_ref_id", "factory_ref_id", "CompanyRefId", "FactoryRefId", "company_id", "factory_id", "org_id", ): if column in values: values[column] = target.factory_id cur.execute( f""" INSERT INTO `{table}` ({",".join(f"`{column}`" for column in columns)}) VALUES ({",".join(["%s"] * len(columns))}) """, tuple(values[column] for column in columns), ) new_id = cur.lastrowid cur.execute( """ INSERT INTO uat_data_migration_map (batch_code,object_type,source_key,target_key,target_tenant_id) VALUES(%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE target_key=VALUES(target_key) """, ( target.target_batch, f"MASTER_{table.upper()}", key, f"{key}:{new_id}", target.tenant_id, ), ) inserted += 1 return inserted def scoped_values( columns: list[str], row: tuple[object, ...], target: Target ) -> dict[str, object]: values = dict(zip(columns, row)) for column in ("tenant_id", "TenantId"): if column in values: values[column] = target.tenant_id for column in ( "company_ref_id", "factory_ref_id", "CompanyRefId", "FactoryRefId", "company_id", "factory_id", "org_id", ): if column in values: values[column] = target.factory_id return values def insert_values( cur: pymysql.cursors.Cursor, table: str, columns: list[str], values: dict[str, object], ) -> int: cur.execute( f""" INSERT INTO `{table}` ({",".join(f"`{column}`" for column in columns)}) VALUES ({",".join(["%s"] * len(columns))}) """, tuple(values[column] for column in columns), ) return int(cur.lastrowid) def execute_structure_closure( cur: pymysql.cursors.Cursor, target: Target, initial_items: list[str] ) -> dict[str, int]: # Expand one BOM level so every copied BOM component also has an item master. placeholders = ",".join(["%s"] * len(initial_items)) component_items: set[str] = set() if initial_items: cur.execute( f""" SELECT DISTINCT ComponentItem FROM ProductStructureMaster WHERE tenant_id IN (%s,%s) AND ParentItem IN ({placeholders}) AND IFNULL(ComponentItem,'')<>'' """, (SOURCE_TENANT, 824585161322565, *initial_items), ) component_items = {str(row[0]) for row in cur.fetchall()} expanded_items = sorted(set(initial_items) | component_items) item_inserted = ensure_master_rows( cur, target, "ItemMaster", "ItemNum", expanded_items, fallback=True ) cur.execute( "SELECT ItemNum,RecID FROM ItemMaster WHERE tenant_id=%s", (target.tenant_id,), ) target_item_ids = {str(row[0]): int(row[1]) for row in cur.fetchall()} bom_columns, bom_primary = table_columns(cur, "ProductStructureMaster") bom_maps: dict[int, int] = {} seen_bom: set[tuple[object, ...]] = set() bom_inserted = 0 if initial_items: for source_tenant in (SOURCE_TENANT, 824585161322565): cur.execute( f""" SELECT `{bom_primary}`,{",".join(f"`{column}`" for column in bom_columns)} FROM ProductStructureMaster WHERE tenant_id=%s AND ParentItem IN ({placeholders}) ORDER BY `{bom_primary}` """, (source_tenant, *initial_items), ) for source_id, *row in cur.fetchall(): values = scoped_values(bom_columns, tuple(row), target) key = ( values.get("ParentItem"), values.get("ComponentItem"), target.factory_id, ) if key in seen_bom: continue seen_bom.add(key) if "ParentMaterialId" in values: values["ParentMaterialId"] = target_item_ids.get( str(values.get("ParentItem")) ) if "ComponentMaterialId" in values: values["ComponentMaterialId"] = target_item_ids.get( str(values.get("ComponentItem")) ) inserted = False try: target_id = insert_values( cur, "ProductStructureMaster", bom_columns, values ) inserted = True except pymysql.IntegrityError as exc: if exc.args[0] != 1062: raise cur.execute( """ SELECT RecID FROM ProductStructureMaster WHERE tenant_id=%s AND CompanyRefId=%s AND FactoryRefId=%s AND ParentItem=%s AND ComponentItem=%s LIMIT 1 """, ( target.tenant_id, target.factory_id, target.factory_id, values.get("ParentItem"), values.get("ComponentItem"), ), ) existing = cur.fetchone() if not existing: raise target_id = int(existing[0]) bom_maps[int(source_id)] = target_id if inserted: bom_inserted += 1 op_inserted = 0 if bom_maps: op_columns, op_primary = table_columns(cur, "ProductStructureOp") ids = list(bom_maps) for start in range(0, len(ids), 500): chunk = ids[start : start + 500] cur.execute( f""" SELECT `{op_primary}`,{",".join(f"`{column}`" for column in op_columns)} FROM ProductStructureOp WHERE ProductStructureMasterId IN ({",".join(["%s"] * len(chunk))}) """, tuple(chunk), ) for _, *row in cur.fetchall(): values = scoped_values(op_columns, tuple(row), target) old_master = int(values["ProductStructureMasterId"]) values["ProductStructureMasterId"] = bom_maps[old_master] try: insert_values(cur, "ProductStructureOp", op_columns, values) op_inserted += 1 except pymysql.IntegrityError as exc: if exc.args[0] != 1062: raise routing_columns, routing_primary = table_columns(cur, "RoutingOpDetail") routing_inserted = 0 seen_routing: set[tuple[object, ...]] = set() if expanded_items: routing_placeholders = ",".join(["%s"] * len(expanded_items)) for source_tenant in (SOURCE_TENANT, 824585161322565): cur.execute( f""" SELECT `{routing_primary}`,{",".join(f"`{column}`" for column in routing_columns)} FROM RoutingOpDetail WHERE tenant_id=%s AND MaterialCode IN ({routing_placeholders}) ORDER BY `{routing_primary}` """, (source_tenant, *expanded_items), ) for _, *row in cur.fetchall(): values = scoped_values(routing_columns, tuple(row), target) key = tuple( values.get(column) for column in ("RouteCode", "MaterialCode", "OperationCode", "SortNo") ) if key in seen_routing: continue seen_routing.add(key) try: insert_values(cur, "RoutingOpDetail", routing_columns, values) routing_inserted += 1 except pymysql.IntegrityError as exc: if exc.args[0] != 1062: raise return { "bom_component_items": len(component_items), "bom_component_items_inserted": item_inserted, "bom_rows_inserted": bom_inserted, "bom_operations_inserted": op_inserted, "routing_rows_inserted": routing_inserted, } def execute_resource_closure( cur: pymysql.cursors.Cursor, target: Target ) -> dict[str, int]: cur.execute( """ SELECT Department FROM DepartmentMaster WHERE tenant_id=%s AND IFNULL(Department,'')<>'' GROUP BY Department ORDER BY MIN(RecID) LIMIT 20 """, (SOURCE_TENANT,), ) departments = [str(row[0]) for row in cur.fetchall()] department_inserted = ensure_master_rows( cur, target, "DepartmentMaster", "Department", departments, fallback=True, ) cur.execute( """ SELECT DISTINCT WorkCenterCode FROM RoutingOpDetail WHERE tenant_id=%s AND IFNULL(WorkCenterCode,'')<>'' """, (target.tenant_id,), ) work_centers = [str(row[0]) for row in cur.fetchall()] cur.execute( """ SELECT DISTINCT COALESCE(NULLIF(StdOp,''),NULLIF(OperationCode,'')) FROM RoutingOpDetail WHERE tenant_id=%s AND COALESCE(NULLIF(StdOp,''),NULLIF(OperationCode,'')) IS NOT NULL """, (target.tenant_id,), ) operations = [str(row[0]) for row in cur.fetchall()] work_center_inserted = ensure_master_rows( cur, target, "WorkCtrMaster", "WorkCtr", work_centers, fallback=True ) operation_inserted = ensure_master_rows( cur, target, "StdOpMaster", "StdOp", operations, fallback=True ) cur.execute( """ SELECT Line FROM LineMaster WHERE tenant_id=%s AND IFNULL(Line,'')<>'' GROUP BY Line ORDER BY MIN(RecID) LIMIT 20 """, (SOURCE_TENANT,), ) lines = [str(row[0]) for row in cur.fetchall()] line_inserted = ensure_master_rows( cur, target, "LineMaster", "Line", lines, fallback=True ) calendar_columns, calendar_primary = table_columns(cur, "ShopCalendarWorkCtr") calendar_inserted = 0 if work_centers or lines: work_placeholders = ",".join(["%s"] * max(1, len(work_centers))) line_placeholders = ",".join(["%s"] * max(1, len(lines))) cur.execute( f""" SELECT {",".join(f"`{column}`" for column in calendar_columns)} FROM ShopCalendarWorkCtr WHERE tenant_id=%s AND (WorkCtr IN ({work_placeholders}) OR ProdLine IN ({line_placeholders})) ORDER BY `{calendar_primary}` """, ( SOURCE_TENANT, *(work_centers or [""]), *(lines or [""]), ), ) seen: set[tuple[object, ...]] = set() for row in cur.fetchall(): values = scoped_values(calendar_columns, tuple(row), target) key = ( values.get("Domain"), values.get("Site"), values.get("WorkCtr"), values.get("WeekDay"), values.get("ProdLine"), ) if key in seen: continue seen.add(key) cur.execute( """ SELECT 1 FROM ShopCalendarWorkCtr WHERE tenant_id=%s AND COALESCE(Domain,'')=COALESCE(%s,'') AND COALESCE(Site,'')=COALESCE(%s,'') AND COALESCE(WorkCtr,'')=COALESCE(%s,'') AND COALESCE(WeekDay,'')=COALESCE(%s,'') AND COALESCE(ProdLine,'')=COALESCE(%s,'') LIMIT 1 """, ( target.tenant_id, values.get("Domain"), values.get("Site"), values.get("WorkCtr"), values.get("WeekDay"), values.get("ProdLine"), ), ) if cur.fetchone(): continue insert_values(cur, "ShopCalendarWorkCtr", calendar_columns, values) calendar_inserted += 1 return { "departments_required": len(departments), "departments_inserted": department_inserted, "work_centers_required": len(work_centers), "work_centers_inserted": work_center_inserted, "operations_required": len(operations), "operations_inserted": operation_inserted, "lines_required": len(lines), "lines_inserted": line_inserted, "calendar_rows_inserted": calendar_inserted, } def execute_work_order_closure( cur: pymysql.cursors.Cursor, target: Target ) -> dict[str, int]: cur.execute( """ INSERT INTO WorkOrdDetail (Domain,Site,LineNum,WorkOrd,Op,ItemNum, QtyRequired,QtytoIssue,QtyIssued,QtyPicked, WorkOrdMasterRecID,tenant_id) SELECT w.Domain,w.Site, ROW_NUMBER() OVER(PARTITION BY w.RecID ORDER BY b.RecID), w.WorkOrd,10,b.ComponentItem, COALESCE(NULLIF(b.Qty,0),1)*COALESCE(NULLIF(w.QtyOrded,0),1), COALESCE(NULLIF(b.Qty,0),1)*COALESCE(NULLIF(w.QtyOrded,0),1), 0,0,w.RecID,w.tenant_id FROM WorkOrdMaster w JOIN ProductStructureMaster b ON b.tenant_id=w.tenant_id AND b.ParentItem=w.ItemNum WHERE w.tenant_id=%s AND NOT EXISTS ( SELECT 1 FROM WorkOrdDetail d WHERE d.tenant_id=w.tenant_id AND d.WorkOrdMasterRecID=w.RecID ) """, (target.tenant_id,), ) material_rows = int(cur.rowcount) cur.execute( """ INSERT INTO WorkOrdRouting (StdOp,Domain,RunCrew,WorkOrd,OP,ItemNum, QtyComplete,QtyOrded,WorkCtr,ParentOp, WorkOrdMasterRecID,tenant_id) SELECT COALESCE(NULLIF(r.StdOp,''),r.OperationCode), LEFT(COALESCE(r.Domain,''),8),1, w.WorkOrd, CASE WHEN r.OperationCode REGEXP '^[0-9]+$' THEN CAST(r.OperationCode AS UNSIGNED) ELSE ROW_NUMBER() OVER(PARTITION BY w.RecID ORDER BY r.SortNo,r.RecID)*10 END, w.ItemNum,0,COALESCE(NULLIF(w.QtyOrded,0),1), LEFT(COALESCE(r.WorkCenterCode,''),8),0, w.RecID,w.tenant_id FROM WorkOrdMaster w JOIN RoutingOpDetail r ON r.tenant_id=w.tenant_id AND r.MaterialCode=w.ItemNum WHERE w.tenant_id=%s AND NOT EXISTS ( SELECT 1 FROM WorkOrdRouting wr WHERE wr.tenant_id=w.tenant_id AND wr.WorkOrdMasterRecID=w.RecID ) """, (target.tenant_id,), ) routing_rows = int(cur.rowcount) return { "work_order_material_rows_inserted": material_rows, "work_order_routing_rows_inserted": routing_rows, } def execute_master_closure(target: Target) -> dict[str, object]: conn = load_connection() try: with conn.cursor() as cur: ensure_audit_tables(cur) cur.execute( """ SELECT DISTINCT item_number FROM crm_seorderentry WHERE tenant_id=%s AND IFNULL(item_number,'')<>'' UNION SELECT DISTINCT product_code FROM mes_morder WHERE tenant_id=%s AND IFNULL(product_code,'')<>'' UNION SELECT DISTINCT icitem_name FROM srm_pr_main WHERE tenant_id=%s AND IFNULL(icitem_name,'')<>'' """, (target.tenant_id, target.tenant_id, target.tenant_id), ) items = [str(row[0]) for row in cur.fetchall()] cur.execute( "SELECT DISTINCT custom_no FROM crm_seorder WHERE tenant_id=%s AND IFNULL(custom_no,'')<>''", (target.tenant_id,), ) customers = [str(row[0]) for row in cur.fetchall()] cur.execute( "SELECT DISTINCT Supp FROM PurOrdMaster WHERE tenant_id=%s AND IFNULL(Supp,'')<>''", (target.tenant_id,), ) suppliers = [str(row[0]) for row in cur.fetchall()] result = { "target": target.code, "items_required": len(items), "items_inserted": ensure_master_rows( cur, target, "ItemMaster", "ItemNum", items, fallback=True ), "customers_required": len(customers), "customers_inserted": ensure_master_rows( cur, target, "CustMaster", "Cust", customers, fallback=True ), "suppliers_required": len(suppliers), "suppliers_inserted": ensure_master_rows( cur, target, "SuppMaster", "Supp", suppliers, fallback=True ), } result.update(execute_structure_closure(cur, target, items)) result.update(execute_resource_closure(cur, target)) result.update(execute_work_order_closure(cur, target)) return result finally: conn.close() def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("targets", nargs="+", choices=TARGETS.keys()) parser.add_argument("--dry-run", action="store_true") parser.add_argument("--master-closure", action="store_true") parser.add_argument("--evidence", type=Path) args = parser.parse_args() if args.master_closure: results = [execute_master_closure(TARGETS[code]) for code in args.targets] else: results = [execute_target(TARGETS[code], args.dry_run) for code in args.targets] payload = {"executed_at": datetime.now().isoformat(timespec="seconds"), "results": results} if args.evidence: args.evidence.parent.mkdir(parents=True, exist_ok=True) args.evidence.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps(payload, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()