AiDOPWarehouse/tools/integration-simulator/catalog/sample_adapters.py

"""三通道共用的对象目录、canonical 样例装载与适配器（任务书 P1-A / P1-F）。

- 样例来源：doc/db/mdp/mock_api/samples/*.json（复用不迁移）；无样例对象用 BUILTIN_SAMPLES。
- 所有业务键统一附加运行编号前缀 SIM-{runId}-，不与真实业务键碰撞。
- adapt_inbound_rows 按 MdpInboundFieldMapper.IdentityFields 的必填/可选字段做别名映射，
  缺值字段进 missing_required，由调用方在预览中显示，不得静默造假业务数据。
"""
from __future__ import annotations

import json
import re
import uuid
from datetime import datetime, timezone, timedelta
from pathlib import Path

SIM_ROOT = Path(__file__).resolve().parents[1]
REPO_ROOT = SIM_ROOT.parents[1]
MOCK_API_DIR = REPO_ROOT / "doc" / "db" / "mdp" / "mock_api"
CATALOG_DIR = SIM_ROOT / "catalog"

RUN_ID_RE = re.compile(r"^\d{14}-[0-9a-f]{6}$")
BIZ_KEY_PREFIX = "SIM-"

# 通用字段别名（小写比较）：把既有样例字段归一为 sourceUpdatedAt
_UPDATED_AT_ALIASES = ("updateTime", "sourceUpdatedAt", "inspectStartTime", "WorkDate", "RctDate", "Date", "sxrq", "effDate")


def new_run_id(now: datetime | None = None) -> str:
    """运行编号 SIM-{yyyyMMddHHmmss}-{shortId} 中的时间戳与短 id 部分。"""
    now = now or datetime.now(timezone(timedelta(hours=8)))
    return f"{now.strftime('%Y%m%d%H%M%S')}-{uuid.uuid4().hex[:6]}"


def biz_prefix(run_id: str) -> str:
    return f"{BIZ_KEY_PREFIX}{run_id}-"


def now_iso() -> str:
    return datetime.now(timezone(timedelta(hours=8))).strftime("%Y-%m-%dT%H:%M:%S+08:00")


# ---------------------------------------------------------------------------
# 对象目录
# ---------------------------------------------------------------------------

_CATALOG_CACHE: dict | None = None


def load_catalog() -> dict:
    global _CATALOG_CACHE
    if _CATALOG_CACHE is None:
        with (CATALOG_DIR / "entities.json").open("r", encoding="utf-8") as f:
            _CATALOG_CACHE = json.load(f)
    return _CATALOG_CACHE


def load_scenarios() -> dict:
    with (CATALOG_DIR / "scenarios.json").open("r", encoding="utf-8") as f:
        return json.load(f)


def find_object(object_code: str) -> dict | None:
    code = (object_code or "").strip().upper()
    for obj in load_catalog()["objects"]:
        if obj["objectCode"] == code:
            return obj
    return None


# ---------------------------------------------------------------------------
# canonical 样例
# ---------------------------------------------------------------------------

BUILTIN_SAMPLES: dict[str, list[dict]] = {
    "CUSTOMER": [
        {"bizKey": "CUST-001", "Cust": "CUST-001", "Name": "模拟客户001", "Domain": "SIM", "Status": "1"}
    ],
    "LOCATION": [
        {"bizKey": "LOC-A01", "Location": "SIM-A-01", "Descr": "模拟库位A01", "Domain": "SIM", "Typed": "WH"}
    ],
    "EMPLOYEE_HEADCOUNT": [
        {"bizKey": "EMP-0001", "Domain": "SIM", "Employee": "EMP-0001", "Name": "模拟员工001",
         "Department": "生产部", "Position": "操作工", "EmploymentStatus": "在岗"}
    ],
    "INVENTORY_OPENING_BALANCE": [
        {"bizKey": "OPEN-INV-001", "idid": "SIM-OPEN-INV-001", "code": "RM-CT-001", "sl": 100,
         "slzx": 100, "shyn": "mock", "shtime": "2026-09-01T00:00:00", "date0": "2026-09-01"}
    ],
    "FINISHED_ONHAND": [
        {"bizKey": "FG-ONHAND-001", "code": "FG-CT-001", "sl": 50, "date0": "2026-09-30", "location": "FG-01"}
    ],
    "FINISHED_OPENING_BALANCE": [
        {"bizKey": "FG-OPEN-001", "code": "FG-CT-001", "sl": 20, "date0": "2026-09-01", "location": "FG-01"}
    ],
}


