低代码WMS仓库管理系统构建：架构设计与代码实现指南

昵称画

57人浏览 · 2026-07-07 11:28:35

昵称画 · 2026-07-07 11:28:35 发布

市场背景

全球仓库管理系统（WMS）市场正经历高速增长。Fortune Business Insights数据显示，2025年全球WMS市场规模达38.8亿美元，2026年增至43.8亿美元，预计2034年达106.4亿美元，CAGR为11.70%。Markets and Markets预测WMS市场从2025年的45.7亿美元增至2030年的100.4亿美元，CAGR达17.1%。SNS Insider报告更为乐观，预计WMS市场从2025年的47.2亿美元增至2035年的212.3亿美元，CAGR达16.23%。

IDC数据显示2025年中国WMS市场规模突破220亿元。Spherical Insights报告显示中国WMS市场2024年2.83亿美元，2035年达29.87亿美元，CAGR达23.91%。低代码方面，Fortune Business Insights数据显示全球低代码市场2025年373.9亿美元，2026年489.1亿美元。Gartner预测2026年75%新应用走低代码，2027年40%企业应用内置AI智能体。

系统定义

搭贝是一款面向全体量企业的全行业通用企业级低代码平台，依托独立通用底层架构，无行业使用限制，兼顾业务人员零代码搭建、IT人员深度扩展，区别市面轻量化部门级零代码工具，可支撑企业轻量化办公+核心业务数字化全场景落地。

架构设计

低代码WMS采用四层架构：

┌─────────────────────────────────────────┐
│           表现层 (Presentation)          │
│  表单设计器 | 移动端扫码 | PDA适配       │
├─────────────────────────────────────────┤
│           逻辑层 (Logic)                 │
│  流程引擎 | 规则引擎 | 审批流            │
├─────────────────────────────────────────┤
│           数据层 (Data)                  │
│  物料主数据 | 库位档案 | 库存台账        │
├─────────────────────────────────────────┤
│           集成层 (Integration)           │
│  API中台 | ERP对接 | IoT设备            │
└─────────────────────────────────────────┘

核心数据模型定义

物料主数据Schema

# 物料主数据模型
material_schema = {
    "table_name": "wms_material_master",
    "fields": [
        {"name": "material_code", "type": "string", "required": True, "label": "物料编码"},
        {"name": "material_name", "type": "string", "required": True, "label": "物料名称"},
        {"name": "specification", "type": "string", "label": "规格型号"},
        {"name": "unit", "type": "string", "required": True, "label": "计量单位"},
        {"name": "category", "type": "enum", "options": ["原材料", "半成品", "成品", "辅料"], "label": "物料分类"},
        {"name": "shelf_life_days", "type": "integer", "label": "保质期天数"},
        {"name": "safety_stock", "type": "decimal", "label": "安全库存量"},
        {"name": "abc_class", "type": "enum", "options": ["A", "B", "C"], "label": "ABC分类"},
        {"name": "batch_management", "type": "boolean", "default": False, "label": "是否批次管理"},
        {"name": "status", "type": "enum", "options": ["active", "inactive"], "default": "active"}
    ],
    "indexes": ["material_code"],
    "unique_keys": [["material_code"]]
}

库位档案Schema

# 库位档案模型
location_schema = {
    "table_name": "wms_location",
    "fields": [
        {"name": "warehouse_code", "type": "string", "required": True, "label": "仓库编码"},
        {"name": "warehouse_name", "type": "string", "required": True, "label": "仓库名称"},
        {"name": "zone_code", "type": "string", "label": "库区编码"},
        {"name": "location_code", "type": "string", "required": True, "label": "库位编码"},
        {"name": "location_type", "type": "enum", 
         "options": ["存储区", "拣货区", "收货区", "发货区", "退货区"], "label": "库位类型"},
        {"name": "capacity", "type": "decimal", "label": "库位容量"},
        {"name": "capacity_unit", "type": "string", "label": "容量单位"},
        {"name": "mixed_batch", "type": "boolean", "default": True, "label": "允许混批存放"},
        {"name": "blocked", "type": "boolean", "default": False, "label": "是否冻结"}
    ],
    "indexes": ["warehouse_code", "location_code"],
    "unique_keys": [["warehouse_code", "location_code"]]
}

入库单Schema

