107 lines
2.7 KiB
Python
107 lines
2.7 KiB
Python
|
#!/usr/bin/env python
|
||
|
|
# -*- coding: utf-8 -*-
|
||
|
|
"""
|
||
|
|
测试数据库功能
|
||
|
|
"""
|
||
|
|
|
||
|
|
from database import Database
|
||
|
|
import json
|
||
|
|
|
||
|
|
# 初始化数据库
|
||
|
|
db = Database("wrench.db")
|
||
|
|
|
||
|
|
# 测试创建工单
|
||
|
|
print("="*60)
|
||
|
|
print("测试创建工单")
|
||
|
|
print("="*60)
|
||
|
|
|
||
|
|
test_order = {
|
||
|
|
"trace_id": "TR20260119001",
|
||
|
|
"process_id": "P001",
|
||
|
|
"process_name": "前轮装配",
|
||
|
|
"product_name": "汽车底盘组件",
|
||
|
|
"station": "装配工位A1",
|
||
|
|
"operator": "张三",
|
||
|
|
"status": "pending",
|
||
|
|
"bolts": [
|
||
|
|
{
|
||
|
|
"bolt_id": 1,
|
||
|
|
"name": "前轮螺栓1",
|
||
|
|
"target_torque": 280,
|
||
|
|
"mode": 1,
|
||
|
|
"torque_tolerance": 0.10,
|
||
|
|
"angle_min": 1,
|
||
|
|
"angle_max": 360,
|
||
|
|
"status": "pending"
|
||
|
|
},
|
||
|
|
{
|
||
|
|
"bolt_id": 2,
|
||
|
|
"name": "前轮螺栓2",
|
||
|
|
"target_torque": 300,
|
||
|
|
"mode": 1,
|
||
|
|
"torque_tolerance": 0.10,
|
||
|
|
"angle_min": 1,
|
||
|
|
"angle_max": 360,
|
||
|
|
"status": "pending"
|
||
|
|
}
|
||
|
|
]
|
||
|
|
}
|
||
|
|
|
||
|
|
if db.create_work_order(test_order):
|
||
|
|
print("[OK] 工单创建成功")
|
||
|
|
else:
|
||
|
|
print("[FAIL] 工单创建失败")
|
||
|
|
|
||
|
|
# 测试查询工单
|
||
|
|
print("\n" + "="*60)
|
||
|
|
print("测试查询工单")
|
||
|
|
print("="*60)
|
||
|
|
|
||
|
|
order = db.get_work_order("TR20260119001", "P001")
|
||
|
|
if order:
|
||
|
|
print(f"[OK] 查询成功: {order['trace_id']} - {order['process_id']}")
|
||
|
|
print(f" 工序名称: {order['process_name']}")
|
||
|
|
print(f" 螺栓数量: {len(order['bolts'])}")
|
||
|
|
else:
|
||
|
|
print("[FAIL] 查询失败：未找到工单")
|
||
|
|
|
||
|
|
# 测试查询可用工单
|
||
|
|
print("\n" + "="*60)
|
||
|
|
print("测试查询可用工单列表")
|
||
|
|
print("="*60)
|
||
|
|
|
||
|
|
available = db.get_available_work_orders("TR20260119001", "P001")
|
||
|
|
print(f"[OK] 找到 {len(available)} 个可用工单")
|
||
|
|
for order in available:
|
||
|
|
print(f" - {order['trace_id']} - {order['process_id']}: {order['process_name']} ({order['bolt_count']}个螺栓)")
|
||
|
|
|
||
|
|
# 从JSON导入数据
|
||
|
|
print("\n" + "="*60)
|
||
|
|
print("从JSON文件导入数据")
|
||
|
|
print("="*60)
|
||
|
|
|
||
|
|
try:
|
||
|
|
with open("work_orders.json", 'r', encoding='utf-8') as f:
|
||
|
|
json_orders = json.load(f)
|
||
|
|
|
||
|
|
count = 0
|
||
|
|
for order in json_orders:
|
||
|
|
if db.create_work_order(order):
|
||
|
|
count += 1
|
||
|
|
|
||
|
|
print(f"[OK] 成功导入 {count}/{len(json_orders)} 个工单")
|
||
|
|
|
||
|
|
# 再次查询
|
||
|
|
available = db.get_available_work_orders("TR20260119001", "P001")
|
||
|
|
print(f"\n[OK] 查询结果: 找到 {len(available)} 个可用工单")
|
||
|
|
for order in available:
|
||
|
|
print(f" - {order['trace_id']} - {order['process_id']}: {order['process_name']}")
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f"[FAIL] 导入失败: {e}")
|
||
|
|
|
||
|
|
print("\n" + "="*60)
|
||
|
|
print("测试完成")
|
||
|
|
print("="*60)
|
||
|
|
|