def load_canonical_rows(obj: dict) -> list[dict]:
    """读取对象 canonical 样例；sample 为 null 时使用内置最小样例。"""
    sample = obj.get("sample")
    if not sample:
        rows = BUILTIN_SAMPLES.get(obj["objectCode"])
        if rows is None:
            raise FileNotFoundError(f"object {obj['objectCode']} has no sample and no builtin rows")
        return [dict(r) for r in rows]
    path = REPO_ROOT / sample
    if not path.exists():
        raise FileNotFoundError(f"sample not found: {sample}")
    with path.open("r", encoding="utf-8") as f:
        data = json.load(f)
    if not isinstance(data, list):
        raise ValueError(f"sample must be a JSON array: {sample}")
    return [dict(r) for r in data]


def _extract_updated_at(row: dict) -> str:
    for alias in _UPDATED_AT_ALIASES:
        for key in row:
            if key.lower() == alias.lower() and row[key]:
                return str(row[key])
    return now_iso()


def _prefix_biz_key(row: dict, run_id: str) -> str:
    """给业务键加 SIM-{runId}- 前缀，返回新 bizKey。兼容 bizKey / Id / id 三种载体。"""
    prefix = biz_prefix(run_id)
    for key in ("bizKey", "Id", "id"):
        if key in row and row[key] is not None:
            raw = str(row[key])
            row[key] = raw if raw.startswith(prefix) else prefix + raw
            return str(row[key])
    row["bizKey"] = prefix + uuid.uuid4().hex[:8]
    return row["bizKey"]


# ---------------------------------------------------------------------------
# 通道适配器
# ---------------------------------------------------------------------------

def adapt_db_rows(obj: dict, run_id: str, rows: list[dict] | None = None) -> list[dict]:
    """DB_SYNC：源表行 = canonical 行 + SIM 前缀业务键 + sourceUpdatedAt 增量列。"""
    rows = [dict(r) for r in (rows if rows is not None else load_canonical_rows(obj))]
    out = []
    for row in rows:
        _prefix_biz_key(row, run_id)
        row.setdefault("Domain", "SIM")
        row["sourceUpdatedAt"] = _extract_updated_at(row)
        out.append(row)
    return out


def adapt_api_rows(obj: dict, run_id: str, rows: list[dict] | None = None) -> list[dict]:
    """API_PULL：与 DB_SYNC 同语义，另保证 bizKey 存在且按 bizKey 排序（cursor 语义）。"""
    rows = adapt_db_rows(obj, run_id, rows)
    for row in rows:
        row["bizKey"] = str(row.get("bizKey") or row.get("Id") or row.get("id"))
    rows.sort(key=lambda r: str(r["bizKey"]))
    return rows


