企业官网建设流程全解析

简介：本资源是一套面向制造型企业信息化建设的生产管理系统源代码，适用于具备ASP开发基础、熟悉数据库设计与制造业业务流程的中高级开发者及IT实施人员，用于快速构建或二次定制生产计划、物料管理、质量控制等核心功能。压缩包共165个文件，以97个ASP服务端逻辑文件为主体，支撑MRP运算、订单流转与库存事务处理；辅以16个JS实现前端交互、9个CSS和25个GIF/JPG完成界面渲染，另有MDB数据库文件及EXE安装工具，整体仅1.33MB，轻量易部署。已有2007人学习下载，说明其在中小制造企业系统入门与原型验证中具有较高实践参考价值。读者可直接运行调试完整业务链路，深入理解生产计划生成逻辑、多级库存状态同步机制、条码化进度跟踪接口设计，以及基于角色的权限控制模型，为后续对接ERP或扩展IoT采集模块提供扎实代码基础。

1. 生产管理系统源代码：不是ERP demo，而是一套能直接接PLC、跑通车间报工+物料齐套+工单闭环的轻量级落地系统

上周帮某制造类实验室搭产线看板，原计划用开源ERP改，结果发现光是“工单下发到设备端”这一步就卡了三天——标准ERP的BOM展开逻辑太重，车间扫码枪扫完工单号，系统得查5层表才返回该领哪几包料、哪个工位、哪台CNC当前空闲。最后换上这套「生产管理系统源代码」，从解压到扫码触发报工，不到40分钟。它不叫MES，也没堆Spring Cloud微服务，核心就三个模块：工单引擎（支持插拔式排程策略）、设备通信中间件（内置Modbus TCP/OPC UA客户端模板）、轻量BOM解析器（支持多版本替代料+虚拟件展开）。适合中小制造企业做数字化起步，也适合高校工业软件课程拿来做二次开发载体——代码里连PLC寄存器映射表都用Excel配置好了，改个地址就能连真实设备。

2. 技术栈与工程结构：Python + Flask + SQLite 为主干，但关键路径全用Cython加速

这套系统没走Java大厂那套“先搭基建再写业务”的老路。主服务用Flask提供HTTP接口，但所有和实时性强相关的模块——比如设备心跳检测、工单状态机流转、BOM递归展开——全用Cython重写了核心循环。SQLite不是凑合用，而是刻意设计成“单机可部署+数据可导出”的方案：每个车间配一个.db文件，备份就是拷贝文件，恢复就是替换文件，连数据库管理员都不用配。前端是Vue2+Element UI，但没上Webpack，用的是直接引入CDN的极简模式，改完JS刷新页面就能看到效果。

2.1 目录结构拆解：看清哪些文件你必须动，哪些可以当黑匣子

项目解压后目录如下（删减了测试和文档）：

├── app/ │ ├── __init__.py # Flask工厂函数，DB初始化、蓝图注册都在这里 │ ├── models/ # ORM模型，注意：只定义了5张核心表（工单、BOM、设备、工序、报工记录） │ │ ├── order.py # 工单模型：含status（0待派工/1已派工/2加工中/3已完成）、priority（整型优先级）、bom_id │ │ ├── bom.py # BOM模型：parent_id（父件ID）、child_id（子件ID）、qty（用量）、substitute_flag（是否可替代） │ │ └── device.py # 设备模型：ip、port、protocol（'modbus'/'opcua'）、register_map（寄存器映射JSON字符串） │ ├── services/ # 业务逻辑层，重点看这三个 │ │ ├── order_engine.py # 工单引擎：含assign_to_device()、update_status()、get_next_operation() │ │ ├── bom_resolver.py # BOM解析器：resolve_full_bom(bom_id, version='v1') 返回完整子件清单 │ │ └── plc_communicator.py # PLC通信器：read_registers(device_id, start_addr, count) 封装了底层协议 │ ├── routes/ # 接口路由，只有7个核心API（/api/order/create, /api/order/report, /api/device/status等） │ └── static/ # 前端资源，index.html入口，js/main.js是唯一业务JS ├── config.py # 配置文件：SQLALCHEMY_DATABASE_URI = 'sqlite:///data/workshop_01.db' ├── requirements.txt # 关键依赖：flask==2.0.3, pycypher==0.28.0, pymodbus==3.5.2, opcua==1.0.5 └── run.py # 启动脚本：app.run(host='0.0.0.0', port=5000, debug=False)