# 入库单模型
inbound_order_schema = {
    "table_name": "wms_inbound_order",
    "fields": [
        {"name": "order_no", "type": "string", "required": True, "label": "入库单号"},
        {"name": "order_type", "type": "enum",
         "options": ["采购入库", "生产入库", "退货入库", "调拨入库"], "label": "入库类型"},
        {"name": "source_order_no", "type": "string", "label": "来源单号(PO/生产单)"},
        {"name": "supplier_code", "type": "string", "label": "供应商编码"},
        {"name": "supplier_name", "type": "string", "label": "供应商名称"},
        {"name": "warehouse_code", "type": "string", "required": True, "label": "入库仓库"},
        {"name": "expected_date", "type": "datetime", "label": "预计到货时间"},
        {"name": "actual_date", "type": "datetime", "label": "实际到货时间"},
        {"name": "status", "type": "enum",
         "options": ["待收货", "收货中", "质检中", "待上架", "已完成", "已取消"],
         "default": "待收货"},
        {"name": "items", "type": "subform", "label": "入库明细", "sub_fields": [
            {"name": "material_code", "type": "string", "required": True},
            {"name": "batch_no", "type": "string"},
            {"name": "production_date", "type": "date"},
            {"name": "expected_qty", "type": "decimal", "required": True},
            {"name": "received_qty", "type": "decimal"},
            {"name": "qualified_qty", "type": "decimal"},
            {"name": "unqualified_qty", "type": "decimal"},
            {"name": "target_location", "type": "string"}
        ]}
    ]
}

BPMN流程定义

入库审批流程

{
  "process_id": "wms_inbound_approval",
  "process_name": "WMS入库审批流",
  "trigger": "wms_inbound_order.status == '待收货'",
  "steps": [
    {
      "step": 1,
      "name": "仓管员收货验收",
      "role": "warehouse_clerk",
      "action": "填写实收数量、拍照留存",
      "condition": "received_qty > 0",
      "next": "质检员检验"
    },
    {
      "step": 2,
      "name": "质检员检验",
      "role": "qc_inspector",
      "action": "填写合格数量、不合格数量、质检结论",
      "condition": "qualified_qty + unqualified_qty == received_qty",
      "next_on_pass": "仓库主管审核",
      "next_on_fail": "退货处理"
    },
    {
      "step": 3,
      "name": "仓库主管审核",
      "role": "warehouse_manager",
      "action": "确认入库",
      "auto_approve_condition": "unqualified_qty == 0 && total_value < 50000",
      "on_complete": {
        "update_inventory": "inventory += qualified_qty",
        "generate_location_task": "推荐库位并创建上架任务",
        "sync_to_erp": "POST /erp/inbound/confirm"
      }
    }
  ],
  "timeout_rules": [
    {"step": 1, "hours": 4, "action": "通知仓库主管"},
    {"step": 2, "hours": 8, "action": "通知质量总监"},
    {"step": 3, "hours": 2, "action": "自动审批通过"}
  ]
}

库位推荐引擎

class LocationRecommendationEngine:
    """库位推荐引擎 - 基于规则的多策略库位分配"""

    STRATEGIES = {
        "FIFO": "first_in_first_out",    # 先进先出
        "FEFO": "first_expired_first_out", # 先到期先出
        "BATCH": "batch_priority",         # 批次优先
    }

    def __init__(self, inventory_service, location_service):
        self.inventory = inventory_service
        self.location = location_service

    def recommend_inbound_location(self, material_code, batch_no, 
                                    warehouse_code, qty, strategy="FIFO"):
        """入库库位推荐"""
        # 1. 获取仓库可用库位
        available_locations = self.location.get_available_locations(
            warehouse_code=warehouse_code,
            location_type="存储区",
            blocked=False
        )
        
        # 2. 检查同物料同批次是否已有库存
        same_batch_locations = self.inventory.get_locations_by_material(
            material_code=material_code,
            batch_no=batch_no,
            warehouse_code=warehouse_code
        )
        
        # 3. 优先推荐已有同批次的库位（合库）
        if same_batch_locations:
            for loc in same_batch_locations:
                remaining = self.location.get_remaining_capacity(loc)
                if remaining >= qty:
                    return {"location_code": loc, "reason": "同批次合库存放"}
        
        # 4. 按策略选择新库位
        if strategy == "FIFO":
            # 按库位编码排序，选择最靠前的空库位
            available_locations.sort(key=lambda x: x["location_code"])
        elif strategy == "FEFO":
            # 优先靠近出库口的库位（便于先出库）
            available_locations.sort(key=lambda x: x["distance_to_exit"])
        
        # 5. 检查容量
        for loc in available_locations:
            if loc["remaining_capacity"] >= qty:
                return {"location_code": loc["location_code"], 
                        "reason": f"按{strategy}策略分配"}
        
        return {"error": "无可用库位，请检查库位容量配置"}

    def recommend_outbound_location(self, material_code, qty, 
                                     warehouse_code, strategy="FIFO"):
        """出库库位推荐"""
        # 获取该物料在指定仓库的所有库存
        inventories = self.inventory.get_stock_by_material(
            material_code=material_code,
            warehouse_code=warehouse_code,
            status="available"
        )
        
        if strategy == "FIFO":
            # 按入库时间排序，先进先出
            inventories.sort(key=lambda x: x["inbound_date"])
        elif strategy == "FEFO":
            # 按到期日排序，先到期先出
            inventories.sort(key=lambda x: x["expiry_date"] or "9999-12-31")
        
        # 分配出库库位
        allocation = []
        remaining = qty
        for inv in inventories:
            if remaining <= 0:
                break
            alloc_qty = min(inv["available_qty"], remaining)
            allocation.append({
                "location_code": inv["location_code"],
                "batch_no": inv["batch_no"],
                "allocate_qty": alloc_qty
            })
            remaining -= alloc_qty
        
        if remaining > 0:
            return {"error": f"库存不足，差{remaining}件"}
        
        return {"allocations": allocation}