# entityCode -> 契约字段（与 MdpInboundFieldMapper.IdentityFields 一致）；("字段", 必填, 别名列表)
# 版本语义：C# 端只有显式 v2 才使用 IdentityFields（中立必填集合）；v1 特化仅存在于
# S5_INVENTORY_TXN 与 MDM_EMPLOYEE_HEADCOUNT。第一阶段目录统一登记 contractVersion=v2，
# 因此本表按 IdentityFields 生成样例，避免“按 v1 造数、按 v2 校验”的口径错配。
INBOUND_SPECS: dict[str, list[tuple[str, bool, list[str]]]] = {
    "MDM_ITEM": [("ItemNum", True, ["itemCode", "item_number", "number", "ItemNum", "wlbm"]),
                 ("Descr", True, ["itemName", "descr", "wlmc"]),
                 ("Drawing", False, []), ("UM", False, ["uom"]), ("ItemType", False, []),
                 ("Status", False, ["status"]), ("Domain", False, ["Domain"])],
    "MDM_CUSTOMER": [("Cust", True, ["customerNo", "Cust"]), ("Name", True, ["customerName", "Name"]),
                     ("Domain", False, ["Domain"]), ("Status", False, ["status"])],
    "MDM_SUPPLIER": [("Supp", True, ["supplierCode", "Supp", "supplier_number"]),
                     ("Name", True, ["supplierName", "Name"]),
                     ("Domain", False, ["Domain"]), ("Status", False, ["status"])],
    "MDM_LOCATION": [("Location", True, ["Location", "location"]), ("Descr", True, ["Descr", "descr", "Address"]),
                     ("Domain", False, ["Domain"]), ("Typed", False, [])],
    "MDM_SOURCE_LIST": [("Supp", True, ["supplier_number", "Supp", "supplierCode"]),
                        ("ItemNum", True, ["number", "itemCode", "ItemNum"]),
                        ("Domain", False, ["Domain"]), ("Status", False, ["status"])],
    "MDM_EMPLOYEE_HEADCOUNT": [("Domain", True, ["Domain"]), ("Employee", True, ["Employee"]),
                               ("Position", True, ["Position"]), ("EmploymentStatus", True, ["EmploymentStatus"]),
                               ("Name", False, ["Name"]), ("Department", False, ["Department"])],
    "S1_SALES_ORDER_ENTRY": [("bill_no", True, ["bill_no", "orderNo"]), ("entry_seq", True, ["entry_seq", "Line"]),
                             ("seorder_id", True, ["seorder_id", "id", "Id"]), ("qty", True, ["qty", "Qty"]),
                             ("item_number", False, ["item_code", "itemCode", "item_number", "ItemNum"]),
                             ("item_name", False, ["itemName"]), ("plan_date", False, ["plan_date"]),
                             ("date", False, ["date"]), ("progress", False, []), ("urgent", False, [])],
    "S1_REQUIREMENT_EXAMINE_RESULT": [("Id", True, ["Id", "id"]), ("bill_no", True, ["bill_no"]),
                                      ("morder_no", True, ["morder_no"]), ("sentry_id", False, ["sentry_id"]),
                                      ("create_time", False, ["create_time"]), ("IsDeleted", False, [])],
    "S1_REQUIREMENT_EXAMINE_DETAIL": [("Id", True, ["Id", "id"]), ("examine_id", True, ["examine_id"]),
                                      ("item_number", True, ["item_number"]), ("lack_qty", True, ["lack_qty", "qty"]),
                                      ("level", True, ["level"]), ("num", False, ["num"]),
                                      ("item_name", False, ["item_name"]), ("needCount", False, []),
                                      ("qty", False, ["qty"]), ("is_use", False, [])],
    "S2_WORK_ORDER_SCHEDULE": [("WorkOrder", True, ["WorkOrd", "work_order", "workOrderNo", "OrderNo"]),
                               ("ItemCode", True, ["ItemNum", "itemCode", "item_number", "componentItem"]),
                               ("QtyOrdered", True, ["Qty", "qty", "QtyReq", "sl"]),
                               ("DocType", False, []), ("DueDate", False, ["WorkDate", "DueDate"]),
                               ("QtyCompleted", False, []), ("Status", False, ["Status", "status", "ztid"]),
                               ("ProdLine", False, ["ProdLine"])],
    "S3_PURCHASE_ORDER": [("PoNo", True, ["poNo", "PurOrd", "OrdNbr"]), ("PoLine", True, ["poLine", "OrdLine", "Line"]),
                          ("ItemCode", True, ["itemCode", "ItemNum"]), ("OrderQty", True, ["qty", "Qty", "QtyOrded"]),
                          ("SupplierCode", False, ["supplierCode", "Supp"]), ("DueDate", False, []),
                          ("OrderDate", False, []), ("Status", False, ["status", "Status"])],
    "S3_PURCHASE_RECEIPT": [("Domain", True, ["Domain"]), ("Receiver", True, ["Receiver", "RctNbr"]),
                            ("Line", True, ["Line", "OrdLine"]), ("ItemCode", True, ["ItemNum", "itemCode"]),
                            ("QtyReceived", True, ["QtyReceived", "qty"]), ("PoNo", False, ["OrdNbr", "poNo"]),
                            ("SupplierCode", False, ["Supp", "supplierCode"]), ("ReceiptDate", False, ["RctDate"])],
    "S4_SHIPMENT": [("ShipmentNo", True, ["shipmentNo", "shddh"]), ("PoNo", True, ["poNo"]),
                    ("ItemCode", True, ["itemCode"]), ("ShipQty", True, ["qty", "Qty"]),
                    ("PoLine", False, ["poLine"]), ("SupplierCode", False, ["supplierCode"]),
                    ("ShipDate", False, ["ShipDate"])],
    "S4_IQC": [("PoNo", True, ["poNo"]), ("PoLine", True, ["poLine"]), ("ItemCode", True, ["itemCode"]),
               ("ReceiptQty", True, ["qty", "ReceiptQty"]), ("SupplierCode", False, ["supplierCode"]),
               ("DefectQty", False, []), ("QcResult", False, ["result", "QcResult"]),
               ("ReceiptDate", False, ["ReceiptDate"])],
    "S4_RETURN": [("PoNo", True, ["poNo", "lydjbh"]), ("PoLine", True, ["poLine", "bbh"]),
                  ("ItemCode", True, ["item_code", "wlbm"]), ("ReturnQty", True, ["qty", "ybl"]),
                  ("SupplierCode", False, ["supplierCode"]), ("ReturnReason", False, ["bz"]),
                  ("ReturnStatus", False, ["status"])],
    "S4_SHORTAGE": [("WorkOrder", True, ["work_order", "WorkOrd"]),
                    ("ItemCode", True, ["component_item_code", "ItemNum"]),
                    ("ShortageQty", True, ["shortage_qty", "lack_qty"]), ("SupplierCode", False, []),
                    ("RiskLevel", False, []), ("NeedDate", False, [])],
    "S5_WORK_ORDER_BOM": [("Domain", True, ["Domain"]), ("OrderNo", True, ["orderNo", "WorkOrd"]),
                          ("ItemCode", True, ["componentItem", "ItemNum", "itemCode"]),
                          ("QtyRequired", True, ["qtyPer", "QtyReq", "qty"]),
                          ("LineNo", False, ["Line"]), ("Unit", False, ["UM", "uom"])],
    "S5_INVENTORY_TXN": [("Id", True, ["Id", "id", "bizKey"]), ("BizDocType", True, ["bizDocType", "transType"]),
                         ("ItemCode", False, ["itemNum", "code"]), ("Qty", False, ["qtyChange", "sl"]),
                         ("ApprovedFlag", False, ["approvedFlag"]), ("VoidFlag", False, ["voidFlag"]),
                         ("SummaryFlag", False, ["summaryFlag"]), ("DocDate", False, ["effDate", "jhdate"]),
                         ("TransTime", False, ["addtime"]), ("Domain", False, ["Domain"]),
                         ("TransType", False, ["transType"]), ("Location", False, ["location"]),
                         ("LotSerial", False, ["lotSerial"]), ("CreateUser", False, ["createUser"]),
                         ("EndBalance", False, ["endBalance"]), ("BeginBalance", False, []),
                         ("DocQty", False, []), ("LineClosedFlag", False, [])],
    "S5_INVENTORY_OPENING_BALANCE": [("Id", True, ["Id", "idid", "bizKey"]), ("idid", False, ["idid"]),
                                     ("code", False, ["code", "ItemNum"]), ("sl", False, ["sl"]),
                                     ("slzx", False, ["slzx"]), ("shyn", False, ["shyn"]),
                                     ("shtime", False, ["shtime"]), ("date0", False, ["date0"])],
    "S5_INVENTORY_BALANCE_MONTHLY": [("PeriodYm", True, ["periodYm"]),
                                     ("AvgBalanceAmount", True, ["avgBalanceAmount"]),
                                     ("CategoryCode", False, []), ("WarehouseCode", False, ["warehouseCode"]),
                                     ("ItemCode", False, ["itemCode"]), ("IssueCostAmount", False, ["issueCostAmount"])],
    "S6_WORK_ORDER_LINE": [("Domain", True, ["Domain"]), ("OrderNo", True, ["WorkOrd", "orderNo"]),
                           ("ItemCode", True, ["ItemNum", "itemCode"]), ("QtyPlanned", True, ["QtyReq", "qty", "Qty"]),
                           ("QtyCompleted", False, []), ("PlanFinishDate", False, ["WorkDate"]),
                           ("ReleaseTime", False, []), ("ClosedFlag", False, []), ("VoidFlag", False, [])],
    "S6_REPORT_TXN": [("Domain", True, ["Domain"]), ("ReportId", True, ["Id", "noid", "ReportId"]),
                      ("WorkOrderNo", True, ["noid", "WorkOrd", "workOrderNo"]),
                      ("ReportDate", True, ["kgdate", "ReportDate"]), ("ReportQty", True, ["sl", "qty", "ReportQty"]),
                      ("StartWorkDate", False, ["kgdate"])],
    "S7_FQC_TASK_TXN": [("BillNo", True, ["billNo", "BillNo"]),
                        ("ProductionOrderNo", True, ["productionOrderNo", "scph"]),
                        ("MaterialCode", True, ["materialCode", "wlbm"]), ("Qty", True, ["qty"]),
                        ("SalesOrderNo", False, ["salesOrderNo"]), ("Domain", False, ["Domain"])],
    "S7_SALES_ORDER_LINE": [("Domain", True, ["Domain"]), ("OrderNo", True, ["bill_no", "orderNo", "OrdNbr"]),
                            ("LineNo", True, ["entry_seq", "Line"]), ("ItemCode", True, ["item_code", "itemCode", "ItemNum"]),
                            ("QtyPlanned", True, ["qty", "Qty"]), ("PlanFinishDate", True, ["plan_date", "WorkDate"]),
                            ("QtyCompleted", False, []), ("ReleaseTime", False, []),
                            ("ClosedFlag", False, []), ("VoidFlag", False, [])],
    "S7_FINISHED_ONHAND": [("Id", True, ["Id", "bizKey"]), ("code", False, ["code", "ItemNum"]),
                           ("sl", False, ["sl", "QtyRec"]), ("date0", False, ["date0", "Date"]),
                           ("location", False, ["location", "LocationTo"])],
    "S7_FINISHED_OPENING_BALANCE": [("Id", True, ["Id", "bizKey"]), ("code", False, ["code"]),
                                    ("sl", False, ["sl"]), ("date0", False, ["date0"]),
                                    ("location", False, ["location"])],
}