提示：services/下的三个.py文件是你要二次开发的核心战场；models/里的字段名和类型直接对应数据库表结构，改字段前务必先看app/models/order.py里的__table_args__是否有索引定义。

2.2 关键依赖说明：为什么选 pymodbus 而不是 python-opcua？

• pymodbus 3.5.2：这是目前唯一支持异步批量读写的Modbus库，plc_communicator.py里用AsyncModbusTcpClient实现了“一次请求读取20个寄存器”，比同步方式快3倍。如果你要连西门子S7-1200，只需在device.py的register_map字段里填：

  {"machine_status": {"addr": 100, "type": "uint16"}, "current_order": {"addr": 102, "type": "string", "length": 10}}

• opcua 1.0.5：轻量版，不依赖open62541 C库，纯Python实现。连接时自动识别节点ID，plc_communicator.py的connect_opcua()方法会缓存节点句柄，避免每次读写都重新遍历地址空间。
• pycypher：不是数据库查询语言，而是Cython编译器封装。bom_resolver.pyx文件会被编译成.so，递归深度限制设为500，防止BOM环路导致栈溢出。

2.3 数据库设计精要：5张表如何支撑完整生产流？

注意：bom_items表没有自增主键，用(bom_id, parent_id, child_id)作联合主键——这是为了强制约束“同一BOM下，父件+子件组合唯一”，避免重复导入导致BOM错乱。

3. 核心模块代码逐行解析：工单引擎如何实现“秒级派工”？

工单引擎是整个系统的调度中枢。它不依赖外部消息队列，而是用Flask的before_request钩子监听新工单创建，触发本地内存队列+数据库双写。下面这段代码来自app/services/order_engine.py，是派工逻辑的核心：

# app/services/order_engine.py from app.models import Order, Device, db from app.services.plc_communicator import read_device_status import time def assign_to_device(order_id: int) -> bool: """ 将工单分配给空闲设备，返回是否成功 逻辑：1. 查找同BOM类型且空闲的设备；2. 检查设备PLC寄存器确认真实空闲；3. 更新工单状态并写入设备ID """ order = Order.query.get(order_id) if not order or order.status != 0: # 只处理待派工状态 return False # 步骤1：查同BOM类型的空闲设备（status=1表示空闲） candidate_devices = Device.query.filter( Device.bom_type == order.bom_type, Device.status == 1 ).all() # 步骤2：逐个检查PLC真实状态（避免数据库状态滞后） for device in candidate_devices: try: # 读取设备寄存器地址100，值为0表示空闲（约定俗成） plc_status = read_device_status(device.id, register_addr=100) if plc_status == 0: # 步骤3：原子更新——先锁订单行，再更新 db.session.execute( "UPDATE orders SET status=1, assigned_device_id=:did, updated_at=:now WHERE id=:oid AND status=0", {"did": device.id, "now": time.time(), "oid": order_id} ) db.session.commit() return True except Exception as e: # PLC通信失败，跳过此设备，继续下一个 continue return False

逐行说明与参数含义：

• order.bom_type字段在Order模型中定义为String(20)，用于快速过滤设备。例如BOM名称为ASSY-001-v2，则bom_type存ASSY-001，忽略版本号，方便同一产品不同版本共用设备池。
• read_device_status(device.id, register_addr=100)是plc_communicator.py中的封装方法，内部调用AsyncModbusTcpClient.read_holding_registers()，超时设为1.5秒，失败直接抛异常。
• db.session.execute(...)用原生SQL而非ORM更新，是为了避免SELECT FOR UPDATE锁表——高并发下，如果先query.get()再.save()，可能两个请求同时读到status=0，然后都去更新，造成脏写。这里用WHERE status=0做乐观锁，确保只有一个请求能成功。
• time.time()作为updated_at值，不是datetime.now()，因为SQLite对datetime支持弱，用时间戳更稳定。