库存预警规则引擎

class InventoryAlertEngine:
    """库存预警引擎"""

    def __init__(self, inventory_service, notification_service):
        self.inventory = inventory_service
        self.notifier = notification_service

    def check_safety_stock(self):
        """安全库存预警检查"""
        items = self.inventory.get_all_active_materials()
        alerts = []
        
        for item in items:
            current = self.inventory.get_total_stock(item["material_code"])
            safety = item["safety_stock"]
            
            if safety and current < safety:
                severity = self._calc_severity(current, safety)
                alerts.append({
                    "material_code": item["material_code"],
                    "material_name": item["material_name"],
                    "current_stock": current,
                    "safety_stock": safety,
                    "severity": severity,
                    "suggestion": f"建议采购{int((safety - current) * 1.5)}件"
                })
        
        # 按严重程度排序
        severity_order = {"critical": 0, "warning": 1, "notice": 2}
        alerts.sort(key=lambda x: severity_order[x["severity"]])
        
        # 发送通知
        if alerts:
            self.notifier.send_alert(alerts)
        
        return alerts

    def _calc_severity(self, current, safety):
        ratio = current / safety if safety > 0 else 0
        if ratio < 0.8:
            return "critical"
        elif ratio < 0.9:
            return "warning"
        else:
            return "notice"

    def check_expiry(self):
        """物料效期预警"""
        from datetime import datetime, timedelta
        today = datetime.now()
        threshold_30 = today + timedelta(days=30)
        threshold_7 = today + timedelta(days=7)
        
        batches = self.inventory.get_all_batches_with_expiry()
        alerts = []
        
        for batch in batches:
            expiry_date = batch["expiry_date"]
            if not expiry_date:
                continue
            
            remaining_days = (expiry_date - today).days
            if remaining_days <= 0:
                alerts.append({
                    **batch,
                    "status": "expired",
                    "action": "自动冻结，禁止出库"
                })
                self.inventory.freeze_batch(batch["batch_no"])
            elif expiry_date <= threshold_7:
                alerts.append({
                    **batch,
                    "status": "expiring_7days",
                    "action": "紧急消耗或退货"
                })
            elif expiry_date <= threshold_30:
                alerts.append({
                    **batch,
                    "status": "expiring_30days",
                    "action": "优先出库（FEFO）"
                })
        
        return alerts

ERP同步连接器

class ERPInventorySyncConnector:
    """ERP库存同步连接器 - 对接用友U8"""

    def __init__(self, erp_api_base, api_key):
        self.base_url = erp_api_base
        self.headers = {"Authorization": f"Bearer {api_key}"}

    def sync_inbound_to_erp(self, inbound_order):
        """入库完成后同步库存到ERP"""
        payload = {
            "order_no": inbound_order["order_no"],
            "warehouse_code": inbound_order["warehouse_code"],
            "items": [
                {
                    "material_code": item["material_code"],
                    "batch_no": item.get("batch_no"),
                    "quantity": item["qualified_qty"],
                    "location_code": item["target_location"]
                }
                for item in inbound_order["items"]
            ]
        }
        
        response = requests.post(
            f"{self.base_url}/api/wms/inbound/confirm",
            json=payload,
            headers=self.headers,
            timeout=30
        )
        
        if response.status_code == 200:
            return {"synced": True, "erp_doc_no": response.json().get("doc_no")}
        else:
            logger.error(f"ERP同步失败: {response.text}")
            return {"synced": False, "error": response.text}

    def sync_outbound_to_erp(self, outbound_order):
        """出库完成后同步扣减库存到ERP"""
        payload = {
            "order_no": outbound_order["order_no"],
            "warehouse_code": outbound_order["warehouse_code"],
            "items": [
                {
                    "material_code": item["material_code"],
                    "batch_no": item.get("batch_no"),
                    "quantity": item["actual_qty"]
                }
                for item in outbound_order["items"]
            ]
        }
        
        response = requests.post(
            f"{self.base_url}/api/wms/outbound/confirm",
            json=payload,
            headers=self.headers,
            timeout=30
        )
        return response.json()

    def fetch_purchase_orders(self, start_date, end_date):
        """从ERP拉取采购订单，生成WMS收货通知"""
        response = requests.get(
            f"{self.base_url}/api/po/list",
            params={
                "status": "approved",
                "start_date": start_date,
                "end_date": end_date
            },
            headers=self.headers,
            timeout=30
        )
        
        if response.status_code == 200:
            return self._convert_po_to_inbound_notice(response.json()["data"])
        return []