def _pick(row_lower: dict[str, object], aliases: list[str]) -> object | None:
    for alias in aliases:
        if alias.lower() in row_lower and row_lower[alias.lower()] not in (None, ""):
            return row_lower[alias.lower()]
    return None


def adapt_inbound_rows(obj: dict, run_id: str, rows: list[dict] | None = None,
                       domain: str = "SIM") -> dict:
    """API_INBOUND：canonical 行 → 契约字段行。返回 rows / missing_required / biz_keys。"""
    inbound = obj.get("apiInbound") or {}
    entity_code = inbound.get("entityCode")
    if not inbound.get("supported") or not entity_code:
        raise ValueError(f"object {obj['objectCode']} does not support API_INBOUND")
    spec = INBOUND_SPECS.get(entity_code)
    if spec is None:
        raise ValueError(f"no inbound spec for {entity_code}")

    rows = [dict(r) for r in (rows if rows is not None else load_canonical_rows(obj))]
    out: list[dict] = []
    missing: set[str] = set()
    biz_keys: list[str] = []
    for row in rows:
        biz_key = _prefix_biz_key(row, run_id)
        biz_keys.append(biz_key)
        row_lower = {str(k).lower(): v for k, v in row.items()}
        mapped: dict[str, object] = {}
        for field, required, aliases in spec:
            value = _pick(row_lower, [field, *aliases])
            if value is None and field == "Domain":
                value = domain
            if value is None and required:
                missing.add(field)
                continue
            if value is not None:
                mapped[field] = value
        # Id 类必填键允许退化为 SIM 业务键
        if "Id" in {f for f, r, _ in spec if r} and "Id" not in mapped:
            mapped["Id"] = biz_key
        mapped["sourceUpdatedAt"] = _extract_updated_at(row)
        out.append(mapped)
    return {"rows": out, "missing_required": sorted(missing), "biz_keys": biz_keys}


def inbound_envelope(rows: list[dict], snapshot_id: str | None = None, seq: int | None = None) -> dict:
    """推数信封，与 push_demo.py / MdpInboundReceiveService 契约一致。"""
    data: dict = {"list": rows}
    if snapshot_id:
        data["snapshotId"] = snapshot_id
    if seq is not None:
        data["seq"] = seq
    return {"data": data}