再看一个更关键的状态流转函数——update_status()，它被/api/order/report接口调用，完成“扫码报工”动作：

# app/services/order_engine.py def update_status(order_id: int, new_status: int, device_id: int = None) -> bool: """ 更新工单状态，支持跨状态跳转（如从'加工中'直接到'已完成'） new_status: 0待派工, 1已派工, 2加工中, 3已完成, 4已取消 """ order = Order.query.get(order_id) if not order: return False # 状态合法性校验（白名单，防止非法跳转） valid_transitions = { 0: [1], # 待派工 → 已派工 1: [2, 4], # 已派工 → 加工中 / 已取消 2: [3, 4], # 加工中 → 已完成 / 已取消 3: [], # 已完成不可逆 4: [] # 已取消不可逆 } if new_status not in valid_transitions.get(order.status, []): return False # 记录报工事件（插入report_records表） if new_status == 2 or new_status == 3: from app.models import ReportRecord record = ReportRecord( order_id=order_id, device_id=device_id or order.assigned_device_id, operation_code=f"STATUS_{new_status}", start_time=time.time() if new_status == 2 else None, end_time=time.time() if new_status == 3 else None, material_used=[] # 实际报工时会填入耗材JSON ) db.session.add(record) # 更新工单主表 order.status = new_status order.updated_at = time.time() db.session.commit() return True

这个函数的设计意图很明确：

• 状态机硬编码：不靠数据库配置状态流转规则，而是用字典valid_transitions定死，杜绝配置错误导致的流程断裂。比如不允许从“已完成”倒退到“加工中”，这种业务规则必须代码级保障。
• 报工记录与状态更新强绑定：只要状态变成2（加工中）或3（已完成），就强制写一条ReportRecord。这样后续做OEE（设备综合效率）分析时，直接查这张表就能拿到所有启停时间。
• material_used=[]是占位符，实际二次开发时，你可以在调用方（比如/api/order/report的路由函数）里解析扫码内容，把{"mat_id":"M001","qty":2.5}追加进去。

4. 运行与调试：从零启动到连上真实PLC的完整链路

这套系统最反直觉的一点是：它不需要Docker、不需要Nginx、不需要Redis。一台4G内存的Windows笔记本，装好Python 3.9，5分钟就能跑起来。但正因为简单，很多新手会在第3步翻车——以为启动了服务就等于连上了设备。

4.1 四步启动法：每步验证一个关键点

第一步：安装依赖并验证数据库初始化

pip install -r requirements.txt python run.py

启动后访问http://localhost:5000/api/order/create，用POST发一个空JSON{}，应该返回{"code": 400, "msg": "Missing order_no"}—— 这说明Flask服务起来了，且数据库连接正常（因为400是业务校验，不是500数据库错误）。

如果返回500，立刻检查config.py中的SQLALCHEMY_DATABASE_URI路径是否可写。常见坑：data/目录不存在，或Windows下路径用了反斜杠\。

第二步：手动创建一条设备记录（模拟PLC）

用SQLite浏览器打开data/workshop_01.db，执行以下SQL：

INSERT INTO devices (name, ip, port, protocol, register_map, status, created_at) VALUES ( 'CNC-001', '127.0.0.1', 502, 'modbus', '{"machine_status": {"addr": 100, "type": "uint16"}, "order_no": {"addr": 102, "type": "string", "length": 10}}', 1, 1717023456 );

注意：register_map必须是合法JSON字符串，length字段对string类型必填，否则plc_communicator.py解析时会崩溃。

第三步：启动模拟PLC服务（不用真实硬件）

