129 lines
4.7 KiB
Python
129 lines
4.7 KiB
Python
# ============================================================
|
||
# SAP 集成桩（moduleId: integ-sap-stub, 可重生 ✅）
|
||
# MD-05 最小切片：进程内 Mock SAP，无真实 RFC/OData。
|
||
# 入站：生产订单 / 库存；出站：开完工、产量、异常（幂等键防重复）。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from datetime import datetime, timedelta
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
_MIRROR_PATH = Path(__file__).resolve().parents[1] / "data" / "sap_mirror.json"
|
||
|
||
|
||
def _today() -> datetime:
|
||
return datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
|
||
|
||
|
||
def default_mirror() -> dict[str, Any]:
|
||
"""演示用 SAP 镜像：含待同步生产订单与库存。"""
|
||
base = _today()
|
||
due1 = (base + timedelta(days=7)).strftime("%Y-%m-%d")
|
||
due2 = (base + timedelta(days=5)).strftime("%Y-%m-%d")
|
||
return {
|
||
"system": "SAP-MOCK",
|
||
"plant": "CNWH",
|
||
"updatedAt": datetime.now().strftime("%Y-%m-%d %H:%M"),
|
||
"productionOrders": [
|
||
{"aufnr": "1000123456", "matnr": "HV-HARNESS", "maktx": "高压线束总成",
|
||
"gamng": 120, "gstrp": due1, "priority": 3, "status": "REL",
|
||
"kitOk": True},
|
||
{"aufnr": "1000123457", "matnr": "PDU-UNIT", "maktx": "PDU配电单元",
|
||
"gamng": 80, "gstrp": due2, "priority": 2, "status": "REL",
|
||
"kitOk": False},
|
||
{"aufnr": "1000123458", "matnr": "CHARGE-GUN", "maktx": "充电枪总成",
|
||
"gamng": 60, "gstrp": due1, "priority": 4, "status": "REL",
|
||
"kitOk": True},
|
||
],
|
||
"stocks": [
|
||
{"matnr": "WIRE-HV", "labst": 15000, "inTransit": 0},
|
||
{"matnr": "TERM-HV", "labst": 8000, "inTransit": 500},
|
||
{"matnr": "BUSBAR", "labst": 2000, "inTransit": 0},
|
||
],
|
||
"receipts": [], # APS→SAP 回写流水
|
||
"idempotency": {}, # key → receipt id
|
||
}
|
||
|
||
|
||
class MockSapClient:
|
||
"""进程内 Mock SAP：读写本地镜像文件，模拟外部系统。"""
|
||
|
||
def __init__(self, path: Path | None = None):
|
||
self.path = path or _MIRROR_PATH
|
||
self.path.parent.mkdir(parents=True, exist_ok=True)
|
||
if not self.path.exists():
|
||
self._save(default_mirror())
|
||
|
||
def _load(self) -> dict:
|
||
try:
|
||
return json.loads(self.path.read_text(encoding="utf-8"))
|
||
except (OSError, json.JSONDecodeError):
|
||
data = default_mirror()
|
||
self._save(data)
|
||
return data
|
||
|
||
def _save(self, data: dict) -> None:
|
||
data["updatedAt"] = datetime.now().strftime("%Y-%m-%d %H:%M")
|
||
self.path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
||
def status(self) -> dict:
|
||
m = self._load()
|
||
return {
|
||
"connected": True,
|
||
"mode": "stub",
|
||
"system": m.get("system"),
|
||
"plant": m.get("plant"),
|
||
"updatedAt": m.get("updatedAt"),
|
||
"openOrders": len(m.get("productionOrders") or []),
|
||
"receiptCount": len(m.get("receipts") or []),
|
||
}
|
||
|
||
def pull_orders(self) -> list[dict]:
|
||
return list(self._load().get("productionOrders") or [])
|
||
|
||
def pull_stocks(self) -> list[dict]:
|
||
return list(self._load().get("stocks") or [])
|
||
|
||
def push_receipt(self, payload: dict, idem_key: str) -> dict:
|
||
"""幂等回写：相同 idem_key 返回原回执，不重复建单。"""
|
||
m = self._load()
|
||
existing = (m.get("idempotency") or {}).get(idem_key)
|
||
if existing:
|
||
receipt = next((r for r in m.get("receipts", []) if r["id"] == existing), None)
|
||
return {"duplicate": True, "receipt": receipt or {"id": existing, "idemKey": idem_key}}
|
||
rid = f"SAP-RCPT-{len(m.get('receipts') or []) + 1:04d}"
|
||
receipt = {
|
||
"id": rid, "idemKey": idem_key, "plant": m.get("plant"),
|
||
"at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
**payload,
|
||
}
|
||
m.setdefault("receipts", []).append(receipt)
|
||
m.setdefault("idempotency", {})[idem_key] = rid
|
||
self._save(m)
|
||
return {"duplicate": False, "receipt": receipt}
|
||
|
||
def reset(self) -> dict:
|
||
data = default_mirror()
|
||
self._save(data)
|
||
return self.status()
|
||
|
||
|
||
_client: MockSapClient | None = None
|
||
|
||
|
||
def get_sap_client() -> MockSapClient:
|
||
global _client
|
||
if _client is None:
|
||
_client = MockSapClient()
|
||
return _client
|
||
|
||
|
||
def reset_sap_client(path: Path | None = None) -> MockSapClient:
|
||
"""测试/演示：重置单例（可选临时镜像路径）。"""
|
||
global _client
|
||
_client = MockSapClient(path)
|
||
_client.reset()
|
||
return _client
|