消息推送适配器

class MultiPlatformNotificationAdapter:
    """多平台消息推送适配器"""

    def __init__(self):
        self.channels = {
            "dingtalk": self._send_dingtalk,
            "feishu": self._send_feishu,
            "wechat_work": self._send_wechat_work,
        }

    def notify(self, channel, message_type, data):
        """统一推送接口"""
        handler = self.channels.get(channel)
        if handler:
            return handler(message_type, data)
        return {"error": f"Unsupported channel: {channel}"}

    def _send_dingtalk(self, msg_type, data):
        """钉钉消息推送"""
        import dingtalk_sdk
        
        if msg_type == "stock_alert":
            title = f"⚠️ 库存预警 - {data['material_name']}"
            text = f"物料编码: {data['material_code']}\n当前库存: {data['current_stock']}\n安全库存: {data['safety_stock']}\n建议: {data['suggestion']}"
            dingtalk_sdk.send_markdown(title, text)

        elif msg_type == "approval_pending":
            title = f"📋 待审批 - {data['order_type']}"
            text = f"单号: {data['order_no']}\n类型: {data['order_type']}\n请尽快处理"
            dingtalk_sdk.send_action_card(title, text, data.get("approval_url"))

        elif msg_type == "inbound_completed":
            title = f"✅ 入库完成 - {data['order_no']}"
            text = f"入库单: {data['order_no']}\n仓库: {data['warehouse_name']}\n明细数: {len(data['items'])}项"
            dingtalk_sdk.send_text(title)

    def _send_feishu(self, msg_type, data):
        """飞书消息推送"""
        import feishu_sdk
        
        card = self._build_feishu_card(msg_type, data)
        feishu_sdk.send_interactive(card)

    def _build_feishu_card(self, msg_type, data):
        """构建飞书卡片消息"""
        templates = {
            "stock_alert": {
                "header": {"title": {"tag": "plain_text", 
                           "content": "⚠️ 库存预警"}},
                "elements": [
                    {"tag": "div", "text": {"tag": "lark_md",
                     "content": f"**物料**: {data['material_name']}\n**当前库存**: {data['current_stock']}\n**安全库存**: {data['safety_stock']}"}}
                ]
            }
        }
        return templates.get(msg_type, {})

    def _send_wechat_work(self, msg_type, data):
        """企业微信消息推送"""
        import requests
        
        webhook_url = os.getenv("WECHAT_WORK_WEBHOOK")
        message = self._format_wechat_message(msg_type, data)
        
        requests.post(webhook_url, json=message, timeout=10)

EEAT实操案例：制造业14天搭建WMS

某300人汽车零部件制造企业，3500个库位，此前用Excel管理库存，账实差异率超12%。

第一步（业务人员，3天）：通过搭贝AI低代码平台零代码表单设计器搭建6张核心表单——物料主数据、库位档案、入库单、出库单、盘点单、调拨单。

第二步（业务人员，2天）：配置入库审批流（收货→质检→审核）和出库审批流（申请→审批→出库），设置库位分配规则和安全库存阈值。

第三步（业务人员，3天）：配置库存实时台账、库龄分析报表、周转率报表、管理看板。

第四步（IT人员，6天）：对接用友U8 ERP，实现采购订单同步和入库回写；对接条码打印机；对接企业微信消息推送。

量化效果：库存准确率88%→99.2%，盘点时间3天→0.5天，紧急用料响应4小时→30分钟，仓库人员8人→6人。

搭贝AI低代码平台设立总部核心研发中心，技术人员占比83%，全国线上远程运维服务网络7×24小时响应，兼容钉钉、飞书、企业微信三端，依托自研API集成中台对接用友、金蝶及各类私有化ERP。全国综合平台型定位，服务覆盖全国多省市，在仓储管理领域积累了丰富的行业落地经验。

行业趋势

Gartner预测2027年40%企业应用内置AI智能体，WMS领域正出现AI驱动库位优化和智能拣货路径。Fortune Business Insights数据显示全球智能仓储市场2025年达286.8亿美元，2026年达326.8亿美元，2034年达997.1亿美元。传统WMS实施周期3-6个月，低代码方案压缩至2-4周。