783 lines
30 KiB
Python
783 lines
30 KiB
Python
# ============================================================
|
||
# MySQL/Navicat SQL 数据包解析（moduleId: importer-sql-pack, 可重生 ✅）
|
||
# 流式读取超大 dump（如松岳 mesdb_pro.sql ~1GB），抽取排产相关表 → flex*
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import re
|
||
from datetime import datetime
|
||
from typing import Any, Iterable
|
||
|
||
# 表 → 角色（松岳 MES / 通用命名）
|
||
_TABLE_KIND: dict[str, str] = {
|
||
"pl_order": "orders",
|
||
"r_plan_order": "orders",
|
||
"pl_order_bom": "bom",
|
||
"md_material": "materials",
|
||
"md_item": "materials",
|
||
"md_equipment": "equipment",
|
||
"em_devices": "equipment",
|
||
"r_production_craftl": "routing",
|
||
"c_project_process_route": "routing",
|
||
"md_craftl": "craft",
|
||
"r_production_procedure": "op_equip",
|
||
"pl_work_order": "work_orders",
|
||
}
|
||
|
||
_CREATE_RE = re.compile(r"CREATE TABLE\s+`([^`]+)`", re.I)
|
||
_INSERT_RE = re.compile(r"INSERT INTO\s+`([^`]+)`\s+VALUES\s*", re.I)
|
||
|
||
# 单表预览/导入上限（防 1GB 全量撑爆内存）
|
||
_DEFAULT_LIMITS = {
|
||
"orders": 800,
|
||
"materials": 4000,
|
||
"bom": 8000,
|
||
"routing": 8000, # 其它工艺上限
|
||
"routing_prefer": 25000, # 订单相关工艺路线优先配额
|
||
"equipment": 800,
|
||
"craft": 2000,
|
||
"op_equip": 4000,
|
||
"work_orders": 3000,
|
||
}
|
||
|
||
|
||
def _split_sql_values(blob: str) -> list[tuple[str, bool]]:
|
||
"""拆分 INSERT VALUES 元组内的字段 → (文本, 是否引号字符串)。"""
|
||
out: list[tuple[str, bool]] = []
|
||
cur: list[str] = []
|
||
i = 0
|
||
in_str = False
|
||
quoted = False
|
||
n = len(blob)
|
||
while i < n:
|
||
ch = blob[i]
|
||
if in_str:
|
||
if ch == "\\" and i + 1 < n:
|
||
cur.append(blob[i + 1])
|
||
i += 2
|
||
continue
|
||
if ch == "'":
|
||
if i + 1 < n and blob[i + 1] == "'":
|
||
cur.append("'")
|
||
i += 2
|
||
continue
|
||
in_str = False
|
||
i += 1
|
||
continue
|
||
cur.append(ch)
|
||
i += 1
|
||
continue
|
||
if ch == "'":
|
||
in_str = True
|
||
quoted = True
|
||
i += 1
|
||
continue
|
||
if ch == ",":
|
||
out.append(("".join(cur).strip(), quoted))
|
||
cur = []
|
||
quoted = False
|
||
i += 1
|
||
continue
|
||
cur.append(ch)
|
||
i += 1
|
||
if cur or blob.endswith(","):
|
||
out.append(("".join(cur).strip(), quoted))
|
||
return out
|
||
|
||
|
||
def _parse_value(raw: str, *, quoted: bool = False) -> Any:
|
||
s = (raw or "").strip()
|
||
if not s or s.upper() == "NULL":
|
||
return None
|
||
if quoted:
|
||
return s # 引号字段保持字符串（料号 030129001 不能丢前导 0）
|
||
if (s.startswith("'") and s.endswith("'")) or (s.startswith('"') and s.endswith('"')):
|
||
return s[1:-1]
|
||
try:
|
||
if "." in s:
|
||
return float(s)
|
||
return int(s)
|
||
except ValueError:
|
||
return s
|
||
|
||
|
||
def _iter_insert_tuples(line: str) -> Iterable[list[Any]]:
|
||
"""从一行或多行 INSERT 语句产出行值列表。"""
|
||
m = _INSERT_RE.search(line)
|
||
if not m:
|
||
return
|
||
rest = line[m.end():].strip()
|
||
if rest.endswith(";"):
|
||
rest = rest[:-1]
|
||
# 多个 (...),(...)
|
||
i = 0
|
||
n = len(rest)
|
||
while i < n:
|
||
while i < n and rest[i] in " \t\r\n,":
|
||
i += 1
|
||
if i >= n or rest[i] != "(":
|
||
break
|
||
depth = 0
|
||
in_str = False
|
||
j = i
|
||
while j < n:
|
||
ch = rest[j]
|
||
if in_str:
|
||
if ch == "\\" and j + 1 < n:
|
||
j += 2
|
||
continue
|
||
if ch == "'":
|
||
if j + 1 < n and rest[j + 1] == "'":
|
||
j += 2
|
||
continue
|
||
in_str = False
|
||
j += 1
|
||
continue
|
||
if ch == "'":
|
||
in_str = True
|
||
j += 1
|
||
continue
|
||
if ch == "(":
|
||
depth += 1
|
||
elif ch == ")":
|
||
depth -= 1
|
||
if depth == 0:
|
||
inner = rest[i + 1:j]
|
||
yield [_parse_value(x, quoted=q) for x, q in _split_sql_values(inner)]
|
||
i = j + 1
|
||
break
|
||
j += 1
|
||
else:
|
||
break
|
||
|
||
|
||
def _parse_create_columns(block: str) -> list[str]:
|
||
"""解析 CREATE TABLE 列名；支持一行多列（测试/精简 dump）与 Navicat 一行一列。"""
|
||
cols: list[str] = []
|
||
for line in block.splitlines():
|
||
s = line.strip().upper()
|
||
if s.startswith(("PRIMARY ", "UNIQUE ", "KEY ", "INDEX ", "CONSTRAINT ")):
|
||
continue
|
||
if s.startswith(")"):
|
||
break
|
||
for m in re.finditer(r"`([^`]+)`\s+[a-zA-Z]", line):
|
||
name = m.group(1)
|
||
if name.upper() in ("PRIMARY", "UNIQUE", "KEY", "INDEX"):
|
||
continue
|
||
cols.append(name)
|
||
return cols
|
||
|
||
|
||
def _due_str(v: Any) -> str:
|
||
if v is None:
|
||
return ""
|
||
if isinstance(v, datetime):
|
||
return v.strftime("%Y-%m-%d")
|
||
s = str(v).strip()
|
||
m = re.match(r"(\d{4}-\d{2}-\d{2})", s)
|
||
return m.group(1) if m else s[:10]
|
||
|
||
|
||
def _row_dict(cols: list[str], vals: list[Any]) -> dict[str, Any]:
|
||
d: dict[str, Any] = {}
|
||
for i, c in enumerate(cols):
|
||
if i < len(vals):
|
||
d[c] = vals[i]
|
||
return d
|
||
|
||
|
||
def scan_sql_pack(
|
||
path: str,
|
||
*,
|
||
limits: dict[str, int] | None = None,
|
||
) -> dict[str, Any]:
|
||
"""流式扫描 SQL dump，返回按 kind 分组的行 + 表统计。"""
|
||
lim = {**_DEFAULT_LIMITS, **(limits or {})}
|
||
schemas: dict[str, list[str]] = {}
|
||
rows_by_table: dict[str, list[dict[str, Any]]] = {}
|
||
counts: dict[str, int] = {}
|
||
creating: str | None = None
|
||
create_buf: list[str] = []
|
||
|
||
size = os.path.getsize(path) if os.path.isfile(path) else 0
|
||
try:
|
||
from server.agent_core.progress import emit_thinking
|
||
except Exception:
|
||
def emit_thinking(*_a, **_k): # type: ignore
|
||
return None
|
||
|
||
emit_thinking("读取 SQL 数据包", f"{os.path.basename(path)} · {size / (1024 ** 2):.0f} MB", pct=1)
|
||
lines_read = 0
|
||
bytes_read = 0
|
||
last_emit = 0
|
||
# 订单上的工艺路线编码：dump 里 pl_order 早于 r_production_craftl，可优先抽取
|
||
prefer_craftl: set[str] = set()
|
||
routing_pref: list[dict[str, Any]] = []
|
||
routing_other: list[dict[str, Any]] = []
|
||
routing_table = "r_production_craftl"
|
||
with open(path, "r", encoding="utf-8", errors="replace") as f:
|
||
for line in f:
|
||
lines_read += 1
|
||
bytes_read += len(line.encode("utf-8", errors="ignore"))
|
||
if creating is not None:
|
||
create_buf.append(line)
|
||
if "ENGINE" in line.upper() and line.strip().endswith(";"):
|
||
schemas[creating] = _parse_create_columns("".join(create_buf))
|
||
creating = None
|
||
create_buf = []
|
||
continue
|
||
|
||
cm = _CREATE_RE.search(line)
|
||
if cm:
|
||
name = cm.group(1)
|
||
if name in _TABLE_KIND:
|
||
creating = name
|
||
create_buf = [line]
|
||
emit_thinking("解析表结构", name, pct=min(40, 5 + len(schemas)))
|
||
continue
|
||
|
||
if "INSERT INTO" not in line.upper():
|
||
if lines_read - last_emit >= 400_000:
|
||
last_emit = lines_read
|
||
pct = min(85.0, 10 + (bytes_read / max(size, 1)) * 70)
|
||
emit_thinking("扫描 SQL 全文", f"已读 {lines_read // 10000 / 10:.1f} 万行", pct=pct)
|
||
continue
|
||
im = _INSERT_RE.search(line)
|
||
if not im:
|
||
continue
|
||
table = im.group(1)
|
||
kind = _TABLE_KIND.get(table)
|
||
if not kind:
|
||
continue
|
||
cols = schemas.get(table) or []
|
||
if not cols:
|
||
continue
|
||
# 支持 INSERT 跨行：VALUES 在下一行，直到分号结束
|
||
insert_blob = line
|
||
while ";" not in insert_blob:
|
||
nxt = f.readline()
|
||
if not nxt:
|
||
break
|
||
lines_read += 1
|
||
bytes_read += len(nxt.encode("utf-8", errors="ignore"))
|
||
insert_blob += nxt
|
||
before = counts.get(table, 0)
|
||
for vals in _iter_insert_tuples(insert_blob):
|
||
counts[table] = counts.get(table, 0) + 1
|
||
row = _row_dict(cols, vals)
|
||
if kind == "orders":
|
||
cc = str(row.get("craftl_code") or "").strip()
|
||
if cc:
|
||
prefer_craftl.add(cc)
|
||
bucket = rows_by_table.setdefault(table, [])
|
||
if len(bucket) < lim.get("orders", 800):
|
||
bucket.append(row)
|
||
elif kind == "routing":
|
||
cc = str(row.get("craftl_code") or row.get("process_id") or "").strip()
|
||
if cc and cc in prefer_craftl:
|
||
if len(routing_pref) < lim.get("routing_prefer", 25000):
|
||
routing_pref.append(row)
|
||
elif len(routing_other) < lim.get("routing", 8000):
|
||
routing_other.append(row)
|
||
else:
|
||
bucket = rows_by_table.setdefault(table, [])
|
||
cap = lim.get(kind, 1000)
|
||
if len(bucket) < cap:
|
||
bucket.append(row)
|
||
if counts.get(table, 0) and before == 0:
|
||
emit_thinking("抽取业务表", f"{table} → {kind}", pct=min(90, 20 + len(counts) * 5))
|
||
if lines_read - last_emit >= 400_000:
|
||
last_emit = lines_read
|
||
pct = min(90.0, 10 + (bytes_read / max(size, 1)) * 75)
|
||
emit_thinking(
|
||
"扫描 SQL 全文",
|
||
f"已读 {lines_read // 10000 / 10:.1f} 万行 · 订单工艺 {len(prefer_craftl)} 条",
|
||
pct=pct,
|
||
)
|
||
|
||
if routing_pref or routing_other:
|
||
rows_by_table[routing_table] = routing_pref + routing_other
|
||
emit_thinking(
|
||
"工艺优先装载",
|
||
f"订单相关 {len(routing_pref)} 步 · 其它 {len(routing_other)} 步 · 覆盖工艺编码 {len(prefer_craftl)}",
|
||
status="done", pct=91,
|
||
)
|
||
|
||
emit_thinking("SQL 扫描完成", f"命中 {len(counts)} 张相关表", status="done", pct=92)
|
||
return {
|
||
"path": path,
|
||
"size": size,
|
||
"schemas": {k: v for k, v in schemas.items()},
|
||
"counts": counts,
|
||
"rowsByTable": rows_by_table,
|
||
"preferCraftl": sorted(prefer_craftl),
|
||
"kinds": sorted({_TABLE_KIND[t] for t in counts}),
|
||
}
|
||
|
||
|
||
def _map_orders(rows: list[dict]) -> list[dict]:
|
||
out = []
|
||
for r in rows:
|
||
if str(r.get("is_delete") or "0") not in ("0", "0.0", "", "None"):
|
||
continue
|
||
ono = str(r.get("code") or r.get("order_code") or "").strip()
|
||
pc = str(r.get("material_code") or r.get("product_code") or "").strip()
|
||
if not ono or not pc:
|
||
continue
|
||
qty = float(r.get("quantity") or r.get("quatity") or r.get("all_quantity") or 0)
|
||
if qty <= 0:
|
||
qty = 1
|
||
due = _due_str(r.get("planned_end_time") or r.get("end_time") or r.get("actual_end_time"))
|
||
if not due:
|
||
due = datetime.now().strftime("%Y-%m-%d")
|
||
out.append({
|
||
"orderNo": ono,
|
||
"productCode": pc,
|
||
"productName": str(r.get("material_name") or r.get("product_name") or pc),
|
||
"quantity": int(qty) if qty == int(qty) else qty,
|
||
"dueDate": due,
|
||
"priority": int(float(r.get("level") or 5) or 5),
|
||
"customerName": str(r.get("customer_business_code") or r.get("dept") or "松岳现场"),
|
||
"status": "RELEASED",
|
||
"drawingNo": str(r.get("drawing_code") or r.get("figure_no") or ""),
|
||
"machineModel": str(r.get("machine_model") or r.get("model") or ""),
|
||
"craftlCode": str(r.get("craftl_code") or "").strip(),
|
||
})
|
||
return out
|
||
|
||
|
||
def _map_materials(rows: list[dict]) -> list[dict]:
|
||
out = []
|
||
seen: set[str] = set()
|
||
for r in rows:
|
||
if str(r.get("is_delete") or "0") not in ("0", "0.0", "", "None"):
|
||
continue
|
||
code = str(r.get("code") or "").strip()
|
||
name = str(r.get("name") or code).strip()
|
||
if not code or code in seen:
|
||
continue
|
||
seen.add(code)
|
||
raw_type = str(r.get("type") or r.get("classify_name") or "").upper()
|
||
type_cn = str(r.get("type") or "")
|
||
if "毛坯" in name or "原料" in type_cn or raw_type in ("RM", "RAW"):
|
||
mtype = "RAW_MATERIAL"
|
||
elif "半" in type_cn or raw_type in ("SF", "SEMI"):
|
||
mtype = "SEMI_FINISHED"
|
||
elif "成品" in type_cn or raw_type in ("FG", "FINISHED"):
|
||
mtype = "FINISHED_PRODUCT"
|
||
else:
|
||
# 松岳料号习惯：01* 毛坯/原料，其余先按原料，有工艺时再升为半成品
|
||
mtype = "RAW_MATERIAL" if code.startswith("01") else "SEMI_FINISHED"
|
||
sourcing = "BUY" if mtype == "RAW_MATERIAL" or "毛坯" in name else "MAKE"
|
||
out.append({
|
||
"code": code, "name": name, "type": mtype,
|
||
"sourcingType": sourcing,
|
||
"unit": str(r.get("unit") or "件").strip('"') or "件",
|
||
"stock": float(r.get("stock") or 0),
|
||
"inTransit": 0, "safetyStock": 0, "procurementLeadTime": 0,
|
||
"spec": str(r.get("specification_type") or r.get("figure_no") or ""),
|
||
})
|
||
return out
|
||
|
||
|
||
def _map_equipment(rows: list[dict]) -> list[dict]:
|
||
out = []
|
||
seen: set[str] = set()
|
||
for r in rows:
|
||
if str(r.get("is_delete") or r.get("delete_flag") or "0") not in ("0", "0.0", "", "None"):
|
||
continue
|
||
code = str(r.get("code") or r.get("devices_code") or "").strip()
|
||
name = str(r.get("name") or r.get("devices_name") or code).strip()
|
||
if not code or code in seen:
|
||
continue
|
||
seen.add(code)
|
||
out.append({
|
||
"code": code, "name": name,
|
||
"capabilities": ["GENERAL"],
|
||
"opStdTime": {"GENERAL": 1.0},
|
||
"movable": False, "moveTimeMin": 0,
|
||
"zone": "ZONE-A", "adaptableMolds": [],
|
||
"availabilityRate": 0.95, "status": "RUNNING",
|
||
})
|
||
return out
|
||
|
||
|
||
def _map_routing(rows: list[dict], craft_product: dict[str, str]) -> list[dict]:
|
||
"""工艺步骤挂到成品：优先 craftl_code→订单/工艺主数据料号，禁止用中间件料号冒充成品。"""
|
||
out = []
|
||
for r in rows:
|
||
if str(r.get("is_delete") or r.get("del_flag") or "0") not in ("0", "0.0", "", "None"):
|
||
continue
|
||
op = str(r.get("procedure_code") or r.get("process_code") or "").strip()
|
||
if not op:
|
||
continue
|
||
craft = str(r.get("craftl_code") or r.get("process_id") or "").strip()
|
||
out_mat = str(r.get("output_material_code") or r.get("product_code") or "").strip()
|
||
# 成品优先：工艺编码映射 > 输出料号（仅当映射缺失）
|
||
pc = craft_product.get(craft) or out_mat or craft
|
||
if not pc:
|
||
continue
|
||
try:
|
||
seq = int(float(r.get("sort_no") or r.get("process_sort") or r.get("serial_number") or 10))
|
||
except (TypeError, ValueError):
|
||
seq = 10
|
||
if seq <= 0:
|
||
seq = 10
|
||
std = r.get("working_hours") or r.get("process_shift_output")
|
||
try:
|
||
std_f = float(std) if std not in (None, "") else None
|
||
except (TypeError, ValueError):
|
||
std_f = None
|
||
# 工时若像小时则转分钟
|
||
if std_f is not None and std_f > 0 and std_f < 24:
|
||
std_f = round(std_f * 60, 2)
|
||
if std_f is None or std_f <= 0:
|
||
std_f = 1.0
|
||
src = "推断"
|
||
else:
|
||
src = "实测"
|
||
pname = str(r.get("craftl_name") or r.get("output_material_name") or pc)
|
||
op_name = str(r.get("procedure_name") or r.get("process_name") or op)
|
||
from server.aps_domain.sourcing import infer_op_sourcing
|
||
sourcing = infer_op_sourcing(op, op_name)
|
||
out.append({
|
||
"productCode": pc,
|
||
"productName": pname,
|
||
"seq": seq,
|
||
"operationCode": op,
|
||
"operationName": op_name,
|
||
"requireMold": False,
|
||
"stdTimePerUnit": std_f,
|
||
"stdTimeSource": src,
|
||
"craftlCode": craft,
|
||
"outputMaterialCode": out_mat,
|
||
"sourcingType": sourcing,
|
||
"isExternal": sourcing == "OUTSOURCE",
|
||
})
|
||
return out
|
||
|
||
|
||
def _expand_routing_for_orders(
|
||
orders: list[dict],
|
||
steps: list[dict],
|
||
) -> list[dict]:
|
||
"""同一 craftl 下的步骤，按订单成品料号再挂一份（多订单共用工艺）。"""
|
||
by_craft: dict[str, list[dict]] = {}
|
||
for s in steps:
|
||
cc = str(s.get("craftlCode") or "").strip()
|
||
if cc:
|
||
by_craft.setdefault(cc, []).append(s)
|
||
out: list[dict] = []
|
||
seen: set[tuple] = set()
|
||
for o in orders:
|
||
pc = str(o.get("productCode") or "").strip()
|
||
cc = str(o.get("craftlCode") or "").strip()
|
||
if not pc:
|
||
continue
|
||
src = by_craft.get(cc) or [s for s in steps if s.get("productCode") == pc]
|
||
for s in src:
|
||
key = (pc, s.get("seq"), s.get("operationCode"))
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
row = dict(s)
|
||
row["productCode"] = pc
|
||
row["productName"] = o.get("productName") or pc
|
||
out.append(row)
|
||
# 保留未能挂到订单的步骤（其它产品）
|
||
for s in steps:
|
||
key = (s.get("productCode"), s.get("seq"), s.get("operationCode"))
|
||
if key not in seen:
|
||
out.append(s)
|
||
seen.add(key)
|
||
return out
|
||
|
||
|
||
def _map_bom(rows: list[dict], order_product: dict[str, str]) -> list[dict]:
|
||
out = []
|
||
for r in rows:
|
||
if str(r.get("is_delete") or "0") not in ("0", "0.0", "", "None"):
|
||
continue
|
||
ono = str(r.get("order_code") or "").strip()
|
||
mc = str(r.get("material_code") or "").strip()
|
||
pc = order_product.get(ono) or str(r.get("machine_model") or "").strip()
|
||
qty = float(r.get("composition_quantity") or r.get("consumption_quantity") or r.get("quantity") or 0)
|
||
if not pc or not mc or qty <= 0:
|
||
continue
|
||
out.append({
|
||
"productCode": pc, "materialCode": mc, "quantity": qty,
|
||
"consumeOp": str(r.get("procedure_code") or ""),
|
||
"isKey": qty >= 1,
|
||
"materialName": str(r.get("material_name") or mc),
|
||
})
|
||
return out
|
||
|
||
|
||
def _apply_op_equip(equip: list[dict], rows: list[dict]) -> None:
|
||
"""用工序-设备关系补全 capabilities。"""
|
||
by_code = {e["code"]: e for e in equip}
|
||
for r in rows:
|
||
if str(r.get("is_delete") or "0") not in ("0", "0.0", "", "None"):
|
||
continue
|
||
ec = str(r.get("equipment_code") or "").strip()
|
||
op = str(r.get("procedure_code") or "").strip()
|
||
if not ec or not op or ec not in by_code:
|
||
continue
|
||
e = by_code[ec]
|
||
caps = list(e.get("capabilities") or [])
|
||
if "GENERAL" in caps and len(caps) == 1:
|
||
caps = []
|
||
if op not in caps:
|
||
caps.append(op)
|
||
e["capabilities"] = caps or ["GENERAL"]
|
||
std = dict(e.get("opStdTime") or {})
|
||
if op not in std:
|
||
std[op] = 1.0
|
||
if "GENERAL" in std and op != "GENERAL" and len(std) > 1:
|
||
std.pop("GENERAL", None)
|
||
e["opStdTime"] = std
|
||
|
||
|
||
def sql_pack_to_flex(scan: dict[str, Any]) -> dict[str, Any]:
|
||
"""把 scan 结果映射为可写入 world 的 flex* 批次。"""
|
||
rbt = scan.get("rowsByTable") or {}
|
||
craft_product: dict[str, str] = {}
|
||
for r in rbt.get("md_craftl") or []:
|
||
code = str(r.get("code") or "").strip()
|
||
pc = str(r.get("product_code") or "").strip()
|
||
if code and pc:
|
||
craft_product[code] = pc
|
||
|
||
orders = _map_orders((rbt.get("pl_order") or []) + (rbt.get("r_plan_order") or []))
|
||
# 订单上的 craftl→成品料号优先（现场口径，覆盖 md_craftl 偏差）
|
||
for o in orders:
|
||
cc = str(o.get("craftlCode") or "").strip()
|
||
pc = str(o.get("productCode") or "").strip()
|
||
if cc and pc:
|
||
craft_product[cc] = pc
|
||
|
||
order_product = {o["orderNo"]: o["productCode"] for o in orders}
|
||
# 订单产品补进物料
|
||
materials = _map_materials((rbt.get("md_material") or []) + (rbt.get("md_item") or []))
|
||
mat_codes = {m["code"] for m in materials}
|
||
for o in orders:
|
||
if o["productCode"] not in mat_codes:
|
||
materials.append({
|
||
"code": o["productCode"], "name": o.get("productName") or o["productCode"],
|
||
"type": "FINISHED_PRODUCT", "unit": "件",
|
||
"stock": 0, "inTransit": 0, "safetyStock": 0, "procurementLeadTime": 0, "spec": "",
|
||
})
|
||
mat_codes.add(o["productCode"])
|
||
|
||
equipment = _map_equipment((rbt.get("md_equipment") or []) + (rbt.get("em_devices") or []))
|
||
_apply_op_equip(equipment, rbt.get("r_production_procedure") or [])
|
||
routing_raw = _map_routing(
|
||
(rbt.get("r_production_craftl") or []) + (rbt.get("c_project_process_route") or []),
|
||
craft_product,
|
||
)
|
||
routing = _expand_routing_for_orders(orders, routing_raw)
|
||
# 设备能力：把订单工艺工序挂到设备（MES 设备表常无工序能力列）
|
||
ops_needed = sorted({str(s.get("operationCode") or "") for s in routing if s.get("operationCode")})
|
||
if ops_needed and equipment:
|
||
for e in equipment:
|
||
caps = [c for c in (e.get("capabilities") or []) if c and c != "GENERAL"]
|
||
if not caps:
|
||
# MES 设备台账常无工序能力列：按已挂工艺放开，否则齐备度永远卡 NO_CAPABLE
|
||
e["capabilities"] = list(ops_needed)
|
||
e["opStdTime"] = {op: float((e.get("opStdTime") or {}).get(op) or 1.0) for op in ops_needed}
|
||
bom = _map_bom(rbt.get("pl_order_bom") or [], order_product)
|
||
|
||
pcs_ord = {o["productCode"] for o in orders}
|
||
pcs_rt = {s["productCode"] for s in routing}
|
||
return {
|
||
"orders": orders,
|
||
"materials": materials,
|
||
"equipment": equipment,
|
||
"routing": routing,
|
||
"bom": bom,
|
||
"stats": {
|
||
"orders": len(orders), "materials": len(materials),
|
||
"equipment": len(equipment), "routing": len(routing), "bom": len(bom),
|
||
"orderProducts": len(pcs_ord),
|
||
"routedProducts": len(pcs_ord & pcs_rt),
|
||
"missingRouting": len(pcs_ord - pcs_rt),
|
||
"rawCounts": scan.get("counts") or {},
|
||
},
|
||
}
|
||
|
||
|
||
def sync_flex_to_classic_master(world: dict[str, Any]) -> dict[str, int]:
|
||
"""Import compatibility entry point; stable objects, supplied fields only."""
|
||
from server.aps_domain.masterdata_sync import reconcile_imported_masterdata
|
||
summary = reconcile_imported_masterdata(world, source="sql-pack")
|
||
params = world.setdefault("flexParams", {})
|
||
params["demoDataCleared"] = True
|
||
params["siteProfile"] = params.get("siteProfile") or "sql-pack"
|
||
return summary
|
||
|
||
|
||
def apply_sql_pack_to_world(world: dict[str, Any], flex: dict[str, Any], *, replace: bool = True) -> dict[str, int]:
|
||
"""将 SQL 映射结果写入 world flex*，并投影到主数据页经典表。"""
|
||
from server.state.seed import ensure_flex_seed, purge_demo_residue
|
||
# 先清演示残留，再保证键存在（不会回填假数据）
|
||
purge_demo_residue(world)
|
||
world.setdefault("flexParams", {})["demoDataCleared"] = True
|
||
ensure_flex_seed(world)
|
||
summary: dict[str, int] = {}
|
||
|
||
def _next(table: str) -> int:
|
||
items = world.get(table) or []
|
||
return max((x.get("id", 0) for x in items if isinstance(x.get("id"), int)), default=0) + 1
|
||
|
||
if replace:
|
||
for k in ("flexOrders", "flexMaterials", "flexEquipment", "flexRoutings", "flexBom"):
|
||
world[k] = []
|
||
# 清掉演示厂班组，否则现场工序（NZ*/WZ*）会被 NO_TEAM 卡死
|
||
world["flexTeams"] = []
|
||
# 经典主数据也整表换新，禁止智能控制器与现场料号并存
|
||
for k in (
|
||
"materials", "boms", "bomItems", "operations", "routings", "routingSteps",
|
||
"lineProducts", "workstationOperations", "changeoverMatrix",
|
||
"factories", "workshops", "lines", "workstations", "equipment",
|
||
):
|
||
world[k] = []
|
||
|
||
# materials first
|
||
fm = world.setdefault("flexMaterials", [])
|
||
for row in flex.get("materials") or []:
|
||
ex = next((m for m in fm if m.get("code") == row["code"]), None)
|
||
if ex:
|
||
ex.update(row)
|
||
else:
|
||
fm.append({"id": _next("flexMaterials"), **row})
|
||
summary["materials"] = summary.get("materials", 0) + 1
|
||
|
||
fo = world.setdefault("flexOrders", [])
|
||
for row in flex.get("orders") or []:
|
||
ex = next((o for o in fo if o.get("orderNo") == row["orderNo"]), None)
|
||
if ex:
|
||
ex.update(row)
|
||
else:
|
||
fo.append({"id": _next("flexOrders"), **row})
|
||
summary["orders"] = summary.get("orders", 0) + 1
|
||
|
||
fe = world.setdefault("flexEquipment", [])
|
||
for row in flex.get("equipment") or []:
|
||
ex = next((e for e in fe if e.get("code") == row["code"]), None)
|
||
if ex:
|
||
ex.update(row)
|
||
else:
|
||
fe.append({"id": _next("flexEquipment"), **row})
|
||
summary["equipment"] = summary.get("equipment", 0) + 1
|
||
|
||
fr = world.setdefault("flexRoutings", [])
|
||
for row in flex.get("routing") or []:
|
||
fr[:] = [s for s in fr if not (
|
||
s.get("productCode") == row["productCode"] and s.get("seq") == row["seq"]
|
||
and s.get("operationCode") == row["operationCode"]
|
||
)]
|
||
fr.append(dict(row))
|
||
summary["routing"] = summary.get("routing", 0) + 1
|
||
|
||
fb = world.setdefault("flexBom", [])
|
||
for row in flex.get("bom") or []:
|
||
fb[:] = [s for s in fb if not (
|
||
s.get("productCode") == row["productCode"]
|
||
and s.get("materialCode") == row["materialCode"]
|
||
)]
|
||
fb.append(dict(row))
|
||
summary["bom"] = summary.get("bom", 0) + 1
|
||
|
||
# 默认班次
|
||
if not world.get("flexCalendar"):
|
||
world["flexCalendar"] = [{
|
||
"shiftCode": "D", "startTime": "08:00", "endTime": "17:00",
|
||
"breaks": [{"start": "12:00", "end": "13:00"}],
|
||
"workdays": [1, 2, 3, 4, 5],
|
||
}]
|
||
summary["calendar"] = 1
|
||
|
||
# 工序库
|
||
ops = world.setdefault("flexOperations", [])
|
||
seen_op = {o.get("code") for o in ops}
|
||
all_ops: list[str] = []
|
||
for row in flex.get("routing") or []:
|
||
code = row.get("operationCode")
|
||
if code and code not in seen_op:
|
||
ops.append({"code": code, "name": row.get("operationName") or code,
|
||
"isBottleneck": False, "changeoverMin": 10})
|
||
seen_op.add(code)
|
||
summary["operations"] = summary.get("operations", 0) + 1
|
||
if code and code not in all_ops:
|
||
all_ops.append(code)
|
||
|
||
# 现场综合班组：覆盖 SQL 工艺全部工序，避免演示厂班组资格卡死
|
||
if all_ops:
|
||
teams = world.setdefault("flexTeams", [])
|
||
plant = next((t for t in teams if t.get("code") == "T-PLANT"), None)
|
||
if plant is None:
|
||
plant = {
|
||
"code": "T-PLANT", "name": "现场综合班组",
|
||
"memberCount": max(8, min(40, len(world.get("flexEquipment") or []) or 8)),
|
||
"supportOps": [], "skillLevel": "L3",
|
||
}
|
||
teams.append(plant)
|
||
summary["teams"] = 1
|
||
for code in all_ops:
|
||
if code not in plant["supportOps"]:
|
||
plant["supportOps"].append(code)
|
||
|
||
# 投影到主数据「工艺模型」页（经典 materials/routings）
|
||
classic = sync_flex_to_classic_master(world)
|
||
for k, v in classic.items():
|
||
summary[f"classic_{k}"] = v
|
||
|
||
# 投影到订单管理页（salesOrders）
|
||
try:
|
||
from server.aps_domain.orders import sync_flex_orders_to_sales
|
||
n = sync_flex_orders_to_sales(world)
|
||
if n:
|
||
summary["salesOrders"] = n
|
||
except Exception:
|
||
pass
|
||
|
||
# 工序/物料供应类型标注（WZ*=委外，毛坯=采购）
|
||
try:
|
||
from server.aps_domain.sourcing import annotate_world_sourcing
|
||
summary["sourcing"] = annotate_world_sourcing(world)
|
||
except Exception:
|
||
pass
|
||
|
||
return summary
|
||
|
||
|
||
def preview_sql_file(path: str) -> dict[str, Any]:
|
||
"""给 folder_pack / 分析用的轻量预览。"""
|
||
scan = scan_sql_pack(path)
|
||
flex = sql_pack_to_flex(scan)
|
||
st = flex["stats"]
|
||
return {
|
||
"filename": os.path.basename(path),
|
||
"kind": "sql-pack",
|
||
"kindCn": "SQL 数据包",
|
||
"size": scan["size"],
|
||
"okCount": st["orders"] + st["materials"] + st["routing"] + st["equipment"],
|
||
"errorCount": 0,
|
||
"tables": [
|
||
{"table": t, "kind": _TABLE_KIND.get(t, ""), "rawCount": c,
|
||
"loaded": len((scan.get("rowsByTable") or {}).get(t) or [])}
|
||
for t, c in sorted((scan.get("counts") or {}).items())
|
||
],
|
||
"fieldMapText": (
|
||
f"订单{st['orders']} · 物料{st['materials']} · 工艺{st['routing']} · "
|
||
f"设备{st['equipment']} · BOM{st['bom']}"
|
||
),
|
||
"samples": [
|
||
f"订单 {o['orderNo']}：{o['productCode']}×{o['quantity']} 交期{o['dueDate']}"
|
||
for o in (flex.get("orders") or [])[:3]
|
||
],
|
||
"flex": flex,
|
||
"canSchedule": st["orders"] > 0 and (st["routing"] > 0 or st["equipment"] > 0),
|
||
}
|