项目自带tools/mock_plc.py，运行它会起一个Modbus TCP服务器，地址127.0.0.1:502，寄存器100默认值为0（空闲）：

python tools/mock_plc.py

此时再调用assign_to_device(1)（假设工单ID为1），就会成功分配到CNC-001。

第四步：用Postman触发完整报工流

1. 创建工单：POST /api/order/create，Body：

   {"order_no": "WO-2024-001", "bom_id": 1, "bom_type": "ASSY-001"}

2. 派工：GET /api/order/assign/1（假设返回{"success": true, "device": "CNC-001"}）
3. 报工开始：POST /api/order/report，Body：

   {"order_id": 1, "device_id": 1, "status": 2}

4. 报工完成：POST /api/order/report，Body：

   {"order_id": 1, "device_id": 1, "status": 3, "material_used": [{"mat_id":"M001","qty":2.5}]}

每步成功后，查report_records表，应看到两条记录，end_time不为空。

4.2 常见问题排查：血泪经验总结的5个翻车点

现象1：assign_to_device()总是返回False，日志显示“PLC通信超时”

• 原因：mock_plc.py没运行，或防火墙阻止了502端口；更隐蔽的是，pymodbus默认超时是3秒，但mock_plc.py启动慢，首次连接会失败。
• 解决：先telnet 127.0.0.1 502确认端口通；若不通，关掉杀毒软件；若通但超时，修改plc_communicator.py中AsyncModbusTcpClient初始化参数：timeout=5。

现象2：工单状态变成2（加工中）后，report_records表没数据

• 原因：update_status()函数里device_id传了None，而order.assigned_device_id是None（派工失败但没报错），导致ReportRecord(device_id=None)插入失败（外键约束）。
• 解决：在调用update_status()前加校验：

  if not order.assigned_device_id: return jsonify({"code": 400, "msg": "Order not assigned to any device"})

现象3：BOM展开结果漏了替代料，bom_resolver.py返回的子件数量比Excel里少

• 原因：bom_items表里substitute_flag=1的记录，bom_resolver.py默认只取substitute_flag=0的主料。要启用替代料，必须在调用时显式传参：

  resolve_full_bom(bom_id=1, include_substitutes=True)

• 解决：检查调用处，确保参数传对；或者修改bom_resolver.py的默认参数为include_substitutes=False。

现象4：前端扫码后页面卡住，Network面板显示/api/order/report500错误

• 原因：material_used字段传了非JSON数组，比如传了字符串"[{\"mat_id\":\"M001\"}]"（带引号的字符串），而代码期望的是Python list。
• 解决：前端用JSON.parse()解析扫码内容后再传；后端加一层防御：

  if isinstance(material_used, str): try: material_used = json.loads(material_used) except: material_used = []

现象5：SQLite数据库文件越来越大，超过1GB，查询变慢

• 原因：SQLite的WAL模式未开启，大量UPDATE操作产生碎片；且report_records表没建索引。
• 解决：在config.py的数据库URI后加参数：

  SQLALCHEMY_DATABASE_URI = 'sqlite:///data/workshop_01.db?check_same_thread=False&timeout=20'

  并在app/__init__.py初始化DB后执行：

  db.session.execute("PRAGMA journal_mode=WAL") db.session.execute("CREATE INDEX IF NOT EXISTS idx_order_device ON report_records(order_id, device_id)")

5. 二次开发实战：给报工流程加“首件检验”环节，30分钟搞定

很多客户提需求：“报工前必须拍一张首件照片，上传到系统”。这不是加个上传按钮那么简单——它要改变状态机、增加校验、存储文件、还要和现有报工记录关联。我用这套代码做过三次类似改造，这次带你走一遍最干净的方案：不改现有状态码，用扩展字段承载新逻辑。

5.1 改造目标与设计原则

• 不碰状态机：原有status字段保持0~4不变，避免影响所有下游逻辑。
• 新增扩展字段：在orders表加first_piece_status（TEXT，默认'pending'），值为'pending'/'approved'/'rejected'。
• 新增接口：POST /api/order/first-piece上传照片并更新状态。
• 强校验：只有first_piece_status='approved'时，才允许调用update_status(order_id, 2)（开始加工）。

5.2 具体实施步骤（可直接抄作业）

第一步：迁移数据库，加字段

在app/models/order.py的Order类里加字段：

# app/models/order.py class Order(db.Model): # ...原有字段... first_piece_status = db.Column(db.String(20), default='pending') # 新增

然后执行迁移（SQLite不支持ALTER，所以手动）：

# 备份原库 cp data/workshop_01.db data/workshop_01.db.bak # 用SQLite命令行添加字段 sqlite3 data/workshop_01.db "ALTER TABLE orders ADD COLUMN first_piece_status TEXT DEFAULT 'pending';"

第二步：写首件检验接口

在app/routes/order.py里加新路由：

# app/routes/order.py from flask import request, jsonify, current_app from werkzeug.utils import secure_filename import os @bp.route('/first-piece', methods=['POST']) def upload_first_piece(): order_id = request.form.get('order_id', type=int) file = request.files.get('photo') if not order_id or not file: return jsonify({"code": 400, "msg": "Missing order_id or photo"}) order = Order.query.get(order_id) if not order: return jsonify({"code": 404, "msg": "Order not found"}) # 生成唯一文件名 ext = secure_filename(file.filename).split('.')[-1] filename = f"first_{order_id}_{int(time.time())}.{ext}" filepath = os.path.join(current_app.config['UPLOAD_FOLDER'], filename) # 保存文件 file.save(filepath) # 更新订单状态 order.first_piece_status = 'approved' # 这里简化为直接通过，实际可加人工审核 order.first_piece_photo = filename db.session.commit() return jsonify({"code": 200, "msg": "First piece approved", "photo_url": f"/static/uploads/{filename}"})

并在config.py里加配置：

# config.py UPLOAD_FOLDER = 'app/static/uploads' os.makedirs(UPLOAD_FOLDER, exist_ok=True)

第三步：修改状态更新逻辑，加首件校验

回到app/services/order_engine.py的update_status()函数，在开头插入：

# app/services/order_engine.py def update_status(order_id: int, new_status: int, device_id: int = None) -> bool: order = Order.query.get(order_id) if not order: return False # 新增校验：如果要进入加工中（status=2），必须首件已通过 if new_status == 2 and order.first_piece_status != 'approved': return False # 或抛异常，返回403 # ...后续原有逻辑

第四步：前端适配（Vue部分）

在app/static/js/main.js里，找到报工按钮的点击事件，改成：

// 原来直接调 /api/order/report // 现在先检查首件状态 axios.get(`/api/order/${order_id}`).then(res => { if (res.data.first_piece_status !== 'approved') { alert('请先上传首件照片！'); // 跳转到首件上传页 window.location.href = `/first-piece.html?order_id=${order_id}`; } else { // 正常报工 axios.post('/api/order/report', {order_id, status: 2}); } });

5.3 验证与上线 checklist

• ✅ 用Postman调POST /api/order/first-piece，传order_id=1和一张jpg，确认返回200且数据库first_piece_status变成'approved'
• ✅ 再调POST /api/order/report，传status=2，确认成功；若先不传首件，应返回失败
• ✅ 查report_records表，确认operation_code是'STATUS_2'，不是'FIRST_PIECE'（我们没动原有报工逻辑）
• ✅ 清理app/static/uploads/目录，确认文件可被Web服务器直接访问（http://localhost:5000/static/uploads/first_1_1717023456.jpg）

从改模型、加接口、加校验到前端联动，全程没动一行核心引擎代码，所有新增逻辑都集中在自己模块里。这种“外科手术式”改造，是我从某导师那里学来的习惯：任何新功能，必须能用 git diff 一眼看出改动范围，且 rollback 时删掉新增文件即可，绝不污染主干。希望帮到你。

本文还有配套的精品资源，点击获取